From 6dad8cc41761731bfe9021f6bd37c14b627f1814 Mon Sep 17 00:00:00 2001 From: Ion Reguera Date: Tue, 6 Oct 2026 11:06:31 +0200 Subject: [PATCH 1/5] test(follow): the controller's run counter and follow relay (869fc8ytz) (red) Every GO carries run_seq; STOP, load and unload bump it before they forward anything; a node's follow message is relayed to every node only while the show runs and only with the current run_seq. The four tests that pinned the GO value now expect it to be a dict in every case. --- tests/test_controller_commands.py | 28 ++- tests/test_cross_node_follow_controller.py | 228 +++++++++++++++++++++ 2 files changed, 248 insertions(+), 8 deletions(-) create mode 100644 tests/test_cross_node_follow_controller.py diff --git a/tests/test_controller_commands.py b/tests/test_controller_commands.py index 89344055..070484ef 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 00000000..969ab8a0 --- /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() From 16a6d6a347200ab9f29b526a854e093169ddcb9b Mon Sep 17 00:00:00 2001 From: Ion Reguera Date: Tue, 6 Oct 2026 11:06:31 +0200 Subject: [PATCH 2/5] feat(follow): the controller relays cross-node Auto follow messages (869fc8ytz) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The controller keeps a run counter, started from the clock. STOP, load and unload bump it before forwarding; every GO ships it to the nodes. A node's 'follow' status message (announce or cancel) is forwarded to every node only while running and only with the current value; the check and the forward share one lock with the bump, so a relayed follow is always queued ahead of the STOP that ends its run. The controller keeps no follow state. The GO value is now always a dict. Nodes read nothing from it but the anchor (_resolve_go_anchor), so the caller's own value is no longer passed on. Plan: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md §3.3-3.4. --- src/cuemsengine/ControllerEngine.py | 68 ++++++++++++++++++++++++++++- 1 file changed, 66 insertions(+), 2 deletions(-) diff --git a/src/cuemsengine/ControllerEngine.py b/src/cuemsengine/ControllerEngine.py index 5e8dd149..a20fad28 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 = {} From fd5261574e52151029b7197a2d40eb42f02b8aaa Mon Sep 17 00:00:00 2001 From: Ion Reguera Date: Tue, 6 Oct 2026 11:06:32 +0200 Subject: [PATCH 3/5] test(follow): node side of the cross-node Auto follow (869fc8ytz) (red) The receive hook (bus order, never the command lock), the receiving node's follow table (announce, pre-arm, wait for the instant, dispatch through a commit gate in go(), cancel by entry state, the 2 s rule), and the owning node's announce/cancel (at the reveal or the last loop; one cancel per announcement; the restart cancel from go()). Includes the tests for the seven findings of the code review of the implementation. --- tests/test_cross_node_follow_hook.py | 164 ++++++ tests/test_cross_node_follow_owner.py | 638 ++++++++++++++++++++++ tests/test_cross_node_follow_receiver.py | 668 +++++++++++++++++++++++ 3 files changed, 1470 insertions(+) create mode 100644 tests/test_cross_node_follow_hook.py create mode 100644 tests/test_cross_node_follow_owner.py create mode 100644 tests/test_cross_node_follow_receiver.py diff --git a/tests/test_cross_node_follow_hook.py b/tests/test_cross_node_follow_hook.py new file mode 100644 index 00000000..3ac1d740 --- /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 00000000..1772d0df --- /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 00000000..9f9f747e --- /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 From 8236720b1d8466970193fac26ecd08d94c7d73d4 Mon Sep 17 00:00:00 2001 From: Ion Reguera Date: Tue, 6 Oct 2026 11:06:32 +0200 Subject: [PATCH 4/5] fix(follow): an Auto follow into a cue on another node starts it (869fc8ytz) An Auto follow (post_go='go_at_end') fired its target only on the nodes where the source is local, so a target on another node never started, silently, on every engine line. The node that owns an AudioCue or VideoCue source now announces the follow instant when the cue reveals (or at the start of its last loop; never for an endless loop), through the controller. Every other node pre-arms what it would start, waits for the instant on MTC and dispatches it there through go_from. ActionCue, FadeCue, CueList and DmxCue sources are local on every node and keep firing their own follow. If the follow does not happen the owner cancels: 'stop' (stop, pause or fade-out action), 'restart' (sent from go()'s commit section when the source is started again), 'error'. A STOP, a load or the yield send nothing; every node handles those itself. A cancel that reaches a target already on stage stops it only within 2 s and only for stop/restart. - cues/FollowRelay.py: both halves of the node side; one entry per source, its own lock, never calls into CueHandler while holding it. - NodeCommunications: a receive hook in bus order, before any command thread; send_operation_nowait. - NodeEngine: run_seq recorded and pending follows dropped at receipt; follow messages applied there, never under the command lock; STOP and load marked applied after they run. - CueHandler: go() gets a commit_gate (refusals, gate, restart cancel, assignments); go_from passes arm_epoch/require_stop_epoch/commit_gate through; stop_dispatched_follow and disarm_if_unchanged (guarded disarm under the cue's arm claim); the reveal stamp; go_threaded wraps the pass so every exit decides its cancel once. - loop_cue: on_last_loop callback for audio and video. Verified on test2 (suite 1109 passed; 40 live runs, 0 LATE, 0 fallback; old-node smoke test; PR #22 drills and a Medina sala1 replay unchanged). Plan: cuems-RELATIONS Plans/2026-10-05-engine-cross-node-auto-follow.md. --- src/cuemsengine/NodeEngine.py | 69 ++- src/cuemsengine/comms/NodeCommunications.py | 35 ++ src/cuemsengine/cues/CueHandler.py | 403 +++++++++++++- src/cuemsengine/cues/FollowRelay.py | 563 ++++++++++++++++++++ src/cuemsengine/cues/loop_cue.py | 20 +- 5 files changed, 1075 insertions(+), 15 deletions(-) create mode 100644 src/cuemsengine/cues/FollowRelay.py diff --git a/src/cuemsengine/NodeEngine.py b/src/cuemsengine/NodeEngine.py index 61d2997d..0bd4a32f 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 PLAYER_HANDLER, _positive_int @@ -97,6 +98,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. @@ -171,6 +176,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: " @@ -182,6 +202,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. @@ -432,7 +492,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 cbc421b6..d3505dd4 100644 --- a/src/cuemsengine/comms/NodeCommunications.py +++ b/src/cuemsengine/comms/NodeCommunications.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: 2026 Stagelab Coop SCCL # SPDX-License-Identifier: GPL-3.0-or-later # SPDX-FileContributor: Adrià Masip +# SPDX-FileContributor: Ion Reguera import asyncio from typing import Any, Callable, Optional @@ -37,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( { @@ -54,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") @@ -107,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 @@ -145,6 +168,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 e6d4db5b..d37ffabd 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 @@ -1165,6 +1187,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. @@ -1187,6 +1210,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 @@ -1298,6 +1326,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 @@ -1306,12 +1351,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 — @@ -1350,7 +1397,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 @@ -1364,7 +1417,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 @@ -1395,8 +1461,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'. @@ -1506,6 +1578,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 @@ -1656,6 +1759,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 @@ -1671,8 +1775,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: @@ -1714,11 +1823,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( @@ -1735,11 +1859,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") @@ -1805,12 +1925,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" @@ -1825,6 +1947,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 00000000..e6159b14 --- /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 bb96021d..d97b62fe 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}" @@ -300,6 +315,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: From 2c358f40e6594a278c194943f7ef0e561e4437bc Mon Sep 17 00:00:00 2001 From: Ion Reguera Date: Tue, 6 Oct 2026 11:06:32 +0200 Subject: [PATCH 5/5] docs(follow): the cross-node Auto follow relay in CLAUDE.md (869fc8ytz) --- CLAUDE.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index 4dc7fc92..847b9985 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.