//! `mcptest exec --connection-server` co-process mode.
//!
//! This command spawns the configured MCP server, owns one MCP
//! connection over its stdio, and serves newline-delimited JSON-RPC
//! requests on its own stdin/stdout. SDK clients (Python pytest plugin,
//! TypeScript Vitest plugin) speak the IPC schema published at
//! `schemas/ipc/v1.json`.
//!
//! Why a thin wrapper in the CLI crate and a fat dispatcher in core:
//! the dispatcher is pure logic over the [`McpBackend`] trait. The CLI
//! supplies the concrete backend (a child process plus a protocol
//! Client) and the stdin/stdout pipes. Keeping the wiring small here
//! means the integration tests under `crates/mcptest-core/tests` cover
//! the dispatcher end-to-end without spawning a child.

use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Mutex;

use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use mcptest_core::exec::dispatcher::BackendError;
use mcptest_core::exec::{run, Dispatcher, McpBackend, SessionOptions};
use mcptest_core::protocol::{
    Client, ClientCapabilities, JsonRpcMessage, ProtocolVersion, Transport,
};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::mpsc;

/// CLI-facing entry point for `mcptest exec --connection-server`.
///
/// Returns the desired process exit code. `0` on a clean session,
/// non-zero when the server connection could not be established (which
/// also gets surfaced as a JSON-RPC error on the first request).
pub async fn run_exec(args: ExecRunArgs) -> Result<i32> {
    if !args.connection_server {
        return Err(anyhow!(
            "mcptest exec requires --connection-server in v1.1; future modes will land in a follow-up ticket"
        ));
    }
    let backend = ChildServerBackend::spawn(&args)
        .await
        .context("failed to spawn upstream MCP server")?;
    let dispatcher = Dispatcher::new(backend);
    let stdin = tokio::io::stdin();
    let stdout = tokio::io::stdout();
    run(
        BufReader::new(stdin),
        stdout,
        dispatcher,
        SessionOptions::default(),
    )
    .await
    .map(|_| 0)
    .map_err(Into::into)
}

/// Inputs for [`run_exec`]. Held as a struct so the CLI parser can pass
/// every relevant flag without expanding the function signature.
#[derive(Debug, Clone, Default)]
pub struct ExecRunArgs {
    /// Required `--connection-server` toggle. Only IPC mode ships in
    /// v1.1; the field stays to leave room for `--http` and friends
    /// later.
    pub connection_server: bool,
    /// Path to the mcptest config (defaults to `mcptest.yaml`).
    pub config: Option<PathBuf>,
    /// Optional path the dispatcher writes a verbatim transcript to.
    /// Used for SDK debugging; production runs leave it `None`.
    pub debug_output: Option<PathBuf>,
    /// Verbose logging on stderr.
    pub verbose: bool,
    /// IPC envelope version pinned by the SDK. Errors out when the
    /// requested version is unknown.
    pub ipc_version: Option<String>,
    /// Disable cache reads and writes for this session.
    pub no_cache: bool,
    /// Disable cassette replay and recording for this session.
    pub no_cassette: bool,
    /// Record new cassettes for any unmocked call.
    pub record_cassettes: bool,
    /// Full command line to spawn the upstream MCP server, in argv form.
    /// Required for v1.1; future config-driven discovery will populate
    /// this from the YAML.
    pub server_command: Vec<String>,
}

/// Backend that owns the spawned child MCP server and drives an MCP
/// protocol [`Client`] over its stdio.
///
/// Kept private to this module because every public knob it needs is
/// already exposed through [`ExecRunArgs`].
struct ChildServerBackend {
    client: Client,
    child: Mutex<Option<Child>>,
    cache_mode: Mutex<String>,
    cassette_mode: Mutex<String>,
    upload_url: Mutex<Option<String>>,
}

impl ChildServerBackend {
    async fn spawn(args: &ExecRunArgs) -> Result<Self> {
        if args.server_command.is_empty() {
            return Err(anyhow!(
                "exec mode needs a server command; pass `--config mcptest.yaml` with `server.command` populated"
            ));
        }
        let (bin, rest) = args
            .server_command
            .split_first()
            .ok_or_else(|| anyhow!("server command is empty"))?;
        let mut child = Command::new(bin)
            .args(rest)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit())
            .spawn()
            .with_context(|| format!("spawn `{bin}`"))?;
        let child_stdin = child
            .stdin
            .take()
            .ok_or_else(|| anyhow!("child stdin was not piped"))?;
        let child_stdout = child
            .stdout
            .take()
            .ok_or_else(|| anyhow!("child stdout was not piped"))?;
        let (outbound_tx, mut outbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (inbound_tx, inbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        // Writer task: drain outbound channel, write framed JSON to
        // the child's stdin. Newline-delimited JSON-RPC framing.
        tokio::spawn(async move {
            let mut stdin = child_stdin;
            while let Some(msg) = outbound_rx.recv().await {
                let Ok(mut buf) = serde_json::to_vec(&msg) else {
                    break;
                };
                buf.push(b'\n');
                if stdin.write_all(&buf).await.is_err() {
                    break;
                }
                if stdin.flush().await.is_err() {
                    break;
                }
            }
        });
        // Reader task: parse newline-delimited JSON envelopes from the
        // child's stdout and push them onto the inbound channel.
        tokio::spawn(async move {
            let mut reader = BufReader::new(child_stdout);
            let mut line = String::new();
            loop {
                line.clear();
                let n = match reader.read_line(&mut line).await {
                    Ok(n) => n,
                    Err(_) => break,
                };
                if n == 0 {
                    break;
                }
                let trimmed = line.trim();
                if trimmed.is_empty() {
                    continue;
                }
                let Ok(msg): Result<JsonRpcMessage, _> = serde_json::from_str(trimmed) else {
                    continue;
                };
                if inbound_tx.send(msg).is_err() {
                    break;
                }
            }
        });
        let transport = Transport::new(outbound_tx, inbound_rx);
        let client = Client::builder(transport)
            .with_client_capabilities(ClientCapabilities::default())
            .build();
        // Drive the MCP handshake eagerly so the first SDK call does
        // not race the server's initialize response.
        client
            .initialize(ProtocolVersion::current())
            .await
            .context("MCP initialize handshake")?;
        Ok(Self {
            client,
            child: Mutex::new(Some(child)),
            cache_mode: Mutex::new(if args.no_cache {
                "off".into()
            } else {
                "readwrite".into()
            }),
            cassette_mode: Mutex::new(match (args.no_cassette, args.record_cassettes) {
                (true, _) => "off".into(),
                (false, true) => "record".into(),
                _ => "replay".into(),
            }),
            upload_url: Mutex::new(None),
        })
    }
}

#[async_trait]
impl McpBackend for ChildServerBackend {
    async fn initialize(&self) -> Result<Value, BackendError> {
        Ok(json!({
            "name": "mcptest-exec",
            "version": mcptest_core::version::semver(),
        }))
    }

    async fn call_tool(&self, tool: &str, arguments: Value) -> Result<Value, BackendError> {
        self.client
            .request("tools/call", json!({"name": tool, "arguments": arguments}))
            .await
            .map_err(map_protocol_error)
    }

    async fn list_tools(&self) -> Result<Value, BackendError> {
        self.client
            .request("tools/list", json!({}))
            .await
            .map_err(map_protocol_error)
    }

    async fn list_resources(&self) -> Result<Value, BackendError> {
        self.client
            .request("resources/list", json!({}))
            .await
            .map_err(map_protocol_error)
    }

    async fn read_resource(&self, uri: &str) -> Result<Value, BackendError> {
        self.client
            .request("resources/read", json!({"uri": uri}))
            .await
            .map_err(map_protocol_error)
    }

    async fn call_prompt(&self, name: &str, arguments: Value) -> Result<Value, BackendError> {
        self.client
            .request("prompts/get", json!({"name": name, "arguments": arguments}))
            .await
            .map_err(map_protocol_error)
    }

    async fn reporter_set_upload(
        &self,
        url: &str,
        _token: Option<&str>,
    ) -> Result<(), BackendError> {
        if let Ok(mut slot) = self.upload_url.lock() {
            *slot = Some(url.to_owned());
        }
        Ok(())
    }

    async fn reporter_flush(&self) -> Result<(), BackendError> {
        // The upload reporter writes to a configured collector in core;
        // for v1.1 we accept the flush, defer the actual HTTPS POST to
        // the runner integration in a follow-up.
        Ok(())
    }

    async fn cache_set_mode(&self, mode: &str) -> Result<(), BackendError> {
        validate_mode(mode, &["read", "write", "readwrite", "off"])?;
        if let Ok(mut slot) = self.cache_mode.lock() {
            *slot = mode.to_owned();
        }
        Ok(())
    }

    async fn cassette_set_mode(&self, mode: &str) -> Result<(), BackendError> {
        validate_mode(mode, &["record", "replay", "off"])?;
        if let Ok(mut slot) = self.cassette_mode.lock() {
            *slot = mode.to_owned();
        }
        Ok(())
    }

    async fn shutdown(&self) -> Result<(), BackendError> {
        if let Ok(mut slot) = self.child.lock() {
            if let Some(mut child) = slot.take() {
                let _ = child.start_kill();
            }
        }
        Ok(())
    }
}

fn validate_mode(mode: &str, accepted: &[&str]) -> Result<(), BackendError> {
    if accepted.contains(&mode) {
        Ok(())
    } else {
        Err(BackendError::InvalidParams(format!(
            "mode `{mode}` not in {accepted:?}"
        )))
    }
}

fn map_protocol_error(err: mcptest_core::protocol::ProtocolError) -> BackendError {
    use mcptest_core::protocol::ProtocolError;
    match err {
        ProtocolError::TransportClosed(reason) => BackendError::Disconnected { reason },
        ProtocolError::ServerError { message, data, .. } => BackendError::Server { message, data },
        other => BackendError::Internal(other.to_string()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn validate_mode_rejects_unknown_value() {
        let err = validate_mode("nope", &["on", "off"]).unwrap_err();
        match err {
            BackendError::InvalidParams(message) => assert!(message.contains("nope")),
            other => panic!("expected InvalidParams, got {other:?}"),
        }
    }

    #[test]
    fn validate_mode_accepts_known_value() {
        validate_mode("on", &["on", "off"]).unwrap();
    }
}
