#![allow(clippy::large_futures)]

//! Integration tests for the RPC server.
//!
//! These tests exercise the full roundtrip: write JSONL requests to a stream,
//! read JSONL responses/notifications back. Uses `tokio::io::duplex` for
//! in-process paired channels.

#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;

use async_trait::async_trait;
use futures::stream;
use meerkat::AgentFactory;
use meerkat_client::{LlmClient, LlmError};
use meerkat_core::{BlobStore, Config, ConfigRuntime, MemoryConfigStore, StopReason};
use meerkat_rpc::server::RpcServer;
use meerkat_rpc::session_runtime::SessionRuntime;
use meerkat_store::MemoryBlobStore;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

// ---------------------------------------------------------------------------
// Mock LLM client
// ---------------------------------------------------------------------------

/// Mock LLM client that echoes the model name from each request.
struct MockLlmClient;

#[async_trait]
impl LlmClient for MockLlmClient {
    fn project_replay_messages(
        &self,
        messages: &[meerkat_core::Message],
    ) -> Result<Vec<meerkat_core::Message>, meerkat_client::LlmError> {
        Ok(messages.to_vec())
    }

    fn stream<'a>(
        &'a self,
        request: &'a meerkat_client::LlmRequest,
    ) -> Pin<Box<dyn futures::Stream<Item = Result<meerkat_client::LlmEvent, LlmError>> + Send + 'a>>
    {
        let model = request.model.clone();
        Box::pin(stream::iter(vec![
            Ok(meerkat_client::LlmEvent::TextDelta {
                delta: format!("Hello from mock [model={model}]"),
                meta: None,
            }),
            Ok(meerkat_client::LlmEvent::Done {
                outcome: meerkat_client::LlmDoneOutcome::Success {
                    stop_reason: StopReason::EndTurn,
                },
            }),
        ]))
    }

    fn provider(&self) -> meerkat_core::Provider {
        meerkat_core::Provider::Other
    }

    async fn health_check(&self) -> Result<(), LlmError> {
        Ok(())
    }
}

#[derive(Default)]
struct RecordingToolClient {
    seen_tools: Mutex<Vec<Vec<String>>>,
}

impl RecordingToolClient {
    fn seen_tools(&self) -> Vec<Vec<String>> {
        self.seen_tools.lock().expect("recording lock").clone()
    }
}

#[async_trait]
impl LlmClient for RecordingToolClient {
    fn project_replay_messages(
        &self,
        messages: &[meerkat_core::Message],
    ) -> Result<Vec<meerkat_core::Message>, meerkat_client::LlmError> {
        Ok(messages.to_vec())
    }

    fn stream<'a>(
        &'a self,
        request: &'a meerkat_client::LlmRequest,
    ) -> Pin<Box<dyn futures::Stream<Item = Result<meerkat_client::LlmEvent, LlmError>> + Send + 'a>>
    {
        self.seen_tools.lock().expect("recording lock").push(
            request
                .tools
                .iter()
                .map(|tool| tool.name.to_string())
                .collect(),
        );
        Box::pin(stream::iter(vec![
            Ok(meerkat_client::LlmEvent::TextDelta {
                delta: "tools recorded".to_string(),
                meta: None,
            }),
            Ok(meerkat_client::LlmEvent::Done {
                outcome: meerkat_client::LlmDoneOutcome::Success {
                    stop_reason: StopReason::EndTurn,
                },
            }),
        ]))
    }

    fn provider(&self) -> meerkat_core::Provider {
        meerkat_core::Provider::Other
    }

    async fn health_check(&self) -> Result<(), LlmError> {
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Set up a server with duplex streams and spawn it in a background task.
///
/// Returns:
/// - `client_writer`: write JSONL requests here
/// - `client_reader`: read JSONL responses/notifications here (wrapped in BufReader)
/// - `server_handle`: JoinHandle for the server task
fn spawn_test_server() -> (
    tokio::io::DuplexStream,
    BufReader<tokio::io::DuplexStream>,
    tokio::task::JoinHandle<Result<(), meerkat_rpc::server::ServerError>>,
) {
    let temp = tempfile::tempdir().unwrap();
    let factory = AgentFactory::new(temp.path().join("sessions"));
    let config = Config::default();
    let store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
    let blob_store: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
    let mut runtime = SessionRuntime::new(
        factory,
        config,
        10,
        meerkat::PersistenceBundle::new(store, None, blob_store),
        meerkat_rpc::router::NotificationSink::noop(),
    );
    let config_store: Arc<dyn meerkat_core::ConfigStore> = Arc::new(MemoryConfigStore::new(
        Config::default(),
        meerkat_models::canonical(),
    ));
    runtime.set_default_llm_client(Some(Arc::new(MockLlmClient)));
    runtime.set_config_runtime(Arc::new(ConfigRuntime::new(
        Arc::clone(&config_store),
        temp.path().join("config_state.json"),
    )));
    let runtime = Arc::new(runtime);

    let (server_reader, client_writer) = tokio::io::duplex(4096);
    let (client_reader, server_writer) = tokio::io::duplex(4096);

    let server_handle = tokio::spawn(async move {
        // Keep temp alive for the duration of the server
        let _temp = temp;
        let reader = BufReader::new(server_reader);
        let mut server = RpcServer::new(reader, server_writer, runtime, config_store);
        server.run().await
    });

    let client_reader = BufReader::new(client_reader);
    (client_writer, client_reader, server_handle)
}

fn spawn_test_server_with_client(
    client: Arc<dyn LlmClient>,
) -> (
    tokio::io::DuplexStream,
    BufReader<tokio::io::DuplexStream>,
    tokio::task::JoinHandle<Result<(), meerkat_rpc::server::ServerError>>,
) {
    let temp = tempfile::tempdir().unwrap();
    let factory = AgentFactory::new(temp.path().join("sessions"));
    let config = Config::default();
    let store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
    let blob_store: Arc<dyn BlobStore> = Arc::new(MemoryBlobStore::new());
    let mut runtime = SessionRuntime::new(
        factory,
        config,
        10,
        meerkat::PersistenceBundle::new(store, None, blob_store),
        meerkat_rpc::router::NotificationSink::noop(),
    );
    let config_store: Arc<dyn meerkat_core::ConfigStore> = Arc::new(MemoryConfigStore::new(
        Config::default(),
        meerkat_models::canonical(),
    ));
    runtime.set_default_llm_client(Some(client));
    runtime.set_config_runtime(Arc::new(ConfigRuntime::new(
        Arc::clone(&config_store),
        temp.path().join("config_state.json"),
    )));
    let runtime = Arc::new(runtime);

    let (server_reader, client_writer) = tokio::io::duplex(4096);
    let (client_reader, server_writer) = tokio::io::duplex(4096);

    let server_handle = tokio::spawn(async move {
        let _temp = temp;
        let reader = BufReader::new(server_reader);
        let mut server = RpcServer::new(reader, server_writer, runtime, config_store);
        server.run().await
    });

    let client_reader = BufReader::new(client_reader);
    (client_writer, client_reader, server_handle)
}

/// Send a JSONL request line.
async fn send_request(writer: &mut tokio::io::DuplexStream, request: &serde_json::Value) {
    let line = format!("{}\n", serde_json::to_string(request).unwrap());
    writer.write_all(line.as_bytes()).await.unwrap();
    writer.flush().await.unwrap();
}

/// Read a single JSONL line and parse it as a JSON value.
async fn read_line_json(reader: &mut BufReader<tokio::io::DuplexStream>) -> serde_json::Value {
    let mut line = String::new();
    reader.read_line(&mut line).await.unwrap();
    assert!(!line.is_empty(), "Expected a JSONL line but got EOF");
    serde_json::from_str(&line).unwrap()
}

/// Read a response (a line that has an "id" field), skipping notifications.
async fn read_response(reader: &mut BufReader<tokio::io::DuplexStream>) -> serde_json::Value {
    loop {
        let value = read_line_json(reader).await;
        // Responses have an "id" field; notifications do not
        if value.get("id").is_some() {
            return value;
        }
        // Otherwise it's a notification - skip it
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

/// 1. Initialize roundtrip: send `initialize`, get capabilities response.
#[tokio::test]
async fn initialize_roundtrip() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    let req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {}
    });
    send_request(&mut writer, &req).await;

    let response = read_response(&mut reader).await;
    assert_eq!(response["id"], 1);
    assert!(
        response["error"].is_null(),
        "Expected success, got error: {response}"
    );
    assert_eq!(response["result"]["server_info"]["name"], "meerkat-rpc");
    assert!(response["result"]["server_info"]["version"].is_string());

    let methods = response["result"]["methods"].as_array().unwrap();
    let method_names: Vec<&str> = methods.iter().map(|m| m.as_str().unwrap()).collect();
    assert!(method_names.contains(&"session/create"));
    assert!(method_names.contains(&"session/external_event"));
    assert!(method_names.contains(&"session/peer_response_terminal"));
    assert!(method_names.contains(&"session/inject_context"));
    assert!(method_names.contains(&"turn/start"));
    assert!(method_names.contains(&"config/get"));
    #[cfg(feature = "mcp")]
    {
        assert!(method_names.contains(&"mcp/add"));
        assert!(method_names.contains(&"mcp/remove"));
        assert!(method_names.contains(&"mcp/reload"));
    }
    #[cfg(not(feature = "mcp"))]
    {
        assert!(!method_names.contains(&"mcp/add"));
        assert!(!method_names.contains(&"mcp/remove"));
        assert!(!method_names.contains(&"mcp/reload"));
    }
    #[cfg(feature = "mob")]
    {
        assert!(method_names.contains(&"mob/spawn_helper"));
        assert!(method_names.contains(&"mob/fork_helper"));
        assert!(method_names.contains(&"mob/force_cancel"));
        assert!(method_names.contains(&"mob/member_status"));
        assert!(method_names.contains(&"mob/ingress_interaction"));
    }

    // Close to trigger EOF
    drop(writer);
    server_handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn deferred_callback_direct_sessions_expose_control_plane_tools_on_first_turn() {
    let client = Arc::new(RecordingToolClient::default());
    let (mut writer, mut reader, server_handle) = spawn_test_server_with_client(client.clone());

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    )
    .await;
    let init_resp = read_response(&mut reader).await;
    assert!(
        init_resp["error"].is_null(),
        "initialize failed: {init_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/register",
            "params": {
                "tools": [{
                    "name": "secret_lookup",
                    "description": "Look up a secret value through a deferred catalog.",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "key": {"type": "string"}
                        },
                        "required": ["key"]
                    }
                }, {
                    "name": "secret_audit",
                    "description": "Audit a secret value through the same deferred catalog.",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "key": {"type": "string"}
                        },
                        "required": ["key"]
                    }
                }]
            }
        }),
    )
    .await;
    let register_resp = read_response(&mut reader).await;
    assert!(
        register_resp["error"].is_null(),
        "tools/register failed: {register_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "session/create",
            "params": {
                "prompt": "Bootstrap deferred session",
                "initial_turn": "deferred",
                "enable_builtins": false,
                "enable_shell": false,
                "enable_memory": false,
                "enable_mob": false
            }
        }),
    )
    .await;
    let create_resp = read_response(&mut reader).await;
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .expect("session_id missing")
        .to_string();

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 4,
            "method": "turn/start",
            "params": {
                "session_id": session_id,
                "prompt": "Inspect the deferred control plane."
            }
        }),
    )
    .await;
    let turn_resp = read_response(&mut reader).await;
    assert!(
        turn_resp["error"].is_null(),
        "turn/start failed: {turn_resp}"
    );

    let seen = client.seen_tools();
    assert_eq!(seen.len(), 1, "expected exactly one LLM call, got {seen:?}");
    let first_call = &seen[0];
    assert!(
        first_call.iter().any(|name| name == "tool_catalog_search"),
        "direct deferred sessions should expose tool_catalog_search, got {first_call:?}"
    );
    assert!(
        first_call.iter().any(|name| name == "tool_catalog_load"),
        "direct deferred sessions should expose tool_catalog_load, got {first_call:?}"
    );
    assert!(
        !first_call.iter().any(|name| name == "secret_lookup"),
        "deferred tools should remain hidden before load, got {first_call:?}"
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn deferred_inline_external_tools_accept_explicit_keep_alive_false_turn() {
    let client = Arc::new(RecordingToolClient::default());
    let (mut writer, mut reader, server_handle) = spawn_test_server_with_client(client.clone());

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    )
    .await;
    let init_resp = read_response(&mut reader).await;
    assert!(
        init_resp["error"].is_null(),
        "initialize failed: {init_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "session/create",
            "params": {
                "prompt": "Bootstrap deferred session",
                "initial_turn": "deferred",
                "enable_builtins": false,
                "enable_shell": false,
                "enable_memory": false,
                "enable_mob": false,
                "keep_alive": false,
                "external_tools": [{
                    "name": "linear_graphql",
                    "description": "Execute GraphQL",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"},
                            "variables": {"type": "object"}
                        },
                        "required": ["query"],
                        "additionalProperties": false
                    }
                }]
            }
        }),
    )
    .await;
    let create_resp = read_response(&mut reader).await;
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .expect("session_id missing")
        .to_string();

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "turn/start",
            "params": {
                "session_id": session_id,
                "prompt": "Inspect the deferred inline callback tool surface.",
                "keep_alive": false
            }
        }),
    )
    .await;
    let turn_resp = read_response(&mut reader).await;
    assert!(
        turn_resp["error"].is_null(),
        "turn/start failed: {turn_resp}"
    );

    let seen = client.seen_tools();
    assert_eq!(seen.len(), 1, "expected exactly one LLM call, got {seen:?}");

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn pipelined_tools_register_commits_before_session_create() {
    let client = Arc::new(RecordingToolClient::default());
    let (mut writer, mut reader, server_handle) = spawn_test_server_with_client(client.clone());

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    )
    .await;
    let init_resp = read_response(&mut reader).await;
    assert!(
        init_resp["error"].is_null(),
        "initialize failed: {init_resp}"
    );

    let register_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/register",
        "params": {
            "tools": [{
                "name": "pipelined_secret_lookup",
                "description": "Look up a secret value registered immediately before create.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "key": {"type": "string"}
                    },
                    "required": ["key"]
                }
            }]
        }
    });
    let create_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "session/create",
        "params": {
            "prompt": "Start after a pipelined callback tool registration.",
            "enable_builtins": false,
            "enable_shell": false,
            "enable_memory": false,
            "enable_mob": false
        }
    });
    let pipelined = format!(
        "{}\n{}\n",
        serde_json::to_string(&register_req).unwrap(),
        serde_json::to_string(&create_req).unwrap()
    );
    writer.write_all(pipelined.as_bytes()).await.unwrap();
    writer.flush().await.unwrap();

    let register_resp = read_response(&mut reader).await;
    assert_eq!(
        register_resp["id"], 2,
        "tools/register must complete before the following pipelined create response: {register_resp}"
    );
    assert!(
        register_resp["error"].is_null(),
        "tools/register failed: {register_resp}"
    );

    let create_resp = read_response(&mut reader).await;
    assert_eq!(create_resp["id"], 3);
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );

    let seen = client.seen_tools();
    assert_eq!(
        seen.len(),
        1,
        "expected exactly one LLM call from session/create, got {seen:?}"
    );
    assert!(
        seen[0].iter().any(|name| name == "pipelined_secret_lookup"),
        "session/create first turn must see the immediately preceding tools/register, got {:?}",
        seen[0]
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn late_registered_deferred_callbacks_keep_control_plane_after_inline_build() {
    let client = Arc::new(RecordingToolClient::default());
    let (mut writer, mut reader, server_handle) = spawn_test_server_with_client(client.clone());

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    )
    .await;
    let init_resp = read_response(&mut reader).await;
    assert!(
        init_resp["error"].is_null(),
        "initialize failed: {init_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/register",
            "params": {
                "tools": [{
                    "name": "secret_lookup",
                    "description": "Look up a secret value through a deferred catalog.",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "key": {"type": "string"}
                        },
                        "required": ["key"]
                    }
                }]
            }
        }),
    )
    .await;
    let register_resp = read_response(&mut reader).await;
    assert!(
        register_resp["error"].is_null(),
        "initial tools/register failed: {register_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "session/create",
            "params": {
                "prompt": "Bootstrap late deferred session",
                "initial_turn": "deferred",
                "enable_builtins": false,
                "enable_shell": false,
                "enable_memory": false,
                "enable_mob": false
            }
        }),
    )
    .await;
    let create_resp = read_response(&mut reader).await;
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .expect("session_id missing")
        .to_string();

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 4,
            "method": "turn/start",
            "params": {
                "session_id": session_id,
                "prompt": "Inspect the current inline callback tool surface."
            }
        }),
    )
    .await;
    let first_turn_resp = read_response(&mut reader).await;
    assert!(
        first_turn_resp["error"].is_null(),
        "initial turn/start failed: {first_turn_resp}"
    );

    let seen = client.seen_tools();
    assert_eq!(
        seen.len(),
        1,
        "expected exactly one initial LLM call, got {seen:?}"
    );
    let initial_call = &seen[0];
    assert!(
        initial_call.iter().any(|name| name == "secret_lookup"),
        "the session should start inline before the adaptive deferred threshold is crossed, got {initial_call:?}"
    );
    assert!(
        initial_call
            .iter()
            .any(|name| name == "tool_catalog_search"),
        "dynamically defer-capable sessions should still precompose tool_catalog_search, got {initial_call:?}"
    );
    assert!(
        initial_call.iter().any(|name| name == "tool_catalog_load"),
        "dynamically defer-capable sessions should still precompose tool_catalog_load, got {initial_call:?}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 5,
            "method": "tools/register",
            "params": {
                "tools": [{
                    "name": "secret_audit",
                    "description": "Audit a secret value through the same deferred catalog.",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "key": {"type": "string"}
                        },
                        "required": ["key"]
                    }
                }]
            }
        }),
    )
    .await;
    let late_register_resp = read_response(&mut reader).await;
    assert!(
        late_register_resp["error"].is_null(),
        "late tools/register failed: {late_register_resp}"
    );

    send_request(
        &mut writer,
        &serde_json::json!({
            "jsonrpc": "2.0",
            "id": 6,
            "method": "turn/start",
            "params": {
                "session_id": session_id,
                "prompt": "Inspect the deferred control plane after a late registration."
            }
        }),
    )
    .await;
    let turn_resp = read_response(&mut reader).await;
    assert!(
        turn_resp["error"].is_null(),
        "turn/start failed: {turn_resp}"
    );

    let seen = client.seen_tools();
    assert_eq!(
        seen.len(),
        2,
        "expected exactly two LLM calls, got {seen:?}"
    );
    let first_call = &seen[1];
    assert!(
        first_call.iter().any(|name| name == "tool_catalog_search"),
        "late-switch deferred sessions should expose tool_catalog_search, got {first_call:?}"
    );
    assert!(
        first_call.iter().any(|name| name == "tool_catalog_load"),
        "late-switch deferred sessions should expose tool_catalog_load, got {first_call:?}"
    );
    assert!(
        !first_call.iter().any(|name| name == "secret_audit"),
        "late-added deferred tools should remain hidden until loaded, got {first_call:?}"
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// mcp/* methods are registered and return contract-typed placeholder responses.
#[cfg(feature = "mcp")]
#[tokio::test]
async fn mcp_live_methods_roundtrip_and_validation() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // Create session for validation against an existing session ID
    let create_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "session/create",
        "params": {"prompt": "hello"}
    });
    send_request(&mut writer, &create_req).await;
    let create_resp = read_response(&mut reader).await;
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .expect("session_id")
        .to_string();

    // mcp/add success
    let add_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "mcp/add",
        "params": {
            "session_id": session_id,
            "server_config": {
                "name": "filesystem",
                "command": "echo",
                "args": [],
                "env": {}
            },
            "persisted": false
        }
    });
    send_request(&mut writer, &add_req).await;
    let add_resp = read_response(&mut reader).await;
    assert!(add_resp["error"].is_null(), "mcp/add failed: {add_resp}");
    assert_eq!(add_resp["result"]["operation"], "add");
    assert_eq!(add_resp["result"]["status"], "staged");
    assert_eq!(add_resp["result"]["persisted"], false);
    assert!(add_resp["result"]["applied_at_turn"].is_null());

    // mcp/reload success (null server_name = reload all)
    let reload_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "mcp/reload",
        "params": {
            "session_id": add_resp["result"]["session_id"],
            "persisted": false
        }
    });
    send_request(&mut writer, &reload_req).await;
    let reload_resp = read_response(&mut reader).await;
    assert!(
        reload_resp["error"].is_null(),
        "mcp/reload failed: {reload_resp}"
    );
    assert_eq!(reload_resp["result"]["operation"], "reload");
    assert_eq!(reload_resp["result"]["status"], "staged");

    // mcp/remove success (persisted=true is accepted but response is persisted=false)
    let remove_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 4,
        "method": "mcp/remove",
        "params": {
            "session_id": reload_resp["result"]["session_id"],
            "server_name": "filesystem",
            "persisted": true
        }
    });
    send_request(&mut writer, &remove_req).await;
    let remove_resp = read_response(&mut reader).await;
    assert!(
        remove_resp["error"].is_null(),
        "mcp/remove failed: {remove_resp}"
    );
    assert_eq!(remove_resp["result"]["operation"], "remove");
    assert_eq!(remove_resp["result"]["status"], "staged");
    assert_eq!(remove_resp["result"]["persisted"], false);

    // invalid params are rejected
    let invalid_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 5,
        "method": "mcp/add",
        "params": {
            "session_id": "not-a-session-id",
            "server_config": {}
        }
    });
    send_request(&mut writer, &invalid_req).await;
    let invalid_resp = read_response(&mut reader).await;
    assert!(invalid_resp["result"].is_null());
    assert!(invalid_resp["error"].is_object());

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 2. Session create and turn start: create session, then start a turn.
#[tokio::test]
async fn session_create_and_turn_start() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // Create session
    let create_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "session/create",
        "params": {"prompt": "Hello"}
    });
    send_request(&mut writer, &create_req).await;

    let create_resp = read_response(&mut reader).await;
    assert_eq!(create_resp["id"], 1);
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .unwrap()
        .to_string();
    assert!(!session_id.is_empty());
    assert!(
        create_resp["result"]["text"]
            .as_str()
            .unwrap()
            .contains("Hello from mock")
    );

    // Start another turn
    let turn_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "turn/start",
        "params": {"session_id": session_id, "prompt": "Follow up"}
    });
    send_request(&mut writer, &turn_req).await;

    let turn_resp = read_response(&mut reader).await;
    assert_eq!(turn_resp["id"], 2);
    assert!(
        turn_resp["error"].is_null(),
        "turn/start failed: {turn_resp}"
    );
    assert_eq!(
        turn_resp["result"]["session_id"].as_str().unwrap(),
        session_id
    );
    assert!(
        turn_resp["result"]["text"]
            .as_str()
            .unwrap()
            .contains("Hello from mock")
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 3. Server shuts down cleanly on EOF.
#[tokio::test]
async fn server_shuts_down_on_eof() {
    let (writer, _reader, server_handle) = spawn_test_server();

    // Immediately close the writer (EOF)
    drop(writer);

    // Server should exit cleanly
    let result = server_handle.await.unwrap();
    assert!(
        result.is_ok(),
        "Server should shut down cleanly on EOF, got: {:?}",
        result.err()
    );
}

/// 4. Malformed JSON returns parse error, then server continues processing.
#[tokio::test]
async fn malformed_json_returns_error_and_continues() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // Send garbage
    writer.write_all(b"this is not json\n").await.unwrap();
    writer.flush().await.unwrap();

    // Should get a parse error response
    let error_resp = read_line_json(&mut reader).await;
    assert!(
        error_resp["error"].is_object(),
        "Expected error response, got: {error_resp}"
    );
    assert_eq!(error_resp["error"]["code"], -32700); // PARSE_ERROR

    // Now send a valid request - server should still work
    let req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 42,
        "method": "initialize",
        "params": {}
    });
    send_request(&mut writer, &req).await;

    let response = read_response(&mut reader).await;
    assert_eq!(response["id"], 42);
    assert!(
        response["error"].is_null(),
        "Expected success after recovery, got: {response}"
    );
    assert_eq!(response["result"]["server_info"]["name"], "meerkat-rpc");

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 5. Config get/patch roundtrip.
#[tokio::test]
async fn config_get_patch_roundtrip() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // Get config
    let get_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "config/get"
    });
    send_request(&mut writer, &get_req).await;

    let get_resp = read_response(&mut reader).await;
    assert_eq!(get_resp["id"], 1);
    assert!(get_resp["error"].is_null(), "config/get failed: {get_resp}");
    let initial_max_tokens = get_resp["result"]["config"]["max_tokens"].as_u64().unwrap();

    // Patch max_tokens
    let new_max_tokens = initial_max_tokens + 1000;
    let patch_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "config/patch",
        "params": {"max_tokens": new_max_tokens}
    });
    send_request(&mut writer, &patch_req).await;

    let patch_resp = read_response(&mut reader).await;
    assert_eq!(patch_resp["id"], 2);
    assert!(
        patch_resp["error"].is_null(),
        "config/patch failed: {patch_resp}"
    );
    assert_eq!(patch_resp["result"]["config"]["max_tokens"], new_max_tokens);

    // Get again and verify
    let get_req2 = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "config/get"
    });
    send_request(&mut writer, &get_req2).await;

    let get_resp2 = read_response(&mut reader).await;
    assert_eq!(get_resp2["id"], 3);
    assert_eq!(get_resp2["result"]["config"]["max_tokens"], new_max_tokens);

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 6. Session list after create: create sessions, list them, verify count.
#[tokio::test]
async fn session_list_after_create() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // Create two sessions
    let create1 = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "session/create",
        "params": {"prompt": "First"}
    });
    send_request(&mut writer, &create1).await;
    let resp1 = read_response(&mut reader).await;
    assert!(resp1["error"].is_null(), "First create failed: {resp1}");
    let sid1 = resp1["result"]["session_id"].as_str().unwrap().to_string();

    let create2 = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "session/create",
        "params": {"prompt": "Second"}
    });
    send_request(&mut writer, &create2).await;
    let resp2 = read_response(&mut reader).await;
    assert!(resp2["error"].is_null(), "Second create failed: {resp2}");
    let sid2 = resp2["result"]["session_id"].as_str().unwrap().to_string();

    // List sessions
    let list_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "session/list"
    });
    send_request(&mut writer, &list_req).await;
    let list_resp = read_response(&mut reader).await;
    assert_eq!(list_resp["id"], 3);
    assert!(
        list_resp["error"].is_null(),
        "session/list failed: {list_resp}"
    );

    let sessions = list_resp["result"]["sessions"].as_array().unwrap();
    assert!(
        sessions.len() >= 2,
        "Expected at least 2 sessions, got {}",
        sessions.len()
    );

    // Both session IDs should appear
    let ids: Vec<&str> = sessions
        .iter()
        .filter_map(|s| s["session_id"].as_str())
        .collect();
    assert!(ids.contains(&sid1.as_str()), "Session 1 not found in list");
    assert!(ids.contains(&sid2.as_str()), "Session 2 not found in list");

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 7. Unknown method returns METHOD_NOT_FOUND error.
#[tokio::test]
async fn unknown_method_returns_error() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    let req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "nonexistent/method",
        "params": {}
    });
    send_request(&mut writer, &req).await;

    let response = read_response(&mut reader).await;
    assert_eq!(response["id"], 1);
    assert!(response["error"].is_object(), "Expected error response");
    assert_eq!(response["error"]["code"], -32601); // METHOD_NOT_FOUND

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// Regression: mcp/add with persisted=true should NOT be rejected — the field
/// is accepted as a real persistence request. This fake session never reaches
/// config mutation because session validation fails first.
/// Without the `mcp` feature the method returns METHOD_NOT_FOUND.
#[tokio::test]
async fn test_mcp_add_persisted_true_not_rejected() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    let init = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 0,
        "method": "initialize",
        "params": {}
    });
    send_request(&mut writer, &init).await;
    let _init_resp = read_response(&mut reader).await;

    let req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "mcp/add",
        "params": {
            "session_id": "00000000-0000-0000-0000-000000000001",
            "server_config": {"name": "test-server", "command": "echo", "args": [], "env": {}},
            "persisted": true
        }
    });
    send_request(&mut writer, &req).await;

    let response = read_response(&mut reader).await;
    assert_eq!(response["id"], 1);
    // Without `mcp` feature: METHOD_NOT_FOUND. With `mcp` feature: session not found
    // (the fake session_id doesn't exist). Either way it's an error, but NOT because
    // of persisted=true.
    assert!(
        response["error"].is_object(),
        "expected error (session not found or method not found): {response}"
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// Regression: mcp/add staged response reports persisted: false when the caller
/// requested a live-only mutation.
#[cfg(feature = "mcp")]
#[tokio::test]
async fn test_mcp_add_staged_response_has_persisted_false() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    let init = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 0,
        "method": "initialize",
        "params": {}
    });
    send_request(&mut writer, &init).await;
    let _init_resp = read_response(&mut reader).await;

    let create = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "session/create",
        "params": { "prompt": "hello" }
    });
    send_request(&mut writer, &create).await;
    let mut session_id = None;
    loop {
        let msg = read_response(&mut reader).await;
        if msg.get("id") == Some(&serde_json::json!(1)) {
            if let Some(result) = msg.get("result") {
                session_id = result
                    .get("session_id")
                    .and_then(|v| v.as_str())
                    .map(String::from);
            }
            break;
        }
    }
    let session_id = session_id.expect("session_id from create");

    let req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "mcp/add",
        "params": {
            "session_id": session_id,
            "server_config": {"name": "test-server", "command": "echo", "args": [], "env": {}},
            "persisted": false
        }
    });
    send_request(&mut writer, &req).await;

    let response = read_response(&mut reader).await;
    assert_eq!(response["id"], 2);
    let result = &response["result"];
    assert_eq!(result["status"], "staged");
    assert_eq!(result["persisted"], false);

    drop(writer);
    server_handle.await.unwrap().unwrap();
}

/// 11. In-session model switching via turn/start with model override.
///
/// Create a session -> run first turn -> verify default model ->
/// run second turn with model override -> verify model name changed.
#[tokio::test]
async fn in_session_model_switch_via_turn_start() {
    let (mut writer, mut reader, server_handle) = spawn_test_server();

    // 1. session/create with initial prompt (materializes the session)
    let create_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "session/create",
        "params": {"prompt": "Hello"}
    });
    send_request(&mut writer, &create_req).await;
    let create_resp = read_response(&mut reader).await;
    assert!(
        create_resp["error"].is_null(),
        "session/create failed: {create_resp}"
    );
    let session_id = create_resp["result"]["session_id"]
        .as_str()
        .unwrap()
        .to_string();
    let first_text = create_resp["result"]["text"].as_str().unwrap();
    assert!(
        first_text.contains("Hello from mock"),
        "First turn should produce mock text, got: {first_text}"
    );
    // Default model is the catalog-owned global default; verify it.
    let expected_default = format!("model={}", meerkat_models::global_default_model());
    assert!(
        first_text.contains(&expected_default),
        "First turn should use the catalog global default model, got: {first_text}"
    );

    // 2. turn/start with model override on the materialized session.
    // Model-only overrides may switch providers between turns when the
    // target model has catalog ownership.
    let turn_req = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "turn/start",
        "params": {
            "session_id": session_id,
            "prompt": "Follow up with new model",
            "model": "claude-sonnet-4-5"
        }
    });
    send_request(&mut writer, &turn_req).await;
    let turn_resp = read_response(&mut reader).await;
    assert!(
        turn_resp["error"].is_null(),
        "turn/start with model override failed: {turn_resp}"
    );
    let switched_text = turn_resp["result"]["text"].as_str().unwrap();
    assert!(
        switched_text.contains("model=claude-sonnet-4-5"),
        "After model switch, response should use claude-sonnet-4-5, got: {switched_text}"
    );

    drop(writer);
    server_handle.await.unwrap().unwrap();
}
