huskies: merge 643_story_web_ui_consumer_for_the_unified_status_broadcaster

server/src/http/ws.rs

@@ -1,6 +1,7 @@
 //! WebSocket transport adapter — accept connection, serialise/deserialise frames,
 //! invoke service methods. No business logic, no inline state transitions.
 
+use crate::config::ProjectConfig;
 use crate::http::context::AppContext;
 use crate::llm::chat;
 use crate::service::ws::{self, WsResponse};
@@ -56,6 +57,18 @@ pub async fn ws_handler(ws: WebSocket, ctx: Data<&Arc<AppContext>>) -> impl poem
         ws::subscribe_watcher(tx.clone(), ctx.clone(), ctx.watcher_tx.subscribe());
         ws::subscribe_reconciliation(tx.clone(), ctx.reconciliation_tx.subscribe());
 
+        // Subscribe to the status broadcaster if web UI consumer is enabled (default: true).
+        let status_enabled = ctx
+            .state
+            .get_project_root()
+            .ok()
+            .and_then(|root| ProjectConfig::load(&root).ok())
+            .map(|c| c.web_ui_status_consumer)
+            .unwrap_or(true);
+        if status_enabled {
+            ws::subscribe_status(tx.clone(), ctx.services.status.subscribe());
+        }
+
         // Map of pending permission request_id -> oneshot responder.
         let mut pending_perms: HashMap<String, oneshot::Sender<PermissionDecision>> =
             HashMap::new();
@@ -230,6 +243,7 @@ pub async fn ws_handler(ws: WebSocket, ctx: Data<&Arc<AppContext>>) -> impl poem
 mod tests {
     use super::*;
     use crate::io::watcher::WatcherEvent;
+    use crate::service::status::StatusEvent;
 
     // ── ws_handler integration tests (real WebSocket connection) ─────
 
@@ -534,4 +548,158 @@ mod tests {
         let (_sink2, _stream2, initial2) = connect_ws(&url).await;
         assert_eq!(initial2["type"], "pipeline_state");
     }
+
+    /// Read the next `status_update` whose story_id or story_name contains `needle`,
+    /// within a timeout.  Skips `log_entry` noise and unrelated status events so
+    /// genuine server log noise cannot cause false positives or negatives.
+    async fn next_status_update_containing(
+        stream: &mut futures::stream::SplitStream<
+            tokio_tungstenite::WebSocketStream<
+                tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+            >,
+        >,
+        needle: &str,
+        timeout_ms: u64,
+    ) -> Option<serde_json::Value> {
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(timeout_ms);
+        loop {
+            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+            if remaining.is_zero() {
+                return None;
+            }
+            let msg = tokio::time::timeout(remaining, stream.next())
+                .await
+                .ok()?
+                .expect("stream ended")
+                .expect("ws error");
+            let val: serde_json::Value = match msg {
+                tungstenite::Message::Text(t) => serde_json::from_str(t.as_ref()).ok()?,
+                _ => continue,
+            };
+            if val["type"] == "status_update" {
+                let event = &val["event"];
+                let story_id = event["story_id"].as_str().unwrap_or("");
+                let story_name = event["story_name"].as_str().unwrap_or("");
+                if story_id.contains(needle) || story_name.contains(needle) {
+                    return Some(val);
+                }
+            }
+            // Skip log_entry and other unrelated messages.
+        }
+    }
+
+    // ── Status broadcaster integration tests ─────────────────────────
+
+    /// Publishing a status event via `services.status` must result in a
+    /// `status_update` WebSocket message with structured fields delivered to the
+    /// connected client.
+    #[tokio::test]
+    async fn ws_handler_forwards_status_events_as_status_update() {
+        let (url, ctx) = start_test_server().await;
+        let (_sink, mut stream, _initial) = connect_ws(&url).await;
+
+        // Use a story ID unique enough that genuine server logs won't match it.
+        ctx.services.status.publish(StatusEvent::StageTransition {
+            story_id: "77_story_status_fwd_test".to_string(),
+            story_name: Some("StatusFwdTest".to_string()),
+            from_stage: "1_backlog".to_string(),
+            to_stage: "2_current".to_string(),
+        });
+
+        // The handler must forward it as a status_update with structured fields.
+        let msg = next_status_update_containing(&mut stream, "StatusFwdTest", 2000)
+            .await
+            .expect("expected a status_update for the status event");
+        assert_eq!(msg["type"], "status_update");
+        let event = &msg["event"];
+        assert_eq!(event["type"], "stage_transition");
+        assert_eq!(event["story_id"], "77_story_status_fwd_test");
+        assert_eq!(event["story_name"], "StatusFwdTest");
+        assert_eq!(event["from_stage"], "1_backlog");
+        assert_eq!(event["to_stage"], "2_current");
+    }
+
+    /// Multi-project isolation: a client connected to project A's server must
+    /// NOT receive status events published on project B's broadcaster.
+    #[tokio::test]
+    async fn ws_handler_multi_project_status_isolation() {
+        // Start two independent servers (each with its own AppContext / Services).
+        let (url_a, ctx_a) = start_test_server().await;
+        let (url_b, _ctx_b) = start_test_server().await;
+
+        let (_sink_a, mut stream_a, _) = connect_ws(&url_a).await;
+        let (_sink_b, mut stream_b, _) = connect_ws(&url_b).await;
+
+        // Use a needle unique enough that genuine server logs won't match.
+        let needle = "ProjAIsolation7734";
+        ctx_a.services.status.publish(StatusEvent::MergeFailure {
+            story_id: "10_story_proj_a_isolation".to_string(),
+            story_name: Some(needle.to_string()),
+            reason: "conflict".to_string(),
+        });
+
+        // Client A must receive the status_update with structured fields.
+        let msg_a = next_status_update_containing(&mut stream_a, needle, 2000)
+            .await
+            .expect("client A should receive the status event");
+        assert_eq!(msg_a["type"], "status_update");
+        assert_eq!(msg_a["event"]["story_name"], needle);
+
+        // Client B must NOT receive any status_update containing the needle.
+        let msg_b = next_status_update_containing(&mut stream_b, needle, 300).await;
+        assert!(
+            msg_b.is_none(),
+            "client B must not receive project A's status event, got: {msg_b:?}"
+        );
+    }
+
+    /// When `web_ui_status_consumer = false` in project.toml, the WebSocket
+    /// handler must not forward status events to the connected client.
+    #[tokio::test]
+    async fn ws_handler_status_consumer_disabled_via_config() {
+        let tmp = tempfile::tempdir().unwrap();
+        let root = tmp.path().to_path_buf();
+
+        // Write a project.toml that disables the web UI status consumer.
+        let huskies_dir = root.join(".huskies");
+        std::fs::create_dir_all(&huskies_dir).unwrap();
+        std::fs::write(
+            huskies_dir.join("project.toml"),
+            "web_ui_status_consumer = false\n",
+        )
+        .unwrap();
+
+        crate::db::ensure_content_store();
+        let ctx = Arc::new(AppContext::new_test(root));
+        let ctx_data = ctx.clone();
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = poem::Route::new()
+            .at("/ws", poem::get(ws_handler))
+            .data(ctx_data);
+        tokio::spawn(async move {
+            let acceptor = poem::listener::TcpAcceptor::from_tokio(listener).unwrap();
+            let _ = poem::Server::new_with_acceptor(acceptor).run(app).await;
+        });
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+        let url = format!("ws://127.0.0.1:{}/ws", addr.port());
+        let (_sink, mut stream, _) = connect_ws(&url).await;
+
+        // Use a unique needle — genuine server logs will never contain this.
+        let needle = "DisabledConsumer9182";
+        ctx.services.status.publish(StatusEvent::StoryBlocked {
+            story_id: "55_story_disabled_consumer".to_string(),
+            story_name: Some(needle.to_string()),
+            reason: "test".to_string(),
+        });
+
+        // Consumer is disabled — no status_update with this needle should arrive.
+        let msg = next_status_update_containing(&mut stream, needle, 500).await;
+        assert!(
+            msg.is_none(),
+            "disabled consumer must not forward status events, got: {msg:?}"
+        );
+    }
 }