Fix: governance-event WebSocket broadcasts never reached anyone (two independent bugs)

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<Self>) {
+        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);
+    }
 }

primeswarm-pchi/src/main.rs

@@ -68,6 +68,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     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<dyn std::error::Error>> {
     // stop_transport needs &mut self, which the Arc above gave up)
     web_handle.abort();
     udp_handle.abort();
+    ws_handle.abort();
 
     println!("PCHI Conductor stopped");
 

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<Vec<PCHIMessage>, ConductorError> {
+    ) -> Result<(Vec<PCHIMessage>, Option<String>), ConductorError> {
         let state_snapshot_hash = snapshot_hash(state);
         let mut events = Vec::new();
         let mut deny: Option<String> = 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")));