use assert_cmd::cargo::cargo_bin;
use serde_json::{Value, json};
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use tempfile::tempdir;

#[test]
fn mcp_server_should_initialize_list_tools_and_run_lifecycle_calls() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let repo_dir = temp.path().join("repo");
    let config_path = temp.path().join("spool.toml");
    let docs_dir = temp.path().join("docs");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::create_dir_all(&repo_dir).unwrap();
    fs::create_dir_all(&docs_dir).unwrap();
    fs::write(
        docs_dir.join("MCP_PROMPTS_ROUND_8_PLAN.md"),
        "Round 8 smoke doc",
    )
    .unwrap();
    fs::write(
        docs_dir.join("SESSION_HANDOFF.md"),
        "Session handoff smoke doc",
    )
    .unwrap();
    fs::write(
        vault_dir.join("10-Projects/spool.md"),
        "# spool\n\ncontext\n",
    )
    .unwrap();
    fs::write(
        &config_path,
        format!(
            r#"[vault]
root = "{}"

[output]
default_format = "markdown"
max_chars = 12000
max_notes = 8

[[projects]]
id = "spool"
name = "spool"
repo_paths = ["{}"]
note_roots = ["10-Projects"]
"#,
            vault_dir.display(),
            repo_dir.display()
        ),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    );
    let initialize = read_message(&mut reader);
    assert_eq!(initialize["result"]["protocolVersion"], json!("2025-03-26"));
    assert_eq!(
        initialize["result"]["capabilities"]["prompts"]["listChanged"],
        json!(false)
    );
    assert_eq!(
        initialize["result"]["capabilities"]["resources"]["listChanged"],
        json!(false)
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {}
        }),
    );
    let tools = read_message(&mut reader);
    assert!(
        tools["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .any(|tool| tool["name"] == "memory_propose")
    );
    assert!(
        tools["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .any(|tool| tool["name"] == "memory_search")
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 21,
            "method": "prompts/list",
            "params": {}
        }),
    );
    let prompts = read_message(&mut reader);
    assert!(
        prompts["result"]["prompts"]
            .as_array()
            .unwrap()
            .iter()
            .any(|prompt| prompt["name"] == "review_lifecycle_queue")
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 22,
            "method": "resources/list",
            "params": {}
        }),
    );
    let resources = read_message(&mut reader);
    assert!(
        resources["result"]["resources"]
            .as_array()
            .unwrap()
            .iter()
            .any(|resource| resource["uri"] == "spool://docs/session-handoff")
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 23,
            "method": "resources/read",
            "params": {
                "uri": "spool://docs/mcp-prompts-round-8"
            }
        }),
    );
    let resource = read_message(&mut reader);
    assert_eq!(
        resource["result"]["contents"][0]["mimeType"],
        json!("text/markdown")
    );
    assert!(
        resource["result"]["contents"][0]["text"]
            .as_str()
            .unwrap()
            .contains("Round 8")
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "memory_propose",
                "arguments": {
                    "title": "测试偏好",
                    "summary": "先 smoke 再收口",
                    "memory_type": "workflow",
                    "scope": "user",
                    "source_ref": "session:1",
                    "user_id": "long",
                    "actor": "codex",
                    "reason": "captured during MCP review",
                    "evidence_refs": ["session:1", "obsidian://workflow"]
                }
            }
        }),
    );
    let propose = read_message(&mut reader);
    let record_id = propose["result"]["structuredContent"]["entry"]["record_id"]
        .as_str()
        .unwrap()
        .to_string();
    assert_eq!(
        propose["result"]["structuredContent"]["entry"]["metadata"]["actor"],
        json!("codex")
    );
    assert_eq!(
        propose["result"]["structuredContent"]["summary"]["state"],
        json!("candidate")
    );
    assert_eq!(
        propose["result"]["structuredContent"]["summary"]["pending_review"],
        json!(true)
    );
    assert_eq!(
        propose["result"]["structuredContent"]["summary"]["actor"],
        json!("codex")
    );
    assert_eq!(
        propose["result"]["structuredContent"]["summary"]["evidence_refs"],
        json!(["session:1", "obsidian://workflow"])
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 4,
            "method": "tools/call",
            "params": {
                "name": "memory_accept",
                "arguments": {
                    "record_id": record_id,
                    "actor": "long",
                    "reason": "approved after review"
                }
            }
        }),
    );
    let accept = read_message(&mut reader);
    assert_eq!(
        accept["result"]["structuredContent"]["entry"]["record"]["state"],
        json!("accepted")
    );
    assert_eq!(
        accept["result"]["structuredContent"]["entry"]["metadata"]["reason"],
        json!("approved after review")
    );
    assert_eq!(
        accept["result"]["structuredContent"]["action"],
        json!("accept")
    );
    assert_eq!(
        accept["result"]["structuredContent"]["summary"]["state"],
        json!("accepted")
    );
    assert_eq!(
        accept["result"]["structuredContent"]["summary"]["pending_review"],
        json!(false)
    );
    assert_eq!(
        accept["result"]["structuredContent"]["summary"]["wakeup_ready"],
        json!(true)
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 5,
            "method": "tools/call",
            "params": {
                "name": "memory_search",
                "arguments": {
                    "task": "spool context",
                    "cwd": repo_dir,
                    "target": "codex"
                }
            }
        }),
    );
    let search = read_message(&mut reader);
    assert!(
        search["result"]["structuredContent"]["rendered"]
            .as_str()
            .unwrap()
            .contains("Codex")
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_lifecycle_reads_should_use_daemon_when_configured() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    fs::create_dir_all(&vault_dir).unwrap();
    let config_path = temp.path().join("spool.toml");
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let create = Command::new(cargo_bin("spool"))
        .args([
            "memory",
            "propose",
            "--config",
            config_path.to_str().unwrap(),
            "--title",
            "测试偏好",
            "--summary",
            "先 smoke 再收口",
            "--memory-type",
            "workflow",
            "--scope",
            "user",
            "--source-ref",
            "session:1",
            "--user-id",
            "long",
        ])
        .output()
        .unwrap();
    assert!(create.status.success());
    let stdout = String::from_utf8(create.stdout).unwrap();
    let record_id = stdout
        .lines()
        .find(|line| line.contains("record_id"))
        .unwrap()
        .split('`')
        .nth(1)
        .unwrap()
        .to_string();

    let daemon_bin = cargo_bin("spool-daemon");
    let (child, mut stdin, mut reader) = start_mcp(&config_path, Some(&daemon_bin));
    initialize_mcp(&mut stdin, &mut reader);

    let review_queue = call_tool(&mut stdin, &mut reader, 2, "memory_review_queue", json!({}));
    assert_eq!(review_queue["result"]["isError"], json!(false));
    assert_eq!(
        review_queue["result"]["structuredContent"]["pending_review"]
            .as_array()
            .unwrap()
            .len(),
        1
    );

    let wakeup_ready = call_tool(&mut stdin, &mut reader, 3, "memory_wakeup_ready", json!({}));
    assert_eq!(wakeup_ready["result"]["isError"], json!(false));

    let get = call_tool(
        &mut stdin,
        &mut reader,
        4,
        "memory_get",
        json!({ "record_id": record_id }),
    );
    assert_eq!(get["result"]["isError"], json!(false));
    assert_eq!(
        get["result"]["structuredContent"]["record"]["record"]["state"],
        json!("candidate")
    );

    let history = call_tool(
        &mut stdin,
        &mut reader,
        5,
        "memory_history",
        json!({ "record_id": record_id }),
    );
    assert_eq!(history["result"]["isError"], json!(false));
    assert_eq!(
        history["result"]["structuredContent"]["history"]
            .as_array()
            .unwrap()
            .len(),
        1
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_lifecycle_reads_should_fallback_when_daemon_is_unavailable() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    fs::create_dir_all(&vault_dir).unwrap();
    let config_path = temp.path().join("spool.toml");
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let create = Command::new(cargo_bin("spool"))
        .args([
            "memory",
            "propose",
            "--config",
            config_path.to_str().unwrap(),
            "--title",
            "测试偏好",
            "--summary",
            "先 smoke 再收口",
            "--memory-type",
            "workflow",
            "--scope",
            "user",
            "--source-ref",
            "session:1",
            "--user-id",
            "long",
        ])
        .output()
        .unwrap();
    assert!(create.status.success());
    let stdout = String::from_utf8(create.stdout).unwrap();
    let record_id = stdout
        .lines()
        .find(|line| line.contains("record_id"))
        .unwrap()
        .split('`')
        .nth(1)
        .unwrap()
        .to_string();

    let missing_daemon = temp.path().join("missing-spool-daemon");
    let (child, mut stdin, mut reader) = start_mcp(&config_path, Some(&missing_daemon));
    initialize_mcp(&mut stdin, &mut reader);

    let review_queue = call_tool(&mut stdin, &mut reader, 2, "memory_review_queue", json!({}));
    assert_eq!(review_queue["result"]["isError"], json!(false));
    assert_eq!(
        review_queue["result"]["structuredContent"]["pending_review"]
            .as_array()
            .unwrap()
            .len(),
        1
    );

    let get = call_tool(
        &mut stdin,
        &mut reader,
        3,
        "memory_get",
        json!({ "record_id": record_id }),
    );
    assert_eq!(get["result"]["isError"], json!(false));
    assert_eq!(
        get["result"]["structuredContent"]["record"]["record"]["state"],
        json!("candidate")
    );

    let history = call_tool(
        &mut stdin,
        &mut reader,
        4,
        "memory_history",
        json!({ "record_id": record_id }),
    );
    assert_eq!(history["result"]["isError"], json!(false));

    stop_mcp(child, stdin);
}

fn start_mcp(
    config_path: &std::path::Path,
    daemon_bin: Option<&std::path::Path>,
) -> (Child, ChildStdin, BufReader<ChildStdout>) {
    let mut command = Command::new(cargo_bin("spool-mcp"));
    command.args(["--config", config_path.to_str().unwrap()]);
    if let Some(daemon_bin) = daemon_bin {
        command.args(["--daemon-bin", daemon_bin.to_str().unwrap()]);
    }
    let mut child = command
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .spawn()
        .unwrap();
    let stdin = child.stdin.take().unwrap();
    let stdout = child.stdout.take().unwrap();
    (child, stdin, BufReader::new(stdout))
}

fn initialize_mcp(stdin: &mut ChildStdin, reader: &mut BufReader<ChildStdout>) {
    write_message(
        stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    );
    let initialize = read_message(reader);
    assert_eq!(initialize["result"]["protocolVersion"], json!("2025-03-26"));
}

fn call_tool(
    stdin: &mut ChildStdin,
    reader: &mut BufReader<ChildStdout>,
    id: i64,
    name: &str,
    arguments: Value,
) -> Value {
    write_message(
        stdin,
        json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": "tools/call",
            "params": {
                "name": name,
                "arguments": arguments
            }
        }),
    );
    read_message(reader)
}

#[test]
fn mcp_memory_import_session_should_error_for_missing_provider_session() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp(&mut stdin, &mut reader);

    let bad_provider = call_tool(
        &mut stdin,
        &mut reader,
        2,
        "memory_import_session",
        json!({ "provider": "opencode", "session_id": "nope", "apply": false }),
    );
    assert!(bad_provider["error"].is_object());

    let missing_session = call_tool(
        &mut stdin,
        &mut reader,
        3,
        "memory_import_session",
        json!({ "provider": "claude", "session_id": "definitely-not-there-zzz", "apply": false }),
    );
    let is_error = missing_session["result"]["isError"]
        .as_bool()
        .unwrap_or(false);
    let has_error = missing_session["error"].is_object();
    assert!(
        is_error || has_error,
        "expected error for missing session: {missing_session}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_should_return_jsonrpc_error_for_malformed_json_and_continue() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    fs::create_dir_all(&vault_dir).unwrap();
    let config_path = temp.path().join("spool.toml");
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);

    stdin.write_all(b"{not json}\n").unwrap();
    stdin.flush().unwrap();
    let invalid = read_message(&mut reader);
    assert_eq!(invalid["jsonrpc"], json!("2.0"));
    assert_eq!(invalid["error"]["code"], json!(-32700));
    assert!(
        invalid["error"]["message"]
            .as_str()
            .unwrap()
            .contains("parse error")
    );

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {}
        }),
    );
    let initialize = read_message(&mut reader);
    assert_eq!(initialize["result"]["protocolVersion"], json!("2025-03-26"));

    stop_mcp(child, stdin);
}

fn stop_mcp(mut child: Child, stdin: ChildStdin) {
    drop(stdin);
    let status = child.wait().unwrap();
    assert!(status.success());
}

#[test]
fn mcp_should_return_jsonrpc_parse_error_and_continue_serving() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let repo_dir = temp.path().join("repo");
    let config_path = temp.path().join("spool.toml");
    let docs_dir = temp.path().join("docs");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::create_dir_all(&repo_dir).unwrap();
    fs::create_dir_all(&docs_dir).unwrap();
    fs::write(
        docs_dir.join("MCP_PROMPTS_ROUND_8_PLAN.md"),
        "Round 8 smoke doc",
    )
    .unwrap();
    fs::write(
        docs_dir.join("SESSION_HANDOFF.md"),
        "Session handoff smoke doc",
    )
    .unwrap();
    fs::write(
        vault_dir.join("10-Projects/spool.md"),
        "# spool\n\ncontext\n",
    )
    .unwrap();
    fs::write(
        &config_path,
        format!(
            r#"[vault]
root = "{}"

[output]
default_format = "markdown"
max_chars = 12000
max_notes = 8

[[projects]]
id = "spool"
name = "spool"
repo_paths = ["{}"]
note_roots = ["10-Projects"]
"#,
            vault_dir.display(),
            repo_dir.display()
        ),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);

    stdin
        .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}\n")
        .unwrap();
    stdin.flush().unwrap();

    let error = read_message(&mut reader);
    assert_eq!(error["jsonrpc"], json!("2.0"));
    assert_eq!(error["error"]["code"], json!(-32700));
    assert!(!error["error"]["message"].as_str().unwrap().is_empty());

    write_message(
        &mut stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "initialize",
            "params": {}
        }),
    );
    let initialize = read_message(&mut reader);
    assert_eq!(initialize["result"]["protocolVersion"], json!("2025-03-26"));

    stop_mcp(child, stdin);
}

fn write_message(stdin: &mut std::process::ChildStdin, value: Value) {
    serde_json::to_writer(&mut *stdin, &value).unwrap();
    stdin.write_all(b"\n").unwrap();
    stdin.flush().unwrap();
}

fn read_message(reader: &mut BufReader<std::process::ChildStdout>) -> Value {
    let mut line = String::new();
    reader.read_line(&mut line).unwrap();
    serde_json::from_str(line.trim()).unwrap()
}

/// Initialize variant that advertises `capabilities.sampling = {}`,
/// flipping `ServerState::client_supports_sampling()` to `true` so
/// `memory_distill_pending` will attempt the reverse sampling call.
fn initialize_mcp_with_sampling(stdin: &mut ChildStdin, reader: &mut BufReader<ChildStdout>) {
    write_message(
        stdin,
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "capabilities": { "sampling": {} }
            }
        }),
    );
    let initialize = read_message(reader);
    assert_eq!(initialize["result"]["protocolVersion"], json!("2025-03-26"));
}

/// Like [`call_tool`], but also fulfills any server→client
/// `sampling/createMessage` reverse requests interleaved before the
/// tool response. `mock_sampling` receives the raw inbound request
/// envelope (so tests can inspect the rendered prompt) and must
/// return a complete JSON-RPC response envelope to write back.
fn call_tool_with_sampling<F>(
    stdin: &mut ChildStdin,
    reader: &mut BufReader<ChildStdout>,
    id: i64,
    name: &str,
    arguments: Value,
    mut mock_sampling: F,
) -> Value
where
    F: FnMut(&Value) -> Value,
{
    write_message(
        stdin,
        json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": "tools/call",
            "params": {
                "name": name,
                "arguments": arguments
            }
        }),
    );
    loop {
        let msg = read_message(reader);
        if msg.get("method").and_then(Value::as_str) == Some("sampling/createMessage") {
            let response = mock_sampling(&msg);
            write_message(stdin, response);
            continue;
        }
        return msg;
    }
}

#[test]
fn mcp_memory_distill_pending_should_persist_self_tag_via_tier1_pipeline() {
    // End-to-end smoke for R4a: an MCP client calls
    // `memory_distill_pending`, spool scans the supplied transcript
    // with Tier 1 heuristics, and the resulting accepted record shows
    // up in `memory_wakeup_ready` immediately afterwards. R4b will add
    // a sampling-first path; this test pins the Tier 1 fallback so we
    // never regress the always-available baseline.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let repo_dir = temp.path().join("repo");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::create_dir_all(&repo_dir).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    // Synthesize a minimal Claude Code transcript with a single user
    // message containing a Chinese self-tag marker. The pipeline
    // promotes that to an accepted memory.
    let transcript_path = temp.path().join("session.jsonl");
    fs::write(
        &transcript_path,
        "{\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":\"记一下：MCP distill_pending 走 Tier 1 启发式\"}}\n",
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp(&mut stdin, &mut reader);

    let response = call_tool(
        &mut stdin,
        &mut reader,
        2,
        "memory_distill_pending",
        json!({
            "cwd": repo_dir.to_str().unwrap(),
            "transcript_path": transcript_path.to_str().unwrap(),
        }),
    );
    let result = response["result"].clone();
    assert_eq!(result["isError"].as_bool(), Some(false));
    let structured = result["structuredContent"].clone();
    assert_eq!(structured["fallback_used"], json!("tier1_heuristic"));
    assert_eq!(structured["sampling_attempted"], json!(false));
    assert_eq!(
        structured["signals_persisted"]
            .as_array()
            .map(|a| a.len())
            .unwrap_or(0),
        1,
        "exactly one self-tag accepted: {result}"
    );

    // Re-read the wakeup pool over the same MCP session: the new
    // record should be reachable without any extra plumbing.
    let wakeup = call_tool(&mut stdin, &mut reader, 3, "memory_wakeup_ready", json!({}));
    let queue = &wakeup["result"]["structuredContent"]["wakeup_ready"];
    let count = queue.as_array().map(|a| a.len()).unwrap_or(0);
    assert!(
        count >= 1,
        "wakeup_ready must include the new accepted record: {wakeup}"
    );

    // Calling distill_pending again with the SAME transcript must be
    // idempotent — the pipeline dedupes against existing summaries.
    let second = call_tool(
        &mut stdin,
        &mut reader,
        4,
        "memory_distill_pending",
        json!({
            "cwd": repo_dir.to_str().unwrap(),
            "transcript_path": transcript_path.to_str().unwrap(),
        }),
    );
    let second_struct = second["result"]["structuredContent"].clone();
    assert_eq!(
        second_struct["signals_persisted"]
            .as_array()
            .map(|a| a.len())
            .unwrap_or(0),
        0,
        "second call must dedupe: {second}"
    );
    assert!(
        second_struct["signals_duplicate_dropped"]
            .as_u64()
            .unwrap_or(0)
            >= 1,
        "second call must report at least one dedupe drop: {second}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_memory_distill_pending_should_reject_relative_cwd() {
    // R4a contract: cwd MUST be absolute. We surface this as a
    // JSON-RPC -32602 (invalid params) so clients can show a clear
    // error instead of silently scanning some unintended directory.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp(&mut stdin, &mut reader);

    let response = call_tool(
        &mut stdin,
        &mut reader,
        2,
        "memory_distill_pending",
        json!({ "cwd": "relative/repo" }),
    );
    assert!(
        response["error"].is_object(),
        "relative cwd must produce a JSON-RPC error: {response}"
    );
    assert_eq!(response["error"]["code"], json!(-32602));

    stop_mcp(child, stdin);
}

#[test]
fn mcp_memory_distill_pending_should_persist_sampling_candidates_when_supported() {
    // R4b end-to-end: the client advertises `sampling` capability,
    // the server issues a `sampling/createMessage` reverse request,
    // the harness mocks the LLM by returning a JSON array of
    // candidates, and `memory_distill_pending` must persist those as
    // ledger candidates (state=candidate) reachable via `memory_get`.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let repo_dir = temp.path().join("repo");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::create_dir_all(&repo_dir).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    // Transcript without any explicit self-tag — sampling is the
    // only path that can produce candidates.
    let transcript_path = temp.path().join("session.jsonl");
    fs::write(
        &transcript_path,
        "{\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":\"random conversation about deployment\"}}\n",
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp_with_sampling(&mut stdin, &mut reader);

    let response = call_tool_with_sampling(
        &mut stdin,
        &mut reader,
        2,
        "memory_distill_pending",
        json!({
            "cwd": repo_dir.to_str().unwrap(),
            "transcript_path": transcript_path.to_str().unwrap(),
        }),
        |request| {
            // Verify the server actually rendered a non-empty
            // sampling prompt before pretending to be the LLM.
            assert_eq!(
                request["method"].as_str(),
                Some("sampling/createMessage"),
                "harness only handles createMessage: {request}"
            );
            let prompt = request["params"]["messages"][0]["content"]["text"]
                .as_str()
                .unwrap_or("");
            assert!(!prompt.is_empty(), "server must send a prompt: {request}");
            json!({
                "jsonrpc": "2.0",
                "id": request["id"].clone(),
                "result": {
                    "role": "assistant",
                    "content": {
                        "type": "text",
                        "text": "[{\"kind\":\"incident\",\"summary\":\"sampling-derived smoke incident\"}]"
                    }
                }
            })
        },
    );

    let result = response["result"].clone();
    assert_eq!(
        result["isError"].as_bool(),
        Some(false),
        "tool must succeed when sampling returns valid JSON: {response}"
    );
    let structured = result["structuredContent"].clone();
    assert_eq!(structured["sampling_attempted"], json!(true));
    let fallback = structured["fallback_used"].as_str().unwrap_or("");
    assert!(
        fallback.starts_with("sampling+tier1_combined")
            || fallback.starts_with("sampling_no_candidates"),
        "fallback_used must reflect sampling success path: {structured}"
    );
    let signals_persisted = structured["candidates_persisted"]
        .as_array()
        .map(|a| a.len())
        .unwrap_or(0);
    assert!(
        signals_persisted >= 1,
        "at least one sampling candidate must be persisted: {structured}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_memory_distill_pending_should_fall_back_to_tier1_when_sampling_rejected() {
    // R4b fallback contract: when the client refuses the sampling
    // request (returns an error envelope with a "rejected" message),
    // the server must record `sampling_attempted = true`, mark
    // `fallback_used` with the `sampling_failed:` prefix, and still
    // produce the Tier 1 self-tag candidate — i.e. sampling rejection
    // never blocks the always-available baseline.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let repo_dir = temp.path().join("repo");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::create_dir_all(&repo_dir).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let transcript_path = temp.path().join("session.jsonl");
    fs::write(
        &transcript_path,
        "{\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":\"记一下：sampling 拒绝降级 Tier 1\"}}\n",
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp_with_sampling(&mut stdin, &mut reader);

    let response = call_tool_with_sampling(
        &mut stdin,
        &mut reader,
        2,
        "memory_distill_pending",
        json!({
            "cwd": repo_dir.to_str().unwrap(),
            "transcript_path": transcript_path.to_str().unwrap(),
        }),
        |request| {
            json!({
                "jsonrpc": "2.0",
                "id": request["id"].clone(),
                "error": {
                    "code": -32603,
                    "message": "user rejected sampling request"
                }
            })
        },
    );

    let result = response["result"].clone();
    assert_eq!(
        result["isError"].as_bool(),
        Some(false),
        "rejection must not surface as a tool error: {response}"
    );
    let structured = result["structuredContent"].clone();
    assert_eq!(structured["sampling_attempted"], json!(true));
    let fallback = structured["fallback_used"].as_str().unwrap_or("");
    assert!(
        fallback.starts_with("sampling_failed:"),
        "fallback_used must record the sampling failure: {structured}"
    );
    let signals_persisted = structured["signals_persisted"]
        .as_array()
        .map(|a| a.len())
        .unwrap_or(0);
    assert_eq!(
        signals_persisted, 1,
        "tier1 must still pick up the self-tag: {structured}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_memory_lint_should_return_clean_report_on_empty_ledger() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(&vault_dir).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp(&mut stdin, &mut reader);

    let tools = call_tool(&mut stdin, &mut reader, 2, "memory_lint", json!({}));
    let result = tools["result"].clone();
    assert_eq!(
        result["total_active_records"].as_u64(),
        Some(0),
        "empty ledger must report zero active records: {tools}"
    );
    assert_eq!(
        result["prune_suggestions"].as_array().map(Vec::len),
        Some(0),
        "empty ledger must have no prune suggestions: {tools}"
    );
    assert_eq!(
        result["broken_cross_refs"].as_array().map(Vec::len),
        Some(0),
        "empty ledger must have no broken cross-refs: {tools}"
    );
    assert_eq!(
        result["orphan_notes"].as_array().map(Vec::len),
        Some(0),
        "empty ledger must have no orphan notes: {tools}"
    );
    let markdown = result["markdown"].as_str().unwrap_or("");
    assert!(
        markdown.contains("知识库干净"),
        "clean report markdown must mention 知识库干净: {markdown}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_memory_lint_should_report_broken_cross_refs() {
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(&vault_dir).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp(&mut stdin, &mut reader);

    // Seed an accepted manual memory whose related_records points at
    // a record_id that does not exist in the ledger — that is exactly
    // what `detect_broken_cross_refs` flags.
    let create = call_tool(
        &mut stdin,
        &mut reader,
        2,
        "memory_record_manual",
        json!({
            "title": "lint 测试记录",
            "summary": "has a broken cross-ref",
            "memory_type": "preference",
            "scope": "user",
            "source_ref": "manual:lint",
            "actor": "long",
            "related_records": ["rec-does-not-exist"]
        }),
    );
    assert_eq!(
        create["result"]["isError"].as_bool(),
        Some(false),
        "seeding record must succeed: {create}"
    );

    let lint = call_tool(&mut stdin, &mut reader, 3, "memory_lint", json!({}));
    let result = lint["result"].clone();
    assert!(
        result["total_active_records"].as_u64().unwrap_or(0) >= 1,
        "active record must be counted: {lint}"
    );
    let broken = result["broken_cross_refs"]
        .as_array()
        .cloned()
        .unwrap_or_default();
    assert!(
        !broken.is_empty(),
        "broken cross-ref must be reported: {lint}"
    );
    assert!(
        broken.iter().any(|b| b["missing_target"] == "rec-does-not-exist"
            && b["field"] == "related_records"),
        "broken cross-ref entry must name the missing target and field: {broken:?}"
    );
    let markdown = result["markdown"].as_str().unwrap_or("");
    assert!(
        markdown.contains("断链"),
        "markdown must include the 断链 section: {markdown}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_lifecycle_write_should_not_block_when_sampling_supported_but_no_cluster() {
    // Regression guard: lifecycle write tool response must arrive promptly
    // even when sampling is advertised. Before the fix, the main dispatch
    // loop called `runtime.block_on(synthesize_with_sampling(...))` which
    // would block for up to 180 s waiting for a sampling response that the
    // test harness never sends. After the fix the compile runs on a
    // dedicated thread and the tool response arrives immediately.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp_with_sampling(&mut stdin, &mut reader);

    // Single record — no cluster, synthesize_with_sampling returns early.
    // The important thing is the response arrives without waiting for a
    // sampling round-trip.
    let start = std::time::Instant::now();
    let response = call_tool(
        &mut stdin,
        &mut reader,
        2,
        "memory_record_manual",
        json!({
            "title": "sampling-nonblock-test",
            "summary": "verifying tool response is not blocked by sampling",
            "memory_type": "preference",
            "scope": "user",
            "source_ref": "test:sampling-nonblock",
            "actor": "test"
        }),
    );
    let elapsed = start.elapsed();

    assert!(
        response["result"]["isError"] == json!(false),
        "record_manual must succeed: {response}"
    );
    // Must complete well under the 180 s sampling timeout.
    assert!(
        elapsed.as_secs() < 10,
        "tool response must not be blocked by sampling timeout: {elapsed:?}"
    );

    stop_mcp(child, stdin);
}

#[test]
fn mcp_lifecycle_write_should_trigger_sampling_compile_when_cluster_ready() {
    // End-to-end: write 3 records sharing the same entity so the cluster
    // detector fires, then accept a 4th. The server must issue a
    // `sampling/createMessage` reverse-call for auto-compile and persist
    // a knowledge candidate in the ledger.
    let temp = tempdir().unwrap();
    let vault_dir = temp.path().join("vault");
    let config_path = temp.path().join("spool.toml");
    fs::create_dir_all(vault_dir.join("10-Projects")).unwrap();
    fs::write(
        &config_path,
        format!("[vault]\nroot = \"{}\"\n", vault_dir.display()),
    )
    .unwrap();

    let (child, mut stdin, mut reader) = start_mcp(&config_path, None);
    initialize_mcp_with_sampling(&mut stdin, &mut reader);

    // Write 3 records sharing entity "rust-testing" so cluster detector fires.
    // Use call_tool_with_sampling for all writes since the 3rd write will
    // trigger auto-compile (cluster size = 3). Return [] to skip synthesis
    // during setup so the knowledge page only appears after the accept.
    for i in 2..=4 {
        let resp = call_tool_with_sampling(
            &mut stdin,
            &mut reader,
            i,
            "memory_record_manual",
            json!({
                "title": format!("rust-testing note {i}"),
                "summary": format!("note {i} about rust-testing patterns and cargo test"),
                "memory_type": "pattern",
                "scope": "user",
                "source_ref": format!("test:cluster-{i}"),
                "actor": "test",
                "entities": ["rust-testing", "cargo"],
                "tags": ["testing", "rust"]
            }),
            |request| {
                // Return empty array — no synthesis during setup phase.
                json!({
                    "jsonrpc": "2.0",
                    "id": request["id"],
                    "result": {
                        "content": { "type": "text", "text": "[]" }
                    }
                })
            },
        );
        assert!(
            resp["result"]["isError"] == json!(false),
            "record {i} must succeed: {resp}"
        );
    }

    // Propose a 4th record then accept it — acceptance triggers auto-compile.
    let propose_resp = call_tool_with_sampling(
        &mut stdin,
        &mut reader,
        5,
        "memory_propose",
        json!({
            "title": "rust-testing synthesis candidate",
            "summary": "proposed pattern about rust-testing",
            "memory_type": "pattern",
            "scope": "user",
            "source_ref": "test:cluster-propose",
            "actor": "ai",
            "entities": ["rust-testing", "cargo"],
            "tags": ["testing", "rust"]
        }),
        |request| {
            json!({
                "jsonrpc": "2.0",
                "id": request["id"],
                "result": { "content": { "type": "text", "text": "[]" } }
            })
        },
    );
    assert!(
        propose_resp["result"]["isError"] == json!(false),
        "propose must succeed: {propose_resp}"
    );
    let record_id = propose_resp["result"]["structuredContent"]["entry"]["record_id"]
        .as_str()
        .unwrap_or("")
        .to_string();
    assert!(
        !record_id.is_empty(),
        "record_id must be present: {propose_resp}"
    );

    // Accept triggers auto-compile; harness fulfills the sampling request.
    let accept_resp = call_tool_with_sampling(
        &mut stdin,
        &mut reader,
        6,
        "memory_accept",
        json!({
            "record_id": record_id,
            "actor": "test",
            "reason": "cluster test acceptance"
        }),
        |request| {
            assert_eq!(
                request["method"].as_str(),
                Some("sampling/createMessage"),
                "expected sampling/createMessage: {request}"
            );
            json!({
                "jsonrpc": "2.0",
                "id": request["id"],
                "result": {
                    "content": {
                        "type": "text",
                        "text": "[{\"title\":\"Rust Testing Patterns\",\"summary\":\"## Overview\\nConsolidated rust-testing patterns.\",\"domain\":\"methodology\",\"tags\":[\"testing\",\"rust\"],\"entities\":[\"rust-testing\",\"cargo\"]}]"
                    }
                }
            })
        },
    );
    assert!(
        accept_resp["result"]["isError"] == json!(false),
        "accept must succeed: {accept_resp}"
    );

    // Give the background compile thread a moment to persist the knowledge page.
    std::thread::sleep(std::time::Duration::from_millis(1000));

    // Verify a knowledge candidate was persisted.
    let queue = call_tool_with_sampling(
        &mut stdin,
        &mut reader,
        7,
        "memory_review_queue",
        json!({}),
        |request| {
            json!({
                "jsonrpc": "2.0",
                "id": request["id"],
                "result": { "content": { "type": "text", "text": "[]" } }
            })
        },
    );
    let records = queue["result"]["structuredContent"]["pending_review"]
        .as_array()
        .cloned()
        .unwrap_or_default();
    let has_knowledge = records
        .iter()
        .any(|r| r["record"]["memory_type"].as_str() == Some("knowledge"));
    assert!(
        has_knowledge,
        "a knowledge candidate must be persisted after sampling compile: {records:?}"
    );

    stop_mcp(child, stdin);
}
