fix(follow): an Auto follow into a cue on another node starts it (869fc8ytz)

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.

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 = {}

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")

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).