//! The whole command line from a workstation (spec §8.6, spectra#21 and

//! #23): the daemon on one machine, the owner, their agents and every

//! command on another, with no ledger file on the workstation at all.

mod common;

use common::{add_agent, delegate, hermetic, run, text, Serving};

use serde_json::{json, Value};

use spectra_core::keys::KeyPair;

use std::path::Path;

use std::process::Stdio;

use std::time::Duration;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

fn json_of(d: &Path, args: &[&str]) -> Value {

    let o = run(d, args);

    assert!(o.status.success(), "{args:?}: {}", text(&o));

    serde_json::from_slice(&o.stdout)

        .unwrap_or_else(|e| panic!("{args:?}: one JSON document ({e}): {}", String::from_utf8_lossy(&o.stdout)))

}

/// A server in `s`, claimed by the owner whose home is `c/owner`.

fn connected(s: &Path, c: &Path) -> (Serving, u16) {

    KeyPair::create_home(&c.join("owner"), "Andy McCall", "").unwrap();

    let owner = KeyPair::load_home(&c.join("owner")).unwrap();

    assert!(run(s, &["init", "--owner", &owner.key.to_hex(), "--name", "earth"]).status.success());

    let (srv, port) = Serving::start_tls(s);

    let url = format!("https://127.0.0.1:{port}");

    assert_eq!(run(c, &["connect", &url]).status.code(), Some(1));

    delegate(&c.join("owner"), &KeyPair::instance(&s.join("data"), "earth").unwrap());

    let o = run(c, &["connect", &url]);

    assert!(o.status.success(), "{}", text(&o));

    (srv, port)

}

#[tokio::test]

async fn every_command_works_from_another_machine() {

    let (sd, cd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

    let (s, c) = (sd.path(), cd.path());

    let (srv, _port) = connected(s, c);

    std::fs::create_dir_all(c.join("oxygen")).unwrap();

    let o = run(c, &["project", "add", c.join("oxygen").to_str().unwrap()]);

    assert!(o.status.success(), "{}", text(&o));

    add_agent(c, "claude-oxygen", &[]);

    add_agent(c, "spare", &[]);

    let id = json_of(c, &["identity", "show", "--json"]);

    assert_eq!(id["instance"]["name"], "earth");

    assert_eq!(id["instance"]["delegated"], true, "{id}");

    let agents = json_of(c, &["agent", "list", "--json"]);

    let claude = agents.as_array().unwrap().iter().find(|a| a["name"] == "claude-oxygen").unwrap();

    assert_eq!(claude["state"], "ok", "{claude}");

    assert_eq!(claude["display"], "claude-oxygen · Andy McCall's agent");

    let projects = json_of(c, &["project", "list", "--json"]);

    assert_eq!(projects[0]["name"], "oxygen");

    assert!(projects[0]["agents"].as_array().unwrap().contains(&json!("claude-oxygen")));

    let o = run(c, &["agent", "disable", "spare"]);

    assert!(o.status.success(), "{}", text(&o));

    assert_eq!(json_of(c, &["say", "--json", "-c", "general", "the release is tomorrow"])["channel"], "#general");

    assert!(json_of(c, &["inbox", "--json"]).is_array());

    assert_eq!(json_of(c, &["search", "--json", "release"]).as_array().unwrap().len(), 1);

    let brief: Value = serde_json::from_slice(&run(c, &["api", "GET", "/api/v1/brief"]).stdout).unwrap();

    assert_eq!(brief["actor"], "human");

    // An agent's shim on the workstation talks to the server.

    let mut cmd = tokio::process::Command::from(hermetic(c));

    let home = c.join("agents/claude-oxygen");

    cmd.args(["mcp", "--agent", "claude-oxygen", "--home", home.to_str().unwrap()])

        .stdin(Stdio::piped())

        .stdout(Stdio::piped())

        .stderr(Stdio::piped())

        .kill_on_drop(true);

    let mut shim = cmd.spawn().unwrap();

    let mut stdin = shim.stdin.take().unwrap();

    let mut lines = BufReader::new(shim.stdout.take().unwrap()).lines();

    for v in [

        json!({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "t", "version": "1" } } }),

        json!({ "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "brief", "arguments": {} } }),

    ] {

        stdin.write_all(format!("{v}\n").as_bytes()).await.unwrap();

    }

    stdin.flush().await.unwrap();

    let mut brief = Value::Null;

    for _ in 0..2 {

        let line = tokio::time::timeout(Duration::from_secs(20), lines.next_line()).await.unwrap().unwrap().unwrap();

        let v: Value = serde_json::from_str(&line).unwrap();

        if v["id"] == 2 {

            brief = v;

        }

    }

    assert!(brief["result"]["content"][0]["text"].as_str().unwrap().contains("You are claude-oxygen · Andy McCall's agent"), "{brief}");

    drop(shim);

    // The ledger stays on the server: verify asks it; check and rebuild refuse here.

    let v = json_of(c, &["admin", "verify", "--json"]);

    assert!(v["findings"].as_array().unwrap().is_empty(), "{v}");

    let o = run(c, &["admin", "check"]);

    assert!(!o.status.success() && text(&o).contains("runs on the server"), "{}", text(&o));

    assert!(!c.join("data").exists(), "the workstation has no data directory");

    srv.stop();

    let o = run(s, &["admin", "verify"]);

    assert!(o.status.success(), "{}", text(&o));

}

/// One HTTP/1.1 exchange with the proxy: status, head, body.

fn http(port: u16, method: &str, path: &str, headers: &[(&str, &str)], body: &str) -> (u16, String, String) {

    use std::io::{Read, Write};

    let mut s = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();

    s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();

    let mut req = format!("{method} {path} HTTP/1.1\r\nconnection: close\r\ncontent-length: {}\r\n", body.len());

    if !headers.iter().any(|(k, _)| k.eq_ignore_ascii_case("host")) {

        req.push_str(&format!("host: 127.0.0.1:{port}\r\n"));

    }

    for (k, v) in headers {

        req.push_str(&format!("{k}: {v}\r\n"));

    }

    req.push_str("\r\n");

    req.push_str(body);

    s.write_all(req.as_bytes()).unwrap();

    let mut out = Vec::new();

    let _ = s.read_to_end(&mut out);

    let text = String::from_utf8_lossy(&out).into_owned();

    let (head, body) = text.split_once("\r\n\r\n").unwrap_or((&text, ""));

    let status = head.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);

    (status, head.to_string(), body.to_string())

}

#[tokio::test]

async fn radar_through_the_loopback_proxy() {

    use std::io::BufRead;

    let (sd, cd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

    let (s, c) = (sd.path(), cd.path());

    let (srv, _port) = connected(s, c);

    let mut open = hermetic(c).args(["open", "--no-browser", "--port", "0"]).stdout(Stdio::piped()).spawn().unwrap();

    let out = open.stdout.take().unwrap();

    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {

        for line in std::io::BufReader::new(out).lines().map_while(Result::ok) {

            if let Some(u) = line.split_whitespace().find(|w| w.contains("/signin?code=")) {

                let _ = tx.send(u.to_string());

            }

        }

    });

    let link = rx.recv_timeout(Duration::from_secs(10)).expect("spectra open printed a link");

    let rest = link.strip_prefix("http://127.0.0.1:").expect("a loopback link");

    let (port, path) = rest.split_once('/').unwrap();

    let port: u16 = port.parse().unwrap();

    // Signing in through the proxy sets the session cookie.

    let (status, head, _) = http(port, "GET", &format!("/{path}"), &[], "");

    assert_eq!(status, 303, "{head}");

    let cookie = head

        .lines()

        .find_map(|l| {

            l.to_ascii_lowercase()

                .starts_with("set-cookie:")

                .then(|| l.split_once(':').unwrap().1.trim().split(';').next().unwrap().to_string())

        })

        .expect("a session cookie");

    // Live and a channel load; the channel's form carries the CSRF token.

    let (status, _, live) = http(port, "GET", "/", &[("cookie", &cookie)], "");

    assert_eq!(status, 200);

    assert!(live.contains("Spectra"), "{live}");

    let (status, _, page) = http(port, "GET", "/c/general", &[("cookie", &cookie)], "");

    assert_eq!(status, 200);

    let csrf = page.split("name=\"_csrf\" value=\"").nth(1).and_then(|r| r.split('"').next()).expect("a CSRF token").to_string();

    // A post from the proxy's own page goes through, as the owner.

    let origin = format!("http://127.0.0.1:{port}");

    let form = format!("_csrf={csrf}&body=hello+from+radar+over+the+proxy");

    let hdrs = [("cookie", cookie.as_str()), ("origin", origin.as_str()), ("content-type", "application/x-www-form-urlencoded")];

    let (status, head, _) = http(port, "POST", "/c/general", &hdrs, &form);

    assert_eq!(status, 303, "{head}");

    let hits = json_of(c, &["search", "--json", "proxy"]);

    assert_eq!(hits.as_array().unwrap().len(), 1, "{hits}");

    assert_eq!(hits[0]["author"], "human");

    // Another site cannot use the proxy: not by Origin, nor by Host.

    let evil = [("cookie", cookie.as_str()), ("origin", "http://evil.example"), ("content-type", "application/x-www-form-urlencoded")];

    assert_eq!(http(port, "POST", "/c/general", &evil, &form).0, 403);

    assert_eq!(http(port, "GET", "/", &[("host", "evil.example"), ("cookie", &cookie)], "").0, 403);

    open.kill().ok();

    open.wait().ok();

    srv.stop();

    let o = run(s, &["admin", "verify"]);

    assert!(o.status.success(), "{}", text(&o));

}