diff --git a/CLAUDE.md b/CLAUDE.md index e76800e..562fbc4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -92,6 +92,11 @@ Terminology: **body / duration** = media playback length (`media.duration` for A - **A stop takes its scheduled chain with it.** `stop`/`pause`/`fade_out` actions call `cancel_pending_descendants`: walks `_target_object` past the target, cancelling every local+enabled cue that has **not** produced output yet (`_revealed`), transitively, stopping at a chain break. A cue already on stage is never touched and the walk continues past it. Each node runs it on its own graph (ActionCues are local everywhere). Cancelled cues also get `remove_cue` — their threads exit through the generation guard, upstream of the normal one. - **Enable/disable mid-pass are honoured.** Nothing downstream re-reads `enabled` once a cue is dispatched, so: disabling a parked cue **cancels** it (`cancel_parked`), and enabling one **rejoins** it (`rejoin_chain`) at the anchor stamped on it when the walk skipped it (`_chain_pass`), dispatched with `unroll=False` so the running chain is not re-dispatched. Refused if its slot already passed (a loud journal line — firing late by surprise is worse), if a STOP ended that pass, or if a stop action already claimed it. - **Auto-follow / auto-pause postwait = an MTC-gated tail after the body**: when `loop_cue` returns, the engine blanks the video (`blank_cue`: `/visible 0`, cue stays armed → STOP-reachable), waits until `body_end + postwait` on the MTC timeline, then ends illumination (`remove_cue`) and — for follow — fires the target via `go_from`. A manual GO during the tail **preempts** the auto-fire. Follow fires at `body_end + postwait`, not at media end. +- **An Auto follow into a cue on another node is relayed** (`cues/FollowRelay.py`, ClickUp 869fc8ytz). Only the nodes where the source is local run it and fire its follow, so a target on another node used to never start, silently. For an **AudioCue or VideoCue** source (ActionCue, FadeCue, CueList and DmxCue are local on every node, so every node already fires their follow) the owning node **announces** the follow instant (`_follow_instant_ms`: the same expression its own fire uses) right after the reveal, or at the start of the last loop for `loop > 1`; never for an endless loop. The controller relays it to every node (`_relay_follow`); the others pre-arm what they would start, wait for the instant on MTC and dispatch it there through `go_from`. If the follow does not happen the owner **cancels**: reason `stop` (stop/pause/fade-out action), `restart` (the source was started again: `go()`'s commit section sends it, the replaced thread notices too late), `error`; nothing for a STOP, a load or the yield, which every node handles itself. A cancel that reaches a target already on stage stops it only within `FOLLOW_CANCEL_GRACE_S` (2 s) and only for `stop`/`restart` -- a cue is never cut for an internal error. + - **Run counter.** The controller's `run_seq` (from the clock) travels in every GO; STOP, load and unload bump it before forwarding. A node records it, and empties its pending follows, in the **receive hook** (`NodeCommunications.set_receive_hook`), i.e. in bus order, before the command threads race for `_command_lock`. Follow messages are applied there too and never take that lock. A message of another run is dropped at the controller and at the node. Between a STOP/load's receipt and the moment the node has run it (`run_command` marks it applied), the node's epochs still belong to the ending run: it announces nothing and reads no epochs for incoming follows until then. + - **`go()` has a `commit_gate`**, evaluated inside the commit section after every other refusal: the follow's dispatch and its cancel are decided in one place. Order inside that section: refusals → gate → restart cancel → assignments. + - **`disarm_if_unchanged`** is the one other place that may disarm a published cue from a background thread (the stopped follow target): it re-checks generation, STOP/load epoch and arm stamp in one `_lock` section and releases the player while holding the cue's **arm claim**. Extend it, don't bypass it (publish-or-abandon contract, 869f9wqpn). + - Version skew: an old controller sends no `run_seq`, so nothing is announced (the bug stays); an old node logs `Command follow not found` for every relayed message. Upgrade the whole cluster together. - `prewait` is applied at exactly one point (`start = arrival + prewait`), never as a wall-clock sleep. The postwait `sleep` in `go_threaded` is **auto-continue only** and no longer paces anything (the chain left at the trigger): it holds the thread past the body when `post > body`, and **it is the sole mechanism producing auto-continue's `prewait + max(body, postwait)` illumination** — `loop_cue` never reads `postwait` and `remove_cue` fires only after it returns — so removing it would silently shorten the highlight to `prewait + body`. Guarded by a test. Cross-cutting invariants: **never auto-stop a running project**; `_reveal_wait` exits on `_stop_requested` (STOP) or a changed `_go_generation` (newer GO/reload) and does **not** bail on a recoverable MTC stall (reveal fires late on resume); anchor comparisons use wrap-accumulated `milliseconds_exact` (24h-safe). See `Plans/postwait-postgo-chain-semantics.md` and `Plans/postwait-engine-only-implementation.md` in the cuems-RELATIONS index repo for the full design/diagnosis. diff --git a/src/cuemsengine/ControllerEngine.py b/src/cuemsengine/ControllerEngine.py index 5e8dd14..a20fad2 100644 --- a/src/cuemsengine/ControllerEngine.py +++ b/src/cuemsengine/ControllerEngine.py @@ -101,6 +101,17 @@ def __init__(self, **kwargs): # nodes. Does NOT force armed=yes — operator decides. self._arm_watchdog: threading.Timer | None = None + # Run counter for cross-node Auto follows (869fc8ytz). Every GO ships + # it; STOP, load and unload bump it BEFORE they forward anything; a + # node's follow message is relayed only if it carries the current + # value. Started from the clock so a restarted controller never + # reuses a value a node still holds. The lock makes "bump" and + # "check, then forward a follow" atomic, so a relayed follow is + # always queued to the nodes ahead of the STOP that ends its run. + # Never held across a wait or the cluster probe. + self._run_seq: int = time.time_ns() // 1_000_000 + self._run_seq_lock = threading.Lock() + super().__init__(**kwargs) self.set_editor_request("") self.set_node_operation_callback() @@ -615,9 +626,56 @@ def status_operation_callback(self, operation: NodeOperation): ) updated += 1 Logger.info(f"Mixer status from {node_uuid}: {updated} channel(s) updated") + elif operation.target == "follow": + self._relay_follow(operation) else: Logger.debug(f"Unknown status target: {operation.target}") + def _bump_run_seq(self, why: str) -> None: + """A run ends here (STOP, load, unload): follow messages sent for it + are no longer relayed. Call BEFORE forwarding the command.""" + with self._run_seq_lock: + self._run_seq += 1 + seq = self._run_seq + Logger.debug(f"run_seq -> {seq} ({why})") + + def _relay_follow(self, operation: NodeOperation) -> None: + """Relay a cross-node Auto follow message (announce or cancel) to + every node (869fc8ytz). The controller keeps no follow state: the + node that owns the source decides, the others act. Dropped when the + sender is not adopted, a cue is not in the loaded project, the show + is not running, or the message belongs to an earlier run.""" + data = operation.data if isinstance(operation.data, dict) else None + if data is None: + Logger.warning(f"Follow message without data from {operation.sender}") + return + what = ( + f"follow {data.get('op')} {data.get('source')} -> {data.get('target')}" + f" at {data.get('seed_ms')} from {operation.sender}" + ) + if operation.sender not in self._adopted_nodes: + Logger.warning(f"Dropping {what}: sender is not an adopted node") + return + if ( + data.get("source") not in self.cue_status + or data.get("target") not in self.cue_status + ): + Logger.warning(f"Dropping {what}: cue not in the loaded project") + return + with self._run_seq_lock: + if ( + self.get_status("running") != "yes" + or data.get("run_seq") != self._run_seq + ): + Logger.info( + f"Dropping {what}: not part of the current run " + f"(run_seq {data.get('run_seq')}, current {self._run_seq}, " + f"running={self.get_status('running')})" + ) + return + self._forward_command_to_nodes("/engine/command/follow", data) + Logger.info(f"Relayed {what}") + ######################### # Editor commands ######################### @@ -970,6 +1028,7 @@ def load_project(self, project_name, context=None, deploy_only=False): return False Logger.info(f"Loading project {project_name}") + self._bump_run_seq("load") self._clear_playback_state() self.reset_script() @@ -1085,10 +1144,13 @@ def go_script(self, value, context=None): # every node fire 4.48s late on test2 (2026-09-25). Without a live # reading, send the GO without an anchor -- each node falls back to # its own MTC, exactly as before this fix. - go_value = value + # Always a dict (869fc8ytz): it carries the run counter that + # cross-node follow messages are checked against. Nodes read nothing + # else from the GO value, so the caller's own `value` is not passed on. + go_value = {"run_seq": self._run_seq} if self.mtc_listener is not None: if self.mtc_listener.is_receiving(): - go_value = {"go_mtc_ms": self.mtc_listener.main_tc.milliseconds_exact} + go_value["go_mtc_ms"] = self.mtc_listener.main_tc.milliseconds_exact else: Logger.error( "GO: controller MTC listener is not receiving -- sending GO " @@ -1376,6 +1438,7 @@ def stop_script(self, value): Logger.info("Script not running, nothing to stop.") return + self._bump_run_seq("stop") self.go_offset = None self.set_status("running", "no") self._clear_playback_state() @@ -1402,6 +1465,7 @@ def unload_project(self, value, context=None): """Unload the current project. Rejects if playback is running.""" if self.get_status("running") == "yes": raise RuntimeError("Cannot unload while running. Stop playback first.") + self._bump_run_seq("unload") self._clear_playback_state() self.reset_script() self.cue_status = {} diff --git a/src/cuemsengine/NodeEngine.py b/src/cuemsengine/NodeEngine.py index b4fb499..6ca296a 100644 --- a/src/cuemsengine/NodeEngine.py +++ b/src/cuemsengine/NodeEngine.py @@ -15,6 +15,7 @@ from .core.BaseEngine import BaseEngine from .cues.CueHandler import CUE_HANDLER, _ArmWalk +from .cues.FollowRelay import FollowRelay from .osc.helpers import add_prefix_to_all from .players import AudioClient, DmxClient, VideoClient from .players.PlayerHandler import ( @@ -101,6 +102,10 @@ class NodeEngine(BaseEngine): """ + # Set in _setup_nng_command_callback; None until then (and in unit tests + # that build the engine without its comms). + _follow_relay = None + # GO anchor tolerance (Badajoz 2026-09-25 fix): how far the controller's # shipped GO instant may disagree with this node's own MTC before # _resolve_go_anchor distrusts it and falls back to a live local read. @@ -175,6 +180,21 @@ def _setup_nng_command_callback(self): self._handle_nng_command ) Logger.info("NNG command callback registered for NodeEngine") + # Cross-node Auto follow (869fc8ytz): announce this node's + # follows, apply the ones relayed from other nodes, in bus order. + self._follow_relay = FollowRelay( + CUE_HANDLER, + node_id=self.cm.node_uuid, + send=self._send_follow, + get_script=lambda: self.script, + get_mtc=lambda: getattr(self, "mtc_listener", None), + get_project_gen=lambda: self._project_generation, + first_local=self._first_local_enabled_in_go_chain, + ) + CUE_HANDLER._follow = self._follow_relay + CUE_HANDLER.communications_thread.set_receive_hook( + self._on_command_received + ) else: Logger.warning( "CUE_HANDLER communications thread: " @@ -186,6 +206,46 @@ def _setup_nng_command_callback(self): ACTION_HANDLER.finalize_node_layer_bindings() ACTION_HANDLER.set_result_sink(self._action_result_sink) + def _on_command_received(self, command_name: str, value) -> bool: + """Receive hook: runs for every command in BUS order, before its + command thread races the others for the command lock (869fc8ytz). + + A follow message is applied here and consumed. GO records the + controller's run counter, STOP and load clear it and every pending + follow -- here, so a follow relayed before a STOP can never be applied + after it. Must not block.""" + relay = self._follow_relay + if relay is None: + return False + if command_name == "follow": + try: + relay.on_message(value) + except Exception as e: + # Consumed anyway: a follow message must never reach the + # command lock. + Logger.error(f"Follow message failed: {e}") + return True + if command_name == "go": + relay.begin_run(value) + elif command_name in ("stop", "load"): + relay.end_run(command_name) + return False + + def _send_follow(self, data: dict) -> None: + """Queue a follow message (announce or cancel) to the controller, + which relays it to every node. Never waits.""" + from .comms.NodesHub import ActionType, NodeOperation, OperationType + + CUE_HANDLER.communications_thread.send_operation_nowait( + NodeOperation( + type=OperationType.STATUS, + action=ActionType.UPDATE, + sender=self.cm.node_uuid, + target="follow", + data=data, + ) + ) + def _handle_nng_command(self, command_name: str, value, address: str = None): """Handle a command received via NNG from ControllerEngine. @@ -436,7 +496,14 @@ def run_command(self, command, value): if command in self.commands_dict.keys(): handler = self.commands_dict[command] if handler is not None: - handler(value) + try: + handler(value) + finally: + if command in ("stop", "load") and self._follow_relay: + # The receive hook counted it received; until + # here this node's epochs still belonged to the + # run it ends (869fc8ytz). + self._follow_relay.stop_applied() return True else: Logger.warning(f"Command {command} has no handler") diff --git a/src/cuemsengine/comms/NodeCommunications.py b/src/cuemsengine/comms/NodeCommunications.py index 27148df..090d1a3 100644 --- a/src/cuemsengine/comms/NodeCommunications.py +++ b/src/cuemsengine/comms/NodeCommunications.py @@ -38,6 +38,9 @@ def __init__( self.nng_hub = NodesHub(hub_address, mode=NodesHub.Mode.DIALER) self.node_id = node_id self._command_callback = command_callback + # Called synchronously, in bus order, before a command thread is + # started; returning True consumes the message (869fc8ytz). + self._receive_hook: Optional[Callable[[str, Any], bool]] = None self.nng_hub.set_receive_callbacks( { @@ -55,6 +58,17 @@ def set_command_callback(self, callback: Callable[[str, Any], None]) -> None: self._command_callback = callback Logger.debug("Command callback set in NodeCommunications") + def set_receive_hook(self, hook: Callable[[str, Any], bool]) -> None: + """Set a hook called for every command in BUS order, on the receiver, + before its command thread is started (869fc8ytz). + + The command threads then race for the engine's command lock, so only + here is the order the controller sent things in still known. The hook + must not block. Returning True consumes the message (no command + thread); an exception is logged and the command still runs. + """ + self._receive_hook = hook + def create_all_tasks(self): """Create async tasks for node communications.""" Logger.info("Starting all tasks in NodeCommunications") @@ -108,6 +122,14 @@ def _handle_command_operation(self, operation: NodeOperation) -> None: Logger.info(f"Received command via NNG: {command_name} = {repr(value)}") + if self._receive_hook is not None: + try: + if self._receive_hook(command_name, value): + return + except Exception as e: + # Never lose a GO, a STOP or a load to a bug in the hook. + Logger.error(f"Receive hook failed for {command_name}: {e}") + if self._command_callback: # Execute command in a separate thread to avoid blocking the NNG # receiver @@ -150,6 +172,18 @@ def send_operation(self, operation: NodeOperation, timeout: Optional[float] = No """ return self.run_coroutine(self.nng_hub.send_operation, operation, timeout) + def send_operation_nowait(self, operation: NodeOperation) -> None: + """Queue a NodeOperation to the controller without waiting for it. + + For senders that must never block: a cue's own thread, or a caller + holding a lock. Raises if the comms loop is not running. + """ + if not self.event_loop: + raise AttributeError(f"{self.name} event loop is not initialized") + asyncio.run_coroutine_threadsafe( + self.nng_hub.send_operation(operation), self.event_loop + ) + def add_player(self, player_id: str, data: dict, timeout: Optional[float] = None): """ Add a player to the OSC hub (thread-safe). diff --git a/src/cuemsengine/cues/CueHandler.py b/src/cuemsengine/cues/CueHandler.py index 76b98b3..623cf88 100644 --- a/src/cuemsengine/cues/CueHandler.py +++ b/src/cuemsengine/cues/CueHandler.py @@ -6,6 +6,7 @@ from __future__ import annotations from dataclasses import dataclass, field +from functools import partial from threading import Event, Lock, Thread, current_thread from time import monotonic, sleep from typing import Callable @@ -59,6 +60,23 @@ class _ArmClaim: waiters: int = 0 +@dataclass +class _FollowOut: + """What one cue thread did about its cross-node Auto follow (869fc8ytz). + + `record` is the (instant_ms, run_seq) it announced; the cue carries the + same tuple as `_follow_out` until whoever takes it (the fire, the thread + leaving, or go() starting the cue again) compare-and-clears it, so one + announcement is followed by at most one cancel. + """ + + revealed: bool = False + record: tuple | None = None + fired: bool = False + yielded: bool = False + error: bool = False + + class CueHandler: """ Singleton class responsible for handling Cue objects. @@ -210,6 +228,10 @@ def remove_armed_cue(self, cue: Cue) -> bool: # 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 @@ -1195,6 +1217,7 @@ def go( 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. @@ -1217,6 +1240,11 @@ def go( 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 @@ -1353,6 +1381,23 @@ def go( ) 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 @@ -1361,12 +1406,14 @@ def go( # thread", which is not an anchor a rejoin can be pinned to. cue._dispatch_arrival_ms = frozen_mtc_ms cue._stop_requested = False - go_gen = getattr(cue, "_go_generation", 0) + 1 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 — @@ -1405,7 +1452,13 @@ def go( return thread def go_from( - self, start_cue: Cue, mtc: MtcListener, seed_ms: float = None + 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 @@ -1419,7 +1472,20 @@ def go_from( 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 @@ -1450,8 +1516,14 @@ def go_from( 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) - return self.go(cue, mtc, seed_ms + sigma_ms) + 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'. @@ -1561,6 +1633,37 @@ def go_threaded( 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 @@ -1722,6 +1825,7 @@ def go_threaded( # 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 @@ -1737,8 +1841,13 @@ def go_threaded( ) 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 = monotonic() + 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: @@ -1780,11 +1889,26 @@ def go_threaded( # 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}") - loop_cue(cue, mtc) + 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( @@ -1801,11 +1925,7 @@ def go_threaded( # 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.) - 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 - fire_seed_ms = base_end_ms + cue.postwait.milliseconds_exact + fire_seed_ms = self._follow_instant_ms(cue, start_ms) tail = ( cue.post_go in ("pause", "go_at_end") @@ -1871,12 +1991,14 @@ def go_threaded( # 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. - go_at_end_thread = self.go_from(cue._target_object, mtc, fire_seed_ms) + 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" @@ -1891,6 +2013,263 @@ def go_threaded( 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 _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") diff --git a/src/cuemsengine/cues/FollowRelay.py b/src/cuemsengine/cues/FollowRelay.py new file mode 100644 index 0000000..e6159b1 --- /dev/null +++ b/src/cuemsengine/cues/FollowRelay.py @@ -0,0 +1,563 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera +"""Cross-node Auto follow (ClickUp 869fc8ytz). + +An Auto follow (post_go='go_at_end') fires its target when the cue ends. Only +the nodes where the cue is local run it, so only they used to fire the follow: +a target that plays on another node never started, silently. + +Now the node that owns the source ANNOUNCES the follow instant when the cue +starts (CueHandler, through `announce`), the controller relays it to every +node, and every other node pre-arms what it would start, waits for that +instant on MTC and dispatches it then, as the owner does for its own cues. +If the follow does not happen (the source was stopped, paused, faded out or +started again), the owner sends a CANCEL for the instant it announced. + +This class is both halves of the node's side: +- owner: `announce` / `cancel` queue the message to the controller; +- receiver: `on_message` applies a relayed message. It is called from the + bus receive hook, in bus order, and never waits: the pre-arm, the wait for + the instant and the stop of a target that must not play run on their own + threads. + +One entry per source cue, in a table with its own lock. Nothing calls into +CueHandler while holding that lock; the only nesting is CueHandler._lock then +the table lock, inside the commit gate `go()` evaluates (`_gate`). + +`run_seq` (the controller's run counter) is set from every GO and cleared at +STOP and load, at receipt; a message of another run is refused, and every +dispatch carries the STOP/load epochs read when its announcement arrived. + +Design: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md. +""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass, field +from time import sleep +from typing import Any, Callable + +from cuemsutils.cues import ActionCue, DmxCue +from cuemsutils.log import Logger + +from .CueHandler import _ArmWalk + +# A cancel that reaches a target which has already started stops it only if it +# started less than this ago, and only when the source was stopped or started +# again. The saved test2 runs showed commands delayed 700-800 ms on a loaded +# node, so the limit sits well above that. Ion, 2026-10-05 (D2). +FOLLOW_CANCEL_GRACE_S = 2.0 +# Two announcements of the same source within this are the same follow (two +# owners of one cue announce the same instant). +FOLLOW_INSTANT_MATCH_MS = 1.0 +# An announcement this late is still honoured (D7), but as an ERROR. +FOLLOW_LATE_ERROR_MS = 10_000.0 +# How often the pre-arm re-checks what it would start until the instant. The +# plan said 0.5 s; test2 (2026-10-06) showed a target still playing its earlier +# pass until 0.3 s before the instant fall back to an arm at dispatch, so it +# checks faster. A round that finds everything loaded costs one chain walk. +FOLLOW_PREARM_PERIOD_S = 0.1 +# Waiter MTC poll, as _wait_mtc. +FOLLOW_POLL_S = 0.02 + +_LIVE = ("waiting", "dispatching", "dispatched") + + +@dataclass(eq=False) +class _Entry: + source_id: str + target_id: str + instant: float + run_seq: int + origins: set + # Read once no STOP or load is pending on this node (None until then). + arm_epoch: int | None + stop_epoch: int | None + project_gen: Any + # waiting -> dispatching -> dispatched | done; any live state -> cancelled + state: str = "waiting" + cue: Any = None + # When the cue went on stage in a pass still playing when the follow + # dispatched it again (MINOR-3 of the code review). + started_before: float | None = None + gen: int | None = None + disarm_epoch: int | None = None + armed_at: float | None = None + lookahead_done: set = field(default_factory=set) + + +class FollowRelay: + def __init__( + self, + cue_handler, + node_id: str, + send: Callable[[dict], None], + get_script: Callable[[], Any], + get_mtc: Callable[[], Any], + get_project_gen: Callable[[], Any], + first_local: Callable[[Any], Any], + spawn: Callable[..., None] | None = None, + sleep: Callable[[float], None] = sleep, + ): + self._ch = cue_handler + self.node_id = node_id + self._send = send + self._get_script = get_script + self._get_mtc = get_mtc + self._get_project_gen = get_project_gen + self._first_local = first_local + self._spawn = spawn or self._spawn_thread + self._sleep = sleep + self._prearm_sleep = sleep + self._lock = threading.Lock() + self._table: dict[str, _Entry] = {} + self._run_seq: int | None = None + self._no_run_logged = False + # STOP/load commands received (receive hook) and applied (after the + # node ran them). In between, this node's epochs still belong to the + # run that is ending: nothing is announced and no epoch is read. + self._stops_received = 0 + self._stops_applied = 0 + + # ------------------------------------------------------------------ + # The run + # ------------------------------------------------------------------ + + @property + def run_seq(self) -> int | None: + return self._run_seq + + @property + def stop_pending(self) -> bool: + """A STOP or a load was received and has not run yet on this node.""" + return self._stops_received > self._stops_applied + + def stop_applied(self) -> None: + """The node has run a STOP or a load (or refused it).""" + with self._lock: + self._stops_applied += 1 + + def begin_run(self, value) -> None: + """At a GO's receipt. Same run: keep the table. New run: empty it. + A GO without an integer run_seq comes from a controller that does + not relay follows: nothing is announced until one that does.""" + seq = value.get("run_seq") if isinstance(value, dict) else None + if type(seq) is not int: + seq = None + with self._lock: + if seq == self._run_seq: + return + self._run_seq = seq + dropped = len(self._table) + self._table.clear() + if seq is None: + Logger.debug("GO carries no run_seq: cross-node follows are off") + else: + Logger.debug(f"Follow run {seq} (dropped {dropped} pending)") + + def end_run(self, why: str) -> None: + """At a STOP's or a load's receipt: every pending follow is gone.""" + with self._lock: + self._stops_received += 1 + self._run_seq = None + dropped = sum(e.state in _LIVE for e in self._table.values()) + self._table.clear() + if dropped: + Logger.info(f"Follow: {why} dropped {dropped} pending follow(s)") + + # ------------------------------------------------------------------ + # Owner: queue the message (never waits) + # ------------------------------------------------------------------ + + def announce(self, cue, instant_ms: float, run_seq: int) -> None: + self._emit("announce", cue, instant_ms, run_seq) + + def cancel(self, cue, instant_ms: float, run_seq: int, reason: str) -> None: + self._emit("cancel", cue, instant_ms, run_seq, reason) + + def _emit(self, op, cue, instant_ms, run_seq, reason=None) -> None: + target = getattr(cue, "_target_object", None) + data = { + "op": op, + "origin": self.node_id, + "run_seq": run_seq, + "source": str(cue.id), + "target": str(getattr(target, "id", "")), + "seed_ms": float(instant_ms), + } + if reason is not None: + data["reason"] = reason + try: + self._send(data) + except Exception as e: + Logger.error( + f"Follow: could not send {op} for {data['source']} -> " + f"{data['target']} at {instant_ms:.0f}ms: {e}" + ) + return + Logger.info( + f"Follow: {op} queued for {data['source']} -> {data['target']} at " + f"{instant_ms:.0f}ms" + (f" ({reason})" if reason else "") + ) + + def warn_no_run(self) -> None: + """The owner had a follow to announce but no run_seq: the controller + does not relay follows (older engine). Said once.""" + if not self._no_run_logged: + self._no_run_logged = True + Logger.info( + "Follow: the controller sent no run_seq with its GO; Auto " + "follows into cues on other nodes are not announced" + ) + + # ------------------------------------------------------------------ + # Receiver: apply a relayed message (receive hook; never waits) + # ------------------------------------------------------------------ + + def on_message(self, data) -> None: + if not isinstance(data, dict): + Logger.warning(f"Follow: ignoring a malformed message {data!r}") + return + op = data.get("op") + if op == "announce": + self._on_announce(data) + elif op == "cancel": + self._on_cancel(data) + else: + Logger.warning(f"Follow: ignoring a message with op {op!r}") + + def _on_announce(self, data: dict) -> None: + source_id = str(data.get("source")) + target_id = str(data.get("target")) + origin = data.get("origin") + what = f"{source_id} -> {target_id} from {origin}" + if origin == self.node_id: + return + script = self._get_script() + mtc = self._get_mtc() + source = script.find(source_id) if script is not None else None + target = script.find(target_id) if script is not None else None + if source is not None and getattr(source, "_local", False): + Logger.debug(f"Follow: {what} ignored, the source is local here") + return + try: + instant = float(data.get("seed_ms")) + except (TypeError, ValueError): + instant = None + refuse = None + if script is None or mtc is None: + refuse = "no project or no timecode on this node" + elif source is None or target is None: + refuse = "cue not in the loaded project" + elif instant is None: + refuse = "no instant" + elif self._run_seq is None or data.get("run_seq") != self._run_seq: + refuse = ( + f"not part of this node's run (run_seq {data.get('run_seq')}, " + f"here {self._run_seq})" + ) + if refuse: + Logger.warning(f"Follow: refusing announce {what}: {refuse}") + return + + # Read before taking the table lock: nothing calls into CueHandler or + # the engine while holding it. While a STOP or a load is pending here + # the epochs still belong to the run that is ending: read them later. + arm_epoch = stop_epoch = None + if not self.stop_pending: + arm_epoch = self._ch.arm_epoch() + stop_epoch = self._ch._last_stop_chain_epoch + project_gen = self._get_project_gen() + with self._lock: + old = self._table.get(source_id) + if ( + old is not None + and old.state in _LIVE + and old.run_seq == self._run_seq + and abs(old.instant - instant) <= FOLLOW_INSTANT_MATCH_MS + ): + old.origins.add(origin) + return + entry = _Entry( + source_id=source_id, + target_id=target_id, + instant=instant, + run_seq=self._run_seq, + origins={origin}, + arm_epoch=arm_epoch, + stop_epoch=stop_epoch, + project_gen=project_gen, + ) + self._table[source_id] = entry + + now = mtc.main_tc.milliseconds_exact + late = now - instant + if late > FOLLOW_LATE_ERROR_MS: + Logger.error( + f"Follow: announce {what} arrived {late:.0f}ms after its " + "instant; starting it anyway, in sync with its beginning cut" + ) + elif late > 0: + Logger.warning( + f"Follow: announce {what} arrived {late:.0f}ms after its " + "instant; starting it at once" + ) + Logger.info(f"Follow: received announce {what} at {instant:.0f}ms") + self._spawn(f"FollowPreArm:{source_id}", self._prearm, entry) + self._spawn(f"FollowWait:{source_id}", self._wait_and_dispatch, entry) + + def _on_cancel(self, data: dict) -> None: + source_id = str(data.get("source")) + origin = data.get("origin") + reason = data.get("reason") + if origin == self.node_id: + return + try: + instant = float(data.get("seed_ms")) + except (TypeError, ValueError): + instant = None + what = f"{source_id} at {instant} from {origin} ({reason})" + with self._lock: + e = self._table.get(source_id) + if ( + e is None + or instant is None + or abs(e.instant - instant) > FOLLOW_INSTANT_MATCH_MS + or e.state not in _LIVE + or data.get("run_seq") != e.run_seq + ): + Logger.debug(f"Follow: cancel {what}: nothing pending") + return + if reason == "error" and origin in e.origins and len(e.origins) > 1: + e.origins.discard(origin) + Logger.info(f"Follow: cancel {what}: other owners remain") + return + state = e.state + e.state = "cancelled" + dispatched = (e.cue, e.gen, e.disarm_epoch, e.armed_at, e.started_before) + if state != "dispatched": + Logger.info(f"Follow: cancel {what} applied before the dispatch") + return + self._spawn( + f"FollowStop:{source_id}", self._stop_dispatched, what, reason, *dispatched + ) + + # ------------------------------------------------------------------ + # Receiver threads + # ------------------------------------------------------------------ + + def _is_current(self, e: _Entry) -> bool: + return ( + self._table.get(e.source_id) is e + and e.state in _LIVE + and self._run_seq == e.run_seq + ) + + def _current(self, e: _Entry) -> bool: + with self._lock: + return self._is_current(e) + + def _epochs_ready(self, e: _Entry) -> bool: + """Read the STOP/load epochs for an entry that arrived while a STOP or + a load was pending, once it has run. False while still pending.""" + if e.arm_epoch is not None: + return True + if self.stop_pending: + return False + arm_epoch = self._ch.arm_epoch() + stop_epoch = self._ch._last_stop_chain_epoch + with self._lock: + if e.arm_epoch is None: + e.arm_epoch, e.stop_epoch = arm_epoch, stop_epoch + return True + + def _prearm(self, e: _Entry) -> None: + """Until the instant, arm what this node would start, every + FOLLOW_PREARM_PERIOD_S: `arm()` skips what is loaded, so each round + only arms what became unloaded since -- a cue whose earlier pass has + just ended, or a target enabled meanwhile.""" + + def should_continue(): + return ( + self._current(e) + and self._get_project_gen() == e.project_gen + and self._ch.arm_epoch() == e.arm_epoch + ) + + mtc = self._get_mtc() + try: + while self._current(e) and not self._epochs_ready(e): + if mtc is None or mtc.main_tc.milliseconds_exact >= e.instant: + return + self._prearm_sleep(FOLLOW_PREARM_PERIOD_S) + while should_continue(): + script = self._get_script() + target = script.find(e.target_id) if script is not None else None + cue = self._first_local(target) if target is not None else None + if cue is not None: + self._arm_round(e, cue, should_continue) + if mtc is None or mtc.main_tc.milliseconds_exact >= e.instant: + return + self._prearm_sleep(FOLLOW_PREARM_PERIOD_S) + except Exception as ex: + Logger.error(f"Follow: pre-arm for {e.source_id} failed: {ex}") + + def _arm_round(self, e: _Entry, cue, should_continue) -> None: + walk = _ArmWalk(should_continue=should_continue, epoch=e.arm_epoch) + if not getattr(cue, "loaded", False): + self._ch.arm(cue, init=True, walk=walk) + # arm() reaches an action's target only when it arms the action itself, + # and arm() on a loaded, disabled cue disarms it even while it plays. + if isinstance(cue, ActionCue) and cue.action_type in ("play", "fade_action"): + at = getattr(cue, "_action_target_object", None) + if ( + at is not None + and getattr(at, "enabled", False) + and getattr(at, "_local", False) + and not getattr(at, "loaded", False) + ): + self._ch.arm(at, init=True, walk=walk) + # The lookahead once per cue, so its depth WARNING is not repeated. + if id(cue) not in e.lookahead_done and getattr(cue, "loaded", False): + e.lookahead_done.add(id(cue)) + self._ch._arm_ahead( + cue, should_continue=should_continue, arm_epoch=e.arm_epoch + ) + + def _wait_and_dispatch(self, e: _Entry) -> None: + mtc = self._get_mtc() + script = self._get_script() + if mtc is None or script is None: + return + source = script.find(e.source_id) + target = script.find(e.target_id) + post_ms = 0.0 + try: + post_ms = source.postwait.milliseconds_exact + except Exception: + pass + check_ms = e.instant - post_ms + snapshot = None + while True: + if not self._current(e): + return + now = mtc.main_tc.milliseconds_exact + if post_ms > 0 and snapshot is None and now >= check_ms: + c = self._first_local(target) + snapshot = (c, getattr(c, "_go_generation", 0)) + if now >= e.instant and self._epochs_ready(e): + break + self._sleep(FOLLOW_POLL_S) + + # The yield: a manual GO of the target during the source's postwait + # already started it; the owner does not fire it again either. + if snapshot is not None and snapshot[0] is not None: + c = self._first_local(target) + if c is snapshot[0] and getattr(c, "_go_generation", 0) != snapshot[1]: + Logger.info( + f"Follow: {e.source_id} -> {e.target_id} yielded: " + f"{c.id} was started by a newer GO" + ) + self._set_state(e, "done") + return + + with self._lock: + if not self._is_current(e): + return + e.state = "dispatching" + try: + thread = self._ch.go_from( + target, + mtc, + e.instant, + arm_epoch=e.arm_epoch, + require_stop_epoch=e.stop_epoch, + commit_gate=lambda cue, gen: self._gate(e, cue, gen), + ) + except Exception as ex: + Logger.error( + f"Follow: dispatch of {e.source_id} -> {e.target_id} failed: {ex}" + ) + self._set_state(e, "done", only_from="dispatching") + return + with self._lock: + if e.state == "dispatching": + e.state = "done" + Logger.info( + f"Follow: {e.source_id} -> {e.target_id}: not dispatched " + "here (nothing local to start, or refused for a STOP)" + ) + elif e.state == "dispatched": + Logger.info( + f"Follow: dispatched {getattr(e.cue, 'id', '?')} for " + f"{e.source_id} at {e.instant:.0f}ms" + + ("" if thread is not None else " (refused)") + ) + + def _gate(self, e: _Entry, cue, gen: int) -> bool: + """Evaluated by go() inside its commit section (CueHandler._lock held). + Refuses an entry that was cancelled, replaced or dropped (a STOP or a + load emptied the table), or that belongs to another run.""" + with self._lock: + if not self._is_current(e) or e.state != "dispatching": + return False + e.cue = cue + e.gen = gen + e.disarm_epoch = self._ch._disarm_epoch + e.armed_at = getattr(cue, "_armed_at", None) + if getattr(cue, "_playing", False) and getattr(cue, "_revealed", False): + # Still on stage from an earlier pass: a cancel must treat it + # as started (go() is about to reset its reveal stamp). + e.started_before = getattr(cue, "_revealed_at", None) + e.state = "dispatched" + return True + + def _set_state(self, e: _Entry, state: str, only_from: str | None = None): + with self._lock: + if only_from is None or e.state == only_from: + e.state = state + + def _stop_dispatched( + self, what, reason, cue, gen, disarm_epoch, armed_at, started_before + ): + may_cut = reason in ("stop", "restart") + try: + outcome, age = self._ch.stop_dispatched_follow( + cue, + gen, + disarm_epoch, + armed_at, + may_cut=may_cut, + grace_s=FOLLOW_CANCEL_GRACE_S, + started_before=started_before, + ) + except Exception as ex: + Logger.error(f"Follow: cancel {what}: stopping {cue.id} failed: {ex}") + return + acted = isinstance(cue, (ActionCue, DmxCue)) and age is not None + if outcome == "owned": + Logger.info(f"Follow: cancel {what}: {cue.id} belongs to a newer GO") + elif outcome == "stopped" and age is None: + Logger.info(f"Follow: cancel {what}: stopped {cue.id} before it started") + elif outcome == "stopped": + Logger.warning( + f"Follow: cancel {what}: stopped {cue.id} {age:.2f}s after it " + "started" + ) + else: + Logger.error( + f"Follow: cancel {what} arrived " + + (f"{age:.2f}s " if age is not None else "") + + f"after {cue.id} started; it was left playing" + ) + if acted and outcome != "owned": + Logger.error( + f"Follow: {type(cue).__name__} {cue.id} had already acted (a DMX " + "scene sent, an action executed); that cannot be undone" + ) + + @staticmethod + def _spawn_thread(name: str, fn, *args) -> None: + threading.Thread(target=fn, args=args, name=name, daemon=True).start() diff --git a/src/cuemsengine/cues/loop_cue.py b/src/cuemsengine/cues/loop_cue.py index 99954ad..90cf07c 100644 --- a/src/cuemsengine/cues/loop_cue.py +++ b/src/cuemsengine/cues/loop_cue.py @@ -25,6 +25,16 @@ CUE_STATUS_UPDATE_HZ = 2 +def _start_of_last_loop(cue, loop_counter: int, on_last_loop) -> None: + """Call on_last_loop at the start of the last of several loops: the + previous iteration has re-based _end_mtc to the final end by now.""" + if on_last_loop is not None and cue.loop > 1 and loop_counter == cue.loop - 1: + try: + on_last_loop() + except Exception as e: + Logger.error(f"Last-loop callback of cue {cue.id} failed: {e}") + + @singledispatch def loop_cue(cue: Cue, mtc: MtcListener): """ @@ -92,7 +102,7 @@ def loop_fadeCue(cue: FadeCue, mtc: MtcListener): @loop_cue.register -def loop_audioCue(cue: AudioCue, mtc: MtcListener): +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 @@ -101,6 +111,9 @@ def loop_audioCue(cue: AudioCue, mtc: MtcListener): 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}" @@ -122,6 +135,7 @@ def loop_audioCue(cue: AudioCue, mtc: MtcListener): 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}" @@ -260,11 +274,12 @@ def loop_dmxCue(cue: DmxCue, mtc: MtcListener): @loop_cue.register -def loop_videoCue(cue: VideoCue, mtc: MtcListener): +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}" @@ -304,6 +319,7 @@ def loop_videoCue(cue: VideoCue, mtc: MtcListener): 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: diff --git a/tests/test_controller_commands.py b/tests/test_controller_commands.py index 8934405..070484e 100644 --- a/tests/test_controller_commands.py +++ b/tests/test_controller_commands.py @@ -333,26 +333,36 @@ def test_forwards_go_mtc_ms_when_mtc_present(self, controller): with patch.object(controller, "_forward_command_to_nodes") as mock_fwd: controller.go_script(None) mock_fwd.assert_called_once_with( - "/engine/command/go", {"go_mtc_ms": 79240.0} + "/engine/command/go", + {"go_mtc_ms": 79240.0, "run_seq": controller._run_seq}, ) - def test_forwards_original_string_value_when_mtc_listener_is_none(self, controller): + def test_forwards_only_run_seq_when_mtc_listener_is_none(self, controller): """The `controller` fixture builds with_mtc=False -- mtc_listener is None, the only reachable no-MTC case in production (go_script already refuses unless armed=="yes", and MTC runs by the time a project can be - armed).""" + armed). + + The GO value is always a dict now (869fc8ytz): it carries the run + counter that cross-node follows are checked against. The caller's + own value is dropped -- nodes never read it; they only read the + anchor, and fall back to their own MTC when it is absent.""" self._armed(controller) assert controller.mtc_listener is None with patch.object(controller, "_forward_command_to_nodes") as mock_fwd: controller.go_script("complex_test") - mock_fwd.assert_called_once_with("/engine/command/go", "complex_test") + mock_fwd.assert_called_once_with( + "/engine/command/go", {"run_seq": controller._run_seq} + ) - def test_forwards_none_value_unchanged_when_mtc_listener_is_none(self, controller): + def test_forwards_only_run_seq_for_none_value_without_mtc(self, controller): self._armed(controller) assert controller.mtc_listener is None with patch.object(controller, "_forward_command_to_nodes") as mock_fwd: controller.go_script(None) - mock_fwd.assert_called_once_with("/engine/command/go", None) + mock_fwd.assert_called_once_with( + "/engine/command/go", {"run_seq": controller._run_seq} + ) def test_forwards_original_value_when_controller_mtc_is_dead(self, controller): """test2, 2026-09-25: the controller's listener never opened a port @@ -360,11 +370,13 @@ def test_forwards_original_value_when_controller_mtc_is_dead(self, controller): node accepted 0.0 as the anchor -- 4.48s late. A listener that is present but not receiving must not be trusted: send the GO without an anchor, so each node falls back to its own MTC (pre-fix - behaviour).""" + behaviour). The run counter still goes out (869fc8ytz).""" self._armed(controller) controller.mtc_listener = Mock() controller.mtc_listener.main_tc.milliseconds_exact = 0.0 controller.mtc_listener.is_receiving.return_value = False with patch.object(controller, "_forward_command_to_nodes") as mock_fwd: controller.go_script(None) - mock_fwd.assert_called_once_with("/engine/command/go", None) + mock_fwd.assert_called_once_with( + "/engine/command/go", {"run_seq": controller._run_seq} + ) diff --git a/tests/test_cross_node_follow_controller.py b/tests/test_cross_node_follow_controller.py new file mode 100644 index 0000000..969ab8a --- /dev/null +++ b/tests/test_cross_node_follow_controller.py @@ -0,0 +1,228 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera +"""Cross-node Auto follow, controller side (ClickUp 869fc8ytz). + +The controller only relays: a node that owns an Auto follow's source announces +the follow instant (or cancels it), and the controller forwards that to every +node. What it adds is the run counter, `run_seq`: + +- it starts from the clock, so a restarted controller never reuses a value a + node still holds; +- STOP, load and unload bump it BEFORE they forward anything to the nodes; +- every GO carries it, so every node knows the run it belongs to; +- a follow message is relayed only while the show runs and only if it carries + the current value. The check and the forward share one lock with the bump, + so a relayed follow can never be queued behind a STOP it predates. + +Design: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md §3.3-3.4. +""" + +import time +from os import environ +from pathlib import Path +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from cuemsengine.comms.NodesHub import ActionType, NodeOperation, OperationType + + +@pytest.fixture(autouse=True) +def set_config_path(): + test_conf_path = Path(__file__).parent / ".." / "dev" / "test_xml_files" + environ["CUEMS_CONF_PATH"] = str(test_conf_path) + + +@pytest.fixture +def controller(): + with ( + patch("cuemsengine.core.BaseEngine.ConfigManager") as MockCM, + patch( + "cuemsengine.core.BaseEngine.BaseEngine.get_controller_ip", + return_value="localhost", + ), + ): + mock_cm_instance = MockCM.return_value + mock_cm_instance.node_conf = { + "uuid": "test-controller-uuid", + "mtc_port": "MTC_MIDI_PORT", + } + mock_cm_instance.library_path = str( + Path(__file__).parent / ".." / "dev" / "test_xml_files" + ) + mock_cm_instance.tmp_path = "/tmp" + + from cuemsengine.ControllerEngine import ControllerEngine + + engine = ControllerEngine(with_mtc=False) + engine.communications_thread = Mock() + engine.communications_thread.broadcast_osc = Mock() + engine.communications_thread.nng_hub = Mock() + yield engine + engine.stop() + + +def _seq_at_forward(controller, address): + """Patch _forward_command_to_nodes and record run_seq at the moment the + given address is forwarded.""" + seen = [] + + def fwd(addr, value): + if addr == address: + seen.append(controller._run_seq) + + return seen, patch.object(controller, "_forward_command_to_nodes", side_effect=fwd) + + +class TestRunSeq: + def test_starts_from_the_clock(self, controller): + now_ms = time.time_ns() // 1_000_000 + assert isinstance(controller._run_seq, int) + assert now_ms - 60_000 <= controller._run_seq <= now_ms + + def test_stop_bumps_it_before_forwarding(self, controller): + controller.set_status("running", "yes") + before = controller._run_seq + seen, p = _seq_at_forward(controller, "/engine/command/stop") + with p: + controller.stop_script(None) + assert seen == [before + 1] + + def test_stop_while_not_running_leaves_it(self, controller): + controller.set_status("running", "no") + before = controller._run_seq + with patch.object(controller, "_forward_command_to_nodes"): + controller.stop_script(None) + assert controller._run_seq == before + + def test_unload_bumps_it_before_forwarding(self, controller): + controller.set_status("running", "no") + before = controller._run_seq + seen, p = _seq_at_forward(controller, "/engine/command/stop") + with p, patch.object(controller, "reset_script"): + controller.unload_project(None) + assert seen == [before + 1] + + def test_load_bumps_it_before_forwarding(self, controller): + controller.set_status("running", "no") + before = controller._run_seq + seen, p = _seq_at_forward(controller, "/engine/command/load") + + def read_script(name): + controller.script = MagicMock() + + with ( + p, + patch.object(controller, "reset_script"), + patch.object(controller, "read_script", side_effect=read_script), + patch.object(controller, "_collect_cue_ids", return_value=[]), + patch.object(controller, "_collect_cue_enabled", return_value={}), + patch.object(controller, "_resolve_cluster_state"), + patch.object(controller, "start_timecode"), + patch.object(controller, "set_show_lock_file"), + ): + assert controller.load_project("p") is True + assert seen == [before + 1] + + def test_every_go_carries_the_current_value(self, controller): + controller.set_status("armed", "yes") + controller.script = Mock() + with patch.object(controller, "_forward_command_to_nodes") as fwd: + controller.go_script(None) + controller.go_script(None) + values = [c.args[1] for c in fwd.call_args_list] + assert [v["run_seq"] for v in values] == [controller._run_seq] * 2 + + +def _follow(sender="node-a", **data): + payload = { + "op": "announce", + "origin": sender, + "source": "src", + "target": "tgt", + "seed_ms": 1000.0, + } + payload.update(data) + return NodeOperation( + type=OperationType.STATUS, + action=ActionType.UPDATE, + sender=sender, + target="follow", + data=payload, + ) + + +class TestFollowRelay: + @pytest.fixture + def running(self, controller): + controller.set_status("running", "yes") + controller._adopted_nodes = {"node-a", "node-b"} + controller.cue_status = {"src": 1, "tgt": 0} + return controller + + def test_forwards_a_current_message_to_every_node(self, running): + op = _follow(run_seq=running._run_seq) + with patch.object(running, "_forward_command_to_nodes") as fwd: + running.status_operation_callback(op) + fwd.assert_called_once_with("/engine/command/follow", op.data) + + def test_forwards_a_cancel_too(self, running): + op = _follow(op="cancel", reason="stop", run_seq=running._run_seq) + with patch.object(running, "_forward_command_to_nodes") as fwd: + running.status_operation_callback(op) + fwd.assert_called_once_with("/engine/command/follow", op.data) + + def test_drops_a_stale_run_seq(self, running): + op = _follow(run_seq=running._run_seq - 1) + with ( + patch.object(running, "_forward_command_to_nodes") as fwd, + patch("cuemsengine.ControllerEngine.Logger") as log, + ): + running.status_operation_callback(op) + fwd.assert_not_called() + assert log.info.called + + def test_drops_when_not_running(self, running): + running.set_status("running", "no") + op = _follow(run_seq=running._run_seq) + with patch.object(running, "_forward_command_to_nodes") as fwd: + running.status_operation_callback(op) + fwd.assert_not_called() + + def test_drops_a_message_sent_before_a_stop(self, running): + """The node's run_seq predates the STOP: it must not reach the nodes + after the STOP, whatever the order the two arrive in.""" + op = _follow(run_seq=running._run_seq) + with patch.object(running, "_forward_command_to_nodes") as fwd: + running.stop_script(None) + running.set_status("running", "yes") # even a new GO's state + running.status_operation_callback(op) + assert all(c.args[0] != "/engine/command/follow" for c in fwd.call_args_list) + + def test_drops_a_non_adopted_sender_with_a_warning(self, running): + op = _follow(sender="stranger", run_seq=running._run_seq) + with ( + patch.object(running, "_forward_command_to_nodes") as fwd, + patch("cuemsengine.ControllerEngine.Logger") as log, + ): + running.status_operation_callback(op) + fwd.assert_not_called() + assert log.warning.called + + def test_drops_an_unknown_cue_with_a_warning(self, running): + op = _follow(target="not-in-project", run_seq=running._run_seq) + with ( + patch.object(running, "_forward_command_to_nodes") as fwd, + patch("cuemsengine.ControllerEngine.Logger") as log, + ): + running.status_operation_callback(op) + fwd.assert_not_called() + assert log.warning.called + + def test_tolerates_a_message_without_data(self, running): + op = _follow(run_seq=running._run_seq) + op.data = None + with patch.object(running, "_forward_command_to_nodes") as fwd: + running.status_operation_callback(op) + fwd.assert_not_called() diff --git a/tests/test_cross_node_follow_hook.py b/tests/test_cross_node_follow_hook.py new file mode 100644 index 0000000..3ac1d74 --- /dev/null +++ b/tests/test_cross_node_follow_hook.py @@ -0,0 +1,164 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera +"""Cross-node Auto follow: the node's receive hook (ClickUp 869fc8ytz). + +NodeCommunications receives one bus message at a time, in order, and then runs +each command on its own thread, where the command lock decides the order. +The follow messages and the run counter must follow the BUS order, so they are +applied in a synchronous hook, before any command thread is started: + +- a 'follow' message is applied there and never reaches the command lock; +- GO records the run counter, STOP and load clear it, there; +- the hook can never stop a GO, a STOP or a load from running: an exception + in it is logged and the command thread still starts. + +Design: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md §3.5. +""" + +import threading +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from cuemsengine.comms.NodeCommunications import NodeCommunications +from cuemsengine.comms.NodesHub import ActionType, NodeOperation, OperationType + +NODE_UUID = "aaaaaaaa-1111-2222-3333-444444444444" +CONTROLLER_UUID = "bbbbbbbb-1111-2222-3333-444444444444" + + +@pytest.fixture +def comms(monkeypatch): + monkeypatch.setattr("cuemsengine.comms.NodeCommunications.NodesHub", MagicMock()) + obj = NodeCommunications(hub_address="tcp://127.0.0.1:0", node_id=NODE_UUID) + obj.nng_hub.send_operation = AsyncMock() + obj._command_callback = MagicMock() + return obj + + +def _op(name, value): + return NodeOperation( + type=OperationType.COMMAND, + action=ActionType.UPDATE, + sender=CONTROLLER_UUID, + target=name, + data={"value": value, "address": f"/engine/command/{name}"}, + ) + + +def _threads_started(comms, op): + started = [] + real = threading.Thread + + def fake_thread(*a, **k): + t = real(*a, **k) + started.append(k.get("name")) + return t + + with patch("threading.Thread", side_effect=fake_thread): + comms._handle_command_operation(op) + return started + + +class TestReceiveHook: + def test_a_consumed_message_starts_no_command_thread(self, comms): + comms.set_receive_hook(MagicMock(return_value=True)) + assert _threads_started(comms, _op("follow", {"op": "announce"})) == [] + comms._command_callback.assert_not_called() + + def test_the_hook_sees_name_and_value_before_the_thread(self, comms): + hook = MagicMock(return_value=False) + comms.set_receive_hook(hook) + started = _threads_started(comms, _op("go", {"run_seq": 7})) + hook.assert_called_once_with("go", {"run_seq": 7}) + assert started == ["NNG-Command-go"] + + def test_a_hook_that_raises_never_loses_the_command(self, comms): + comms.set_receive_hook(MagicMock(side_effect=RuntimeError("boom"))) + with patch("cuemsengine.comms.NodeCommunications.Logger") as log: + started = _threads_started(comms, _op("stop", None)) + assert started == ["NNG-Command-stop"] + assert log.error.called + + def test_no_hook_keeps_todays_behaviour(self, comms): + assert _threads_started(comms, _op("go", None)) == ["NNG-Command-go"] + + +class _Relay: + def __init__(self): + self.calls = [] + + def begin_run(self, value): + self.calls.append(("begin_run", value)) + + def end_run(self, why): + self.calls.append(("end_run", why)) + + def on_message(self, value): + self.calls.append(("on_message", value)) + + +class TestNodeEngineHook: + """NodeEngine._on_command_received, called from the hook.""" + + def _node(self): + from cuemsengine.NodeEngine import NodeEngine + + node = object.__new__(NodeEngine) + node._follow_relay = _Relay() + return node + + def test_follow_is_consumed(self): + node = self._node() + assert node._on_command_received("follow", {"op": "cancel"}) is True + assert node._follow_relay.calls == [("on_message", {"op": "cancel"})] + + @pytest.mark.parametrize("value", [{"run_seq": 5, "go_mtc_ms": 1.0}, None, "x"]) + def test_go_records_the_run_and_still_runs(self, value): + node = self._node() + assert node._on_command_received("go", value) is False + assert node._follow_relay.calls == [("begin_run", value)] + + @pytest.mark.parametrize("name", ["stop", "load"]) + def test_stop_and_load_end_the_run_and_still_run(self, name): + node = self._node() + assert node._on_command_received(name, None) is False + assert node._follow_relay.calls == [("end_run", name)] + + @pytest.mark.parametrize("name", ["setnextcue", "cue_enabled", "deploy"]) + def test_other_commands_are_untouched(self, name): + node = self._node() + assert node._on_command_received(name, "v") is False + assert node._follow_relay.calls == [] + + def test_without_a_relay_nothing_is_consumed(self): + node = self._node() + node._follow_relay = None + assert node._on_command_received("follow", {}) is False + + +class TestReviewFixes: + def test_a_follow_handler_that_raises_is_still_consumed(self): + """MINOR-6: never let a follow message reach the command lock.""" + from cuemsengine.NodeEngine import NodeEngine + + node = object.__new__(NodeEngine) + node._follow_relay = MagicMock() + node._follow_relay.on_message.side_effect = RuntimeError("bug") + with patch("cuemsengine.NodeEngine.Logger") as log: + assert node._on_command_received("follow", {}) is True + assert log.error.called + + @pytest.mark.parametrize("name", ["stop", "load"]) + def test_stop_and_load_are_marked_applied_after_they_run(self, name): + """MINOR-4: the node's own STOP/load has run (or was refused).""" + from cuemsengine.NodeEngine import NodeEngine + + node = object.__new__(NodeEngine) + node._command_lock = threading.Lock() + node._follow_relay = MagicMock() + node.commands_dict = {name: MagicMock(side_effect=RuntimeError("x"))} + with pytest.raises(RuntimeError): + node.run_command(name, None) + node._follow_relay.stop_applied.assert_called_once() diff --git a/tests/test_cross_node_follow_owner.py b/tests/test_cross_node_follow_owner.py new file mode 100644 index 0000000..1772d0d --- /dev/null +++ b/tests/test_cross_node_follow_owner.py @@ -0,0 +1,638 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera +"""Cross-node Auto follow, owning side and CueHandler hooks (ClickUp 869fc8ytz). + +The node where an AudioCue or VideoCue with Auto follow is local announces the +follow instant when the cue starts (or at the start of its last loop), so the +other nodes can start their part of the target on time. The instant comes +from the same helper the cue's own fire uses. Leaving without firing sends a +cancel, except when every node already knows (a STOP, a load, the yield) or +when go() already sent it (the cue was started again). Only the current pass +announces, and the outstanding record is compare-and-cleared, so exactly one +cancel can follow one announcement. + +Also here: go()'s commit gate, the restart cancel's place in the commit +section, go_from passing the dispatch guards through, the guarded stop of a +follow target and its guarded disarm. + +Design: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md +§3.2, §3.5. +""" + +import sys +from threading import Lock +from types import SimpleNamespace +from unittest.mock import MagicMock, Mock, patch +from uuid import uuid4 + +import pytest + +sys.modules.setdefault("cuemsutils.tools.Osc_nodes_hub", Mock()) + +from cuemsutils.cues import DmxCue, VideoCue # noqa: E402 +from cuemsutils.tools.CTimecode import CTimecode # noqa: E402 + +from cuemsengine.cues.CueHandler import CueHandler # noqa: E402 + + +class _FakeMtc: + def __init__(self, ms=0.0): + self.main_tc = SimpleNamespace( + milliseconds_exact=float(ms), + milliseconds_rounded=int(ms), + framerate=25.0, + frames=0, + ) + + def advance(self, seconds): + self.main_tc.milliseconds_exact += float(seconds) * 1000.0 + self.main_tc.milliseconds_rounded = int(self.main_tc.milliseconds_exact) + + +def _tc(ms): + return CTimecode(start_seconds=ms / 1000.0) + + +def _video(post_go="go_at_end", target="default", loop=1, postwait=0, cls=VideoCue): + cue = cls.__new__(cls) + cue.id = str(uuid4()) + cue.enabled = True + cue.post_go = post_go + cue.prewait = CTimecode(start_seconds=0) + cue.postwait = CTimecode(start_seconds=postwait) + cue.loop = loop + cue._local = True + cue.loaded = True + cue._stop_requested = False + cue._go_generation = 1 + cue._revealed = False + cue._start_mtc = None + cue._end_mtc = None + if target == "default": + target = SimpleNamespace(id=str(uuid4()), _go_generation=0) + cue._target_object = target + return cue + + +def _relay(run_seq=7): + relay = MagicMock() + relay.run_seq = run_seq + relay.stop_pending = False + return relay + + +def _ch(relay=None): + ch = object.__new__(CueHandler) + ch._lock = Lock() + ch._armed_cues = [] + ch._armed_cues_set = set() + ch.communications_thread = MagicMock() + ch.disarm = MagicMock() + ch.arm = MagicMock() + ch._arm_ahead = MagicMock() + ch.go = MagicMock(return_value=None) + ch.go_from = MagicMock(return_value=None) + ch._next_local_fire = MagicMock(return_value=(None, 0.0)) + ch._follow = relay + return ch + + +def _run( + ch, + cue, + mtc, + body_ms=2000, + loop_effect=None, + order=None, + reveal_effect=None, + blank_effect=None, + **kw, +): + """Drive go_threaded with the players patched out. run_cue sets the + cue's start and end as the real one does (start = arrival + prewait).""" + calls = order if order is not None else [] + + def run_cue(c, m, start_ms): + calls.append("run_cue") + c._start_mtc = _tc(start_ms) + c._end_mtc = _tc(start_ms + body_ms) + + def loop_cue(c, m, on_last_loop=None): + calls.append("loop") + if loop_effect: + loop_effect(c, m, on_last_loop) + m.main_tc.milliseconds_exact = c._end_mtc.milliseconds_exact + + kw.setdefault("go_gen", cue._go_generation) + kw.setdefault("arm_epoch", ch._disarm_epoch) + ch._arm_ahead.side_effect = lambda *a, **k: calls.append("arm_ahead") + ch._follow and setattr( + ch._follow.announce, "side_effect", lambda *a: calls.append("announce") + ) + with ( + patch("cuemsengine.cues.CueHandler.run_cue", side_effect=run_cue), + patch( + "cuemsengine.cues.CueHandler.reveal_cue", + side_effect=lambda *a: ( + calls.append("reveal"), + reveal_effect and reveal_effect(), + ), + ), + patch("cuemsengine.cues.CueHandler.loop_cue", side_effect=loop_cue), + patch( + "cuemsengine.cues.CueHandler.blank_cue", + side_effect=lambda *a: blank_effect and blank_effect(), + ), + patch( + "cuemsengine.cues.CueHandler.sleep", side_effect=lambda s: mtc.advance(s) + ), + ): + ch.go_threaded(cue, mtc, 1000.0, **kw) + return calls + + +# ── announce ────────────────────────────────────────────────────────────── + + +class TestAnnounce: + def test_announces_the_fire_instant_after_the_reveal_before_the_arm_ahead(self): + relay = _relay() + ch, cue = _ch(relay), _video(postwait=0.5) + order = _run(ch, cue, _FakeMtc(1000), body_ms=2000) + relay.announce.assert_called_once_with(cue, 3500.0, 7) + assert order.index("reveal") < order.index("announce") + assert order.index("announce") < order.index("arm_ahead") + + def test_the_cue_fires_at_the_announced_instant(self): + relay = _relay() + ch, cue = _ch(relay), _video(postwait=0.5) + _run(ch, cue, _FakeMtc(1000), body_ms=2000) + args = ch.go_from.call_args.args + assert args[0] is cue._target_object and args[2] == 3500.0 + relay.cancel.assert_not_called() + assert getattr(cue, "_follow_out", None) is None + + @pytest.mark.parametrize( + "why", ["pause", "no_target", "infinite_loop", "dmx", "no_relay", "no_run"] + ) + def test_does_not_announce(self, why): + relay = _relay(run_seq=None if why == "no_run" else 7) + kw = {} + if why == "pause": + kw["post_go"] = "pause" + elif why == "no_target": + kw["target"] = None + elif why == "infinite_loop": + kw["loop"] = 0 + elif why == "dmx": + kw["cls"] = DmxCue + ch = _ch(None if why == "no_relay" else relay) + cue = _video(**kw) + _run(ch, cue, _FakeMtc(1000)) + relay.announce.assert_not_called() + + def test_a_cue_that_never_reveals_does_not_announce(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + with patch.object(ch, "_reveal_wait", return_value="stopped"): + _run(ch, cue, _FakeMtc(1000), order=[]) + relay.announce.assert_not_called() + + def test_a_replaced_pass_does_not_announce(self): + """A newer go() took the cue between the reveal and the announce.""" + relay = _relay() + ch, cue = _ch(relay), _video() + _run( + ch, + cue, + _FakeMtc(1000), + reveal_effect=lambda: setattr(cue, "_go_generation", 2), + ) + relay.announce.assert_not_called() + + def test_a_looping_cue_announces_once_at_its_last_loop(self): + relay = _relay() + ch, cue = _ch(relay), _video(loop=3) + + def loops(c, m, on_last_loop): + assert on_last_loop is not None + relay.announce.assert_not_called() # not at the reveal + c._end_mtc = _tc(7000) # the last loop's end, rebased + on_last_loop() + + _run(ch, cue, _FakeMtc(1000), body_ms=2000, loop_effect=loops) + relay.announce.assert_called_once_with(cue, 7000.0, 7) + + def test_a_looping_cue_without_follow_gets_no_callback(self): + ch, cue = _ch(_relay()), _video(loop=3, post_go="pause") + seen = [] + _run(ch, cue, _FakeMtc(1000), loop_effect=lambda c, m, cb: seen.append(cb)) + assert seen == [None] + + def test_an_instant_that_moved_is_an_error(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + def stretch(c, m, cb): + c._end_mtc = _tc(c._end_mtc.milliseconds_exact + 500) + + with patch("cuemsengine.cues.CueHandler.Logger") as log: + _run(ch, cue, _FakeMtc(1000), loop_effect=stretch) + assert any("announced" in str(c) for c in log.error.call_args_list) + + +# ── leaving without firing ──────────────────────────────────────────────── + + +class TestLeave: + def test_a_stop_action_cancels_with_reason_stop(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + def stop_action(c, m, cb): + c._stop_requested = True + c._go_generation += 1 + + _run(ch, cue, _FakeMtc(1000), body_ms=2000, loop_effect=stop_action) + relay.cancel.assert_called_once_with(cue, 3000.0, 7, "stop") + ch.go_from.assert_not_called() + + def test_a_pause_cancels_with_reason_stop(self): + relay = _relay() + ch, cue = _ch(relay), _video() + _run( + ch, + cue, + _FakeMtc(1000), + loop_effect=lambda c, m, cb: setattr(c, "_stop_requested", True), + ) + relay.cancel.assert_called_once_with(cue, 3000.0, 7, "stop") + + def test_a_stop_or_load_sends_nothing(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + def stop(c, m, cb): + ch._disarm_epoch += 1 + c._stop_requested = True + c._go_generation += 1 + + _run(ch, cue, _FakeMtc(1000), loop_effect=stop) + relay.cancel.assert_not_called() + assert getattr(cue, "_follow_out", None) is None + + def test_a_restart_lets_go_send_the_cancel(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + def restart(c, m, cb): + with ch._lock: + ch._take_follow_on_restart_locked(c) # what go() does + c._go_generation += 1 + + _run(ch, cue, _FakeMtc(1000), loop_effect=restart) + relay.cancel.assert_called_once_with(cue, 3000.0, 7, "restart") + + def test_an_exception_cancels_with_reason_error_and_still_raises(self): + relay = _relay() + ch, cue = _ch(relay), _video() + + def boom(c, m, cb): + raise RuntimeError("player died") + + with pytest.raises(RuntimeError): + _run(ch, cue, _FakeMtc(1000), loop_effect=boom) + relay.cancel.assert_called_once_with(cue, 3000.0, 7, "error") + + def test_the_yield_sends_nothing(self): + relay = _relay() + ch, cue = _ch(relay), _video(postwait=0.5) + + def manual_go_of_target(): # during the postwait tail + cue._target_object._go_generation += 1 + + _run(ch, cue, _FakeMtc(1000), blank_effect=manual_go_of_target) + ch.go_from.assert_not_called() + relay.cancel.assert_not_called() + + def test_an_error_in_the_own_fire_does_not_cancel(self): + relay = _relay() + ch, cue = _ch(relay), _video() + ch.go_from.side_effect = RuntimeError("fire failed") + with pytest.raises(RuntimeError): + _run(ch, cue, _FakeMtc(1000)) + relay.cancel.assert_not_called() + + +# ── go(): commit gate and restart cancel ───────────────────────────────── + + +def _plain(id="c", **kw): + c = SimpleNamespace(id=id, loaded=True, enabled=True, _local=True) + for k, v in kw.items(): + setattr(c, k, v) + return c + + +class TestGoCommit: + def _go(self, ch, cue, **kw): + with patch.object(CueHandler, "go_threaded"): + return ch.go(cue, _FakeMtc(), 0.0, **kw) + + def test_restart_sends_a_cancel_for_the_outstanding_announcement(self): + relay = _relay() + ch = _ch(relay) + del ch.go + cue = _plain(_follow_out=(5000.0, 7), _target_object=None) + assert self._go(ch, cue) is not None + relay.cancel.assert_called_once_with(cue, 5000.0, 7, "restart") + assert cue._follow_out is None + + def test_an_outstanding_announcement_of_an_old_run_is_dropped_silently(self): + relay = _relay() + ch = _ch(relay) + del ch.go + cue = _plain(_follow_out=(5000.0, 6)) + self._go(ch, cue) + relay.cancel.assert_not_called() + assert cue._follow_out is None + + def test_a_refused_dispatch_sends_nothing_and_keeps_the_record(self): + relay = _relay() + ch = _ch(relay) + del ch.go + ch._last_stop_chain_epoch = 1 + cue = _plain(_follow_out=(5000.0, 7)) + assert self._go(ch, cue, require_stop_epoch=0) is None + relay.cancel.assert_not_called() + assert cue._follow_out == (5000.0, 7) + + def test_the_gate_sees_the_generation_about_to_be_assigned(self): + ch = _ch() + del ch.go + cue = _plain(_go_generation=4) + gate = MagicMock(return_value=True) + self._go(ch, cue, commit_gate=gate) + gate.assert_called_once_with(cue, 5) + assert cue._go_generation == 5 + + def test_a_refusing_gate_commits_nothing(self): + relay = _relay() + ch = _ch(relay) + del ch.go + cue = _plain(_go_generation=4, _follow_out=(5000.0, 7), _playing=False) + assert self._go(ch, cue, commit_gate=MagicMock(return_value=False)) is None + assert cue._go_generation == 4 and cue._playing is False + relay.cancel.assert_not_called() + + def test_a_gate_that_raises_refuses(self): + ch = _ch() + del ch.go + cue = _plain(_go_generation=4) + with patch("cuemsengine.cues.CueHandler.Logger") as log: + out = self._go(ch, cue, commit_gate=MagicMock(side_effect=ValueError())) + assert out is None and log.error.called + + def test_go_clears_the_start_stamp(self): + ch = _ch() + del ch.go + cue = _plain(_revealed_at=123.0) + self._go(ch, cue) + assert cue._revealed_at is None + + +class TestGoFrom: + def test_passes_the_guards_only_when_set(self): + ch = _ch() + del ch.go_from + cue = _plain(post_go="pause") + ch.go_from(cue, _FakeMtc(), 10.0) + ch.go.assert_called_once_with(cue, ch.go.call_args.args[1], 10.0) + + def test_passes_the_guards_through(self): + ch = _ch() + del ch.go_from + cue = _plain(post_go="pause") + gate = MagicMock() + ch.go_from( + cue, _FakeMtc(), 10.0, arm_epoch=3, require_stop_epoch=2, commit_gate=gate + ) + assert ch.go.call_args.kwargs == { + "arm_epoch": 3, + "require_stop_epoch": 2, + "commit_gate": gate, + } + + +# ── the guarded stop and disarm of a follow target ─────────────────────── + + +class TestStopDispatchedFollow: + def _target(self, revealed_ago=None, gen=9): + from time import monotonic + + cue = _plain( + _go_generation=gen, + _stop_requested=False, + _playing=True, + _revealed_at=None if revealed_ago is None else monotonic() - revealed_ago, + _armed_at=42.0, + ) + return cue + + def _ch(self): + ch = _ch() + ch._disarm_epoch = 3 + ch.cancel_pending_descendants = MagicMock() + ch._end_illumination = MagicMock() + ch.disarm_if_unchanged = MagicMock(return_value=True) + return ch + + def _stop(self, ch, cue, may_cut=True): + with patch("cuemsengine.cues.CueHandler.sleep"): + return ch.stop_dispatched_follow( + cue, 9, 3, 42.0, may_cut=may_cut, grace_s=2.0 + ) + + def test_a_newer_go_owns_it(self): + ch, cue = self._ch(), self._target(gen=10) + assert self._stop(ch, cue)[0] == "owned" + assert cue._stop_requested is False + + def test_a_stop_owns_it(self): + ch, cue = self._ch(), self._target() + ch._disarm_epoch = 4 + assert self._stop(ch, cue)[0] == "owned" + + def test_not_started_is_stopped(self): + ch, cue = self._ch(), self._target() + assert self._stop(ch, cue)[0] == "stopped" + assert cue._stop_requested and cue._go_generation == 10 and not cue._playing + ch.cancel_pending_descendants.assert_called_once_with(cue) + ch.disarm_if_unchanged.assert_called_once_with( + cue, 10, 3, 42.0, reason="follow_cancelled" + ) + + def test_started_inside_the_grace_is_stopped_when_it_may_cut(self): + ch, cue = self._ch(), self._target(revealed_ago=0.5) + outcome, age = self._stop(ch, cue) + assert outcome == "stopped" and 0.4 < age < 1.0 + + def test_started_inside_the_grace_is_left_for_an_error(self): + ch, cue = self._ch(), self._target(revealed_ago=0.5) + assert self._stop(ch, cue, may_cut=False)[0] == "left" + assert cue._stop_requested is False + + def test_started_after_the_grace_is_left(self): + ch, cue = self._ch(), self._target(revealed_ago=3.0) + assert self._stop(ch, cue)[0] == "left" + ch.disarm_if_unchanged.assert_not_called() + + +class TestDisarmIfUnchanged: + def _setup(self): + ch = object.__new__(CueHandler) + ch._lock = Lock() + ch._disarm_epoch = 3 + ch.communications_thread = MagicMock() + cue = _plain(_go_generation=10, _armed_at=42.0, _playing=False) + ch._armed_cues = [cue] + ch._armed_cues_set = {cue.id} + return ch, cue + + def test_unpublishes_and_releases(self): + ch, cue = self._setup() + seen = [] + + def release(c, reason, what="Disarmed"): + seen.append(ch._arming_registry().get(c.id) is not None) + + with patch.object(ch, "_release_cue_resources", side_effect=release): + assert ch.disarm_if_unchanged(cue, 10, 3, 42.0, reason="x") is True + assert cue.loaded is False and ch._armed_cues == [] + assert seen == [True] # released while holding the cue's claim + assert ch._arming_registry() == {} + + @pytest.mark.parametrize("change", ["gen", "epoch", "rearmed", "unloaded"]) + def test_refuses_when_anything_moved(self, change): + ch, cue = self._setup() + if change == "gen": + cue._go_generation = 11 + elif change == "epoch": + ch._disarm_epoch = 4 + elif change == "rearmed": + cue._armed_at = 43.0 + else: + cue.loaded = False + with patch.object(ch, "_release_cue_resources") as rel: + assert ch.disarm_if_unchanged(cue, 10, 3, 42.0, reason="x") is False + rel.assert_not_called() + + def test_refuses_while_an_arm_is_in_flight(self): + ch, cue = self._setup() + ch._arming_registry()[cue.id] = object() + with patch.object(ch, "_release_cue_resources") as rel: + assert ch.disarm_if_unchanged(cue, 10, 3, 42.0, reason="x") is False + rel.assert_not_called() + + def test_the_claim_is_released_even_if_the_release_fails(self): + ch, cue = self._setup() + with patch.object(ch, "_release_cue_resources", side_effect=RuntimeError()): + with pytest.raises(RuntimeError): + ch.disarm_if_unchanged(cue, 10, 3, 42.0, reason="x") + assert ch._arming_registry() == {} + + +# ── code review of the implementation (2026-10-06) ─────────────────────── + + +class TestReviewFixes: + def test_the_start_stamp_is_set_with_the_reveal_decision(self): + """MINOR-1: a cancel taking the lock right after the reveal decision + must already see the cue as started.""" + relay = _relay() + ch, cue = _ch(relay), _video() + seen = [] + _run( + ch, + cue, + _FakeMtc(1000), + reveal_effect=lambda: seen.append(getattr(cue, "_revealed_at", None)), + ) + assert seen and seen[0] is not None + + def test_a_restart_just_before_the_fire_skips_the_fire(self): + """MINOR-2: go() took the announcement (restart) between the tail and + the fire; the replaced pass must not fire its follow.""" + relay = _relay() + ch, cue = _ch(relay), _video() + + def restart(*a, **k): + with ch._lock: + ch._take_follow_on_restart_locked(cue) + cue._go_generation += 1 + + ch.communications_thread.remove_cue.side_effect = restart + _run(ch, cue, _FakeMtc(1000)) + ch.go_from.assert_not_called() + relay.cancel.assert_called_once_with(cue, 3000.0, 7, "restart") + + def test_without_an_announcement_the_fire_is_unchanged(self): + """The generation re-check only applies to an announced follow; a + same-node or unannounced follow fires as before.""" + ch, cue = _ch(None), _video() + + def bump(*a, **k): + cue._go_generation += 1 + + ch.communications_thread.remove_cue.side_effect = bump + _run(ch, cue, _FakeMtc(1000)) + ch.go_from.assert_called_once() + + def test_no_announcement_while_a_stop_is_pending(self): + """MINOR-4: a STOP was received but not applied yet; this pass is + about to be stopped and must not announce into the next run.""" + relay = _relay() + relay.stop_pending = True + ch, cue = _ch(relay), _video() + _run(ch, cue, _FakeMtc(1000)) + relay.announce.assert_not_called() + + +class TestStopDispatchedFollowReview: + def _ch(self): + ch = _ch() + ch._disarm_epoch = 3 + ch.cancel_pending_descendants = MagicMock() + ch._end_illumination = MagicMock() + ch.disarm_if_unchanged = MagicMock(return_value=True) + return ch + + def test_revealed_without_a_stamp_counts_as_started(self): + ch = self._ch() + cue = _plain(_go_generation=9, _revealed=True, _revealed_at=None) + with patch("cuemsengine.cues.CueHandler.sleep"): + out = ch.stop_dispatched_follow(cue, 9, 3, 42.0, may_cut=False, grace_s=2.0) + assert out[0] == "left" + + def test_a_cue_already_on_stage_before_the_follow_counts_as_started(self): + """MINOR-3: the follow restarted a cue that was playing; its new pass + has not revealed yet, but its output has been on stage for minutes.""" + from time import monotonic + + ch = self._ch() + cue = _plain(_go_generation=9, _revealed=False, _revealed_at=None) + with patch("cuemsengine.cues.CueHandler.sleep"): + out = ch.stop_dispatched_follow( + cue, + 9, + 3, + 42.0, + may_cut=True, + grace_s=2.0, + started_before=monotonic() - 120.0, + ) + assert out[0] == "left" and out[1] > 100 + ch.disarm_if_unchanged.assert_not_called() diff --git a/tests/test_cross_node_follow_receiver.py b/tests/test_cross_node_follow_receiver.py new file mode 100644 index 0000000..9f9f747 --- /dev/null +++ b/tests/test_cross_node_follow_receiver.py @@ -0,0 +1,668 @@ +# SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL +# SPDX-License-Identifier: GPL-3.0-or-later +# SPDX-FileContributor: Ion Reguera +"""Cross-node Auto follow, receiving side (ClickUp 869fc8ytz). + +A node that does NOT own an Auto follow's source learns the follow instant from +the owner's announcement, relayed by the controller. It keeps one entry per +source cue, pre-arms what it would start, waits for the instant on MTC and +dispatches then, through `go_from`, with a commit gate that `go()` evaluates +inside its own commit section. The owner's cancel is applied by the state of +the entry: + + waiting / dispatching -> cancelled (the waiter ends, or the gate refuses) + dispatched -> handed to CueHandler.stop_dispatched_follow, which + applies the 2 s rule under its own lock + +Messages are applied in bus order, in the receive hook; nothing here waits. + +Design: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md §3.5. +""" + +import sys +from types import SimpleNamespace +from unittest.mock import MagicMock, Mock, patch + +import pytest + +sys.modules.setdefault("cuemsutils.tools.Osc_nodes_hub", Mock()) + +from cuemsutils.tools.CTimecode import CTimecode # noqa: E402 + +from cuemsengine.core.BaseEngine import BaseEngine # noqa: E402 +from cuemsengine.cues.FollowRelay import FollowRelay # noqa: E402 + +ME, OWNER, OTHER = "node-me", "node-owner", "node-other" + + +class _Mtc: + def __init__(self, ms=0.0): + self.main_tc = SimpleNamespace(milliseconds_exact=float(ms)) + + def set(self, ms): + self.main_tc.milliseconds_exact = float(ms) + + +def _cue(id, local=False, post_go="pause", target=None, postwait=0, **kw): + c = SimpleNamespace( + id=id, + _local=local, + enabled=True, + loaded=False, + post_go=post_go, + postwait=CTimecode(start_seconds=postwait), + _target_object=target, + _go_generation=0, + ) + for k, v in kw.items(): + setattr(c, k, v) + return c + + +class _Script: + def __init__(self, *cues): + self.cues = {str(c.id): c for c in cues} + + def find(self, id): + return self.cues.get(str(id)) + + +class _Rig: + """A FollowRelay wired to fakes. Threads are not started: `spawned` + records them and the test runs them by hand.""" + + def __init__(self, script, ms=0.0, run_seq=7): + self.mtc = _Mtc(ms) + self.ch = MagicMock() + self.ch._disarm_epoch = 3 + self.ch._last_stop_chain_epoch = 2 + self.ch.arm_epoch.side_effect = lambda: self.ch._disarm_epoch + self.ch.go_from.return_value = None + self.ch.stop_dispatched_follow.return_value = ("stopped", None) + self.sent = [] + self.spawned = [] + self.gen = [1] + self.script = script + self.relay = FollowRelay( + self.ch, + node_id=ME, + send=self.sent.append, + get_script=lambda: self.script, + get_mtc=lambda: self.mtc, + get_project_gen=lambda: self.gen[0], + first_local=BaseEngine._first_local_enabled_in_go_chain, + spawn=lambda name, fn, *a: self.spawned.append((name, fn, a)), + sleep=lambda s: self.mtc.set(self.mtc.main_tc.milliseconds_exact + 20), + ) + if run_seq is not None: + self.relay.begin_run({"run_seq": run_seq}) + + def announce(self, source="S", target="T", seed=1000.0, origin=OWNER, run_seq=7): + self.relay.on_message( + { + "op": "announce", + "origin": origin, + "run_seq": run_seq, + "source": source, + "target": target, + "seed_ms": seed, + } + ) + + def cancel(self, source="S", target="T", seed=1000.0, origin=OWNER, reason="stop"): + self.relay.on_message( + { + "op": "cancel", + "origin": origin, + "run_seq": 7, + "source": source, + "target": target, + "seed_ms": seed, + "reason": reason, + } + ) + + def run(self, prefix): + """Run the spawned threads whose name starts with prefix.""" + for name, fn, a in list(self.spawned): + if name.startswith(prefix): + fn(*a) + + def entry(self, source="S"): + return self.relay._table.get(source) + + +@pytest.fixture +def basic(): + """S (video, on the owner) --Auto follow--> T (local here).""" + t = _cue("T", local=True) + s = _cue("S", local=False, post_go="go_at_end", target=t) + return _Rig(_Script(s, t)) + + +class TestAnnounce: + def test_creates_an_entry_and_starts_prearm_and_waiter(self, basic): + basic.announce() + e = basic.entry() + assert e.state == "waiting" and e.instant == 1000.0 and e.origins == {OWNER} + assert e.arm_epoch == 3 and e.stop_epoch == 2 + names = sorted(n.split(":")[0] for n, _, _ in basic.spawned) + assert names == ["FollowPreArm", "FollowWait"] + + def test_ignores_its_own_announcement(self, basic): + basic.announce(origin=ME) + assert basic.entry() is None and basic.spawned == [] + + def test_ignores_a_source_local_here(self): + t = _cue("T", local=True) + s = _cue("S", local=True, post_go="go_at_end", target=t) + rig = _Rig(_Script(s, t)) + rig.announce() + assert rig.entry() is None and rig.spawned == [] + + @pytest.mark.parametrize( + "why", ["no_script", "no_mtc", "unknown_cue", "stale_run", "no_run"] + ) + def test_refuses_with_a_warning(self, basic, why): + kw = {} + if why == "no_script": + basic.script = None + elif why == "no_mtc": + basic.mtc = None + elif why == "unknown_cue": + kw["target"] = "nope" + elif why == "stale_run": + kw["run_seq"] = 6 + elif why == "no_run": + basic.relay.end_run("stop") + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.announce(**kw) + assert basic.entry() is None and basic.spawned == [] + assert log.warning.called + + def test_tolerates_a_message_that_is_not_a_dict(self, basic): + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.relay.on_message("garbage") + basic.relay.on_message(None) + assert log.warning.call_count == 2 + + def test_same_instant_from_another_owner_joins_the_entry(self, basic): + basic.announce() + basic.announce(origin=OTHER, seed=1000.4) + assert basic.entry().origins == {OWNER, OTHER} + assert len(basic.spawned) == 2 + + def test_a_new_instant_replaces_the_entry(self, basic): + basic.announce() + old = basic.entry() + basic.announce(seed=5000.0) + assert basic.entry() is not old and basic.entry().instant == 5000.0 + assert not basic.relay._is_current(old) + assert len(basic.spawned) == 4 + + def test_a_late_announcement_is_logged(self, basic): + basic.mtc.set(1500.0) + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.announce() + assert log.warning.called and not log.error.called + assert basic.entry().state == "waiting" + + def test_a_very_late_announcement_is_an_error_but_still_kept(self, basic): + basic.mtc.set(1000.0 + 10_001.0) + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.announce() + assert log.error.called + assert basic.entry().state == "waiting" + + +class TestWaiter: + def test_dispatches_at_the_instant_with_the_stored_epochs_and_a_gate(self, basic): + basic.announce() + basic.mtc.set(1000.0) + basic.run("FollowWait") + args, kw = basic.ch.go_from.call_args + assert args[0].id == "T" and args[1] is basic.mtc and args[2] == 1000.0 + assert kw["arm_epoch"] == 3 and kw["require_stop_epoch"] == 2 + assert callable(kw["commit_gate"]) + + def test_the_gate_records_the_dispatch(self, basic): + def go_from(target, mtc, seed, **kw): + cue = basic.script.find("T") + cue._armed_at = 42.0 + assert kw["commit_gate"](cue, 9) is True + return "thread" + + basic.ch.go_from.side_effect = go_from + basic.announce() + basic.mtc.set(1000.0) + basic.run("FollowWait") + e = basic.entry() + assert e.state == "dispatched" and e.cue.id == "T" and e.gen == 9 + assert e.disarm_epoch == 3 and e.armed_at == 42.0 + + def test_waits_for_the_instant(self, basic): + basic.announce() + basic.mtc.set(900.0) + basic.run("FollowWait") # the fake sleep advances 20 ms per poll + assert basic.mtc.main_tc.milliseconds_exact >= 1000.0 + basic.ch.go_from.assert_called_once() + + def test_a_cancel_while_waiting_ends_the_waiter(self, basic): + basic.announce() + basic.cancel() + basic.mtc.set(1000.0) + basic.run("FollowWait") + basic.ch.go_from.assert_not_called() + + def test_a_stop_while_waiting_ends_the_waiter(self, basic): + basic.announce() + basic.relay.end_run("stop") + basic.mtc.set(1000.0) + basic.run("FollowWait") + basic.ch.go_from.assert_not_called() + + def test_nothing_local_to_start_marks_it_done(self, basic): + basic.announce() + basic.mtc.set(1000.0) + basic.run("FollowWait") + assert basic.entry().state == "done" + + def test_an_exception_in_the_dispatch_is_logged_and_done(self, basic): + basic.ch.go_from.side_effect = RuntimeError("re-arm failed") + basic.announce() + basic.mtc.set(1000.0) + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.run("FollowWait") + assert log.error.called and basic.entry().state == "done" + + def test_yields_to_a_manual_go_during_the_postwait(self): + t = _cue("T", local=True) + s = _cue("S", post_go="go_at_end", target=t, postwait=0.5) + rig = _Rig(_Script(s, t)) + rig.announce(seed=1000.0) + + def sleep(_): + ms = rig.mtc.main_tc.milliseconds_exact + 20 + rig.mtc.set(ms) + if ms >= 700: + t._go_generation = 5 # the operator started T at ~700 ms + + rig.relay._sleep = sleep + rig.mtc.set(400.0) + rig.run("FollowWait") + rig.ch.go_from.assert_not_called() + assert rig.entry().state == "done" + + def test_does_not_yield_without_a_manual_go(self): + t = _cue("T", local=True) + s = _cue("S", post_go="go_at_end", target=t, postwait=0.5) + rig = _Rig(_Script(s, t)) + rig.announce(seed=1000.0) + rig.mtc.set(400.0) + rig.run("FollowWait") + rig.ch.go_from.assert_called_once() + + +class TestGate: + def _gate(self, rig): + captured = {} + + def go_from(target, mtc, seed, **kw): + captured["gate"] = kw["commit_gate"] + + rig.ch.go_from.side_effect = go_from + rig.announce() + rig.mtc.set(1000.0) + rig.run("FollowWait") + rig.entry().state = "dispatching" # as during go() + return captured["gate"] + + def test_refuses_a_cancelled_entry(self, basic): + gate = self._gate(basic) + basic.cancel() + assert gate(basic.script.find("T"), 1) is False + + def test_refuses_a_replaced_entry(self, basic): + gate = self._gate(basic) + basic.announce(seed=9000.0) + assert gate(basic.script.find("T"), 1) is False + + def test_refuses_after_a_stop_emptied_the_table(self, basic): + gate = self._gate(basic) + basic.relay.end_run("stop") + assert gate(basic.script.find("T"), 1) is False + + def test_refuses_after_a_new_run(self, basic): + gate = self._gate(basic) + basic.relay.begin_run({"run_seq": 8}) + assert gate(basic.script.find("T"), 1) is False + + +class TestCancel: + def _dispatched(self, rig, gen=9): + def go_from(target, mtc, seed, **kw): + cue = rig.script.find("T") + kw["commit_gate"](cue, gen) + return "thread" + + rig.ch.go_from.side_effect = go_from + rig.announce() + rig.mtc.set(1000.0) + rig.run("FollowWait") + assert rig.entry().state == "dispatched" + + def test_without_an_entry_does_nothing(self, basic): + basic.cancel() + assert basic.spawned == [] + + def test_for_another_instant_does_nothing(self, basic): + basic.announce() + basic.cancel(seed=2000.0) + assert basic.entry().state == "waiting" + + def test_an_error_from_one_of_two_owners_only_removes_that_owner(self, basic): + basic.announce() + basic.announce(origin=OTHER) + basic.cancel(origin=OTHER, reason="error") + assert basic.entry().state == "waiting" and basic.entry().origins == {OWNER} + + def test_while_waiting_cancels_the_entry(self, basic): + basic.announce() + basic.cancel() + assert basic.entry().state == "cancelled" + + def test_a_dispatched_cue_goes_to_the_guarded_stop(self, basic): + self._dispatched(basic) + basic.cancel(reason="restart") + basic.run("FollowStop") + args, kw = basic.ch.stop_dispatched_follow.call_args + assert args[0].id == "T" and args[1:] == (9, 3, None) + assert kw["may_cut"] is True and kw["grace_s"] == 2.0 + assert basic.entry().state == "cancelled" + + def test_an_error_cancel_may_not_cut_a_started_cue(self, basic): + self._dispatched(basic) + basic.cancel(reason="error") + basic.run("FollowStop") + assert basic.ch.stop_dispatched_follow.call_args.kwargs["may_cut"] is False + + def test_a_second_cancel_is_a_no_op(self, basic): + self._dispatched(basic) + basic.cancel() + basic.cancel() + stops = [n for n, _, _ in basic.spawned if n.startswith("FollowStop")] + assert len(stops) == 1 + + def test_arrival_order_is_applied_order(self, basic): + """Announce then cancel, as the bus delivers them: never dispatched.""" + basic.announce() + basic.cancel() + basic.mtc.set(1000.0) + basic.run("FollowWait") + basic.ch.go_from.assert_not_called() + + +class TestRun: + def test_the_same_run_keeps_its_entries(self, basic): + basic.announce() + basic.relay.begin_run({"run_seq": 7, "go_mtc_ms": 5.0}) + assert basic.entry() is not None + + def test_a_new_run_empties_the_table(self, basic): + basic.announce() + basic.relay.begin_run({"run_seq": 8}) + assert basic.entry() is None and basic.relay.run_seq == 8 + + @pytest.mark.parametrize("value", [None, "complex_test", {"run_seq": True}, {}]) + def test_a_go_without_a_run_counter_means_no_relay(self, basic, value): + basic.relay.begin_run(value) + assert basic.relay.run_seq is None + + def test_end_run_empties_everything(self, basic): + basic.announce() + basic.relay.end_run("stop") + assert basic.relay.run_seq is None and basic.relay._table == {} + + +class TestPreArm: + def test_arms_the_first_local_cue_with_the_entry_epoch(self, basic): + basic.announce() + basic.ch.arm.side_effect = lambda c, init, walk: setattr(c, "loaded", True) + basic.run("FollowPreArm") + args, kw = basic.ch.arm.call_args + assert args[0].id == "T" and kw["walk"].epoch == 3 + basic.ch._arm_ahead.assert_called_once() + + def test_runs_until_the_instant_and_rearms_what_became_unloaded(self, basic): + basic.announce(seed=100.0) + t = basic.script.find("T") + t.loaded = True # still playing an earlier pass + basic.ch.arm.side_effect = lambda c, init, walk: setattr(c, "loaded", True) + rounds = [] + + def sleep(_): + rounds.append(1) + basic.mtc.set(basic.mtc.main_tc.milliseconds_exact + 50) + if len(rounds) == 1: + t.loaded = False # the earlier pass ended and disarmed + + basic.relay._prearm_sleep = sleep + basic.run("FollowPreArm") + assert [c.args[0].id for c in basic.ch.arm.call_args_list] == ["T"] + + def test_arms_an_enabled_play_target(self): + from uuid import uuid4 + + from cuemsutils.cues import ActionCue + + x = _cue("X", local=True) + p = ActionCue.__new__(ActionCue) + p.id = str(uuid4()) + p.enabled = True + p.post_go = "pause" + p.action_type = "play" + p._local = True + p.loaded = True + p._target_object = None + p._go_generation = 0 + p._action_target_object = x + s = _cue("S", post_go="go_at_end", target=p) + rig = _Rig(_Script(s, p, x)) + rig.announce(target=str(p.id)) + rig.run("FollowPreArm") + assert "X" in [c.args[0].id for c in rig.ch.arm.call_args_list] + + def test_skips_a_disabled_play_target(self): + from uuid import uuid4 + + from cuemsutils.cues import ActionCue + + x = _cue("X", local=True, enabled=False) + p = ActionCue.__new__(ActionCue) + p.id = str(uuid4()) + p.enabled = True + p.post_go = "pause" + p.action_type = "play" + p._local = True + p.loaded = True + p._target_object = None + p._go_generation = 0 + p._action_target_object = x + s = _cue("S", post_go="go_at_end", target=p) + rig = _Rig(_Script(s, p, x)) + rig.announce(target=str(p.id)) + rig.run("FollowPreArm") + assert "X" not in [c.args[0].id for c in rig.ch.arm.call_args_list] + + def test_stops_when_the_entry_is_cancelled(self, basic): + basic.announce(seed=10_000.0) + basic.cancel(seed=10_000.0) + basic.run("FollowPreArm") + basic.ch.arm.assert_not_called() + + +class TestOwnerSend: + def test_announce_sends_the_message(self, basic): + t = basic.script.find("T") + s = _cue("S2", local=True, post_go="go_at_end", target=t) + basic.relay.announce(s, 1234.5, 7) + assert basic.sent == [ + { + "op": "announce", + "origin": ME, + "run_seq": 7, + "source": "S2", + "target": "T", + "seed_ms": 1234.5, + } + ] + + def test_cancel_carries_its_reason(self, basic): + t = basic.script.find("T") + s = _cue("S2", local=True, post_go="go_at_end", target=t) + basic.relay.cancel(s, 1234.5, 7, "stop") + assert basic.sent[0]["op"] == "cancel" and basic.sent[0]["reason"] == "stop" + + def test_a_failed_send_is_an_error_not_an_exception(self, basic): + basic.relay._send = MagicMock(side_effect=RuntimeError("no loop")) + t = basic.script.find("T") + s = _cue("S2", local=True, post_go="go_at_end", target=t) + with patch("cuemsengine.cues.FollowRelay.Logger") as log: + basic.relay.announce(s, 1.0, 7) + assert log.error.called + + +class TestReviewFixes: + def test_a_cancelled_entry_does_not_swallow_a_new_announcement(self, basic): + """MINOR-5: cancel(restart, I) then announce(I) for the same instant: + the new pass must still play here.""" + basic.announce() + basic.cancel(reason="restart") + basic.announce() + assert basic.entry().state == "waiting" + assert len([n for n, _, _ in basic.spawned if n.startswith("FollowWait")]) == 2 + + def test_a_cancel_of_another_run_is_ignored(self, basic): + """MINOR-7.""" + basic.announce() + basic.relay.on_message( + { + "op": "cancel", + "origin": OWNER, + "run_seq": 6, + "source": "S", + "target": "T", + "seed_ms": 1000.0, + "reason": "stop", + } + ) + assert basic.entry().state == "waiting" + + def test_the_gate_records_a_cue_already_on_stage(self, basic): + """MINOR-3: the follow restarts a cue still playing an earlier pass.""" + from time import monotonic + + t = basic.script.find("T") + t._playing, t._revealed, t._revealed_at = True, True, monotonic() - 30 + + def go_from(target, mtc, seed, **kw): + kw["commit_gate"](t, 9) + + basic.ch.go_from.side_effect = go_from + basic.announce() + basic.mtc.set(1000.0) + basic.run("FollowWait") + assert basic.entry().started_before == t._revealed_at + basic.cancel(reason="error") + basic.run("FollowStop") + kw = basic.ch.stop_dispatched_follow.call_args.kwargs + assert kw["started_before"] == t._revealed_at + + def test_stop_pending_defers_the_epochs_and_the_dispatch(self, basic): + """MINOR-4: a STOP was received but has not run yet on this node: the + epochs are read only once it has, and nothing is dispatched before.""" + basic.relay.end_run("stop") # received + basic.relay.begin_run({"run_seq": 7}) # the next GO, received + assert basic.relay.stop_pending + basic.announce() + e = basic.entry() + assert e.arm_epoch is None + basic.mtc.set(1000.0) + polls = [] + + def sleep(_): + polls.append(1) + if len(polls) == 3: + basic.ch._disarm_epoch = 4 # the STOP runs ... + basic.ch._last_stop_chain_epoch = 5 + basic.relay.stop_applied() # ... and is applied + + basic.relay._sleep = sleep + basic.run("FollowWait") + kw = basic.ch.go_from.call_args.kwargs + assert kw["arm_epoch"] == 4 and kw["require_stop_epoch"] == 5 + + def test_stop_pending_blocks_the_prearm(self, basic): + basic.relay.end_run("stop") + basic.relay.begin_run({"run_seq": 7}) + basic.announce(seed=60.0) + basic.run("FollowPreArm") + basic.ch.arm.assert_not_called() + + +class TestRealGoAndGate: + """MINOR-8: the real go() evaluating the real gate, both locks nested.""" + + def _ch(self): + from threading import Lock + + from cuemsengine.cues.CueHandler import CueHandler + + ch = object.__new__(CueHandler) + ch._lock = Lock() + ch._armed_cues = [] + ch._armed_cues_set = set() + ch._arm_ahead = MagicMock() + ch.arm = MagicMock() + return ch + + def _rig(self, ch): + t = _cue("T", local=True, loaded=True) + s = _cue("S", post_go="go_at_end", target=t) + rig = _Rig(_Script(s, t)) + rig.relay._ch = ch + return rig, t + + def test_dispatches_through_the_gate(self): + from cuemsengine.cues.CueHandler import CueHandler + + ch = self._ch() + rig, t = self._rig(ch) + rig.announce() + rig.mtc.set(1000.0) + with patch.object(CueHandler, "go_threaded"): + rig.run("FollowWait") + e = rig.entry() + assert e.state == "dispatched" and e.cue is t and e.gen == t._go_generation + + def test_a_cancel_while_inside_go_refuses_the_commit(self): + """The cancel lands while go() is in its fallback arm: the gate, run + later inside go()'s commit section, refuses.""" + from cuemsengine.cues.CueHandler import CueHandler + + ch = self._ch() + rig, t = self._rig(ch) + t.loaded = False + + def arm_and_cancel(c, init=False, epoch=None, **kw): + rig.cancel() # arrives on the bus during the arm + c.loaded = True + return True + + ch.arm.side_effect = arm_and_cancel + rig.announce() + rig.mtc.set(1000.0) + with patch.object(CueHandler, "go_threaded") as gt: + rig.run("FollowWait") + gt.assert_not_called() + assert rig.entry().state == "cancelled" and t._go_generation == 0