From d86621eacf61122b08f68074439548bd94b66242 Mon Sep 17 00:00:00 2001 From: vdmo Date: Thu, 1 Oct 2026 00:23:41 +1000 Subject: [PATCH] fix(conductor): governance-event WS broadcasts never reached anyone; DENY events were dropped from broadcast even when they did MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found while preparing to build a TiXL bridge on top of the governance broadcast path, before writing a single line of bridge code — exactly the discipline that caught the Resolume bridge's bugs. Two independent bugs, both silent: 1. TransportLayer's WebSocket server (port 8889) binds via start_transport, but nothing ever called accept_websocket() in a loop. ws_clients was permanently empty, so broadcast_websocket (called from process_message for every fired rule since the real-governance-signed-receipts change) iterated zero clients and did nothing — every governance-event broadcast has gone nowhere since that feature shipped. Same bug shape as the earlier UDP receive-loop fix: a bound socket is not a served one. PCHIConductor::run_ws_accept_loop closes this, spawned from main.rs alongside the UDP loop. 2. Independent of (1): RuleEngine::evaluate built the events Vec correctly but, on a DENY, returned Err(ConductorError::RuleError(..)) directly — discarding the already-built events before process_message ever got to broadcast them. The DENY itself was still correctly signed and persisted (that path doesn't touch events), so GET /governance/export always had it — but nothing was ever broadcast live for the single most important event type. evaluate() now returns (events, Option) instead of using Err for this; process_message broadcasts every event first, then converts deny_reason into the Err it returns to its own caller, preserving the existing "a denial is a real error to the caller" contract. Added a real regression test: connects an actual WebSocket client to a running conductor, triggers a real DENY, and asserts the client receives it — not a mock, an actual client on an actual socket. It failed against the pre-fix code (correctly reproducing bug 1); passes now. Updated two existing rules.rs tests for evaluate()'s new return shape. Verified live, not just in the test suite: ran the real binary, a real Python websocket-client, and a real triggering UDP packet — `RECEIVED: governance_event {'gate_state': 'DENY', ...}` — before this fix, that client would have received nothing, forever. 30/30 tests pass; clippy clean. Co-Authored-By: Claude Sonnet 5 --- primeswarm-pchi/src/conductor.rs | 90 +++++++++++++++++++++++++++++++- primeswarm-pchi/src/main.rs | 7 +++ primeswarm-pchi/src/rules.rs | 44 ++++++++++------ 3 files changed, 123 insertions(+), 18 deletions(-) diff --git a/primeswarm-pchi/src/conductor.rs b/primeswarm-pchi/src/conductor.rs index 1e9cc56..d68c41e 100644 --- a/primeswarm-pchi/src/conductor.rs +++ b/primeswarm-pchi/src/conductor.rs @@ -158,9 +158,13 @@ impl PCHIConductor { // returned Ok regardless of a DENY), so a caller had no way to // learn a denial actually happened. Every fired rule is signed // into the governance log before this returns, denial or not. - let events = self.rule_engine.evaluate(&state, &self.governance)?; + let (events, deny_reason) = self.rule_engine.evaluate(&state, &self.governance)?; drop(state); + // Broadcast every event — including a DENY's — before turning a + // denial into the error this function returns. evaluate() used to + // return Err directly on a deny, which discarded events before + // they were ever broadcast; see evaluate()'s doc comment. for event in events { if let Ok(bytes) = serde_json::to_vec(&event) { if let Err(e) = self.transport.broadcast_websocket(&bytes).await { @@ -169,6 +173,10 @@ impl PCHIConductor { } } + if let Some(msg) = deny_reason { + return Err(ConductorError::RuleError(msg)); + } + Ok(()) } @@ -197,6 +205,30 @@ impl PCHIConductor { self.transport.start(bind_addr).await } + /// Accept WebSocket connections on the transport layer's dedicated port + /// (8889) and register each with `TransportLayer` so + /// `broadcast_websocket` — the mechanism `process_message` uses to push + /// governance events out in real time — actually has clients to send + /// to. `start_transport` only opens the listening socket; until this + /// ran, nothing ever called `accept_websocket`, so `ws_clients` was + /// permanently empty and every broadcast silently went nowhere. Same + /// bug shape as the UDP receive loop above, found the same way: by + /// actually testing the path end-to-end instead of assuming a bound + /// socket means a served one. + pub async fn run_ws_accept_loop(self: Arc) { + loop { + match self.transport.accept_websocket().await { + Ok((id, stream)) => { + self.transport.add_ws_client(id, stream).await; + } + Err(e) => { + error!("WebSocket accept loop stopping: {}", e); + return; + } + } + } + } + /// Receive PCHI messages over UDP and feed them into `process_message` /// in a loop until the socket errors or is stopped. `start_transport` /// only opens the sockets — until this ran, nothing anything sent to @@ -299,4 +331,60 @@ mod tests { assert!(conductor.process_message(msg).await.is_ok()); let _ = std::fs::remove_dir_all(&dir); } + + #[tokio::test] + async fn governance_events_actually_reach_a_connected_ws_client() { + // Regression test for the bug this fixes: broadcast_websocket had a + // real client registry and real send logic, but nothing ever called + // accept_websocket to populate it — every broadcast silently went + // nowhere, undetected because no test exercised a real WS + // connection end-to-end. bind_addr must end in exactly ":8888" — + // start_transport derives the WS port via bind_addr.replace( + // ":8888", ":8889"), a plain substring replace with no other + // validation. + let (mut conductor, dir) = test_conductor().await; + conductor.start_transport("127.0.0.1:8888").await.unwrap(); + let conductor = Arc::new(conductor); + let ws_task = tokio::spawn(conductor.clone().run_ws_accept_loop()); + + // Give the accept loop a moment to actually be polling accept(). + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let (ws_stream, _) = tokio_tungstenite::connect_async("ws://127.0.0.1:8889") + .await + .expect("client should connect to the now-accepting WS server"); + let (_, mut read) = futures_util::StreamExt::split(ws_stream); + + // Give the server a moment to register the new client before the + // triggering message is processed — otherwise the broadcast could + // race the registration. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let msg = PCHIMessage::new( + MessageType::ControlParameter, + "test".to_string(), + Payload::ControlParameter(ControlParameterData { + target_id: "tentacle_system".to_string(), + parameter: "total_curl".to_string(), + value: serde_json::json!(10.0), + }), + ); + // Expected to Err (a real DENY) — the point here is what the WS + // client receives, not this call's return value. + let _ = conductor.process_message(msg).await; + + let received = tokio::time::timeout(std::time::Duration::from_secs(5), futures_util::StreamExt::next(&mut read)) + .await + .expect("timed out waiting for a governance event over the WS connection") + .expect("stream ended without a message") + .expect("WS read error"); + + let text = received.into_text().expect("expected a text frame"); + let parsed: serde_json::Value = serde_json::from_str(&text).expect("expected valid PCHIMessage JSON"); + assert_eq!(parsed["type"], "governance_event"); + assert_eq!(parsed["payload"]["data"]["gate_state"], "DENY"); + + ws_task.abort(); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/primeswarm-pchi/src/main.rs b/primeswarm-pchi/src/main.rs index a7d7a46..001c507 100644 --- a/primeswarm-pchi/src/main.rs +++ b/primeswarm-pchi/src/main.rs @@ -68,6 +68,12 @@ async fn main() -> Result<(), Box> { let udp_handle = tokio::spawn(conductor.clone().run_udp_receive_loop()); println!("Governing incoming UDP traffic on 127.0.0.1:8888"); + // Accept WebSocket clients on 8889 so broadcast_websocket (governance + // events) actually reaches someone. Until this ran, the socket was + // bound but nothing ever accepted a connection. + let ws_handle = tokio::spawn(conductor.clone().run_ws_accept_loop()); + println!("Accepting governance-event WebSocket clients on ws://127.0.0.1:8889"); + println!("PCHI Conductor running. Press Ctrl+C to stop."); // Wait for shutdown signal @@ -78,6 +84,7 @@ async fn main() -> Result<(), Box> { // stop_transport needs &mut self, which the Arc above gave up) web_handle.abort(); udp_handle.abort(); + ws_handle.abort(); println!("PCHI Conductor stopped"); diff --git a/primeswarm-pchi/src/rules.rs b/primeswarm-pchi/src/rules.rs index 8fc9e31..1a1af72 100644 --- a/primeswarm-pchi/src/rules.rs +++ b/primeswarm-pchi/src/rules.rs @@ -51,18 +51,27 @@ impl RuleEngine { } /// Evaluate every rule against the current state. For each rule whose - /// condition is true, records a signed receipt in `gov` and returns a - /// `GovernanceEvent` PCHI message for the conductor to broadcast. A - /// `deny` action also short-circuits the whole evaluation with - /// `ConductorError::RuleError` — same fail-closed semantics the - /// original hardcoded engine had — but only after every fired rule up - /// to that point has already been signed and persisted, so a denial is - /// never lost from the chain. + /// condition is true, records a signed receipt in `gov` and builds a + /// `GovernanceEvent` PCHI message for the conductor to broadcast. + /// + /// Returns `(events, deny_reason)` rather than using `Err` for a deny. + /// An earlier version returned `Err(ConductorError::RuleError(msg))` + /// directly from here, which discarded the already-built `events` — + /// every governance event from an evaluation that included a DENY was + /// silently never broadcast, only ever signed into the log. Caught by + /// `conductor::tests::governance_events_actually_reach_a_connected_ws_client`, + /// which connects a real WebSocket client and asserts it actually + /// receives the event — the DENY was correctly in the chain the whole + /// time (verifiable via `/governance/export`), just never live. + /// `process_message` broadcasts every event first, then turns + /// `deny_reason` into the `Err` it returns to its own caller — so a + /// denial is still never silently treated as success, and is now also + /// never silently dropped from telemetry. pub fn evaluate( &self, state: &SceneState, gov: &GovernanceLog, - ) -> Result, ConductorError> { + ) -> Result<(Vec, Option), ConductorError> { let state_snapshot_hash = snapshot_hash(state); let mut events = Vec::new(); let mut deny: Option = None; @@ -109,10 +118,7 @@ impl RuleEngine { } } - if let Some(msg) = deny { - return Err(ConductorError::RuleError(msg)); - } - Ok(events) + Ok((events, deny)) } } @@ -171,8 +177,13 @@ mod tests { let mut state = SceneState::new(); state.set_control_parameter("tentacle_system.total_curl".to_string(), 10.0); - let result = engine.evaluate(&state, &gov); - assert!(result.is_err(), "curl over threshold should deny"); + let (events, deny_reason) = engine.evaluate(&state, &gov).unwrap(); + assert!(deny_reason.is_some(), "curl over threshold should deny"); + assert!( + events.iter().any(|e| matches!(&e.payload, + pchi_schema::Payload::GovernanceEvent(d) if d.gate_state == "DENY")), + "the DENY must still be in the returned events, not just the deny_reason" + ); let export = gov.export(); let receipts = export["receipts"].as_array().unwrap(); @@ -205,9 +216,8 @@ mod tests { // Should not error: no curl/animation_speed/coherence_gap set, so // only the equilibrium check (Unknown -> false, doesn't fire) and // the bpm/kick evolve rule fire. - let result = engine.evaluate(&state, &gov); - assert!(result.is_ok()); - let events = result.unwrap(); + let (events, deny_reason) = engine.evaluate(&state, &gov).unwrap(); + assert!(deny_reason.is_none()); assert!(events.iter().any(|e| matches!(&e.payload, pchi_schema::Payload::GovernanceEvent(d) if d.gate_state == "ALLOW")));