huskies/server/src/gateway/tests.rs

//! Integration tests for the gateway route tree and service interactions.

use super::*;
use crate::service::gateway::{GatewayConfig, GatewayState, ProjectEntry};
use std::collections::BTreeMap;
use std::path::PathBuf;

fn make_test_state() -> Arc<GatewayState> {
    let mut projects = BTreeMap::new();
    projects.insert("test".into(), ProjectEntry::with_url("http://test:3001"));
    let config = GatewayConfig {
        projects,
        sled_tokens: BTreeMap::new(),
        release_channels: BTreeMap::new(),
    };
    Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap())
}

#[test]
fn gateway_route_tree_builds_without_panic() {
    let state = make_test_state();
    let _route = build_gateway_route(state);
}

// ── Tests that exercised internal functions have been moved to their
// ── respective service/gateway modules.  The integration tests that use
// ── poem::test::TestClient and mock HTTP servers remain here since they
// ── test the combined HTTP + service interaction through real routes.

#[tokio::test]
async fn crdt_sync_handshake_then_register_writes_crdt_node() {
    crate::crdt_state::init_for_test();
    let state = make_test_state();

    // Generate a valid join token via the tokens endpoint.
    let token_app = poem::Route::new()
        .at(
            "/gateway/tokens",
            poem::post(gateway_generate_token_handler),
        )
        .data(state.clone());
    let cli = poem::test::TestClient::new(token_app);
    let resp = cli.post("/gateway/tokens").send().await;
    assert_eq!(resp.0.status(), poem::http::StatusCode::OK);
    let body: serde_json::Value = resp.0.into_body().into_json().await.unwrap();
    let token = body["token"].as_str().unwrap().to_string();

    // Token must be pending before the upgrade.
    assert!(state.pending_tokens.read().await.contains_key(&token));

    // Call register_agent directly (the service function exercised by the handler).
    let node = crate::service::gateway::register_agent(
        &state,
        &token,
        "test-label".into(),
        "ws://test:9000".into(),
    )
    .await
    .unwrap();

    // Token is consumed after registration.
    assert!(state.pending_tokens.read().await.is_empty());

    // CRDT node was written and is visible via list_agents.
    let agents = crate::service::gateway::list_agents();
    assert!(
        agents.iter().any(|n| n.node_id == node.node_id),
        "Registered node must appear in list_agents"
    );

    // remove_agent tombstones the node.
    assert!(crate::service::gateway::remove_agent(&node.node_id));
    let alive = crate::service::gateway::list_agents();
    assert!(
        !alive.iter().any(|n| n.node_id == node.node_id),
        "Tombstoned node must not appear in list_agents"
    );
}

#[tokio::test]
async fn generate_token_creates_pending_token() {
    let state = make_test_state();
    let app = poem::Route::new()
        .at(
            "/gateway/tokens",
            poem::post(gateway_generate_token_handler),
        )
        .data(state.clone());
    let cli = poem::test::TestClient::new(app);
    let resp = cli.post("/gateway/tokens").send().await;
    assert_eq!(resp.0.status(), poem::http::StatusCode::OK);
    let body: serde_json::Value = resp.0.into_body().into_json().await.unwrap();
    let token = body["token"].as_str().unwrap();
    assert!(!token.is_empty());
    let tokens = state.pending_tokens.read().await;
    assert!(tokens.contains_key(token));
}

// ── init_project integration tests ──────────────────────────────────

#[tokio::test]
async fn init_project_scaffolds_huskies_dir() {
    let dir = tempfile::tempdir().unwrap();
    let state = make_test_state();
    let result = gateway::init_project(&state, dir.path().to_str().unwrap(), None, None).await;
    assert!(
        result.is_ok(),
        "init_project should succeed: {:?}",
        result.err()
    );
    assert!(dir.path().join(".huskies").exists());
    assert!(dir.path().join(".huskies/project.toml").exists());
    assert!(dir.path().join(".huskies/agents.toml").exists());
    assert!(dir.path().join("script/test").exists());
}

#[tokio::test]
async fn init_project_creates_wizard_state() {
    let dir = tempfile::tempdir().unwrap();
    let state = make_test_state();
    gateway::init_project(&state, dir.path().to_str().unwrap(), None, None)
        .await
        .unwrap();
    let wizard_state_path = dir.path().join(".huskies/wizard_state.json");
    assert!(wizard_state_path.exists());
    let content = std::fs::read_to_string(&wizard_state_path).unwrap();
    let v: serde_json::Value = serde_json::from_str(&content).unwrap();
    assert!(v.get("steps").is_some());
    assert!(v.get("completed").is_some());
}

#[tokio::test]
async fn init_project_already_initialised_returns_error() {
    let dir = tempfile::tempdir().unwrap();
    std::fs::create_dir_all(dir.path().join(".huskies")).unwrap();
    let state = make_test_state();
    let result = gateway::init_project(&state, dir.path().to_str().unwrap(), None, None).await;
    assert!(result.is_err());
}

#[tokio::test]
async fn init_project_missing_path_returns_error() {
    let state = make_test_state();
    let result = gateway::init_project(&state, "", None, None).await;
    assert!(result.is_err());
}

#[tokio::test]
async fn init_project_registers_in_projects_toml_when_name_and_url_given() {
    let dir = tempfile::tempdir().unwrap();
    let config_dir = tempfile::tempdir().unwrap();
    let mut projects = BTreeMap::new();
    projects.insert(
        "existing".into(),
        ProjectEntry::with_url("http://existing:3001"),
    );
    let config = GatewayConfig {
        projects,
        sled_tokens: BTreeMap::new(),
        release_channels: BTreeMap::new(),
    };
    let state = Arc::new(GatewayState::new(config, config_dir.path().to_path_buf(), 3000).unwrap());

    let result = gateway::init_project(
        &state,
        dir.path().to_str().unwrap(),
        Some("new-project"),
        Some("http://new-project:3002"),
    )
    .await;
    assert!(result.is_ok());

    let projects = state.projects.read().await;
    assert!(projects.contains_key("new-project"));
    assert_eq!(
        projects["new-project"].url.as_deref(),
        Some("http://new-project:3002")
    );
}

#[tokio::test]
async fn init_project_duplicate_name_returns_error() {
    let dir = tempfile::tempdir().unwrap();
    let mut projects = BTreeMap::new();
    projects.insert("taken".into(), ProjectEntry::with_url("http://taken:3001"));
    let config = GatewayConfig {
        projects,
        sled_tokens: BTreeMap::new(),
        release_channels: BTreeMap::new(),
    };
    let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap());

    let result = gateway::init_project(
        &state,
        dir.path().to_str().unwrap(),
        Some("taken"),
        Some("http://new:3002"),
    )
    .await;
    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 crate::service::gateway::ProjectEntry;

    let mut projects = BTreeMap::new();
    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(),
        release_channels: BTreeMap::new(),
    };
    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;
    assert!(result.is_ok());
    assert!(project_dir.path().join(".huskies").exists());

    let wizard_path = project_dir.path().join(".huskies/wizard_state.json");
    assert!(wizard_path.exists());

    // Proxy call over the sled-uplink WS.
    let proxy_body = serde_json::to_vec(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": { "name": "wizard_status", "arguments": {} }
    }))
    .unwrap();
    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();
    let result = resp_json.get("result");
    assert!(result.is_some());
    let text = result
        .and_then(|r| r.get("content"))
        .and_then(|c| c.get(0))
        .and_then(|c| c.get("text"))
        .and_then(|t| t.as_str())
        .unwrap_or("");
    let wizard: serde_json::Value = serde_json::from_str(text).unwrap();
    assert!(wizard.get("steps").is_some());
}

// ── Aggregate pipeline status integration tests ─────────────────────

#[tokio::test]
async fn aggregate_pipeline_status_integration_healthy_and_unreachable() {
    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 healthy_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 pipeline_json = serde_json::to_string(&serde_json::json!({
                "active": [
                    { "story_id": "1_story_a", "name": "A", "stage": "current" },
                    { "story_id": "2_story_b", "name": "B", "stage": "qa" },
                    { "story_id": "3_story_c", "name": "C", "stage": "current", "blocked": true, "retry_count": 5 },
                ],
                "backlog": [{ "story_id": "4_story_d", "name": "D" }],
                "backlog_count": 1
            }))
            .unwrap();
            let body = serde_json::to_vec(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 1,
                "result": {
                    "content": [{ "type": "text", "text": pipeline_json }]
                }
            }))
            .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.len()
            );
            let _ = stream.write_all(header.as_bytes()).await;
            let _ = stream.write_all(&body).await;
        }
    });

    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    let unreachable_url = "http://127.0.0.1:1".to_string();

    let mut project_urls = BTreeMap::new();
    project_urls.insert("healthy-project".to_string(), healthy_url);
    project_urls.insert("broken-project".to_string(), unreachable_url);

    let client = reqwest::Client::new();
    let statuses = gateway::fetch_all_project_pipeline_statuses(&project_urls, &client).await;

    assert!(statuses.contains_key("healthy-project"));
    assert!(statuses.contains_key("broken-project"));

    let healthy = &statuses["healthy-project"];
    assert!(healthy.get("error").is_none());
    assert_eq!(healthy["counts"]["backlog"], 1);
    assert_eq!(healthy["counts"]["current"], 2);
    assert_eq!(healthy["counts"]["qa"], 1);

    let blocked = healthy["blocked"].as_array().unwrap();
    assert_eq!(blocked.len(), 1);
    assert_eq!(blocked[0]["story_id"], "3_story_c");

    let broken = &statuses["broken-project"];
    assert!(broken.get("error").is_some());
}

// ── Gateway broadcaster forwarder tests ─────────────────────────────

#[tokio::test]
async fn broadcaster_forwarder_forwards_events_with_project_tag() {
    use crate::chat::{ChatTransport, MessageId};
    use crate::service::events::StoredEvent;
    use async_trait::async_trait;

    type CallLog = Arc<std::sync::Mutex<Vec<(String, String)>>>;

    struct MockTransport {
        calls: CallLog,
    }

    #[async_trait]
    impl ChatTransport for MockTransport {
        async fn send_message(
            &self,
            room_id: &str,
            plain: &str,
            _html: &str,
        ) -> Result<MessageId, String> {
            self.calls
                .lock()
                .unwrap()
                .push((room_id.to_string(), 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 (tx, rx) =
        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(16);
    gateway::spawn_gateway_broadcaster_forwarder(
        transport as Arc<dyn crate::chat::ChatTransport>,
        vec!["!room:example.org".to_string()],
        rx,
    );

    // Give the forwarder task a moment to start.
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    let event = crate::service::gateway::GatewayStatusEvent {
        project: "my-project".to_string(),
        event: StoredEvent::StageTransition {
            story_id: "7_story_x".to_string(),
            story_name: String::new(),
            from_stage: "2_current".to_string(),
            to_stage: "3_qa".to_string(),
            timestamp_ms: 100,
        },
    };
    tx.send(event).unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let messages = calls.lock().unwrap();
    assert_eq!(messages.len(), 1, "Expected exactly one notification");
    let (room, plain) = &messages[0];
    assert_eq!(room, "!room:example.org");
    assert!(
        plain.starts_with("[my-project]"),
        "Expected [my-project] prefix; got: {plain}"
    );
    assert!(
        plain.contains("#7"),
        "Expected story number #7; got: {plain}"
    );
}

#[tokio::test]
async fn broadcaster_forwarder_resubscribes_on_lag() {
    use crate::chat::{ChatTransport, MessageId};
    use crate::service::events::StoredEvent;
    use async_trait::async_trait;

    type Counter = Arc<std::sync::Mutex<usize>>;

    struct CountTransport {
        count: Counter,
    }

    #[async_trait]
    impl ChatTransport for CountTransport {
        async fn send_message(
            &self,
            _room_id: &str,
            _plain: &str,
            _html: &str,
        ) -> Result<MessageId, String> {
            *self.count.lock().unwrap() += 1;
            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 count: Counter = Arc::new(std::sync::Mutex::new(0));
    let transport = Arc::new(CountTransport {
        count: Arc::clone(&count),
    });

    // Use a tiny channel (capacity 1) so the second send causes a Lagged error.
    let (tx, rx) =
        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(1);

    // Flood the channel to trigger Lagged before the forwarder task starts.
    let make_event = |n: u64| crate::service::gateway::GatewayStatusEvent {
        project: "p".to_string(),
        event: StoredEvent::StageTransition {
            story_id: format!("{n}_story"),
            story_name: String::new(),
            from_stage: "2_current".to_string(),
            to_stage: "3_qa".to_string(),
            timestamp_ms: n,
        },
    };
    // Send 3 events to overflow the capacity-1 channel before the task runs.
    let _ = tx.send(make_event(1));
    let _ = tx.send(make_event(2));
    let _ = tx.send(make_event(3));

    gateway::spawn_gateway_broadcaster_forwarder(
        transport as Arc<dyn crate::chat::ChatTransport>,
        vec!["!r:x.org".to_string()],
        rx,
    );

    // Send one more event after the forwarder subscribes; it should arrive.
    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    tx.send(make_event(4)).unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    // After Lagged + resubscribe, the forwarder must still process event 4.
    let received = *count.lock().unwrap();
    assert!(
        received >= 1,
        "Expected at least one event after Lagged resubscribe; got {received}"
    );
}

/// Story 1200 AC4: identical disk-space warnings arriving from different
/// sleds within the rate window must collapse into a single forwarded chat
/// message.
#[tokio::test]
async fn broadcaster_forwarder_dedupes_identical_disk_warnings_from_different_sleds() {
    use crate::chat::{ChatTransport, MessageId};
    use crate::service::events::StoredEvent;
    use async_trait::async_trait;

    type CallLog = Arc<std::sync::Mutex<Vec<(String, String)>>>;

    struct MockTransport {
        calls: CallLog,
    }

    #[async_trait]
    impl ChatTransport for MockTransport {
        async fn send_message(
            &self,
            room_id: &str,
            plain: &str,
            _html: &str,
        ) -> Result<MessageId, String> {
            self.calls
                .lock()
                .unwrap()
                .push((room_id.to_string(), 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 (tx, rx) =
        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(16);
    gateway::spawn_gateway_broadcaster_forwarder(
        transport as Arc<dyn crate::chat::ChatTransport>,
        vec!["!room:example.org".to_string()],
        rx,
    );

    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    let warning = |host_id: &str| crate::service::gateway::GatewayStatusEvent {
        project: host_id.to_string(),
        event: StoredEvent::DiskSpaceWarning {
            level: "warn".to_string(),
            free_bytes: 45_000_000_000,
            target_bytes: 10_000_000_000,
            worktrees_bytes: 5_000_000_000,
            host_id: host_id.to_string(),
            timestamp_ms: 100,
        },
    };

    // Two different sleds both observe the same "warn" level within the
    // dedupe window — only the first should be forwarded.
    tx.send(warning("sled-a")).unwrap();
    tx.send(warning("sled-b")).unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let messages = calls.lock().unwrap();
    assert_eq!(
        messages.len(),
        1,
        "Expected identical disk warnings from different sleds to dedupe to one message"
    );
}

/// Non-disk-space events (e.g. stage transitions) must never be deduped, even
/// when several arrive back-to-back — only disk-space warnings/recoveries
/// share a dedupe key (story 1200 AC4).
#[tokio::test]
async fn broadcaster_forwarder_does_not_dedupe_non_disk_events() {
    use crate::chat::{ChatTransport, MessageId};
    use crate::service::events::StoredEvent;
    use async_trait::async_trait;

    type CallLog = Arc<std::sync::Mutex<Vec<(String, String)>>>;

    struct MockTransport {
        calls: CallLog,
    }

    #[async_trait]
    impl ChatTransport for MockTransport {
        async fn send_message(
            &self,
            room_id: &str,
            plain: &str,
            _html: &str,
        ) -> Result<MessageId, String> {
            self.calls
                .lock()
                .unwrap()
                .push((room_id.to_string(), 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 (tx, rx) =
        tokio::sync::broadcast::channel::<crate::service::gateway::GatewayStatusEvent>(16);
    gateway::spawn_gateway_broadcaster_forwarder(
        transport as Arc<dyn crate::chat::ChatTransport>,
        vec!["!room:example.org".to_string()],
        rx,
    );

    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    let transition = |n: u64| crate::service::gateway::GatewayStatusEvent {
        project: "p".to_string(),
        event: StoredEvent::StageTransition {
            story_id: format!("{n}_story"),
            story_name: String::new(),
            from_stage: "2_current".to_string(),
            to_stage: "3_qa".to_string(),
            timestamp_ms: n,
        },
    };
    tx.send(transition(1)).unwrap();
    tx.send(transition(2)).unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let messages = calls.lock().unwrap();
    assert_eq!(
        messages.len(),
        2,
        "Non-disk events must not be deduped against each other"
    );
}

// ── BotConfig tests ─────────────────────────────────────────────────

#[test]
fn bot_config_loads_from_gateway_config_dir() {
    use crate::chat::transport::matrix::BotConfig;

    let tmp = tempfile::tempdir().unwrap();
    let huskies_dir = tmp.path().join(".huskies");
    std::fs::create_dir_all(&huskies_dir).unwrap();
    std::fs::write(
        huskies_dir.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());
    assert!(config.is_some());
    let config = config.unwrap();
    assert_eq!(
        config.homeserver.as_deref(),
        Some("https://matrix.example.com")
    );
}

#[test]
fn bot_config_absent_returns_none_in_gateway_mode() {
    use crate::chat::transport::matrix::BotConfig;
    let tmp = tempfile::tempdir().unwrap();
    let config = BotConfig::load(tmp.path());
    assert!(config.is_none());
}

#[test]
fn bot_config_disabled_returns_none_in_gateway_mode() {
    use crate::chat::transport::matrix::BotConfig;
    let tmp = tempfile::tempdir().unwrap();
    let huskies_dir = tmp.path().join(".huskies");
    std::fs::create_dir_all(&huskies_dir).unwrap();
    std::fs::write(
        huskies_dir.join("bot.toml"),
        r#"
homeserver = "https://matrix.example.com"
username = "@bot:example.com"
password = "secret"
room_ids = ["!abc:example.com"]
enabled = false
"#,
    )
    .unwrap();

    let config = BotConfig::load(tmp.path());
    assert!(config.is_none());
}

// ── Gateway MCP SSE proxy integration tests ──────────────────────────

#[tokio::test]
async fn gateway_mcp_sse_proxy_streams_progress_and_final_response() {
    let mut mock_sled = mockito::Server::new_async().await;

    let prog1 = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "notifications/progress",
        "params": { "progressToken": "tok1", "progress": 1.0 }
    });
    let prog2 = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "notifications/progress",
        "params": { "progressToken": "tok1", "progress": 2.0 }
    });
    let final_resp = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "result": { "content": [{ "type": "text", "text": "tests passed" }] }
    });
    let sse_body = format!("data: {prog1}\n\ndata: {prog2}\n\ndata: {final_resp}\n\n");

    let _mock = mock_sled
        .mock("POST", "/mcp")
        .with_status(200)
        .with_header("content-type", "text/event-stream")
        .with_body(&sse_body)
        .create_async()
        .await;

    let mut projects = BTreeMap::new();
    projects.insert("sled".to_string(), ProjectEntry::with_url(mock_sled.url()));
    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": 1,
        "method": "tools/call",
        "params": {
            "name": "run_tests",
            "arguments": {},
            "_meta": { "progressToken": "tok1" }
        }
    }))
    .unwrap();

    let resp = cli
        .post("/mcp")
        .header("content-type", "application/json")
        .header("accept", "text/event-stream")
        .body(rpc_body)
        .send()
        .await;

    let body = resp.0.into_body().into_string().await.unwrap();

    let data_lines: Vec<&str> = body
        .lines()
        .filter_map(|l| l.strip_prefix("data: "))
        .collect();

    assert_eq!(
        data_lines.len(),
        3,
        "Expected 3 SSE events (2 progress + 1 final); got {}: {:?}",
        data_lines.len(),
        body
    );

    let ev1: serde_json::Value =
        serde_json::from_str(data_lines[0]).expect("event 1 is valid JSON");
    assert_eq!(
        ev1["method"], "notifications/progress",
        "event 1 must be a progress notification"
    );
    assert_eq!(ev1["params"]["progress"], 1.0);

    let ev2: serde_json::Value =
        serde_json::from_str(data_lines[1]).expect("event 2 is valid JSON");
    assert_eq!(
        ev2["method"], "notifications/progress",
        "event 2 must be a progress notification"
    );
    assert_eq!(ev2["params"]["progress"], 2.0);

    let ev3: serde_json::Value =
        serde_json::from_str(data_lines[2]).expect("event 3 is valid JSON");
    assert_eq!(ev3["id"], 1, "event 3 must be the final JSON-RPC response");
    assert!(
        ev3.get("result").is_some(),
        "event 3 must carry a result field"
    );
}

/// 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() {
    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 {
            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 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());
    let cli = poem::test::TestClient::new(app);

    let rpc_body = serde_json::to_vec(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": { "name": "run_tests", "arguments": {} }
    }))
    .unwrap();

    let resp = cli
        .post("/mcp")
        .header("content-type", "application/json")
        .body(rpc_body)
        .send()
        .await;

    let ct = resp
        .0
        .headers()
        .get("content-type")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    assert!(
        ct.contains("application/json"),
        "Non-SSE path must return application/json; got: {ct}"
    );

    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!(
        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}"
    );
}

// ── Story 899: MCP-over-WS uplink integration ────────────────────────────────

/// Build a `SledConnection` plus a spawned "mock sled" task that:
///
/// * Reads outbound `mcp_request` envelopes off the connection's channel.
/// * Invokes the supplied closure to build a `result` value for each request.
/// * Resolves the matching in-flight oneshot directly (the same effect the WS
///   handler has when it receives an `mcp_response` from a real sled).
///
/// Returns the registered `SledConnection`.
fn spawn_mock_sled<F>(handler: F) -> crate::service::gateway::SledConnection
where
    F: Fn(&serde_json::Value) -> serde_json::Value + Send + Sync + 'static,
{
    use crate::service::gateway::SledConnection;
    use std::sync::atomic::AtomicI64;

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
    let last_heartbeat_ms = Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis()));
    let in_flight = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new()));

    let conn = SledConnection {
        tx,
        last_heartbeat_ms,
        in_flight: Arc::clone(&in_flight),
    };
    let handler = Arc::new(handler);
    tokio::spawn(async move {
        while let Some(env) = rx.recv().await {
            if env.msg_type != "mcp_request" {
                continue;
            }
            let body_str = env
                .payload
                .get("body")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            let body_json: serde_json::Value = match serde_json::from_str(&body_str) {
                Ok(v) => v,
                Err(_) => serde_json::Value::Null,
            };
            let result = handler(&body_json);
            let response = serde_json::json!({
                "jsonrpc": "2.0",
                "id": body_json.get("id").cloned().unwrap_or(serde_json::Value::Null),
                "result": result,
            });
            if let Some(oneshot_tx) = in_flight.lock().await.remove(&env.req_id) {
                let _ = oneshot_tx.send(response);
            }
        }
    });
    conn
}

/// AC 9: gateway switches active project, calls tools/list and tools/call
/// against a sled connected only via WS uplink, gets correct responses.
#[tokio::test]
async fn ws_only_sled_handles_tools_list_and_tools_call() {
    use crate::service::gateway::ProjectEntry;

    // Project entry with NO url — WS-only.
    let mut projects = BTreeMap::new();
    projects.insert(
        "ws-only".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(),
        release_channels: BTreeMap::new(),
    };
    let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap());

    let conn = spawn_mock_sled(|body| {
        let method = body.get("method").and_then(|m| m.as_str()).unwrap_or("");
        match method {
            "tools/list" => serde_json::json!({
                "tools": [
                    { "name": "my_tool", "description": "test" }
                ]
            }),
            "tools/call" => serde_json::json!({
                "content": [{ "type": "text", "text": "called ok" }]
            }),
            _ => serde_json::json!({ "echo": method }),
        }
    });
    state
        .register_sled_connection("ws-only".to_string(), conn)
        .await;

    // tools/list via proxy_active_mcp.
    let body = serde_json::to_vec(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/list",
        "params": {}
    }))
    .unwrap();
    let resp = state.proxy_active_mcp(&body).await.expect("ws proxy works");
    let resp_json: serde_json::Value = serde_json::from_slice(&resp).unwrap();
    assert_eq!(
        resp_json["result"]["tools"][0]["name"], "my_tool",
        "tools/list response must come from the WS-connected sled, not HTTP"
    );

    // tools/call via proxy_active_mcp.
    let body = serde_json::to_vec(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": { "name": "my_tool", "arguments": {} }
    }))
    .unwrap();
    let resp = state.proxy_active_mcp(&body).await.expect("ws proxy works");
    let resp_json: serde_json::Value = serde_json::from_slice(&resp).unwrap();
    assert_eq!(
        resp_json["result"]["content"][0]["text"], "called ok",
        "tools/call response must come from the WS-connected sled, not HTTP"
    );
}

/// AC 10: two sleds connected to one gateway concurrently; gateway routes
/// calls to the right sled based on active project.
#[tokio::test]
async fn two_concurrent_sleds_are_routed_by_active_project() {
    use crate::service::gateway::ProjectEntry;

    let mut projects = BTreeMap::new();
    projects.insert(
        "alpha".into(),
        ProjectEntry {
            url: None,
            auth_token: Some("alpha-tok".into()),
            ssh_port: None,
            host_path: None,
            expected_node_id: None,
        },
    );
    projects.insert(
        "beta".into(),
        ProjectEntry {
            url: None,
            auth_token: Some("beta-tok".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 alpha_conn = spawn_mock_sled(|_body| {
        serde_json::json!({
            "content": [{ "type": "text", "text": "from-alpha" }]
        })
    });
    let beta_conn = spawn_mock_sled(|_body| {
        serde_json::json!({
            "content": [{ "type": "text", "text": "from-beta" }]
        })
    });
    state
        .register_sled_connection("alpha".to_string(), alpha_conn)
        .await;
    state
        .register_sled_connection("beta".to_string(), beta_conn)
        .await;

    // Switch to alpha.
    gateway::switch_project(&state, "alpha").await.unwrap();
    let body = serde_json::to_vec(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": { "name": "noop", "arguments": {} }
    }))
    .unwrap();
    let resp = state.proxy_active_mcp(&body).await.expect("ws proxy works");
    let resp_json: serde_json::Value = serde_json::from_slice(&resp).unwrap();
    assert_eq!(
        resp_json["result"]["content"][0]["text"], "from-alpha",
        "When active project is alpha, calls must route to the alpha sled"
    );

    // Switch to beta.
    gateway::switch_project(&state, "beta").await.unwrap();
    let resp = state.proxy_active_mcp(&body).await.expect("ws proxy works");
    let resp_json: serde_json::Value = serde_json::from_slice(&resp).unwrap();
    assert_eq!(
        resp_json["result"]["content"][0]["text"], "from-beta",
        "When active project is beta, calls must route to the beta sled"
    );
}