Merge pull request #19 from obscuremind/claude/clever-shannon-9grkqt

feat(cluster): Phase 4 — signed command channel (COMMANDS flow)

docs/adr/0004-cluster-api.md

@@ -1,6 +1,6 @@
 # ADR 0004 — Cluster API between MAIN and load balancers: the panel's contract
 
-- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3 (authoritative telemetry, the 1 s liveness loop, MAIN endpoint changes) is too. Phases 4–11 are not.
+- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3 (authoritative telemetry, the 1 s liveness loop, MAIN endpoint changes) is too. Phase 4 has its command channel (RPCs and viewer kills); root commands are not in yet. Phases 5–11 are not.
 - **Date:** 2026-09-25
 - **Plan:** `docs/superpowers/specs/2026-09-21-main-lb-api-communication-design.md` (MAIN ↔ LB API communication, revision 3 plus corrections).
 - **Extension side:** `xcvm_core` ADR-002, "Cluster API: the extension's half of MAIN ↔ LB communication", cluster API version 1.
@@ -88,6 +88,8 @@ The plan numbers its Phase 1 migrations 026–032. 026 and 027 were already take
 | `token_rekey` | POST | node signature (epoch 0, no MAC), body SEALed to the panel box key, a challenge | `active`, once a minute | panel-signed (`pre`): a new epoch's sealed token |
 | `hello` | POST | session | `active`, `quarantined` | BOX: state, mode, flows, proto, policy |
 | `heartbeat` | POST | session | `active`, `quarantined` | BOX: state, mode, flows, `pending` |
+| `commands` | POST | session | `active` | BOX: signed commands after `after_seq`, held up to `wait_ms` (≤ 20 s) |
+| `ack` | POST | session | `active`, `quarantined` | BOX: `ok` (own commands only) |
 
 A session request is checked in this order. Nothing is written, not even the nonce, before the MAC and, for token operations, the node signature have verified:
 
@@ -269,6 +271,32 @@ This was checked with nginx 1.24:
 - `nginx -t` passes with and without the file.
 - On the old port, only `/cluster/v1/` reaches PHP.
 
+### Commands (Phase 4, first increment)
+
+`CommandBus` queues MAIN → node commands in `cluster_commands`, FIFO per node by `seq`. Each command is a typed JSON record signed by the panel with tag `cmd`:
+
+```text
+{"v":1, "type", "exp", "iat", "cmd_id", "seq", "node_uuid", "gen", "dedupe_key", "args"}
+```
+
+The extension derives the class from `type`. Kills and stops are restrictive and sign without a licence; the rest need it. A `dedupe_key` replaces a not-yet-acked command for the same desired state. Commands expire (`conn.*` 5 min, default 10 min), and `cron:cluster` prunes them.
+
+The flow, for a node whose COMMANDS flow is on (toggled per node on the Cluster Nodes page):
+
+1. **Poll.** The agent holds a `commands` long-poll. Each poll holds a PHP worker on MAIN; the plan's bus replaces that later.
+2. **Checks.** The agent checks each command: the panel signature under its pinned key, its own uuid and generation, `seq` above its persisted high-water `cmd_seq`, and `exp` on MAIN's clock.
+3. **Run.** It runs the command through `console.php cluster:exec`. `cluster:exec` checks the signature again and runs `node.rpc{action}` with the legacy `/api` handlers (`InternalApiController::runCommand`, actions limited to `NodeRpc::ACTIONS`) and `conn.kill_worker`.
+4. **Ack.** The agent `ack`s the command with the result (≤ 64 KB) and raises its high-water. MAIN raises `cluster_nodes.cmd_seq`.
+5. **Refusals.** A refused command is acked as refused, with the reason.
+
+On MAIN, `Domain\Cluster\ClusterRoute` sits behind the Phase 0 seams:
+
+- `NodeRpc::request()` becomes a signed `node.rpc`, and MAIN waits for its ack up to the caller's timeout.
+- `NodeRpc::broadcast()` and `SignalDispatcher::kill()` queue without waiting.
+- Nodes without the flow, and every LB, keep the legacy transport.
+
+Root actions (`NodeActions`) stay on the signals table until `cluster:root` and the root-owned panel-key pin exist.
+
 ### Extension updates
 
 `console.php xcvm_core` rolls back an update whose cluster API falls outside the panel's range when the installed one was inside it. `console.php xcvm_core status` reports what is loaded. Installing an exact pinned version needs versioned paths in the binaries repo; that prerequisite is still open.

src/Cli/Commands/ClusterExecCommand.php

@@ -0,0 +1,105 @@
+<?php
+
+namespace XcVm\Cli\Commands;
+
+use XcVm\Cli\CommandInterface;
+use XcVm\Core\Cluster\Crypto\Enc;
+use XcVm\Core\Cluster\Crypto\PanelSig;
+use XcVm\Core\Cluster\NodeRpc;
+use XcVm\Domain\Server\ServerRepository;
+use XcVm\Public\Controllers\Api\InternalApiController;
+
+/**
+ * ClusterExecCommand — run one MAIN command on this node (Phase 4). The agent
+ * pipes a command it has verified, `{"doc": "<json>", "sig": "<b64url>"}`, on
+ * stdin; this checks it again against the panel key the agent pinned
+ * (config/cluster/agent.json), runs it and writes its result to stdout for
+ * the agent's ack. Exit 0 means done.
+ *
+ * - `node.rpc {action, …}` — one of NodeRpc::ACTIONS, through the handlers of
+ *   the legacy /api (InternalApiController::runCommand);
+ * - `conn.kill_worker {pid, rtmp}` — a viewer's PHP worker (this user's
+ *   processes only) or an RTMP client.
+ *
+ * Runs as xc_vm; root actions are not taken here.
+ *
+ * Usage: `console.php cluster:exec < command.json`
+ *
+ * @package XC_VM_CLI_Commands
+ */
+class ClusterExecCommand implements CommandInterface {
+	/** Commands older than their exp by more than this (LB clock) are refused. */
+	public const SKEW = 300;
+
+	public function getName(): string {
+		return 'cluster:exec';
+	}
+
+	public function getDescription(): string {
+		return 'Run one signed MAIN command (from xc_agent) on this node';
+	}
+
+	public function execute(array $rArgs): int {
+		$rIn = json_decode((string) stream_get_contents(STDIN), true);
+		$rState = json_decode((string) @file_get_contents(CONFIG_PATH . 'cluster/agent.json'), true);
+		$rCmd = self::verify(is_array($rIn) ? $rIn : [], is_array($rState) ? $rState : [], time());
+		if (is_string($rCmd)) {
+			fwrite(STDERR, 'cluster:exec: ' . $rCmd . "\n");
+			return 2;
+		}
+		return self::run($rCmd);
+	}
+
+	/**
+	 * Check a command against the agent's pinned panel key and identity.
+	 *
+	 * @param array<string, mixed> $rIn {doc, sig}
+	 * @param array<string, mixed> $rState the agent's state (base64 fields)
+	 * @return array<string, mixed>|string the command, or why it is refused
+	 */
+	public static function verify(array $rIn, array $rState, int $rNow): array|string {
+		$rPub = base64_decode((string) ($rState['panel_sign_pub'] ?? ''), true);
+		$rDoc = is_string($rIn['doc'] ?? null) ? $rIn['doc'] : '';
+		$rSig = Enc::b64urlDecode((string) ($rIn['sig'] ?? ''));
+		if ($rPub === false || !PanelSig::verify($rPub, 'cmd', $rDoc, (string) $rSig)) {
+			return 'bad signature';
+		}
+		$rCmd = json_decode($rDoc, true);
+		if (!is_array($rCmd) || ($rCmd['node_uuid'] ?? null) !== ($rState['node_uuid'] ?? '') || (int) ($rCmd['exp'] ?? 0) + self::SKEW < $rNow) {
+			return 'not for this node, or expired';
+		}
+		return $rCmd;
+	}
+
+	/** @param array<string, mixed> $rCmd */
+	public static function run(array $rCmd): int {
+		$rArgs = is_array($rCmd['args'] ?? null) ? $rCmd['args'] : [];
+		switch ($rCmd['type'] ?? '') {
+			case 'node.rpc':
+				if (!in_array($rArgs['action'] ?? null, NodeRpc::ACTIONS, true)) {
+					fwrite(STDERR, "cluster:exec: unknown action\n");
+					return 2;
+				}
+				(new InternalApiController())->runCommand($rArgs);
+				return 0;
+
+			case 'conn.kill_worker':
+				$rPID = (int) ($rArgs['pid'] ?? 0);
+				if ($rPID <= 0) {
+					return 2;
+				}
+				if (!empty($rArgs['rtmp'])) {
+					$rUrl = (string) (ServerRepository::getAll()[SERVER_ID]['rtmp_mport_url'] ?? '');
+					if ($rUrl !== '') {
+						@file_get_contents($rUrl . 'control/drop/client?clientid=' . $rPID, false, stream_context_create(['http' => ['timeout' => 2]]));
+					}
+				} elseif (file_exists('/proc/' . $rPID)) {
+					posix_kill($rPID, 9);
+				}
+				echo json_encode(['result' => true]);
+				return 0;
+		}
+		fwrite(STDERR, "cluster:exec: unknown command type\n");
+		return 2;
+	}
+}

src/Cli/CronJobs/ClusterCronJob.php

@@ -9,6 +9,7 @@ use XcVm\Core\Cluster\NodeRole;
 use XcVm\Core\Config\SettingsManager;
 use XcVm\Domain\Cluster\ClusterAudit;
 use XcVm\Domain\Cluster\ClusterEndpoint;
+use XcVm\Domain\Cluster\CommandBus;
 use XcVm\Domain\Cluster\EnrolCodeService;
 use XcVm\Domain\Cluster\LivenessService;
 use XcVm\Domain\Cluster\NonceStore;
@@ -60,6 +61,7 @@ class ClusterCronJob implements CommandInterface {
 			'epochs' => static fn() => TokenService::prune(),
 			'nonces' => static fn() => NonceStore::purge(),
 			'enrol_codes' => static fn() => EnrolCodeService::prune(),
+			'commands' => static fn() => CommandBus::prune(),
 			// MAIN's old HTTP ports past their 7 days: release them in nginx.
 			'endpoint' => static function () {
 				if (ClusterEndpoint::prune(SettingsManager::getAll()) && defined('SERVER_ID')) {

src/Core/Cluster/NodeRpc.php

@@ -3,6 +3,7 @@
 namespace XcVm\Core\Cluster;
 
 use XcVm\Core\Http\ApiClient;
+use XcVm\Domain\Cluster\ClusterRoute;
 
 /**
  * Node RPC
@@ -42,6 +43,13 @@ final class NodeRpc {
 		if (self::$rTransport !== null) {
 			return (self::$rTransport)('request', [$rServerID], $rData, $rTimeout);
 		}
+		// A node with the COMMANDS flow gets a signed node.rpc command (MAIN only).
+		if (class_exists(ClusterRoute::class)) {
+			[$rRouted, $rResult] = ClusterRoute::rpc($rServerID, $rData, $rTimeout);
+			if ($rRouted) {
+				return $rResult;
+			}
+		}
 		return ApiClient::systemRequest($rServerID, $rData, $rTimeout);
 	}
 
@@ -58,6 +66,18 @@ final class NodeRpc {
 		if (self::$rTransport !== null) {
 			return (self::$rTransport)('broadcast', array_map('intval', $rServerIDs), $rData, 0);
 		}
+		if (class_exists(ClusterRoute::class)) {
+			$rLegacy = [];
+			foreach ($rServerIDs as $rServerID) {
+				if (!ClusterRoute::send((int) $rServerID, $rData)[0]) {
+					$rLegacy[] = $rServerID;
+				}
+			}
+			if ($rLegacy === []) {
+				return ['result' => true];
+			}
+			$rServerIDs = $rLegacy;
+		}
 		return ApiClient::asyncRequest($rServerIDs, $rData);
 	}
 

src/Core/Cluster/SignalDispatcher.php

@@ -2,6 +2,8 @@
 
 namespace XcVm\Core\Cluster;
 
+use XcVm\Domain\Cluster\ClusterRoute;
+
 /**
  * Signal Dispatcher
  *
@@ -36,6 +38,13 @@ final class SignalDispatcher {
 		if ($rPID <= 0) {
 			return false;
 		}
+		// A node with the COMMANDS flow gets a signed conn.kill_worker (MAIN only).
+		if (class_exists(ClusterRoute::class)) {
+			[$rRouted, $rQueued] = ClusterRoute::kill($rServerID, $rPID, $rRTMP);
+			if ($rRouted) {
+				return $rQueued;
+			}
+		}
 		$rRow = ['pid' => $rPID, 'server_id' => $rServerID];
 		if ($rRTMP) {
 			$rRow['rtmp'] = 1;

src/Core/Localization/lang/ar.ini

@@ -1882,6 +1882,10 @@ cluster_telemetry_help = "تدفق القياس عن بُعد: عند تفعيل
 cluster_telemetry_on_done = "تم تفعيل تدفق القياس: وكيل العقدة يرسل الإحصاءات الآن."
 cluster_telemetry_off_done = "تم تعطيل تدفق القياس: يعود watchdog العقدة إلى الإرسال."
 cluster_fleet_silence = "صمتت معظم العقد في وقت واحد: يشك MAIN في شبكته أو في مجمع الواجهة، ويحتفظ بحالة كل عقدة دون وضعها غير متصلة حتى ينتهي الصمت."
+cluster_commands_flow = "الأوامر"
+cluster_commands_help = "تدفق الأوامر: عند تفعيله يرسل MAIN إلى هذه العقدة أوامر موقّعة (استدعاءات مثل تشغيل/إيقاف البث والفحص والمسح وقطع المشاهدين) عبر واجهة العنقود بدلًا من /api وجدول الإشارات."
+cluster_commands_on_done = "تم تفعيل تدفق الأوامر: يرسل MAIN الآن أوامر موقّعة إلى هذه العقدة."
+cluster_commands_off_done = "تم تعطيل تدفق الأوامر: يعود MAIN إلى استخدام /api وإشارات العقدة."
 restreamer_bypass_proxy = "وكيل تجاوز Restreamer"
 channel_sorting_type = "نوع فرز القناة"
 sort_vod_by_date = "فرز VOD حسب التاريخ"

src/Core/Localization/lang/bg.ini

@@ -1604,6 +1604,10 @@ cluster_telemetry_help = "Поток на телеметрията: при вк
 cluster_telemetry_on_done = "Потокът на телеметрията е включен: агентът на възела вече отчита статистиките."
 cluster_telemetry_off_done = "Потокът на телеметрията е изключен: watchdog на възела отново отчита."
 cluster_fleet_silence = "Повечето възли замлъкнаха едновременно: MAIN подозира собствената си мрежа или API пула и запазва състоянието на всеки възел, без да го маркира офлайн, докато тишината отмине."
+cluster_commands_flow = "Команди"
+cluster_commands_help = "Поток на командите: при включване MAIN изпраща на възела подписани команди (RPC като старт/стоп на потоци, проби и сканиране, и прекъсване на зрители) през API на клъстера вместо през неговия /api и таблицата със сигнали."
+cluster_commands_on_done = "Потокът на командите е включен: MAIN вече изпраща подписани команди до възела."
+cluster_commands_off_done = "Потокът на командите е изключен: MAIN отново използва /api и сигналите на възела."
 fanout_debug = "Дебъг дневник Fanout"
 fanout_debug_off = "Изкл."
 fanout_debug_all = "Всички категории"

src/Core/Localization/lang/de.ini

@@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Telemetrie-Flow: an, MAIN bezieht die Statistiken dies
 cluster_telemetry_on_done = "Telemetrie-Flow an: Der Agent des Knotens meldet jetzt seine Statistiken."
 cluster_telemetry_off_done = "Telemetrie-Flow aus: Der Watchdog des Knotens meldet wieder."
 cluster_fleet_silence = "Die meisten Knoten sind gleichzeitig verstummt: MAIN vermutet sein eigenes Netz oder den API-Pool und hält den Zustand jedes Knotens, statt ihn offline zu setzen, bis die Stille endet."
+cluster_commands_flow = "Befehle"
+cluster_commands_help = "Befehls-Flow: an, sendet MAIN diesem Knoten signierte Befehle (RPCs wie Stream-Start/Stopp, Probes und Scans sowie Zuschauer-Kills) über die Cluster-API statt über seine /api und die Signaltabelle."
+cluster_commands_on_done = "Befehls-Flow an: MAIN sendet diesem Knoten jetzt signierte Befehle."
+cluster_commands_off_done = "Befehls-Flow aus: MAIN nutzt wieder die /api und die Signale des Knotens."
 fanout_debug = "Fanout-Debug-Protokoll"
 fanout_debug_off = "Aus"
 fanout_debug_all = "Alle Kategorien"

src/Core/Localization/lang/en.ini

@@ -1893,6 +1893,10 @@ cluster_telemetry_help = "Telemetry flow: when on, MAIN takes this node's stats
 cluster_telemetry_on_done = "Telemetry flow on: the node's agent now reports its stats."
 cluster_telemetry_off_done = "Telemetry flow off: the node's watchdog reports again."
 cluster_fleet_silence = "Most nodes went silent at once: MAIN suspects its own network or API pool, and holds every node's state instead of marking it offline until the silence clears."
+cluster_commands_flow = "Commands"
+cluster_commands_help = "Commands flow: when on, MAIN sends this node signed commands (RPCs such as stream start/stop, probes and scans, and viewer kills) over the cluster API instead of its /api and the signals table."
+cluster_commands_on_done = "Commands flow on: MAIN now sends this node signed commands."
+cluster_commands_off_done = "Commands flow off: MAIN uses the node's legacy /api and signals again."
 fanout_debug = "Fanout Debug Log"
 fanout_debug_off = "Off"
 fanout_debug_all = "All categories"

src/Core/Localization/lang/es.ini

@@ -1610,6 +1610,10 @@ cluster_telemetry_help = "Flujo de telemetría: activado, MAIN toma las estadís
 cluster_telemetry_on_done = "Flujo de telemetría activado: el agente del nodo informa ahora sus estadísticas."
 cluster_telemetry_off_done = "Flujo de telemetría desactivado: el watchdog del nodo vuelve a informar."
 cluster_fleet_silence = "La mayoría de los nodos quedaron en silencio a la vez: MAIN sospecha de su propia red o del pool de la API y mantiene el estado de cada nodo, sin marcarlos offline, hasta que pase el silencio."
+cluster_commands_flow = "Comandos"
+cluster_commands_help = "Flujo de comandos: activado, MAIN envía a este nodo comandos firmados (RPC como iniciar/detener streams, sondeos y escaneos, y expulsión de espectadores) por la API del clúster, en lugar de su /api y la tabla de señales."
+cluster_commands_on_done = "Flujo de comandos activado: MAIN envía ahora comandos firmados a este nodo."
+cluster_commands_off_done = "Flujo de comandos desactivado: MAIN vuelve a usar la /api y las señales del nodo."
 fanout_debug = "Registro de depuración Fanout"
 fanout_debug_off = "Desactivado"
 fanout_debug_all = "Todas las categorías"

src/Core/Localization/lang/fr.ini

@@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Flux de télémétrie : activé, MAIN reçoit les stat
 cluster_telemetry_on_done = "Flux de télémétrie activé : l'agent du nœud rapporte désormais ses statistiques."
 cluster_telemetry_off_done = "Flux de télémétrie désactivé : le watchdog du nœud rapporte de nouveau."
 cluster_fleet_silence = "La plupart des nœuds se sont tus en même temps : MAIN soupçonne son propre réseau ou le pool de l'API et conserve l'état de chaque nœud, sans les marquer hors ligne, jusqu'à la fin du silence."
+cluster_commands_flow = "Commandes"
+cluster_commands_help = "Flux de commandes : activé, MAIN envoie à ce nœud des commandes signées (RPC comme démarrer/arrêter un flux, sondes et scans, et déconnexion de spectateurs) par l'API du cluster, au lieu de sa /api et de la table des signaux."
+cluster_commands_on_done = "Flux de commandes activé : MAIN envoie désormais des commandes signées à ce nœud."
+cluster_commands_off_done = "Flux de commandes désactivé : MAIN utilise de nouveau la /api et les signaux du nœud."
 fanout_debug = "Journal de débogage Fanout"
 fanout_debug_off = "Désactivé"
 fanout_debug_all = "Toutes les catégories"

src/Core/Localization/lang/pt.ini

@@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Fluxo de telemetria: ligado, o MAIN recebe as estatís
 cluster_telemetry_on_done = "Fluxo de telemetria ligado: o agente do nó passa a reportar as estatísticas."
 cluster_telemetry_off_done = "Fluxo de telemetria desligado: o watchdog do nó volta a reportar."
 cluster_fleet_silence = "A maioria dos nós ficou em silêncio ao mesmo tempo: o MAIN suspeita da sua própria rede ou do pool da API e mantém o estado de cada nó, sem os marcar como offline, até o silêncio passar."
+cluster_commands_flow = "Comandos"
+cluster_commands_help = "Fluxo de comandos: ligado, o MAIN envia a este nó comandos assinados (RPCs como iniciar/parar streams, probes e scans, e kills de espetadores) pela API do cluster, em vez da /api e da tabela de sinais."
+cluster_commands_on_done = "Fluxo de comandos ligado: o MAIN passa a enviar comandos assinados a este nó."
+cluster_commands_off_done = "Fluxo de comandos desligado: o MAIN volta a usar a /api e os sinais do nó."
 fanout_debug = "Registo de depuração Fanout"
 fanout_debug_off = "Desligado"
 fanout_debug_all = "Todas as categorias"

src/Core/Localization/lang/ru.ini

@@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Поток телеметрии: при включен
 cluster_telemetry_on_done = "Поток телеметрии включён: статистику узла теперь передаёт агент."
 cluster_telemetry_off_done = "Поток телеметрии выключен: статистику снова передаёт watchdog узла."
 cluster_fleet_silence = "Большинство узлов замолчали одновременно: MAIN подозревает свою сеть или пул API и сохраняет состояние каждого узла, не помечая их offline, пока тишина не закончится."
+cluster_commands_flow = "Команды"
+cluster_commands_help = "Поток команд: при включении MAIN отправляет узлу подписанные команды (RPC: запуск/остановка потоков, пробы и сканирование, отключение зрителей) через API кластера вместо его /api и таблицы сигналов."
+cluster_commands_on_done = "Поток команд включён: MAIN теперь отправляет узлу подписанные команды."
+cluster_commands_off_done = "Поток команд выключен: MAIN снова использует /api и сигналы узла."
 fanout_debug = "Отладочный лог Fanout"
 fanout_debug_off = "Выкл."
 fanout_debug_all = "Все категории"

src/Domain/Cluster/ClusterAdmin.php

@@ -104,14 +104,18 @@ final class ClusterAdmin {
 
 				case 'telemetry_on':
 				case 'telemetry_off':
+				case 'commands_on':
+				case 'commands_off':
 					$rNode = NodeRegistry::byServer($rServerID);
 					if ($rNode === null || !in_array($rNode['state'], ['active', 'quarantined'], true)) {
 						return ['type' => 'info', 'message' => 'cluster_not_enrolled'];
 					}
-					$rFlows = $rAction === 'telemetry_on' ? ((int) $rNode['flows'] | NodeRegistry::FLOW_TELEMETRY) : ((int) $rNode['flows'] & ~NodeRegistry::FLOW_TELEMETRY);
+					[$rName, $rSwitch] = explode('_', $rAction);
+					$rBit = $rName === 'telemetry' ? NodeRegistry::FLOW_TELEMETRY : NodeRegistry::FLOW_COMMANDS;
+					$rFlows = $rSwitch === 'on' ? ((int) $rNode['flows'] | $rBit) : ((int) $rNode['flows'] & ~$rBit);
 					NodeRegistry::update($rServerID, ['flows' => $rFlows]);
 					ClusterAudit::log('node.flows', $rServerID, ['flows' => $rFlows, 'was' => (int) $rNode['flows']], $rUserID === null ? 'admin' : 'admin:' . $rUserID);
-					return ['type' => 'success', 'message' => $rAction === 'telemetry_on' ? 'cluster_telemetry_on_done' : 'cluster_telemetry_off_done'];
+					return ['type' => 'success', 'message' => 'cluster_' . $rName . '_' . $rSwitch . '_done'];
 
 				case 'revoke':
 					return NodeRegistry::revoke($rServerID, $rCrypto, $rUserID === null ? 'admin' : 'admin:' . $rUserID)

src/Domain/Cluster/ClusterApi.php

@@ -11,7 +11,8 @@ use XcVm\Core\Cluster\Crypto\SessionKeys;
 
 /**
  * MAIN's `/cluster/v1/<op>` API (Phase 2: health, challenge, enrol_complete,
- * enrol_code, enrol_code_status, token_refresh, token_rekey, hello, heartbeat). Transport-free: handle() takes the request
+ * enrol_code, enrol_code_status, token_refresh, token_rekey, hello, heartbeat;
+ * Phase 4: commands, ack). Transport-free: handle() takes the request
  * as an array and returns status, headers and body, so it is tested without
  * a web server; Public/cluster/index.php is the HTTP shell around it.
  *
@@ -40,6 +41,8 @@ final class ClusterApi {
 		'enrol_code' => ['POST', true, null],
 		'enrol_code_status' => ['POST', false, null],
 		'hello' => ['POST', false, ['active', 'quarantined']],
+		'commands' => ['POST', false, ['active']],
+		'ack' => ['POST', false, ['active', 'quarantined']],
 		'heartbeat' => ['POST', false, ['active', 'quarantined']],
 	];
 
@@ -139,6 +142,8 @@ final class ClusterApi {
 			'token_refresh' => self::tokenRefresh($rCrypto, $rNode, $rKeys, $rCtx, $rH, $rPayload),
 			'hello' => self::hello($rNode, $rKeys, $rCtx, $rH, $rPayload, $rSettings, $rMain),
 			'heartbeat' => self::heartbeat($rNode, $rKeys, $rCtx, $rH, $rPayload, $rSettings),
+			'commands' => self::commands($rNode, $rKeys, $rCtx, $rPayload),
+			'ack' => self::ack($rCrypto, $rNode, $rKeys, $rCtx, $rH, $rPayload),
 		};
 	}
 
@@ -439,6 +444,37 @@ final class ClusterApi {
 		]);
 	}
 
+	/** Longest a `commands` long-poll is held (ms). */
+	public const COMMANDS_WAIT_MAX_MS = 20000;
+
+	/**
+	 * `commands`: the node's queued commands after its high-water, panel-signed
+	 * (`cmd`) each. Held up to `wait_ms` while there are none, so a command
+	 * reaches the node within a poll step of being queued.
+	 */
+	private static function commands(array $rNode, SessionKeys $rKeys, string $rCtx, array $rP): array {
+		$rAfter = max(0, (int) ($rP['after_seq'] ?? 0));
+		$rWait = max(0, min(self::COMMANDS_WAIT_MAX_MS, (int) ($rP['wait_ms'] ?? 0)));
+		$rDeadline = microtime(true) + $rWait / 1000;
+		while (true) {
+			$rCommands = CommandBus::pending((int) $rNode['server_id'], $rAfter);
+			if ($rCommands !== [] || microtime(true) >= $rDeadline) {
+				break;
+			}
+			usleep(250000);
+		}
+		return ClusterReply::boxed($rKeys, $rCtx, ['commands' => $rCommands, 'main_time_ms' => ClusterClock::nowMs()]);
+	}
+
+	/** `ack`: a command's outcome, accepted only for this node's own commands. */
+	private static function ack(ClusterCrypto $rCrypto, array $rNode, SessionKeys $rKeys, string $rCtx, array $rH, array $rP): array {
+		$rCmdID = is_string($rP['cmd_id'] ?? null) && preg_match('/^[0-9a-f]{32}$/', (string) $rP['cmd_id']) ? (string) $rP['cmd_id'] : null;
+		if ($rCmdID === null || !CommandBus::ack((int) $rNode['server_id'], $rCmdID, !empty($rP['ok']), is_string($rP['result'] ?? null) ? (string) $rP['result'] : '')) {
+			return DenialFactory::deny($rCrypto, 400, 'BAD_REQUEST', $rH['node'], $rH['nonce']);
+		}
+		return ClusterReply::boxed($rKeys, $rCtx, ['ok' => true, 'main_time_ms' => ClusterClock::nowMs()]);
+	}
+
 	/** Map an extension refusal to a signed denial. */
 	private static function refusal(ClusterCrypto $rCrypto, string $rReason, array $rNode, array $rH): array {
 		return match (true) {

src/Domain/Cluster/ClusterRoute.php

@@ -0,0 +1,96 @@
+<?php
+
+namespace XcVm\Domain\Cluster;
+
+use XcVm\Core\Cluster\Crypto\ClusterCrypto;
+use XcVm\Core\Cluster\Crypto\ClusterCryptoFactory;
+use XcVm\Core\Config\SettingsManager;
+
+/**
+ * Where MAIN's calls to a node go: the node's signed command channel when its
+ * COMMANDS flow is on, else the legacy transport (the caller's own). Core's
+ * seams (NodeRpc, SignalDispatcher) ask here first; on load balancers this
+ * class is not in the build and they go legacy directly.
+ *
+ * Every method returns [routed, result]: routed false means "not a command
+ * node, use the legacy path".
+ */
+final class ClusterRoute {
+	/** @var (callable(): ClusterCrypto)|null */
+	private static $rCrypto = null;
+
+	/**
+	 * An RPC answered by the node: `node.rpc{action}`, waiting for its ack.
+	 *
+	 * @param array<string, mixed> $rData NodeRpc payload (action + args)
+	 * @return array{0: bool, 1: mixed} the node's raw result, or null on failure/timeout
+	 */
+	public static function rpc(int $rServerID, array $rData, int $rTimeout): array {
+		$rCrypto = self::target($rServerID);
+		if ($rCrypto === null) {
+			return [false, null];
+		}
+		try {
+			$rCmdID = CommandBus::enqueue($rCrypto, $rServerID, 'node.rpc', $rData);
+		} catch (\Throwable) {
+			return [true, null];
+		}
+		$rOutcome = CommandBus::await($rCmdID, max(1, $rTimeout));
+		return [true, $rOutcome !== null && $rOutcome[0] ? $rOutcome[1] : null];
+	}
+
+	/**
+	 * An RPC sent without waiting (NodeRpc::broadcast).
+	 *
+	 * @param array<string, mixed> $rData
+	 * @return array{0: bool, 1: bool} queued?
+	 */
+	public static function send(int $rServerID, array $rData): array {
+		$rCrypto = self::target($rServerID);
+		if ($rCrypto === null) {
+			return [false, false];
+		}
+		try {
+			CommandBus::enqueue($rCrypto, $rServerID, 'node.rpc', $rData);
+			return [true, true];
+		} catch (\Throwable) {
+			return [true, false];
+		}
+	}
+
+	/**
+	 * Kill a viewer's worker on its node (SignalDispatcher::kill): restrictive,
+	 * so it is signed even without a licence.
+	 *
+	 * @return array{0: bool, 1: bool}
+	 */
+	public static function kill(int $rServerID, int $rPID, bool $rRTMP): array {
+		$rCrypto = self::target($rServerID);
+		if ($rCrypto === null) {
+			return [false, false];
+		}
+		try {
+			CommandBus::enqueue($rCrypto, $rServerID, 'conn.kill_worker', ['pid' => $rPID, 'rtmp' => $rRTMP]);
+			return [true, true];
+		} catch (\Throwable) {
+			return [true, false];
+		}
+	}
+
+	/** Tests: supply the extension handle. Null restores the factory. */
+	public static function useCrypto(?callable $rFactory): void {
+		self::$rCrypto = $rFactory;
+	}
+
+	/** The extension, when this node takes commands; null for the legacy path. */
+	private static function target(int $rServerID): ?ClusterCrypto {
+		try {
+			if (empty(SettingsManager::get('cluster_api_enabled')) || !CommandBus::accepts(NodeRegistry::byServer($rServerID))) {
+				return null;
+			}
+			return self::$rCrypto !== null ? (self::$rCrypto)() : ClusterCryptoFactory::create();
+		} catch (\Throwable) {
+			return null; // no cluster tables, no extension: the legacy path
+		}
+	}
+}

src/Domain/Cluster/CommandBus.php

@@ -0,0 +1,199 @@
+<?php
+
+namespace XcVm\Domain\Cluster;
+
+use XcVm\Core\Cluster\Crypto\ClusterCrypto;
+use XcVm\Core\Cluster\Crypto\Enc;
+use XcVm\Infrastructure\Database\DatabaseAware;
+
+/**
+ * MAIN → node commands (plan, section 7, Phase 4). A command is a typed JSON
+ * record, never shell, signed by the panel with tag `cmd`:
+ *
+ * ```text
+ * {"v":1, "type", "exp", "iat", "cmd_id", "seq", "node_uuid", "gen", "dedupe_key", "args"}
+ * ```
+ *
+ * The extension derives its class from `type`: restrictive ones (kills,
+ * stops) sign without a licence, granting ones need it. Commands queue in
+ * `cluster_commands`, FIFO per node by `seq`, and reach the node's agent
+ * through the `commands` long-poll; the agent checks the signature, its own
+ * uuid and generation, `seq` above its high-water and `exp`, runs the
+ * command and `ack`s it with the result. Delivery is at least once.
+ *
+ * Only nodes with the COMMANDS flow on get commands; the others keep the
+ * legacy transport (NodeRpc's HTTP, the signals table).
+ */
+final class CommandBus {
+	use DatabaseAware;
+
+	/** Lifetime of a command by type prefix (seconds). */
+	public const TTL = ['conn.' => 300, 'node.root' => 86400, 'default' => 600];
+
+	/** Types this increment carries. */
+	public const TYPES = ['node.rpc', 'conn.kill_worker'];
+
+	/** Types that are restrictive (always signable); the extension decides, this is informational. */
+	public const RESTRICTIVE = ['conn.drop', 'conn.drop_line', 'conn.kill_worker', 'stream.stop', 'vod.stop', 'token.rotate_now', 'node.quarantine', 'node.fence', 'resync', 'config.changed'];
+
+	/** Largest result stored from an ack. */
+	public const MAX_RESULT = 65536;
+
+	/**
+	 * Is this node taking commands over the cluster API?
+	 *
+	 * @param array<string, mixed>|null $rNode cluster_nodes row
+	 */
+	public static function accepts(?array $rNode): bool {
+		return $rNode !== null && $rNode['state'] === 'active' && (int) $rNode['mode'] >= 1 && ((int) $rNode['flows'] & NodeRegistry::FLOW_COMMANDS) !== 0;
+	}
+
+	/**
+	 * Sign and queue a command. Returns its cmd_id.
+	 *
+	 * @param array<string, mixed> $rArgs
+	 */
+	public static function enqueue(ClusterCrypto $rCrypto, int $rServerID, string $rType, array $rArgs, ?string $rDedupeKey = null, ?int $rTtl = null): string {
+		if (!in_array($rType, self::TYPES, true)) {
+			throw new \InvalidArgumentException('Unknown command type: ' . $rType);
+		}
+		$rNode = NodeRegistry::byServer($rServerID);
+		if ($rNode === null) {
+			throw new \InvalidArgumentException('Server ' . $rServerID . ' is not an enrolled node');
+		}
+		$rNow = ClusterClock::now();
+		$rTtl ??= self::ttl($rType);
+		$rCmdID = bin2hex(random_bytes(16));
+		for ($rAttempt = 0; ; $rAttempt++) {
+			self::db()->query('SELECT MAX(`seq`) AS `seq` FROM `cluster_commands` WHERE `server_id` = ?;', $rServerID);
+			$rSeq = max((int) (self::db()->get_row()['seq'] ?? 0), (int) $rNode['cmd_seq']) + 1;
+			$rDoc = (string) json_encode([
+				'v' => 1, 'type' => $rType, 'exp' => $rNow + $rTtl, 'iat' => $rNow, 'cmd_id' => $rCmdID, 'seq' => $rSeq,
+				'node_uuid' => (string) $rNode['node_uuid'], 'gen' => (int) $rNode['gen'], 'dedupe_key' => $rDedupeKey, 'args' => (object) $rArgs,
+			], JSON_UNESCAPED_SLASHES);
+			$rSig = $rCrypto->sign('cmd', $rDoc);
+			if ($rDedupeKey !== null) {
+				// A newer desired state supersedes a command not yet acked.
+				self::db()->query("DELETE FROM `cluster_commands` WHERE `server_id` = ? AND `dedupe_key` = ? AND `state` <> 'acked';", $rServerID, $rDedupeKey);
+			}
+			try {
+				self::db()->query(
+					'INSERT INTO `cluster_commands` (`server_id`, `seq`, `cmd_id`, `type`, `action`, `class`, `dedupe_key`, `payload`, `sig`, `state`, `created_at`, `exp`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);',
+					$rServerID,
+					$rSeq,
+					$rCmdID,
+					$rType,
+					isset($rArgs['action']) ? substr((string) $rArgs['action'], 0, 32) : null,
+					in_array($rType, self::RESTRICTIVE, true) ? 'R' : 'G',
+					$rDedupeKey,
+					$rDoc,
+					$rSig,
+					'queued',
+					$rNow,
+					$rNow + $rTtl
+				);
+				return $rCmdID;
+			} catch (\Throwable $rE) {
+				if ($rAttempt >= 3) {
+					throw $rE; // a concurrent enqueue took this seq three times
+				}
+			}
+		}
+	}
+
+	/**
+	 * Commands for a node after its high-water, oldest first; marks them delivered.
+	 *
+	 * @return list<array{doc: string, sig: string, seq: int}>
+	 */
+	public static function pending(int $rServerID, int $rAfterSeq, int $rLimit = 50): array {
+		self::db()->query(
+			"SELECT `id`, `seq`, `payload`, `sig` FROM `cluster_commands` WHERE `server_id` = ? AND `seq` > ? AND `state` IN ('queued', 'delivered') AND `exp` > ? ORDER BY `seq` ASC LIMIT " . max(1, min(200, $rLimit)) . ';',
+			$rServerID,
+			$rAfterSeq,
+			ClusterClock::now()
+		);
+		$rRows = self::db()->get_rows();
+		$rOut = [];
+		$rIDs = [];
+		foreach ($rRows as $rRow) {
+			$rOut[] = ['doc' => (string) $rRow['payload'], 'sig' => Enc::b64url((string) $rRow['sig']), 'seq' => (int) $rRow['seq']];
+			$rIDs[] = (int) $rRow['id'];
+		}
+		if ($rIDs !== []) {
+			self::db()->query("UPDATE `cluster_commands` SET `state` = 'delivered', `delivered_at` = ? WHERE `state` = 'queued' AND `id` IN (" . implode(',', $rIDs) . ');', ClusterClock::now());
+		}
+		return $rOut;
+	}
+
+	/**
+	 * A node's acknowledgement: only for its own commands. Raises its
+	 * high-water (cmd_seq) so a restored queue is not replayed.
+	 */
+	public static function ack(int $rServerID, string $rCmdID, bool $rOk, string $rResult): bool {
+		self::db()->query('SELECT `seq`, `state` FROM `cluster_commands` WHERE `server_id` = ? AND `cmd_id` = ?;', $rServerID, $rCmdID);
+		$rRow = self::db()->num_rows() > 0 ? self::db()->get_row() : null;
+		if ($rRow === null) {
+			return false;
+		}
+		if (in_array($rRow['state'], ['acked', 'failed'], true)) {
+			return true; // a repeated ack
+		}
+		self::db()->query(
+			'UPDATE `cluster_commands` SET `state` = ?, `acked_at` = ?, `result` = ? WHERE `server_id` = ? AND `cmd_id` = ?;',
+			$rOk ? 'acked' : 'failed',
+			ClusterClock::now(),
+			substr($rResult, 0, self::MAX_RESULT),
+			$rServerID,
+			$rCmdID
+		);
+		self::db()->query('UPDATE `cluster_nodes` SET `cmd_seq` = ? WHERE `server_id` = ? AND `cmd_seq` < ?;', (int) $rRow['seq'], $rServerID, (int) $rRow['seq']);
+		return true;
+	}
+
+	/**
+	 * A command's outcome once acked: [ok, result], or null while pending.
+	 *
+	 * @return array{0: bool, 1: string}|null
+	 */
+	public static function result(string $rCmdID): ?array {
+		self::db()->query('SELECT `state`, `result` FROM `cluster_commands` WHERE `cmd_id` = ?;', $rCmdID);
+		$rRow = self::db()->num_rows() > 0 ? self::db()->get_row() : null;
+		if ($rRow === null || !in_array($rRow['state'], ['acked', 'failed'], true)) {
+			return null;
+		}
+		return [$rRow['state'] === 'acked', (string) $rRow['result']];
+	}
+
+	/**
+	 * Wait for a command's outcome (the admin side of an RPC).
+	 *
+	 * @return array{0: bool, 1: string}|null null on timeout
+	 */
+	public static function await(string $rCmdID, float $rTimeout, int $rPollMs = 100): ?array {
+		$rDeadline = microtime(true) + $rTimeout;
+		do {
+			$rResult = self::result($rCmdID);
+			if ($rResult !== null) {
+				return $rResult;
+			}
+			usleep($rPollMs * 1000);
+		} while (microtime(true) < $rDeadline);
+		return self::result($rCmdID);
+	}
+
+	/** Drop expired commands and outcomes older than a day. */
+	public static function prune(): void {
+		$rNow = ClusterClock::now();
+		self::db()->query("DELETE FROM `cluster_commands` WHERE (`exp` <= ? AND `state` IN ('queued', 'delivered')) OR (`acked_at` IS NOT NULL AND `acked_at` < ?);", $rNow, $rNow - 86400);
+	}
+
+	private static function ttl(string $rType): int {
+		foreach (self::TTL as $rPrefix => $rSeconds) {
+			if ($rPrefix !== 'default' && str_starts_with($rType, $rPrefix)) {
+				return $rSeconds;
+			}
+		}
+		return self::TTL['default'];
+	}
+}

src/Public/Controllers/Api/InternalApiController.php

@@ -62,6 +62,19 @@ class InternalApiController {
 		$this->dispatch($rAction, $rRequest, $rSettings);
 	}
 
+	/**
+	 * Run one system action for a signed `node.rpc` command (cluster:exec on
+	 * the node, Phase 4): the same handlers as the legacy /api, without its
+	 * password and source-IP checks, which the command's panel signature
+	 * replaces. Output is the action's response, as /api would send it.
+	 *
+	 * @param array<string, mixed> $rRequest action + arguments
+	 */
+	public function runCommand(array $rRequest): void {
+		$this->deny = false;
+		$this->dispatch((string) ($rRequest['action'] ?? ''), $rRequest, SettingsManager::getAll());
+	}
+
 	private function dispatch($rAction, $rRequest, $rSettings) {
 		switch ($rAction) {
 			case 'view_log':

src/Public/Views/admin/cluster_nodes.php

@@ -83,6 +83,7 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . '
                     <th><?= $language::get('status'); ?></th>
                     <th><?= $language::get('cluster_mode'); ?></th>
                     <th><?= $language::get('cluster_telemetry_flow'); ?></th>
+                    <th><?= $language::get('cluster_commands_flow'); ?></th>
                     <th><?= $language::get('cluster_epoch'); ?></th>
                     <th><?= $language::get('cluster_token_expires'); ?></th>
                     <th><?= $language::get('cluster_last_seen'); ?></th>
@@ -92,7 +93,7 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . '
             </thead>
             <tbody>
                 <?php if (empty($clusterNodes)): ?>
-                    <tr><td colspan="9" class="text-center text-body-secondary"><?= $language::get('cluster_no_nodes'); ?></td></tr>
+                    <tr><td colspan="10" class="text-center text-body-secondary"><?= $language::get('cluster_no_nodes'); ?></td></tr>
                 <?php endif; ?>
                 <?php foreach ($clusterNodes as $rNode): ?>
                     <tr>
@@ -107,17 +108,19 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . '
                             <?php endif; ?>
                         </td>
                         <td><?= (int) $rNode['mode']; ?></td>
+                        <?php foreach (['telemetry' => 1, 'commands' => 2] as $rFlowName => $rFlowBit): ?>
                         <td>
-                            <?php $rTel = ((int) $rNode['flows'] & 1) === 1; ?>
+                            <?php $rOn = ((int) $rNode['flows'] & $rFlowBit) === $rFlowBit; ?>
                             <?php if (in_array($rNode['state'], ['active', 'quarantined'], true)): ?>
                                 <form method="POST" class="d-inline">
                                     <input type="hidden" name="server_id" value="<?= (int) $rNode['server_id']; ?>">
-                                    <button type="submit" name="cluster_action" value="<?= $rTel ? 'telemetry_off' : 'telemetry_on'; ?>" class="btn btn-sm <?= $rTel ? 'btn-label-success' : 'btn-label-secondary'; ?>" title="<?= $language::get('cluster_telemetry_help'); ?>"><?= $language::get($rTel ? 'cluster_flow_on' : 'cluster_flow_off'); ?></button>
+                                    <button type="submit" name="cluster_action" value="<?= $rFlowName . ($rOn ? '_off' : '_on'); ?>" class="btn btn-sm <?= $rOn ? 'btn-label-success' : 'btn-label-secondary'; ?>" title="<?= $language::get('cluster_' . $rFlowName . '_help'); ?>"><?= $language::get($rOn ? 'cluster_flow_on' : 'cluster_flow_off'); ?></button>
                                 </form>
                             <?php else: ?>
                                 <span class="text-body-secondary">—</span>
                             <?php endif; ?>
                         </td>
+                        <?php endforeach; ?>
                         <td><?= (int) $rNode['epoch']; ?> <span class="small text-body-secondary">(gen <?= (int) $rNode['gen']; ?>)</span></td>
                         <td><?= $rWhen($rNode['token_exp'] === null ? null : (int) $rNode['token_exp']); ?></td>
                         <td><?= $rWhen($rNode['last_seen_at'] === null ? null : intdiv((int) $rNode['last_seen_at'], 1000)); ?></td>

tests/Unit/ClusterApiTest.php

@@ -53,7 +53,7 @@ final class ClusterApiTest extends TestCase {
 
 	protected function setUp(): void {
 		$this->rDb = new TestDb();
-		foreach (['029_create_cluster_nodes', '032_create_cluster_audit'] as $rName) {
+		foreach (['029_create_cluster_nodes', '030_create_cluster_commands', '032_create_cluster_audit'] as $rName) {
 			$this->rDb->exec($this->ddl((string) file_get_contents(dirname(__DIR__, 2) . '/src/migrations/database/up/' . $rName . '.sql')));
 		}
 		$this->rDb->exec('ALTER TABLE `cluster_node_epochs` ADD COLUMN `agent_eph_pub` binary(32) DEFAULT NULL');
@@ -675,4 +675,73 @@ final class ClusterApiTest extends TestCase {
 		[$rRes, $rReq] = $this->rekey($this->challenge(), random_bytes(32));
 		$this->denial($rRes, 403, 'NODE_REVOKED', $rReq);
 	}
+
+	public function testCommandsAreDeliveredSignedAndAcked(): void {
+		$rKeys = $this->active();
+		$rCmd = \XcVm\Domain\Cluster\CommandBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'get_pids']);
+		[$rRes, $rCtx] = $this->call('commands', ['after_seq' => 0, 'wait_ms' => 0], 1, $rKeys);
+		$rOut = $this->reply($rRes, $rCtx, $rKeys);
+		$this->assertCount(1, $rOut['commands']);
+		$rOne = $rOut['commands'][0];
+		$this->assertTrue(PanelSig::verify($this->rCrypto->info()['panel_sign_pub'], 'cmd', $rOne['doc'], (string) Enc::b64urlDecode($rOne['sig'])), 'panel-signed, tag cmd');
+		$rDoc = json_decode($rOne['doc'], true);
+		$this->assertSame(['node.rpc', 1, $this->rUuid, 1, $rCmd, ['action' => 'get_pids']], [$rDoc['type'], $rDoc['seq'], $rDoc['node_uuid'], $rDoc['gen'], $rDoc['cmd_id'], $rDoc['args']]);
+
+		// Nothing after the high-water.
+		[$rRes, $rCtx] = $this->call('commands', ['after_seq' => 1, 'wait_ms' => 0], 1, $rKeys);
+		$this->assertSame([], $this->reply($rRes, $rCtx, $rKeys)['commands']);
+
+		[$rRes, $rCtx] = $this->call('ack', ['cmd_id' => $rCmd, 'ok' => true, 'result' => '[1,2]'], 1, $rKeys);
+		$this->assertTrue($this->reply($rRes, $rCtx, $rKeys)['ok']);
+		$this->assertSame([true, '[1,2]'], \XcVm\Domain\Cluster\CommandBus::result($rCmd));
+		$this->assertSame(1, (int) NodeRegistry::byServer(self::SID)['cmd_seq'], 'the high-water rises with the ack');
+		[$rRes, $rCtx] = $this->call('commands', ['after_seq' => 0, 'wait_ms' => 0], 1, $rKeys);
+		$this->assertSame([], $this->reply($rRes, $rCtx, $rKeys)['commands'], 'an acked command is not delivered again');
+
+		// An ack for a command that is not this node's is refused.
+		[$rRes, , $rReq] = $this->call('ack', ['cmd_id' => str_repeat('ab', 16), 'ok' => true], 1, $rKeys);
+		$this->denial($rRes, 400, 'BAD_REQUEST', $rReq);
+	}
+
+	public function testQuarantineFreezesCommands(): void {
+		$rKeys = $this->active();
+		NodeRegistry::update(self::SID, ['state' => 'quarantined']);
+		[$rRes, , $rReq] = $this->call('commands', ['after_seq' => 0], 1, $rKeys);
+		$this->denial($rRes, 409, 'NOT_ACTIVE', $rReq);
+	}
+
+	public function testCommandsExpireAndDedupe(): void {
+		$this->active();
+		$rBus = \XcVm\Domain\Cluster\CommandBus::class;
+		$rFirst = $rBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'stream', 'function' => 'start', 'stream_ids' => [1]], 'stream:1');
+		$rSecond = $rBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'stream', 'function' => 'stop', 'stream_ids' => [1]], 'stream:1');
+		$rPending = $rBus::pending(self::SID, 0);
+		$this->assertCount(1, $rPending, 'a newer desired state replaces the unacked one');
+		$this->assertSame($rSecond, json_decode($rPending[0]['doc'], true)['cmd_id']);
+		$this->assertNotSame($rFirst, $rSecond);
+		ClusterClock::fix($this->rT0 + 601000);
+		$this->assertSame([], $rBus::pending(self::SID, 0), 'expired');
+		$rBus::prune();
+		$this->rDb->query('SELECT COUNT(*) AS `n` FROM `cluster_commands`');
+		$this->assertSame(0, (int) $this->rDb->get_row()['n']);
+	}
+
+	public function testRoutingFollowsTheCommandsFlow(): void {
+		$this->active();
+		SettingsManager::set($this->rSettings);
+		\XcVm\Domain\Cluster\ClusterRoute::useCrypto(fn() => $this->rCrypto);
+		try {
+			$this->assertSame([false, false], \XcVm\Domain\Cluster\ClusterRoute::kill(self::SID, 123, false), 'flow off: legacy');
+			NodeRegistry::update(self::SID, ['flows' => NodeRegistry::FLOW_COMMANDS]);
+			$this->assertSame([true, true], \XcVm\Domain\Cluster\ClusterRoute::kill(self::SID, 123, false));
+			$this->assertSame([true, true], \XcVm\Domain\Cluster\ClusterRoute::send(self::SID, ['action' => 'free_temp']));
+			$rStart = microtime(true);
+			$this->assertSame([true, null], \XcVm\Domain\Cluster\ClusterRoute::rpc(self::SID, ['action' => 'get_pids'], 1), 'no ack: a timeout, as an offline node');
+			$this->assertLessThan(3, microtime(true) - $rStart);
+			$rTypes = array_map(static fn($rC) => json_decode($rC['doc'], true)['type'], \XcVm\Domain\Cluster\CommandBus::pending(self::SID, 0));
+			$this->assertSame(['conn.kill_worker', 'node.rpc', 'node.rpc'], $rTypes);
+		} finally {
+			\XcVm\Domain\Cluster\ClusterRoute::useCrypto(null);
+		}
+	}
 }

tests/Unit/ClusterExecCommandTest.php

@@ -0,0 +1,47 @@
+<?php
+
+use PHPUnit\Framework\TestCase;
+use XcVm\Cli\Commands\ClusterExecCommand;
+use XcVm\Core\Cluster\Crypto\Enc;
+use XcVm\Tests\Support\FakeClusterCrypto;
+
+/**
+ * cluster:exec runs a command only when it carries the panel's `cmd`
+ * signature under the key the agent pinned, names this node and is not stale.
+ */
+final class ClusterExecCommandTest extends TestCase {
+	private FakeClusterCrypto $rCrypto;
+
+	private array $rState;
+
+	protected function setUp(): void {
+		$this->rCrypto = new FakeClusterCrypto();
+		$this->rState = ['node_uuid' => '0f8fad5b-d9cb-469f-a165-70867728950e', 'panel_sign_pub' => base64_encode($this->rCrypto->info()['panel_sign_pub'])];
+	}
+
+	private function command(array $rOverride = [], string $rTag = 'cmd'): array {
+		$rDoc = (string) json_encode($rOverride + ['v' => 1, 'type' => 'node.rpc', 'exp' => 1800000600, 'iat' => 1800000000, 'cmd_id' => str_repeat('a', 32), 'seq' => 1, 'node_uuid' => $this->rState['node_uuid'], 'gen' => 1, 'dedupe_key' => null, 'args' => ['action' => 'get_pids']]);
+		return ['doc' => $rDoc, 'sig' => Enc::b64url($this->rCrypto->sign($rTag, $rDoc))];
+	}
+
+	public function testAGenuineCommandVerifies(): void {
+		$rCmd = ClusterExecCommand::verify($this->command(), $this->rState, 1800000000);
+		$this->assertIsArray($rCmd);
+		$this->assertSame('get_pids', $rCmd['args']['action']);
+	}
+
+	public function testForgedOrMisaddressedCommandsAreRefused(): void {
+		$rGood = $this->command();
+		$this->assertSame('bad signature', ClusterExecCommand::verify(['doc' => str_replace('get_pids', 'kill_pid', $rGood['doc'])] + $rGood, $this->rState, 1800000000));
+		$this->assertSame('bad signature', ClusterExecCommand::verify($this->command([], 'den'), $this->rState, 1800000000), 'another tag');
+		$this->assertSame('bad signature', ClusterExecCommand::verify($rGood, ['panel_sign_pub' => base64_encode(str_repeat("\1", 32))] + $this->rState, 1800000000));
+		$this->assertIsString(ClusterExecCommand::verify($this->command(['node_uuid' => '11111111-1111-4111-a111-111111111111']), $this->rState, 1800000000));
+		$this->assertIsString(ClusterExecCommand::verify($rGood, $this->rState, 1800000600 + ClusterExecCommand::SKEW + 1), 'stale');
+	}
+
+	public function testUnknownTypesAndActionsDoNothing(): void {
+		$this->assertSame(2, ClusterExecCommand::run(['type' => 'node.root', 'args' => ['action' => 'reboot']]));
+		$this->assertSame(2, ClusterExecCommand::run(['type' => 'node.rpc', 'args' => ['action' => 'view_log']]), 'not in NodeRpc::ACTIONS');
+		$this->assertSame(2, ClusterExecCommand::run(['type' => 'conn.kill_worker', 'args' => ['pid' => 0]]));
+	}
+}