//! `spectra mcp --agent NAME`: the stdio shim (spec §15.1). A client starts
//! it as an ordinary stdio MCP server; it signs every request with the
//! agent's key from the agent's own home and forwards it to the daemon over
//! the Unix socket, so a client's configuration is one line with nothing
//! secret in it. Nothing is written to stdout but protocol.

use super::Ctx;
use anyhow::{bail, Context, Result};
use serde_json::{json, Value as Json};
use spectra_core::keys::KeyPair;
use spectra_core::ledger::Ledger;
use spectra_mcp::client::{rpc_result, Client};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::Mutex;

/// The revisions the shim speaks to its client (spec §15.1); 2026-07-28
/// requests are passed through as they are.
pub const STDIO_VERSIONS: &[&str] = &["2025-11-25", "2025-06-18", "2025-03-26"];

pub async fn run(ctx: &Ctx, agent: &str, home: Option<PathBuf>) -> Result<i32> {
    let home = match home {
        Some(h) => h,
        None if ctx.remote()?.is_some() => {
            bail!("with Spectra on another machine, name the agent's home: spectra mcp --agent {agent} --home DIR")
        }
        None => {
            let ledger = Ledger::open_read_only(&ctx.db()?).await?;
            let h: Option<String> =
                sqlx::query_scalar("SELECT home FROM agents WHERE name = ?").bind(agent).fetch_optional(ledger.pool()).await?;
            PathBuf::from(h.with_context(|| format!("no agent named {agent} (spectra agent list)"))?)
        }
    };
    let key = KeyPair::load_home(&home).with_context(|| format!("reading {agent}'s key in {}", home.display()))?;
    if key.identity.name != agent {
        bail!("{} holds {}'s key, not {agent}'s", home.display(), key.identity.name);
    }
    let client = Arc::new(Client::at(ctx.endpoint()?, Some(key)));
    let version = Arc::new(Mutex::new(STDIO_VERSIONS[0].to_string()));
    let out = Arc::new(Mutex::new(tokio::io::stdout()));
    let mut lines = BufReader::new(tokio::io::stdin()).lines();
    let mut tasks = tokio::task::JoinSet::new();
    while let Some(line) = lines.next_line().await? {
        if line.trim().is_empty() {
            continue;
        }
        let (client, version, out) = (client.clone(), version.clone(), out.clone());
        // Each request on its own: a `wait` must not hold up the others.
        tasks.spawn(async move {
            if let Some(reply) = forward(&client, &version, &line).await {
                let mut o = out.lock().await;
                let _ = o.write_all(format!("{reply}\n").as_bytes()).await;
                let _ = o.flush().await;
            }
        });
    }
    while tasks.join_next().await.is_some() {}
    Ok(0)
}

/// Forward one message; the reply to write, if it was a request.
async fn forward(client: &Client, version: &Mutex<String>, line: &str) -> Option<Json> {
    let req: Json = match serde_json::from_str(line) {
        Ok(v) => v,
        Err(e) => return Some(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32700, "message": format!("not JSON: {e}") } })),
    };
    let id = req.get("id").cloned();
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
    let asked_2026 = method == "server/discover"
        || req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str) == Some("2026-07-28");
    let v = if asked_2026 {
        "2026-07-28".to_string()
    } else if method == "initialize" {
        let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or(STDIO_VERSIONS[2]);
        let chosen = if STDIO_VERSIONS.contains(&asked) { asked } else { STDIO_VERSIONS[0] };
        *version.lock().await = chosen.to_string();
        chosen.to_string()
    } else {
        version.lock().await.clone()
    };
    let r = match client.mcp(&v, &req).await {
        Ok(r) => r,
        Err(e) => return id.map(|id| json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra is not reachable: {e:#}. Is `spectra serve` running?") } })),
    };
    let id = id?;
    if r.status == 202 {
        return None;
    }
    match rpc_result(&r) {
        Ok(v) => Some(v),
        Err(e) => Some(
            json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra answered {}: {e:#}", r.status) } }),
        ),
    }
}