From de7d22cb9f862f8a454fbad2fd56c3f0aaffe17d Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Thu, 16 Jul 2026 18:04:33 +0000 Subject: [PATCH] =?UTF-8?q?huskies:=20merge=201180=20story=20Sled=E2=86=94?= =?UTF-8?q?gateway=20goes=20WS-only:=20remove=20the=20deprecated=20HTTP=20?= =?UTF-8?q?fallback=20paths?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/src/chat/transport/matrix/bot/run.rs | 31 +- .../transport/matrix/config/tests_core.rs | 41 -- .../src/chat/transport/matrix/config/types.rs | 28 - server/src/gateway/mod.rs | 3 +- server/src/gateway/tests.rs | 582 +++++------------- server/src/http/events.rs | 202 ------ server/src/http/mod.rs | 23 +- server/src/main.rs | 6 - server/src/service/events/buffer.rs | 115 +--- server/src/service/events/io.rs | 80 --- server/src/service/events/mod.rs | 28 +- server/src/service/gateway/io.rs | 90 +-- server/src/service/gateway/mod.rs | 27 +- 13 files changed, 203 insertions(+), 1053 deletions(-) delete mode 100644 server/src/http/events.rs delete mode 100644 server/src/service/events/io.rs diff --git a/server/src/chat/transport/matrix/bot/run.rs b/server/src/chat/transport/matrix/bot/run.rs index 8ba3b7ff..ad69d2bc 100644 --- a/server/src/chat/transport/matrix/bot/run.rs +++ b/server/src/chat/transport/matrix/bot/run.rs @@ -190,21 +190,6 @@ pub async fn run_bot( let notif_room_ids = target_room_ids.clone(); let notif_project_root = project_root.clone(); let announce_room_ids = target_room_ids.clone(); - // Clone values needed by the gateway notification poller (only used in gateway mode). - let poller_room_ids: Vec = target_room_ids.iter().map(|r| r.to_string()).collect(); - let poller_project_urls: std::collections::BTreeMap = - if let Some(ref store) = gateway_projects_store { - store - .read() - .await - .iter() - .filter_map(|(name, entry)| entry.url.clone().map(|url| (name.clone(), url))) - .collect() - } else { - std::collections::BTreeMap::new() - }; - let poller_poll_interval = config.aggregated_notifications_poll_interval_secs; - let poller_enabled = config.aggregated_notifications_enabled; let persisted = load_history(project_root); slog!( @@ -371,21 +356,9 @@ pub async fn run_bot( notif_project_root, ); - // In gateway mode, spawn the cross-project notification poller. - // It polls every registered project's `/api/events` endpoint and forwards - // new events to the configured gateway rooms with a `[project-name]` prefix. - // The poller is controlled by the gateway-level `aggregated_notifications_enabled` - // flag in bot.toml — set it to `false` to disable without touching per-project configs. - if !poller_project_urls.is_empty() && poller_enabled { - crate::gateway::spawn_gateway_notification_poller( - Arc::clone(&transport), - poller_room_ids, - poller_project_urls, - poller_poll_interval, - ); - } - // Forwarder task: post gateway events to Matrix rooms with `[project-name]` prefix. + // Project nodes push events over the WS uplink (story 899/1179); the + // gateway no longer polls per-project `/api/events` over HTTP (story 1180). if let Some(event_rx) = gateway_event_rx_for_forwarder { let broadcast_room_ids: Vec = announce_room_ids.iter().map(|r| r.to_string()).collect(); diff --git a/server/src/chat/transport/matrix/config/tests_core.rs b/server/src/chat/transport/matrix/config/tests_core.rs index 068fc137..53840695 100644 --- a/server/src/chat/transport/matrix/config/tests_core.rs +++ b/server/src/chat/transport/matrix/config/tests_core.rs @@ -282,47 +282,6 @@ require_verified_devices = true ); } -#[test] -fn aggregated_notifications_enabled_defaults_to_true() { - let tmp = tempfile::tempdir().unwrap(); - let sk = tmp.path().join(".huskies"); - fs::create_dir_all(&sk).unwrap(); - fs::write( - sk.join("bot.toml"), - r#" -homeserver = "https://matrix.example.com" -username = "@bot:example.com" -password = "secret" -room_ids = ["!abc:example.com"] -enabled = true -"#, - ) - .unwrap(); - let config = BotConfig::load(tmp.path()).unwrap(); - assert!(config.aggregated_notifications_enabled); -} - -#[test] -fn aggregated_notifications_enabled_can_be_set_to_false() { - let tmp = tempfile::tempdir().unwrap(); - let sk = tmp.path().join(".huskies"); - fs::create_dir_all(&sk).unwrap(); - fs::write( - sk.join("bot.toml"), - r#" -homeserver = "https://matrix.example.com" -username = "@bot:example.com" -password = "secret" -room_ids = ["!abc:example.com"] -enabled = true -aggregated_notifications_enabled = false -"#, - ) - .unwrap(); - let config = BotConfig::load(tmp.path()).unwrap(); - assert!(!config.aggregated_notifications_enabled); -} - #[test] fn load_reads_ambient_rooms() { let tmp = tempfile::tempdir().unwrap(); diff --git a/server/src/chat/transport/matrix/config/types.rs b/server/src/chat/transport/matrix/config/types.rs index 87430b00..88c3c46d 100644 --- a/server/src/chat/transport/matrix/config/types.rs +++ b/server/src/chat/transport/matrix/config/types.rs @@ -9,14 +9,6 @@ pub(super) fn default_permission_timeout_secs() -> u64 { 120 } -pub(super) fn default_aggregated_notifications_poll_interval_secs() -> u64 { - 5 -} - -pub(super) fn default_aggregated_notifications_enabled() -> bool { - true -} - /// Default coalesce window for the chat dispatcher (1 500 ms). pub(super) fn default_coalesce_window_ms() -> u64 { 1_500 @@ -173,26 +165,6 @@ pub struct BotConfig { #[serde(default)] pub discord_allowed_users: Vec, - /// How often (in seconds) the gateway polls each project server's - /// `/api/events` endpoint to aggregate cross-project notifications. - /// - /// Only used when the gateway's bot is enabled. Defaults to 5 seconds. - #[serde(default = "default_aggregated_notifications_poll_interval_secs")] - pub aggregated_notifications_poll_interval_secs: u64, - - /// Whether the gateway-level aggregated cross-project notification stream - /// is enabled. When `false`, the gateway will not poll per-project - /// servers for events even if the bot is otherwise enabled. - /// - /// Set this in the **gateway's** `bot.toml` (not in per-project configs). - /// Adding a new project to `projects.toml` never requires touching - /// per-project bot configs — the aggregated stream picks it up - /// automatically once this flag is `true` (the default). - /// - /// Defaults to `true`. - #[serde(default = "default_aggregated_notifications_enabled")] - pub aggregated_notifications_enabled: bool, - /// Duration in milliseconds of the chat dispatcher's coalesce window. /// /// Messages for the same session arriving within this window are diff --git a/server/src/gateway/mod.rs b/server/src/gateway/mod.rs index a219f734..d78fd887 100644 --- a/server/src/gateway/mod.rs +++ b/server/src/gateway/mod.rs @@ -18,8 +18,7 @@ use std::sync::Arc; pub use crate::service::gateway::{ GatewayConfig, GatewayState as GatewayStateType, GatewayStatusEvent, ProjectEntry, broadcast_status_event, fetch_all_project_pipeline_statuses, format_aggregate_status_compact, - spawn_gateway_broadcaster_forwarder, spawn_gateway_notification_poller, - subscribe_status_events, + spawn_gateway_broadcaster_forwarder, subscribe_status_events, }; /// Build the complete gateway route tree. diff --git a/server/src/gateway/tests.rs b/server/src/gateway/tests.rs index 7673b5a1..30732a45 100644 --- a/server/src/gateway/tests.rs +++ b/server/src/gateway/tests.rs @@ -96,219 +96,6 @@ async fn generate_token_creates_pending_token() { assert!(tokens.contains_key(token)); } -// ── Notification poller integration tests ──────────────────────────── - -#[tokio::test] -async fn gateway_notification_poller_continues_when_one_project_unreachable() { - use crate::chat::{ChatTransport, MessageId}; - use crate::service::events::StoredEvent; - use async_trait::async_trait; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - type CallLog = Arc>>; - - struct MockTransport { - calls: CallLog, - } - - #[async_trait] - impl ChatTransport for MockTransport { - async fn send_message( - &self, - _room_id: &str, - plain: &str, - _html: &str, - ) -> Result { - self.calls.lock().unwrap().push(plain.to_string()); - Ok("id".to_string()) - } - - async fn edit_message( - &self, - _room_id: &str, - _id: &str, - _plain: &str, - _html: &str, - ) -> Result<(), String> { - Ok(()) - } - - async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { - Ok(()) - } - } - - let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new())); - let transport = Arc::new(MockTransport { - calls: Arc::clone(&calls), - }); - - let event = vec![StoredEvent::StoryBlocked { - story_id: "10_story_ok".to_string(), - story_name: String::new(), - reason: "retry limit".to_string(), - timestamp_ms: 500, - }]; - let event_body = serde_json::to_vec(&event).unwrap(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let good_port = listener.local_addr().unwrap().port(); - let good_url = format!("http://127.0.0.1:{good_port}"); - tokio::spawn(async move { - for _ in 0..4 { - if let Ok((mut stream, _)) = listener.accept().await { - let mut buf = vec![0u8; 4096]; - let _ = stream.read(&mut buf).await; - let header = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - event_body.len() - ); - let _ = stream.write_all(header.as_bytes()).await; - let _ = stream.write_all(&event_body).await; - } - } - }); - - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - - let bad_url = "http://127.0.0.1:1".to_string(); - - let mut project_urls = BTreeMap::new(); - project_urls.insert("good-project".to_string(), good_url); - project_urls.insert("unreachable-project".to_string(), bad_url); - - gateway::spawn_gateway_notification_poller( - transport as Arc, - vec!["!room:example.org".to_string()], - project_urls, - 1, - ); - - tokio::time::sleep(std::time::Duration::from_millis(1500)).await; - - let messages = calls.lock().unwrap(); - assert!( - !messages.is_empty(), - "Expected notifications from the reachable project; got none" - ); - let has_good = messages - .iter() - .any(|m| m.contains("[good-project]") && m.contains("#10")); - assert!( - has_good, - "Expected a notification from [good-project]; got: {messages:?}" - ); - let has_bad = messages.iter().any(|m| m.contains("[unreachable-project]")); - assert!( - !has_bad, - "Unreachable project must not produce notifications; got: {messages:?}" - ); -} - -#[tokio::test] -async fn gateway_notification_poller_sends_only_to_configured_gateway_rooms() { - use crate::chat::{ChatTransport, MessageId}; - use crate::service::events::StoredEvent; - use async_trait::async_trait; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - type RoomLog = Arc>>; - - struct RoomCapture { - rooms: RoomLog, - } - - #[async_trait] - impl ChatTransport for RoomCapture { - async fn send_message( - &self, - room_id: &str, - _plain: &str, - _html: &str, - ) -> Result { - self.rooms.lock().unwrap().push(room_id.to_string()); - Ok("id".to_string()) - } - - async fn edit_message( - &self, - _room_id: &str, - _id: &str, - _plain: &str, - _html: &str, - ) -> Result<(), String> { - Ok(()) - } - - async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { - Ok(()) - } - } - - let rooms: RoomLog = Arc::new(std::sync::Mutex::new(Vec::new())); - let transport = Arc::new(RoomCapture { - rooms: Arc::clone(&rooms), - }); - - let event = vec![StoredEvent::MergeFailure { - story_id: "5_story_x".to_string(), - story_name: String::new(), - reason: "conflict".to_string(), - timestamp_ms: 300, - }]; - let event_body = serde_json::to_vec(&event).unwrap(); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = listener.local_addr().unwrap().port(); - let url = format!("http://127.0.0.1:{port}"); - tokio::spawn(async move { - for _ in 0..4 { - if let Ok((mut stream, _)) = listener.accept().await { - let mut buf = vec![0u8; 4096]; - let _ = stream.read(&mut buf).await; - let header = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - event_body.len() - ); - let _ = stream.write_all(header.as_bytes()).await; - let _ = stream.write_all(&event_body).await; - } - } - }); - - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - - const GATEWAY_ROOM: &str = "!gateway:example.org"; - #[allow(dead_code)] - const PER_PROJECT_ROOM: &str = "!project:example.org"; - - let mut project_urls = BTreeMap::new(); - project_urls.insert("myproj".to_string(), url); - - gateway::spawn_gateway_notification_poller( - transport as Arc, - vec![GATEWAY_ROOM.to_string()], - project_urls, - 1, - ); - - tokio::time::sleep(std::time::Duration::from_millis(1500)).await; - - let room_calls = rooms.lock().unwrap(); - assert!( - !room_calls.is_empty(), - "Expected at least one notification; got none" - ); - for room in room_calls.iter() { - assert_eq!( - room, GATEWAY_ROOM, - "Notification must only go to the gateway room, not {room}" - ); - } - assert!( - !room_calls.iter().any(|r| r == PER_PROJECT_ROOM), - "Per-project room must not receive gateway aggregated notifications" - ); -} - // ── init_project integration tests ────────────────────────────────── #[tokio::test] @@ -413,42 +200,25 @@ async fn init_project_duplicate_name_returns_error() { assert!(result.is_err()); } +/// story 1180: MCP proxying is now WS-uplink-only, so this test registers a +/// mock sled connection (rather than a mockito/raw-TCP HTTP mock) to verify +/// `init_project` scaffolding followed by an MCP `tools/call` for +/// `wizard_status` routed over the live uplink. #[tokio::test] async fn init_project_then_wizard_status_integration() { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let mock_port = listener.local_addr().unwrap().port(); - let mock_url = format!("http://127.0.0.1:{mock_port}"); - - tokio::spawn(async move { - if let Ok((mut stream, _)) = listener.accept().await { - let mut buf = vec![0u8; 4096]; - let _ = stream.read(&mut buf).await; - let body = serde_json::json!({ - "jsonrpc": "2.0", - "id": 1, - "result": { - "content": [{ - "type": "text", - "text": "{\"steps\":[{\"id\":\"scaffold\",\"title\":\"Scaffold\",\"status\":\"confirmed\"}],\"completed\":false}" - }] - } - }); - let body_bytes = serde_json::to_vec(&body).unwrap(); - let header = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - body_bytes.len() - ); - let _ = stream.write_all(header.as_bytes()).await; - let _ = stream.write_all(&body_bytes).await; - } - }); - - tokio::time::sleep(std::time::Duration::from_millis(10)).await; + use crate::service::gateway::ProjectEntry; let mut projects = BTreeMap::new(); - projects.insert("mock-project".into(), ProjectEntry::with_url(mock_url)); + projects.insert( + "mock-project".into(), + ProjectEntry { + url: None, + auth_token: Some("secret".into()), + ssh_port: None, + host_path: None, + expected_node_id: None, + }, + ); let config = GatewayConfig { projects, sled_tokens: BTreeMap::new(), @@ -457,6 +227,18 @@ async fn init_project_then_wizard_status_integration() { let config_dir = tempfile::tempdir().unwrap(); let state = Arc::new(GatewayState::new(config, config_dir.path().to_path_buf(), 3000).unwrap()); + let conn = spawn_mock_sled(|_body| { + serde_json::json!({ + "content": [{ + "type": "text", + "text": "{\"steps\":[{\"id\":\"scaffold\",\"title\":\"Scaffold\",\"status\":\"confirmed\"}],\"completed\":false}" + }] + }) + }); + state + .register_sled_connection("mock-project".to_string(), conn) + .await; + let project_dir = tempfile::tempdir().unwrap(); let result = gateway::init_project(&state, project_dir.path().to_str().unwrap(), None, None).await; @@ -466,8 +248,7 @@ async fn init_project_then_wizard_status_integration() { let wizard_path = project_dir.path().join(".huskies/wizard_state.json"); assert!(wizard_path.exists()); - // Proxy call to the mock server. - let active_url = state.active_url().await.unwrap(); + // Proxy call over the sled-uplink WS. let proxy_body = serde_json::to_vec(&serde_json::json!({ "jsonrpc": "2.0", "id": 2, @@ -475,7 +256,7 @@ async fn init_project_then_wizard_status_integration() { "params": { "name": "wizard_status", "arguments": {} } })) .unwrap(); - let proxy_resp = gateway::io::proxy_mcp_call(&state.client, &active_url, &proxy_body).await; + let proxy_resp = state.proxy_active_mcp(&proxy_body).await; assert!(proxy_resp.is_ok()); let resp_json: serde_json::Value = serde_json::from_slice(&proxy_resp.unwrap()).unwrap(); @@ -560,155 +341,6 @@ async fn aggregate_pipeline_status_integration_healthy_and_unreachable() { assert!(broken.get("error").is_some()); } -// ── Multi-project notification poller integration ──────────────────── - -#[tokio::test] -async fn gateway_notification_poller_delivers_events_from_two_projects_with_project_tags() { - use crate::chat::{ChatTransport, MessageId}; - use crate::service::events::StoredEvent; - use async_trait::async_trait; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - type CallLog = Arc>>; - - struct MockTransport { - calls: CallLog, - } - - #[async_trait] - impl ChatTransport for MockTransport { - async fn send_message( - &self, - room_id: &str, - plain: &str, - html: &str, - ) -> Result { - self.calls.lock().unwrap().push(( - room_id.to_string(), - plain.to_string(), - html.to_string(), - )); - Ok("mock-id".to_string()) - } - - async fn edit_message( - &self, - _room_id: &str, - _id: &str, - _plain: &str, - _html: &str, - ) -> Result<(), String> { - Ok(()) - } - - async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { - Ok(()) - } - } - - let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new())); - let transport = Arc::new(MockTransport { - calls: Arc::clone(&calls), - }); - - let alpha_events = vec![StoredEvent::StageTransition { - story_id: "1_story_alpha".to_string(), - story_name: String::new(), - from_stage: "2_current".to_string(), - to_stage: "3_qa".to_string(), - timestamp_ms: 100, - }]; - let alpha_body = serde_json::to_vec(&alpha_events).unwrap(); - let alpha_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let alpha_port = alpha_listener.local_addr().unwrap().port(); - let alpha_url = format!("http://127.0.0.1:{alpha_port}"); - tokio::spawn(async move { - for _ in 0..4 { - if let Ok((mut stream, _)) = alpha_listener.accept().await { - let mut buf = vec![0u8; 4096]; - let _ = stream.read(&mut buf).await; - let header = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - alpha_body.len() - ); - let _ = stream.write_all(header.as_bytes()).await; - let _ = stream.write_all(&alpha_body).await; - } - } - }); - - let beta_events = vec![StoredEvent::MergeFailure { - story_id: "2_story_beta".to_string(), - story_name: String::new(), - reason: "merge conflict in lib.rs".to_string(), - timestamp_ms: 200, - }]; - let beta_body = serde_json::to_vec(&beta_events).unwrap(); - let beta_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let beta_port = beta_listener.local_addr().unwrap().port(); - let beta_url = format!("http://127.0.0.1:{beta_port}"); - tokio::spawn(async move { - for _ in 0..4 { - if let Ok((mut stream, _)) = beta_listener.accept().await { - let mut buf = vec![0u8; 4096]; - let _ = stream.read(&mut buf).await; - let header = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - beta_body.len() - ); - let _ = stream.write_all(header.as_bytes()).await; - let _ = stream.write_all(&beta_body).await; - } - } - }); - - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - - let mut project_urls = BTreeMap::new(); - project_urls.insert("alpha".to_string(), alpha_url); - project_urls.insert("beta".to_string(), beta_url); - - gateway::spawn_gateway_notification_poller( - transport as Arc, - vec!["!room:example.org".to_string()], - project_urls, - 1, - ); - - tokio::time::sleep(std::time::Duration::from_millis(1500)).await; - - let calls = calls.lock().unwrap(); - assert!( - !calls.is_empty(), - "Expected at least one notification; got none" - ); - - let plains: Vec<&str> = calls.iter().map(|(_, p, _)| p.as_str()).collect(); - - let alpha_notification = plains - .iter() - .any(|p| p.contains("[alpha]") && p.contains("1")); - let beta_notification = plains - .iter() - .any(|p| p.contains("[beta]") && p.contains("merge conflict")); - - assert!( - alpha_notification, - "Expected a notification from [alpha] containing story ID '1'; got: {plains:?}" - ); - assert!( - beta_notification, - "Expected a notification from [beta] containing 'merge conflict'; got: {plains:?}" - ); - - for (room_id, _, _) in calls.iter() { - assert_eq!( - room_id, "!room:example.org", - "All notifications must go to the gateway room" - ); - } -} - // ── Gateway broadcaster forwarder tests ───────────────────────────── #[tokio::test] @@ -1046,26 +678,26 @@ async fn gateway_mcp_sse_proxy_streams_progress_and_final_response() { ); } +/// Non-SSE `tools/call` requests must be routed over the live sled-uplink WS +/// (story 1180: the gateway no longer falls back to HTTP for MCP proxying) +/// and return a plain `application/json` body. #[tokio::test] async fn gateway_mcp_post_without_sse_returns_plain_json() { - let mut mock_sled = mockito::Server::new_async().await; - - let json_resp = serde_json::json!({ - "jsonrpc": "2.0", - "id": 2, - "result": { "content": [{ "type": "text", "text": "done" }] } - }); - - let _mock = mock_sled - .mock("POST", "/mcp") - .with_status(200) - .with_header("content-type", "application/json") - .with_body(serde_json::to_string(&json_resp).unwrap()) - .create_async() - .await; + use crate::service::gateway::ProjectEntry; + // WS-only project entry — no URL, proving the response comes from the + // sled-uplink connection and not an HTTP fallback. let mut projects = BTreeMap::new(); - projects.insert("sled".to_string(), ProjectEntry::with_url(mock_sled.url())); + projects.insert( + "sled".to_string(), + ProjectEntry { + url: None, + auth_token: Some("secret".into()), + ssh_port: None, + host_path: None, + expected_node_id: None, + }, + ); let config = GatewayConfig { projects, sled_tokens: BTreeMap::new(), @@ -1073,6 +705,13 @@ async fn gateway_mcp_post_without_sse_returns_plain_json() { }; let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap()); + let conn = spawn_mock_sled( + |_body| serde_json::json!({ "content": [{ "type": "text", "text": "done" }] }), + ); + state + .register_sled_connection("sled".to_string(), conn) + .await; + let app = poem::Route::new() .at("/mcp", poem::post(gateway_mcp_post_handler)) .data(state.clone()); @@ -1106,9 +745,122 @@ async fn gateway_mcp_post_without_sse_returns_plain_json() { let body: serde_json::Value = resp.0.into_body().into_json().await.unwrap(); assert_eq!(body["id"], 2); + assert_eq!( + body["result"]["content"][0]["text"], "done", + "Expected result in plain JSON response, routed over the sled-uplink WS" + ); +} + +/// story 1180 AC1/AC3/AC4: when a project has no live sled-uplink connection, +/// `proxy_active_mcp` must fail fast with an actionable error naming the +/// sled — no HTTP fallback attempt, no hang. +#[tokio::test] +async fn proxy_active_mcp_with_no_live_connection_fails_fast_naming_sled() { + use crate::service::gateway::ProjectEntry; + + let mut projects = BTreeMap::new(); + projects.insert( + "offline-sled".to_string(), + ProjectEntry { + url: None, + auth_token: Some("secret".into()), + ssh_port: None, + host_path: None, + expected_node_id: None, + }, + ); + let config = GatewayConfig { + projects, + sled_tokens: BTreeMap::new(), + release_channels: BTreeMap::new(), + }; + let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap()); + + // No sled connection is ever registered — the uplink is down. + let body = serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/list", + "params": {} + })) + .unwrap(); + + let err = tokio::time::timeout( + std::time::Duration::from_secs(2), + state.proxy_active_mcp(&body), + ) + .await + .expect("proxy_active_mcp must fail fast, not hang, when the sled is disconnected") + .expect_err("must return an error when no live uplink connection exists"); + assert!( - body.get("result").is_some(), - "Expected result in plain JSON response" + err.contains("offline-sled"), + "error must name the disconnected sled; got: {err}" + ); + assert!( + !err.to_lowercase().contains("http"), + "error must not mention HTTP fallback; got: {err}" + ); +} + +/// Bot chat commands (e.g. `tools/call` proxied through the gateway MCP HTTP +/// route) against a disconnected sled must surface an immediate JSON-RPC +/// error response rather than hanging — story 1180 AC4. +#[tokio::test] +async fn gateway_mcp_post_against_disconnected_sled_returns_error_response_fast() { + use crate::service::gateway::ProjectEntry; + + let mut projects = BTreeMap::new(); + projects.insert( + "offline-sled".to_string(), + ProjectEntry { + url: None, + auth_token: Some("secret".into()), + ssh_port: None, + host_path: None, + expected_node_id: None, + }, + ); + let config = GatewayConfig { + projects, + sled_tokens: BTreeMap::new(), + release_channels: BTreeMap::new(), + }; + let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap()); + + let app = poem::Route::new() + .at("/mcp", poem::post(gateway_mcp_post_handler)) + .data(state.clone()); + let cli = poem::test::TestClient::new(app); + + let rpc_body = serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { "name": "get_pipeline_status", "arguments": {} } + })) + .unwrap(); + + let resp = tokio::time::timeout( + std::time::Duration::from_secs(2), + cli.post("/mcp") + .header("content-type", "application/json") + .body(rpc_body) + .send(), + ) + .await + .expect("request against a disconnected sled must fail fast, not hang"); + + let body: serde_json::Value = resp.0.into_body().into_json().await.unwrap(); + assert_eq!(body["id"], 3); + assert!( + body.get("error").is_some(), + "Expected a JSON-RPC error for a disconnected sled; got: {body}" + ); + let msg = body["error"]["message"].as_str().unwrap_or(""); + assert!( + msg.contains("offline-sled"), + "error message must name the sled; got: {msg}" ); } diff --git a/server/src/http/events.rs b/server/src/http/events.rs deleted file mode 100644 index 4b3aca06..00000000 --- a/server/src/http/events.rs +++ /dev/null @@ -1,202 +0,0 @@ -//! Per-project event buffer and `GET /api/events` HTTP endpoint. -//! -//! The gateway polls `/api/events?since={ts_ms}` on each registered project -//! server to aggregate cross-project pipeline notifications into a single -//! gateway chat channel. Each project server buffers up to 500 events in -//! memory and serves them via this endpoint. -//! -//! Domain logic lives in `service::events`; this module is a thin HTTP -//! adapter: extract query params → call service → shape response. - -#[cfg(test)] -pub use crate::service::events::StoredEvent; -pub use crate::service::events::{EventBuffer, subscribe_to_watcher}; -// MAX_BUFFER_SIZE is used in tests via `use super::*`. -#[cfg(test)] -pub use crate::service::events::MAX_BUFFER_SIZE; - -use poem::web::{Data, Query}; -use poem::{Response, handler, http::StatusCode}; -use serde::Deserialize; - -/// Query parameters for `GET /api/events`. -#[derive(Deserialize)] -pub struct EventsQuery { - /// Return only events with `timestamp_ms` strictly greater than this value. - /// Defaults to `0` (return all buffered events). - #[serde(default)] - pub since: u64, -} - -/// `GET /api/events?since={ts_ms}` -/// -/// Returns a JSON array of [`StoredEvent`] objects recorded after `since` ms. -/// The gateway polls this endpoint on each registered project server to build -/// an aggregated cross-project notification stream. -#[handler] -pub fn events_handler( - Query(params): Query, - Data(buffer): Data<&EventBuffer>, -) -> Response { - let events = crate::service::events::events_since(buffer, params.since); - let body = serde_json::to_vec(&events).unwrap_or_else(|_| b"[]".to_vec()); - Response::builder() - .status(StatusCode::OK) - .header(poem::http::header::CONTENT_TYPE, "application/json") - .body(body) -} - -#[cfg(test)] -mod tests { - use super::*; - use tokio::sync::broadcast; - - #[test] - fn event_buffer_push_and_retrieve() { - let buf = EventBuffer::new(); - buf.push(StoredEvent::MergeFailure { - story_id: "42_story_x".to_string(), - story_name: String::new(), - reason: "conflict".to_string(), - timestamp_ms: 1000, - }); - buf.push(StoredEvent::StoryBlocked { - story_id: "43_story_y".to_string(), - story_name: String::new(), - reason: "retry limit".to_string(), - timestamp_ms: 2000, - }); - - let all = buf.events_since(0); - assert_eq!(all.len(), 2); - - let after_1000 = buf.events_since(1000); - assert_eq!(after_1000.len(), 1); - assert!(matches!(after_1000[0], StoredEvent::StoryBlocked { .. })); - } - - #[test] - fn event_buffer_evicts_oldest_when_full() { - let buf = EventBuffer::new(); - for i in 0..MAX_BUFFER_SIZE + 1 { - buf.push(StoredEvent::MergeFailure { - story_id: format!("{i}_story_x"), - story_name: String::new(), - reason: "x".to_string(), - timestamp_ms: i as u64, - }); - } - // Buffer must not exceed MAX_BUFFER_SIZE. - assert_eq!(buf.events_since(0).len(), MAX_BUFFER_SIZE); - // Oldest entry (timestamp_ms == 0) should have been evicted. - assert!(buf.events_since(0).iter().all(|e| e.timestamp_ms() > 0)); - } - - #[test] - fn stage_transition_timestamp_ms_accessor() { - let e = StoredEvent::StageTransition { - story_id: "1".to_string(), - story_name: String::new(), - from_stage: "2_current".to_string(), - to_stage: "3_qa".to_string(), - timestamp_ms: 9999, - }; - assert_eq!(e.timestamp_ms(), 9999); - } - - #[tokio::test] - async fn subscribe_to_watcher_stores_work_item_with_from_stage() { - let buf = EventBuffer::new(); - let (tx, rx) = broadcast::channel(16); - - subscribe_to_watcher(buf.clone(), rx); - - tx.send(crate::io::watcher::WatcherEvent::WorkItem { - stage: "3_qa".to_string(), - item_id: "42_story_foo".to_string(), - action: "qa".to_string(), - commit_msg: "huskies: qa 42_story_foo".to_string(), - from_stage: Some("2_current".to_string()), - }) - .unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - - let events = buf.events_since(0); - assert_eq!(events.len(), 1); - assert!(matches!(events[0], StoredEvent::StageTransition { .. })); - if let StoredEvent::StageTransition { - ref story_id, - ref from_stage, - ref to_stage, - .. - } = events[0] - { - assert_eq!(story_id, "42_story_foo"); - assert_eq!(from_stage, "2_current"); - assert_eq!(to_stage, "3_qa"); - } - } - - #[tokio::test] - async fn subscribe_to_watcher_ignores_work_item_without_from_stage() { - let buf = EventBuffer::new(); - let (tx, rx) = broadcast::channel(16); - - subscribe_to_watcher(buf.clone(), rx); - - // Synthetic event: no from_stage. - tx.send(crate::io::watcher::WatcherEvent::WorkItem { - stage: "2_current".to_string(), - item_id: "99_story_syn".to_string(), - action: "start".to_string(), - commit_msg: "huskies: start 99_story_syn".to_string(), - from_stage: None, - }) - .unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - - assert_eq!(buf.events_since(0).len(), 0); - } - - #[tokio::test] - async fn subscribe_to_watcher_stores_merge_failure() { - let buf = EventBuffer::new(); - let (tx, rx) = broadcast::channel(16); - - subscribe_to_watcher(buf.clone(), rx); - - tx.send(crate::io::watcher::WatcherEvent::MergeFailure { - story_id: "42_story_foo".to_string(), - reason: "merge conflict".to_string(), - }) - .unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - - let events = buf.events_since(0); - assert_eq!(events.len(), 1); - assert!(matches!(events[0], StoredEvent::MergeFailure { .. })); - } - - #[tokio::test] - async fn subscribe_to_watcher_stores_story_blocked() { - let buf = EventBuffer::new(); - let (tx, rx) = broadcast::channel(16); - - subscribe_to_watcher(buf.clone(), rx); - - tx.send(crate::io::watcher::WatcherEvent::StoryBlocked { - story_id: "43_story_bar".to_string(), - reason: "retry limit exceeded".to_string(), - }) - .unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - - let events = buf.events_since(0); - assert_eq!(events.len(), 1); - assert!(matches!(events[0], StoredEvent::StoryBlocked { .. })); - } -} diff --git a/server/src/http/mod.rs b/server/src/http/mod.rs index 513d8cca..7290008d 100644 --- a/server/src/http/mod.rs +++ b/server/src/http/mod.rs @@ -3,8 +3,6 @@ pub mod agents_sse; /// Shared application context threaded through handlers. pub mod context; -/// Server-sent event stream for pipeline/watcher events. -pub mod events; /// Node identity endpoint (public key, node ID). pub mod identity; /// Model Context Protocol (MCP) HTTP endpoint and tool modules. @@ -86,7 +84,6 @@ pub fn build_routes( whatsapp_ctx: Option>, slack_ctx: Option>, port: u16, - event_buffer: Option, ) -> impl poem::Endpoint { let ctx_arc = std::sync::Arc::new(ctx); @@ -118,10 +115,6 @@ pub fn build_routes( .at("/oauth/status", get(oauth::oauth_status)) .at("/debug/crdt", get(debug_crdt_handler)); - if let Some(buf) = event_buffer { - route = route.at("/api/events", get(events::events_handler).data(buf)); - } - route = route .at("/api/upgrade", post(upgrade_trigger_handler)) .at("/api/artifacts/:filename", get(serve_artifact_handler)); @@ -388,7 +381,7 @@ mod tests { fn build_routes_constructs_without_panic() { let tmp = tempfile::tempdir().unwrap(); let ctx = context::AppContext::new_test(tmp.path().to_path_buf()); - let _endpoint = build_routes(ctx, None, None, 3001, None); + let _endpoint = build_routes(ctx, None, None, 3001); } #[test] @@ -407,7 +400,7 @@ mod tests { async fn artifact_endpoint_rejects_traversal_and_misses() { let tmp = tempfile::tempdir().unwrap(); let ctx = context::AppContext::new_test(tmp.path().to_path_buf()); - let app = build_routes(ctx, None, None, 3001, None); + let app = build_routes(ctx, None, None, 3001); let cli = poem::test::TestClient::new(app); let resp = cli.get("/api/artifacts/..").send().await; @@ -424,7 +417,7 @@ mod tests { async fn version_endpoint_reports_version_and_git_hash() { let tmp = tempfile::tempdir().unwrap(); let ctx = context::AppContext::new_test(tmp.path().to_path_buf()); - let app = build_routes(ctx, None, None, 3001, None); + let app = build_routes(ctx, None, None, 3001); let cli = poem::test::TestClient::new(app); let resp = cli.get("/api/version").send().await; @@ -447,14 +440,6 @@ mod tests { // ensuring the port parameter flows through to OAuthState. let tmp = tempfile::tempdir().unwrap(); let ctx = context::AppContext::new_test(tmp.path().to_path_buf()); - let _endpoint = build_routes(ctx, None, None, 9999, None); - } - - #[test] - fn build_routes_with_event_buffer_constructs_without_panic() { - let tmp = tempfile::tempdir().unwrap(); - let ctx = context::AppContext::new_test(tmp.path().to_path_buf()); - let buf = events::EventBuffer::new(); - let _endpoint = build_routes(ctx, None, None, 3001, Some(buf)); + let _endpoint = build_routes(ctx, None, None, 9999); } } diff --git a/server/src/main.rs b/server/src/main.rs index 76c033cf..cecd31fa 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -222,7 +222,6 @@ async fn main() -> Result<(), std::io::Error> { let watcher_rx_for_whatsapp = watcher_tx.subscribe(); let watcher_rx_for_slack = watcher_tx.subscribe(); let watcher_rx_for_discord = watcher_tx.subscribe(); - let watcher_rx_for_events = watcher_tx.subscribe(); let permission_registry = service::permission_router::ResponderRegistry::new(); service::permission_router::spawn_permission_router(perm_rx, Arc::clone(&permission_registry)); @@ -365,10 +364,6 @@ async fn main() -> Result<(), std::io::Error> { event_trigger_store, }; - // Per-project event buffer for the gateway's `/api/events` poller. - let event_buffer = crate::http::events::EventBuffer::new(); - crate::http::events::subscribe_to_watcher(event_buffer.clone(), watcher_rx_for_events); - // Gateway relay task (pushes StatusEvents to a configured gateway). startup::tick_loop::spawn_gateway_relay(&startup_root, Arc::clone(&services.status)); @@ -380,7 +375,6 @@ async fn main() -> Result<(), std::io::Error> { bot_ctxs.whatsapp_ctx.clone(), bot_ctxs.slack_ctx.clone(), port, - Some(event_buffer), ); // Permanent liveness heartbeat — stops when the tokio runtime freezes, diff --git a/server/src/service/events/buffer.rs b/server/src/service/events/buffer.rs index 120600fa..934d8f38 100644 --- a/server/src/service/events/buffer.rs +++ b/server/src/service/events/buffer.rs @@ -1,16 +1,12 @@ -//! Pure event-buffer types — no side effects. +//! Pure event types — no side effects. //! -//! `StoredEvent` and `EventBuffer` contain only data-transformation and -//! structural logic; all I/O (clocks, spawned tasks) lives in `io.rs`. +//! `StoredEvent` is the wire format shared by the gateway's WS event-push +//! relay (`gateway_relay.rs`, `http/gateway/websocket.rs`) — data-transformation +//! only; all I/O lives in `io.rs`. use serde::{Deserialize, Serialize}; -use std::collections::VecDeque; -use std::sync::{Arc, Mutex}; -/// Maximum number of events retained in the in-memory buffer. -pub const MAX_BUFFER_SIZE: usize = 500; - -/// A pipeline event stored in the event buffer with a timestamp. +/// A pipeline event pushed to the gateway over the WS event-push relay. #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum StoredEvent { @@ -62,88 +58,10 @@ impl StoredEvent { } } -/// Shared, thread-safe ring buffer of recent pipeline events. -/// -/// Wrapped in `Arc` so it can be shared between the background subscriber -/// task and the HTTP handler. The inner `Mutex` guards the `VecDeque`. -#[derive(Clone, Debug)] -pub struct EventBuffer(Arc>>); - -impl EventBuffer { - /// Create a new, empty event buffer. - pub fn new() -> Self { - EventBuffer(Arc::new(Mutex::new(VecDeque::new()))) - } - - /// Append an event to the buffer, evicting the oldest entry if the buffer - /// exceeds [`MAX_BUFFER_SIZE`]. - pub fn push(&self, event: StoredEvent) { - let mut buf = self.0.lock().unwrap(); - if buf.len() >= MAX_BUFFER_SIZE { - buf.pop_front(); - } - buf.push_back(event); - } - - /// Return all events whose `timestamp_ms` is strictly greater than `since_ms`. - pub fn events_since(&self, since_ms: u64) -> Vec { - let buf = self.0.lock().unwrap(); - buf.iter() - .filter(|e| e.timestamp_ms() > since_ms) - .cloned() - .collect() - } -} - -impl Default for EventBuffer { - fn default() -> Self { - Self::new() - } -} - #[cfg(test)] mod tests { use super::*; - #[test] - fn push_and_retrieve_events() { - let buf = EventBuffer::new(); - buf.push(StoredEvent::MergeFailure { - story_id: "42_story_x".to_string(), - story_name: String::new(), - reason: "conflict".to_string(), - timestamp_ms: 1000, - }); - buf.push(StoredEvent::StoryBlocked { - story_id: "43_story_y".to_string(), - story_name: String::new(), - reason: "retry limit".to_string(), - timestamp_ms: 2000, - }); - - let all = buf.events_since(0); - assert_eq!(all.len(), 2); - - let after_1000 = buf.events_since(1000); - assert_eq!(after_1000.len(), 1); - assert!(matches!(after_1000[0], StoredEvent::StoryBlocked { .. })); - } - - #[test] - fn evicts_oldest_when_full() { - let buf = EventBuffer::new(); - for i in 0..MAX_BUFFER_SIZE + 1 { - buf.push(StoredEvent::MergeFailure { - story_id: format!("{i}_story_x"), - story_name: String::new(), - reason: "x".to_string(), - timestamp_ms: i as u64, - }); - } - assert_eq!(buf.events_since(0).len(), MAX_BUFFER_SIZE); - assert!(buf.events_since(0).iter().all(|e| e.timestamp_ms() > 0)); - } - #[test] fn timestamp_ms_accessor_for_all_variants() { let variants = [ @@ -171,27 +89,4 @@ mod tests { assert_eq!(variants[1].timestamp_ms(), 200); assert_eq!(variants[2].timestamp_ms(), 300); } - - #[test] - fn events_since_filters_by_timestamp() { - let buf = EventBuffer::new(); - for ts in [100u64, 200, 300] { - buf.push(StoredEvent::MergeFailure { - story_id: "x".to_string(), - story_name: String::new(), - reason: "r".to_string(), - timestamp_ms: ts, - }); - } - // strictly greater than 100 - let result = buf.events_since(100); - assert_eq!(result.len(), 2); - assert!(result.iter().all(|e| e.timestamp_ms() > 100)); - } - - #[test] - fn default_creates_empty_buffer() { - let buf = EventBuffer::default(); - assert_eq!(buf.events_since(0).len(), 0); - } } diff --git a/server/src/service/events/io.rs b/server/src/service/events/io.rs deleted file mode 100644 index a4cf1dde..00000000 --- a/server/src/service/events/io.rs +++ /dev/null @@ -1,80 +0,0 @@ -//! Events I/O wrappers — the ONLY place in `service/events/` that may perform -//! side effects such as reading the system clock or spawning async tasks. - -use crate::io::watcher::WatcherEvent; -use tokio::sync::broadcast; - -use super::buffer::{EventBuffer, StoredEvent}; - -/// Returns the current Unix timestamp in milliseconds. -pub(super) fn now_ms() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0) -} - -/// Spawn a background task that consumes [`WatcherEvent`] broadcasts and -/// stores relevant events in `buffer`. -/// -/// Only [`WatcherEvent::WorkItem`] (with a known `from_stage`), -/// [`WatcherEvent::MergeFailure`], and [`WatcherEvent::StoryBlocked`] -/// variants are stored. All other variants are silently ignored. -pub fn subscribe_to_watcher(buffer: EventBuffer, mut rx: broadcast::Receiver) { - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(WatcherEvent::WorkItem { - stage, - item_id, - from_stage, - .. - }) => { - if let Some(from) = from_stage { - let story_name = lookup_story_name(&item_id); - buffer.push(StoredEvent::StageTransition { - story_id: item_id, - story_name, - from_stage: from, - to_stage: stage, - timestamp_ms: now_ms(), - }); - } - } - Ok(WatcherEvent::MergeFailure { story_id, reason }) => { - let story_name = lookup_story_name(&story_id); - buffer.push(StoredEvent::MergeFailure { - story_id, - story_name, - reason, - timestamp_ms: now_ms(), - }); - } - Ok(WatcherEvent::StoryBlocked { story_id, reason }) => { - let story_name = lookup_story_name(&story_id); - buffer.push(StoredEvent::StoryBlocked { - story_id, - story_name, - reason, - timestamp_ms: now_ms(), - }); - } - Ok(_) => {} - Err(broadcast::error::RecvError::Lagged(n)) => { - crate::slog!("[events] Subscriber lagged, skipped {n} events"); - } - Err(broadcast::error::RecvError::Closed) => { - crate::slog!("[events] Watcher channel closed; stopping event subscriber"); - break; - } - } - } - }); -} - -/// Look up the human-readable story name from the CRDT, or empty when absent. -fn lookup_story_name(story_id: &str) -> String { - crate::crdt_state::read_item(story_id) - .map(|view| view.name().to_string()) - .unwrap_or_default() -} diff --git a/server/src/service/events/mod.rs b/server/src/service/events/mod.rs index b7c2de45..7c53f203 100644 --- a/server/src/service/events/mod.rs +++ b/server/src/service/events/mod.rs @@ -1,28 +1,23 @@ //! Events service — public API for the events domain. //! -//! This module re-exports the pure buffer types from `buffer.rs` and the -//! side-effectful watcher subscription from `io.rs`. HTTP handlers call -//! these exports instead of containing the logic inline. +//! Re-exports [`StoredEvent`], the wire format shared by the gateway's +//! WS-based event-push relay (`gateway_relay.rs` on the project side, +//! `http/gateway/websocket.rs`'s event-push handler on the gateway side). //! //! Conventions: `docs/architecture/service-modules.md` -/// Bounded in-memory event ring buffer for SSE streaming. +/// Pure event types — no side effects. pub mod buffer; -pub(super) mod io; -pub use buffer::{EventBuffer, StoredEvent}; -// Re-exported for tests (http::events uses it via `use super::*`). -#[allow(unused_imports)] -pub use buffer::MAX_BUFFER_SIZE; -pub use io::subscribe_to_watcher; +pub use buffer::StoredEvent; // ── Error type ──────────────────────────────────────────────────────────────── /// Typed errors returned by `service::events` functions. /// -/// Events operations on the in-memory buffer are infallible; this enum -/// exists to satisfy the module convention and to accommodate future -/// error cases (e.g. persistence). +/// Events operations are currently infallible; this enum exists to satisfy +/// the module convention and to accommodate future error cases (e.g. +/// persistence). #[allow(dead_code)] #[derive(Debug)] pub enum Error { @@ -37,10 +32,3 @@ impl std::fmt::Display for Error { } } } - -// ── Public API ──────────────────────────────────────────────────────────────── - -/// Return all events in `buffer` recorded after `since_ms` milliseconds. -pub fn events_since(buffer: &EventBuffer, since_ms: u64) -> Vec { - buffer.events_since(since_ms) -} diff --git a/server/src/service/gateway/io.rs b/server/src/service/gateway/io.rs index a5605f9d..15c885c0 100644 --- a/server/src/service/gateway/io.rs +++ b/server/src/service/gateway/io.rs @@ -7,7 +7,7 @@ use super::config::{GatewayConfig, ProjectEntry}; pub use reqwest::Client; use serde_json::{Value, json}; -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use std::path::Path; // ── Config I/O ─────────────────────────────────────────────────────────────── @@ -239,28 +239,6 @@ pub fn read_installed_manifest( // ── MCP proxy I/O ─────────────────────────────────────────────────────────── -/// Proxy a raw MCP request body to the given project URL. -pub async fn proxy_mcp_call( - client: &Client, - base_url: &str, - request_bytes: &[u8], -) -> Result, String> { - let mcp_url = format!("{}/mcp", base_url.trim_end_matches('/')); - - let resp = client - .post(&mcp_url) - .header("Content-Type", "application/json") - .body(request_bytes.to_vec()) - .send() - .await - .map_err(|e| format!("failed to reach {mcp_url}: {e}"))?; - - resp.bytes() - .await - .map(|b| b.to_vec()) - .map_err(|e| format!("failed to read response from {mcp_url}: {e}")) -} - /// Proxy an MCP `tools/call` request to the sled with `Accept: text/event-stream` /// and return the raw response for streaming. No per-request timeout is applied /// so long-running tool calls (e.g. `run_tests`, up to 1200 s) are not cut short. @@ -563,72 +541,6 @@ pub fn spawn_gateway_broadcaster_forwarder( }); } -/// Spawn a background task that polls events from all project servers. -pub fn spawn_gateway_notification_poller( - transport: std::sync::Arc, - room_ids: Vec, - project_urls: BTreeMap, - poll_interval_secs: u64, -) { - tokio::spawn(async move { - let client = Client::builder() - .timeout(std::time::Duration::from_secs(10)) - .build() - .unwrap_or_else(|_| Client::new()); - let interval = std::time::Duration::from_secs(poll_interval_secs.max(1)); - - let mut last_ts: HashMap = project_urls - .keys() - .map(|name| (name.clone(), 0u64)) - .collect(); - - loop { - for (project_name, base_url) in &project_urls { - let since = last_ts.get(project_name).copied().unwrap_or(0); - let url = format!("{base_url}/api/events?since={since}"); - - let response = match client.get(&url).send().await { - Ok(r) => r, - Err(e) => { - crate::slog!( - "[gateway-poller] {project_name}: unreachable ({e}); skipping" - ); - continue; - } - }; - - let events: Vec = match response.json().await { - Ok(v) => v, - Err(e) => { - crate::slog!( - "[gateway-poller] {project_name}: failed to parse events: {e}" - ); - continue; - } - }; - - for event in &events { - let ts = event.timestamp_ms(); - if ts > *last_ts.get(project_name).unwrap_or(&0) { - last_ts.insert(project_name.clone(), ts); - } - - let (plain, html) = super::polling::format_gateway_event(project_name, event); - for room_id in &room_ids { - if let Err(e) = transport.send_message(room_id, &plain, &html).await { - crate::slog!( - "[gateway-poller] Failed to send notification to {room_id}: {e}" - ); - } - } - } - } - - tokio::time::sleep(interval).await; - } - }); -} - // ── Gateway bot spawn ─────────────────────────────────────────────────────── /// Re-export type alias for the active project lock. diff --git a/server/src/service/gateway/mod.rs b/server/src/service/gateway/mod.rs index a90610c2..84e1f0f3 100644 --- a/server/src/service/gateway/mod.rs +++ b/server/src/service/gateway/mod.rs @@ -24,7 +24,6 @@ pub use config::{GatewayConfig, ProjectEntry}; pub use identity::{IdentityCheck, check_identity}; pub use io::{ fetch_all_project_pipeline_statuses, probe_identity, spawn_gateway_broadcaster_forwarder, - spawn_gateway_notification_poller, }; use crate::http::context::PermissionForward; @@ -364,21 +363,25 @@ impl GatewayState { } } - /// Proxy an MCP request to the active project, preferring the live - /// sled-uplink WebSocket when available (story 899 AC 2) and falling - /// back to HTTP otherwise. + /// Proxy an MCP request to the active project over its live sled-uplink + /// WebSocket (story 899 AC 2). + /// + /// The gateway is WS-only for MCP proxying (story 1180): when no live + /// uplink connection exists for the active project, this returns an + /// immediate, actionable error naming the sled rather than falling back + /// to HTTP. Callers (e.g. bot chat commands) surface this error straight + /// to the user instead of hanging on an unreachable HTTP endpoint. /// /// Returns the raw response body bytes ready to be relayed to the caller. pub async fn proxy_active_mcp(&self, bytes: &[u8]) -> Result, String> { - if let Some(conn) = self.active_sled_connection().await { - return proxy_mcp_via_ws(&conn, bytes).await; + let name = self.active_project.read().await.clone(); + match self.active_sled_connection().await { + Some(conn) => proxy_mcp_via_ws(&conn, bytes).await, + None => Err(format!( + "sled '{name}' has no live WS uplink connection; \ + ensure the sled is running and connected to this gateway" + )), } - let url = self.active_url().await.map_err(|e| e.to_string())?; - crate::slog!( - "[gateway] MCP proxy: WS uplink unavailable, falling back to HTTP \ - (deprecated, will be removed once all sleds are WS-only)" - ); - crate::service::gateway::io::proxy_mcp_call(&self.client, &url, bytes).await } }