mentedb-mcp 0.5.46

//! End-to-end tests for the lifecycle hook integration: daemon HTTP surface,
//! hook binary stdin/stdout contract, and the full prompt -> stop -> recall
//! loop against a real embedded database.
#![cfg(feature = "local")]

use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

const BIN: &str = env!("CARGO_BIN_EXE_mentedb-mcp");

/// Each test here spawns a daemon that loads a local embedding model, which is
/// far too heavy to run several at once: the parallel default thrashes loading
/// N models and daemons miss their health deadline. Serialize the suite so
/// exactly one daemon warms at a time (they all pass serially).
static E2E_SERIAL: std::sync::Mutex<()> = std::sync::Mutex::new(());

fn serial() -> std::sync::MutexGuard<'static, ()> {
    E2E_SERIAL.lock().unwrap_or_else(|e| e.into_inner())
}

#[derive(serde::Deserialize)]
struct DaemonInfo {
    port: u16,
    pid: u32,
    token: String,
}

/// Kills the daemon (spawned directly or by hook auto-spawn) on drop.
struct DaemonGuard {
    child: Option<Child>,
    data_dir: PathBuf,
}

impl Drop for DaemonGuard {
    fn drop(&mut self) {
        if let Some(info) = read_info(&self.data_dir) {
            let _ = Command::new("kill").arg(info.pid.to_string()).status();
        }
        if let Some(child) = self.child.as_mut() {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

fn read_info(data_dir: &Path) -> Option<DaemonInfo> {
    let raw = std::fs::read_to_string(data_dir.join("daemon.json")).ok()?;
    serde_json::from_str(&raw).ok()
}

fn wait_for_daemon(data_dir: &Path, timeout: Duration) -> DaemonInfo {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        if let Some(info) = read_info(data_dir) {
            let health = ureq_get(&format!("http://127.0.0.1:{}/health", info.port));
            if health.is_some() {
                return info;
            }
        }
        std::thread::sleep(Duration::from_millis(250));
    }
    panic!("daemon did not become healthy within {timeout:?}");
}

fn ureq_get(url: &str) -> Option<String> {
    let output = Command::new("curl")
        .args(["-s", "-f", "-m", "2", url])
        .output()
        .ok()?;
    if output.status.success() {
        Some(String::from_utf8_lossy(&output.stdout).to_string())
    } else {
        None
    }
}

fn daemon_post(info: &DaemonInfo, path: &str, body: &serde_json::Value) -> (u32, String) {
    let output = Command::new("curl")
        .args([
            "-s",
            "-m",
            "30",
            "-o",
            "-",
            "-w",
            "\n%{http_code}",
            "-X",
            "POST",
            "-H",
            "Content-Type: application/json",
            "-H",
            &format!("x-mentedb-token: {}", info.token),
            "-d",
            &body.to_string(),
            &format!("http://127.0.0.1:{}{}", info.port, path),
        ])
        .output()
        .expect("curl failed");
    let raw = String::from_utf8_lossy(&output.stdout).to_string();
    let (body, code) = raw.rsplit_once('\n').unwrap_or(("", "0"));
    (code.trim().parse().unwrap_or(0), body.to_string())
}

fn run_hook(data_dir: &Path, event: &str, payload: &serde_json::Value) -> String {
    let mut child = Command::new(BIN)
        .args([
            "--data-dir",
            data_dir.to_str().unwrap(),
            "--local",
            "hook",
            event,
        ])
        // Keep tests hermetic: never let the hook reconcile the developer's
        // real Claude Code settings; the self-update path has its own test
        // with an isolated home.
        .env("MENTEDB_HOOK_NO_SELF_UPDATE", "1")
        // Deterministic, fast embedder: skip the Candle model download so the
        // suite is reproducible and exercises the BM25 keyword recall path.
        // The auto-spawned daemon inherits this from the hook process.
        .env("MENTEDB_FORCE_HASH_EMBEDDINGS", "1")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
        .expect("failed to run hook");
    child
        .stdin
        .as_mut()
        .unwrap()
        .write_all(payload.to_string().as_bytes())
        .unwrap();
    let output = child.wait_with_output().expect("hook did not exit");
    assert!(
        output.status.success(),
        "hook must always exit 0, got {:?}",
        output.status
    );
    String::from_utf8_lossy(&output.stdout).to_string()
}

fn spawn_daemon(data_dir: &Path) -> DaemonGuard {
    let child = Command::new(BIN)
        .args(["--data-dir", data_dir.to_str().unwrap(), "daemon"])
        .env("MENTEDB_FORCE_HASH_EMBEDDINGS", "1")
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("failed to spawn daemon");
    DaemonGuard {
        child: Some(child),
        data_dir: data_dir.to_path_buf(),
    }
}

#[test]
fn daemon_serves_turn_and_context_with_auth() {
    let _serial = serial();
    let dir = tempfile::tempdir().unwrap();
    let _guard = spawn_daemon(dir.path());
    let info = wait_for_daemon(dir.path(), Duration::from_secs(300));

    // Unauthorized without the token.
    let output = Command::new("curl")
        .args([
            "-s",
            "-o",
            "/dev/null",
            "-w",
            "%{http_code}",
            "-X",
            "POST",
            "-H",
            "Content-Type: application/json",
            "-d",
            "{\"prompt\":\"x\"}",
            &format!("http://127.0.0.1:{}/v1/context", info.port),
        ])
        .output()
        .unwrap();
    assert_eq!(String::from_utf8_lossy(&output.stdout), "401");

    // Store a turn through the full pipeline.
    let (code, body) = daemon_post(
        &info,
        "/v1/turn",
        &serde_json::json!({
            "user_message": "I always deploy with blue green releases on Fridays",
            "assistant_response": "Noted, blue green on Fridays.",
            "turn_id": 1,
        }),
    );
    assert_eq!(code, 200, "turn failed: {body}");
    let parsed: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(parsed["ok"], true);

    // Recall it.
    let (code, body) = daemon_post(
        &info,
        "/v1/context",
        &serde_json::json!({ "prompt": "how do I deploy releases" }),
    );
    assert_eq!(code, 200);
    let ctx: serde_json::Value = serde_json::from_str(&body).unwrap();
    let memories = ctx["memories"].as_array().unwrap();
    assert!(
        memories
            .iter()
            .any(|m| m["content"].as_str().unwrap_or("").contains("blue green")),
        "stored turn must be recallable, got: {ctx}"
    );
}

#[test]
fn hook_full_turn_loop_with_autospawn() {
    let _serial = serial();
    let dir = tempfile::tempdir().unwrap();
    // No daemon started: the first hook call must auto-spawn it.
    let guard = DaemonGuard {
        child: None,
        data_dir: dir.path().to_path_buf(),
    };

    let session = "sess-hook-e2e";

    // Turn 1: user prompt (stashes prompt, may return no context on an empty DB).
    let _ = run_hook(
        dir.path(),
        "user-prompt",
        &serde_json::json!({
            "session_id": session,
            "prompt": "remember that my project codename is nightjar",
            "hook_event_name": "UserPromptSubmit",
        }),
    );
    // The auto-spawned daemon may still be loading on the very first call;
    // wait until it registers before the stop hook.
    let _info = wait_for_daemon(dir.path(), Duration::from_secs(300));

    // Retry the prompt hook now that the daemon is up (first call may have
    // timed out waiting on model download).
    let _ = run_hook(
        dir.path(),
        "user-prompt",
        &serde_json::json!({
            "session_id": session,
            "prompt": "remember that my project codename is nightjar",
            "hook_event_name": "UserPromptSubmit",
        }),
    );

    // Stop hook: stores the turn.
    let out = run_hook(
        dir.path(),
        "stop",
        &serde_json::json!({
            "session_id": session,
            "last_assistant_message": "Got it, codename nightjar.",
            "hook_event_name": "Stop",
        }),
    );
    assert!(out.trim().is_empty(), "stop hook must print nothing: {out}");

    // Session state advanced.
    let state_raw =
        std::fs::read_to_string(dir.path().join(format!("hook_sessions/{session}.json"))).unwrap();
    let state: serde_json::Value = serde_json::from_str(&state_raw).unwrap();
    assert_eq!(state["turn_id"], 1);
    assert!(state["pending_prompt"].is_null());

    // Turn 2: the injection policy deliberately suppresses turns this fresh
    // (they are still in the model's context window), so the prompt hook
    // must print nothing.
    let out = run_hook(
        dir.path(),
        "user-prompt",
        &serde_json::json!({
            "session_id": session,
            "prompt": "what is my project codename",
            "hook_event_name": "UserPromptSubmit",
        }),
    );
    assert!(
        out.trim().is_empty(),
        "fresh same-session turns are echo and must not be injected, got: {out}"
    );

    // The turn is stored asynchronously now: the Stop hook spools it and a
    // detached flusher sends it to the daemon, so the turn end never blocks on
    // the write. Poll until it is retrievable (eventual consistency), below the
    // injection policy.
    let info = wait_for_daemon(dir.path(), Duration::from_secs(30));
    let mut recalled = serde_json::Value::Null;
    let mut found = false;
    for _ in 0..40 {
        let (code, body) = daemon_post(
            &info,
            "/v1/context",
            &serde_json::json!({ "prompt": "what is my project codename", "limit": 8 }),
        );
        if code == 200 {
            recalled = serde_json::from_str(&body).unwrap_or(serde_json::Value::Null);
            let memories = recalled["memories"].as_array().cloned().unwrap_or_default();
            if memories.iter().any(|m| {
                m["content"]
                    .as_str()
                    .is_some_and(|c| c.contains("nightjar"))
            }) {
                found = true;
                break;
            }
        }
        std::thread::sleep(Duration::from_millis(250));
    }
    assert!(
        found,
        "stored turn must eventually be retrievable from the daemon, last: {recalled}"
    );

    drop(guard);
}

#[test]
fn post_tool_use_captures_action_live() {
    let _serial = serial();
    let dir = tempfile::tempdir().unwrap();
    let _guard = spawn_daemon(dir.path());
    let info = wait_for_daemon(dir.path(), Duration::from_secs(300));

    // A PostToolUse hook for a file edit stores an action note immediately,
    // with no prior user-prompt/stop turn.
    let out = run_hook(
        dir.path(),
        "post-tool-use",
        &serde_json::json!({
            "session_id": "sess-ptu",
            "tool_name": "Edit",
            "tool_input": { "file_path": "src/payments.rs" },
            "cwd": "/tmp/myproj",
            "hook_event_name": "PostToolUse",
        }),
    );
    assert!(out.trim().is_empty(), "post-tool-use prints nothing: {out}");

    // The action is immediately recallable (flushed on write).
    let (code, body) = daemon_post(
        &info,
        "/v1/context",
        &serde_json::json!({ "prompt": "what files were changed for payments" }),
    );
    assert_eq!(code, 200);
    let ctx: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert!(
        ctx["memories"]
            .as_array()
            .unwrap()
            .iter()
            .any(|m| m["content"].as_str().unwrap_or("").contains("payments.rs")),
        "captured action must be recallable, got: {ctx}"
    );

    // A read-only command is not captured (noise filter).
    run_hook(
        dir.path(),
        "post-tool-use",
        &serde_json::json!({
            "session_id": "sess-ptu",
            "tool_name": "Bash",
            "tool_input": { "command": "git status" },
            "hook_event_name": "PostToolUse",
        }),
    );

    // PreCompact flush never errors.
    let out = run_hook(
        dir.path(),
        "pre-compact",
        &serde_json::json!({ "session_id": "sess-ptu", "trigger": "auto" }),
    );
    assert!(out.trim().is_empty());
}

#[test]
fn hook_tolerates_garbage_input() {
    let _serial = serial();
    let dir = tempfile::tempdir().unwrap();
    // Malformed JSON, no daemon, no cloud: must still exit 0 with no output.
    let mut child = Command::new(BIN)
        .args([
            "--data-dir",
            dir.path().to_str().unwrap(),
            "--local",
            "hook",
            "stop",
        ])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
        .unwrap();
    child
        .stdin
        .as_mut()
        .unwrap()
        .write_all(b"this is not json")
        .unwrap();
    let output = child.wait_with_output().unwrap();
    assert!(output.status.success());
    assert!(output.stdout.is_empty());
}

#[test]
fn pre_tool_use_injects_action_rules_before_commit() {
    let _serial = serial();
    use mentedb::prelude::*;
    use mentedb_core::types::AgentId;

    let dir = tempfile::tempdir().unwrap();

    // Seed action rules directly in the engine, then close it so the daemon
    // (single writer) can open the same directory.
    {
        let mut db = mentedb::MenteDb::open(dir.path()).unwrap();
        // Seed with the SAME embedder the daemon uses under the test's forced
        // hash mode (dim 128), so the stored vectors and the daemon's query
        // vectors share a space. The commit rule is then reached by the BM25
        // keyword path (the daemon passes the action text), which is the
        // robustness this suite verifies: action recall works even when the
        // embedder is the non-semantic hash fallback.
        db.set_embedder(Box::new(mentedb_embedding::HashEmbeddingProvider::new(128)));
        let seed = |content: &str, ty: MemoryType| {
            let emb = db.embed_text(content).ok().flatten().unwrap_or_default();
            db.store(MemoryNode::new(
                AgentId::nil(),
                ty,
                content.to_string(),
                emb,
            ))
            .unwrap();
        };
        seed(
            "when running git commit, never add Co-Authored-By trailers to the commit message",
            MemoryType::Procedural,
        );
        seed(
            "when opening a pull request, the PR description uses Summary and Verification sections",
            MemoryType::Procedural,
        );
        seed(
            "the user prefers dark mode in their editor",
            MemoryType::Semantic,
        );
        // Persist the indexes (written by close, not drop) so the daemon
        // reopening this directory sees them.
        db.close().unwrap();
    }

    let _guard = spawn_daemon(dir.path());
    wait_for_daemon(dir.path(), Duration::from_secs(300));

    // The exact command shape agents run, global flag value containing the
    // word commit included.
    let out = run_hook(
        dir.path(),
        "pre-tool-use",
        &serde_json::json!({
            "session_id": "sess-pre",
            "tool_name": "Bash",
            "tool_input": { "command": "git -c commit.gpgsign=false commit -m \"fix: x\"" },
            "cwd": "/tmp/myproj",
            "hook_event_name": "PreToolUse",
        }),
    );
    let parsed: serde_json::Value =
        serde_json::from_str(out.trim()).expect("pre-tool-use must print valid JSON");
    let hso = &parsed["hookSpecificOutput"];
    assert_eq!(hso["hookEventName"], "PreToolUse");
    let ctx = hso["additionalContext"].as_str().unwrap_or_default();
    assert!(
        ctx.contains("Co-Authored-By"),
        "commit rule must be recalled by meaning for a commit command, got: {ctx}"
    );
    assert!(
        !ctx.contains("Summary and Verification"),
        "the PR rule is a different action and must not lead for a commit, got: {ctx}"
    );
    assert!(
        !ctx.contains("dark mode"),
        "an ordinary preference must not surface as an action rule, got: {ctx}"
    );
    assert!(
        hso.get("permissionDecision").is_none(),
        "the hook must never emit a permission decision"
    );

    // A clearly unrelated action surfaces no governing rule. (Fine
    // distinctions between sibling commands, git log vs git commit, are
    // embedder-quality dependent and covered deterministically by the engine's
    // controlled-embedding action_rules tests; here we assert the robust case:
    // an action with no bearing on the stored rules stays silent.)
    let out = run_hook(
        dir.path(),
        "pre-tool-use",
        &serde_json::json!({
            "session_id": "sess-pre",
            "tool_name": "Bash",
            "tool_input": { "command": "ls -la /tmp/project" },
            "hook_event_name": "PreToolUse",
        }),
    );
    assert!(
        out.trim().is_empty(),
        "an unrelated command must not surface action rules: {out}"
    );

    // Non-Bash tools stay silent even if their input mentions git.
    let out = run_hook(
        dir.path(),
        "pre-tool-use",
        &serde_json::json!({
            "session_id": "sess-pre",
            "tool_name": "Edit",
            "tool_input": { "file_path": "git_commit.rs" },
            "hook_event_name": "PreToolUse",
        }),
    );
    assert!(out.trim().is_empty(), "non-Bash tools are ignored: {out}");
}

#[test]
fn pre_tool_use_is_silent_with_no_rules_and_tolerates_garbage() {
    let _serial = serial();
    let dir = tempfile::tempdir().unwrap();
    let _guard = spawn_daemon(dir.path());
    wait_for_daemon(dir.path(), Duration::from_secs(300));

    // A commit with zero stored rules injects nothing.
    let out = run_hook(
        dir.path(),
        "pre-tool-use",
        &serde_json::json!({
            "session_id": "sess-empty",
            "tool_name": "Bash",
            "tool_input": { "command": "git commit -m x" },
            "hook_event_name": "PreToolUse",
        }),
    );
    assert!(out.trim().is_empty(), "no rules means no output: {out}");

    // Garbage stdin never breaks the tool call: exit 0, no output. run_hook
    // asserts the exit status itself.
    let out = run_hook(
        dir.path(),
        "pre-tool-use",
        &serde_json::json!("not an object"),
    );
    assert!(out.trim().is_empty());
}

#[test]
fn session_start_self_updates_hook_registrations() {
    let _serial = serial();
    // A settings file written by an older version (five events, no
    // pre-tool-use) must gain the missing hook on session start, while a
    // config dir that never ran setup stays untouched.
    let home = tempfile::tempdir().unwrap();
    let data_dir = tempfile::tempdir().unwrap();

    let ours = home.path().join(".claude");
    std::fs::create_dir_all(&ours).unwrap();
    let old_settings = serde_json::json!({
        "hooks": {
            "UserPromptSubmit": [
                { "hooks": [ { "type": "command", "command": "npx -y mentedb-mcp@latest hook user-prompt" } ] }
            ],
            "Stop": [
                { "hooks": [ { "type": "command", "command": "npx -y mentedb-mcp@latest hook stop" } ] }
            ],
            "SessionStart": [
                { "matcher": "startup|resume|compact",
                  "hooks": [ { "type": "command", "command": "npx -y mentedb-mcp@latest hook session-start" } ] }
            ],
            "PostToolUse": [
                { "matcher": "Write|Edit|MultiEdit|NotebookEdit|Bash",
                  "hooks": [ { "type": "command", "command": "npx -y mentedb-mcp@latest hook post-tool-use" } ] }
            ],
            "PreCompact": [
                { "hooks": [ { "type": "command", "command": "npx -y mentedb-mcp@latest hook pre-compact" } ] }
            ]
        }
    });
    std::fs::write(
        ours.join("settings.json"),
        serde_json::to_string_pretty(&old_settings).unwrap(),
    )
    .unwrap();

    // A second profile with hooks from some other tool and no mentedb:
    // the consent guard must leave it alone.
    let theirs = home.path().join(".claude-other");
    std::fs::create_dir_all(&theirs).unwrap();
    let foreign = r#"{ "hooks": { "Stop": [ { "hooks": [ { "type": "command", "command": "other-tool hook stop" } ] } ] } }"#;
    std::fs::write(theirs.join("settings.json"), foreign).unwrap();

    let run = |label: &str| {
        let mut child = Command::new(BIN)
            .args([
                "--data-dir",
                data_dir.path().to_str().unwrap(),
                "--local",
                "hook",
                "session-start",
            ])
            .env("HOME", home.path())
            .env_remove("CLAUDE_CONFIG_DIR")
            .env_remove("MENTEDB_HOOK_NO_SELF_UPDATE")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null())
            .spawn()
            .expect("failed to run hook");
        child
            .stdin
            .as_mut()
            .unwrap()
            .write_all(br#"{"session_id":"self-update"}"#)
            .unwrap();
        let out = child.wait_with_output().expect("hook did not exit");
        assert!(out.status.success(), "{label}: hook must exit 0");
        String::from_utf8_lossy(&out.stdout).to_string()
    };

    let out = run("first");
    let updated = std::fs::read_to_string(ours.join("settings.json")).unwrap();
    assert!(
        updated.contains("hook pre-tool-use"),
        "missing hook must be added: {updated}"
    );
    assert!(
        updated.contains("hook user-prompt"),
        "existing hooks must survive"
    );
    assert!(
        out.contains("refreshed its Claude Code hooks"),
        "session output must mention the refresh: {out}"
    );
    assert_eq!(
        std::fs::read_to_string(theirs.join("settings.json")).unwrap(),
        foreign,
        "profiles without mentedb hooks must never be touched"
    );
    let marker = std::fs::read_to_string(data_dir.path().join("hooks_version")).unwrap();
    assert_eq!(marker.trim(), env!("CARGO_PKG_VERSION"));

    // Second run: marker matches, nothing changes, no refresh notice.
    let before = std::fs::read_to_string(ours.join("settings.json")).unwrap();
    let out = run("second");
    let after = std::fs::read_to_string(ours.join("settings.json")).unwrap();
    assert_eq!(before, after, "reconcile must be once per version");
    assert!(
        !out.contains("refreshed its Claude Code hooks"),
        "no repeat notice: {out}"
    );
}

/// A backend that accepts the connection and never answers (a hung server)
/// must not hold the user's prompt: the hook gives up at its recall budget,
/// exits 0, and the prompt goes through, long before the client's 30s hook
/// kill that discards everything.
#[test]
fn user_prompt_fails_open_fast_when_cloud_hangs() {
    let _guard = serial();
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let hold = std::thread::spawn(move || {
        let mut held = Vec::new();
        for stream in listener.incoming().take(4).flatten() {
            held.push(stream);
        }
        std::thread::sleep(Duration::from_secs(30));
        drop(held);
    });

    let home = tempfile::tempdir().unwrap();
    let data_dir = tempfile::tempdir().unwrap();
    let payload = serde_json::json!({
        "session_id": "hang-test",
        "hook_event_name": "UserPromptSubmit",
        "prompt": "how do I deploy the gateway",
    });
    let started = std::time::Instant::now();
    let mut child = Command::new(BIN)
        .args([
            "--data-dir",
            data_dir.path().to_str().unwrap(),
            "hook",
            "user-prompt",
        ])
        .env("HOME", home.path())
        .env("MENTEDB_API_KEY", "test-key")
        .env("MENTEDB_API_URL", format!("http://127.0.0.1:{port}"))
        .env("MENTEDB_HOOK_RECALL_BUDGET_MS", "1500")
        .env("MENTEDB_HOOK_NO_SELF_UPDATE", "1")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
        .expect("failed to run hook");
    child
        .stdin
        .as_mut()
        .unwrap()
        .write_all(payload.to_string().as_bytes())
        .unwrap();
    let output = child.wait_with_output().expect("hook did not exit");
    let elapsed = started.elapsed();

    assert!(
        output.status.success(),
        "hook must exit 0, got {:?}",
        output.status
    );
    assert!(
        elapsed < Duration::from_secs(6),
        "hook held the prompt for {elapsed:?} against a hung backend"
    );
    drop(hold);
}

/// A fake cloud that answers every tool call with success and records, per
/// request, the tool name and how many memories it carried.
fn fake_cloud() -> (u16, std::sync::Arc<std::sync::Mutex<Vec<(String, usize)>>>) {
    fake_cloud_with_delay(Duration::ZERO)
}

/// `fake_cloud` that takes `per_memory` to answer for each memory a call
/// carries, like a server that embeds and checks every note.
fn fake_cloud_with_delay(
    per_memory: Duration,
) -> (u16, std::sync::Arc<std::sync::Mutex<Vec<(String, usize)>>>) {
    use std::io::{BufRead, BufReader, Read};
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
    let log = std::sync::Arc::clone(&seen);
    std::thread::spawn(move || {
        for mut stream in listener.incoming().flatten() {
            let mut reader = BufReader::new(stream.try_clone().unwrap());
            let mut len = 0usize;
            loop {
                let mut line = String::new();
                if reader.read_line(&mut line).unwrap_or(0) == 0 {
                    break;
                }
                let lower = line.to_ascii_lowercase();
                if let Some(v) = lower.strip_prefix("content-length:") {
                    len = v.trim().parse().unwrap_or(0);
                }
                if line == "\r\n" {
                    break;
                }
            }
            let mut body = vec![0u8; len];
            let _ = reader.read_exact(&mut body);
            let req: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
            let name = req["name"].as_str().unwrap_or("").to_string();
            let n = req["arguments"]["memories"]
                .as_array()
                .map(|a| a.len())
                .unwrap_or(1);
            log.lock().unwrap().push((name, n));
            std::thread::sleep(per_memory * n as u32);
            let resp = r#"{"content":[{"type":"text","text":"{}"}],"is_error":false}"#;
            let _ = write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                resp.len(),
                resp
            );
        }
    });
    (port, seen)
}

/// A queued backlog of action notes drains in a few batch calls, not one
/// round trip (and one server side embed, write and disk sync) per note.
#[test]
fn queued_notes_flush_as_batches() {
    let _guard = serial();
    let (port, seen) = fake_cloud();
    let home = tempfile::tempdir().unwrap();
    let data_dir = tempfile::tempdir().unwrap();
    let backlog: String = (0..120)
        .map(|i| {
            serde_json::json!({ "kind": "note", "content": format!("edited file {i}"), "project": "p" })
                .to_string()
                + "\n"
        })
        .collect();
    std::fs::write(data_dir.path().join("spool.jsonl"), backlog).unwrap();

    let payload = serde_json::json!({
        "session_id": "batch-test",
        "tool_name": "Edit",
        "tool_input": { "file_path": "src/lib.rs" },
        "cwd": "/tmp/p",
        "hook_event_name": "PostToolUse",
    });
    let mut child = Command::new(BIN)
        .args([
            "--data-dir",
            data_dir.path().to_str().unwrap(),
            "hook",
            "post-tool-use",
        ])
        .env("HOME", home.path())
        .env("MENTEDB_API_KEY", "test-key")
        .env("MENTEDB_API_URL", format!("http://127.0.0.1:{port}"))
        .env("MENTEDB_HOOK_NO_SELF_UPDATE", "1")
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("failed to run hook");
    child
        .stdin
        .as_mut()
        .unwrap()
        .write_all(payload.to_string().as_bytes())
        .unwrap();
    assert!(child.wait().unwrap().success());

    let calls = seen.lock().unwrap().clone();
    let batches: Vec<usize> = calls
        .iter()
        .filter(|(name, _)| name == "store_memories")
        .map(|(_, n)| *n)
        .collect();
    assert_eq!(batches, vec![50, 50, 20], "calls: {calls:?}");
    let left = std::fs::read_to_string(data_dir.path().join("spool.jsonl")).unwrap_or_default();
    assert!(
        left.trim().is_empty(),
        "spool should be drained, left: {left}"
    );
}

/// A server too slow to store a full batch within the hook's batch timeout
/// must not jam the queue: the hook halves the batch after a timeout, grows it
/// back after successes, and the backlog drains over a few hooks.
#[test]
fn a_slow_server_gets_smaller_batches_and_the_backlog_drains() {
    let _guard = serial();
    // 60ms per note: 50 notes take 3s, over the 2s batch timeout set below.
    let (port, seen) = fake_cloud_with_delay(Duration::from_millis(60));
    let home = tempfile::tempdir().unwrap();
    let data_dir = tempfile::tempdir().unwrap();
    let backlog: String = (0..120)
        .map(|i| {
            serde_json::json!({ "kind": "note", "content": format!("edited file {i}"), "project": "p" })
                .to_string()
                + "\n"
        })
        .collect();
    std::fs::write(data_dir.path().join("spool.jsonl"), backlog).unwrap();
    // A different file each run: the hook skips a note it just saw, and a
    // skipped note does not flush the spool.
    let payload = |run: usize| {
        serde_json::json!({
            "session_id": "slow-batch-test",
            "tool_name": "Edit",
            "tool_input": { "file_path": format!("src/file{run}.rs") },
            "cwd": "/tmp/p",
            "hook_event_name": "PostToolUse",
        })
    };
    let spool_left = || {
        std::fs::read_to_string(data_dir.path().join("spool.jsonl"))
            .unwrap_or_default()
            .lines()
            .filter(|l| !l.trim().is_empty())
            .count()
    };
    let mut runs = 0;
    while spool_left() > 0 && runs < 12 {
        runs += 1;
        let mut child = Command::new(BIN)
            .args([
                "--data-dir",
                data_dir.path().to_str().unwrap(),
                "hook",
                "post-tool-use",
            ])
            .env("HOME", home.path())
            .env("MENTEDB_API_KEY", "test-key")
            .env("MENTEDB_API_URL", format!("http://127.0.0.1:{port}"))
            .env("MENTEDB_HOOK_NO_SELF_UPDATE", "1")
            .env("MENTEDB_HOOK_BATCH_TIMEOUT_MS", "2000")
            .stdin(Stdio::piped())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("failed to run hook");
        child
            .stdin
            .as_mut()
            .unwrap()
            .write_all(payload(runs).to_string().as_bytes())
            .unwrap();
        assert!(child.wait().unwrap().success());
    }
    let batches: Vec<usize> = seen
        .lock()
        .unwrap()
        .iter()
        .filter(|(name, _)| name == "store_memories")
        .map(|(_, n)| *n)
        .collect();
    assert_eq!(
        &batches[..2],
        &[50, 25],
        "halves after the first timeout: {batches:?}"
    );
    assert!(
        batches.iter().all(|&n| n <= 50),
        "never above the cap: {batches:?}"
    );
    assert_eq!(
        spool_left(),
        0,
        "backlog drained within {runs} hooks: {batches:?}"
    );
    let learned: usize = std::fs::read_to_string(data_dir.path().join("spool_batch"))
        .unwrap()
        .trim()
        .parse()
        .unwrap();
    assert!(learned < 50, "the smaller size is remembered: {learned}");
}