cctop 0.39.7

//! An MCP server over stdio, so an agent can ask what the other agents are doing.
//!
//! cctop already knows something no single agent can find out for itself: what
//! every *other* harness on the machine is working on. The dashboard shows that
//! to a person. This shows it to an agent, which is the same data answering a
//! different question — "has someone already touched this file", "what was the
//! last plan anyone made here", "which session is burning the money".
//!
//! Deliberately read-only. Every tool here answers a question; none of them
//! start, stop, or type at anything. An agent that can drive other agents is a
//! different and much larger proposition than one that can see them, and the
//! visibility is the part with no downside.
//!
//! The protocol is JSON-RPC 2.0, one message per line, `initialize` then
//! `tools/list` then `tools/call`. No SDK: three methods and a fixed tool
//! schema is less code than a dependency to carry.
//!
//! ponytail: no `notifications/*` beyond swallowing `initialized`, and no
//! server-initiated messages. Nothing here changes without being asked.
//!
//! One tool blocks: `wait_for_session` is `cctop wait` behind a bounded
//! timeout ([`MAX_WAIT`]), so one agent can hand off to another and pick up
//! when it is done. It still only reads.
//!
//! ponytail: that wait does not bind a hook socket, so it hears an agent's
//! own reports only once a transcript or a running cctop's record on the rmux
//! session reflects them — a second or two later than `cctop wait` would. A
//! socket here would outlive the call: its accept thread cannot be stopped,
//! and one kept for the server's lifetime would pile up every event on the
//! machine between calls, for a server that answers a handful of them.

use cctop_core::loader::Loader;
use cctop_core::pricing::Plan;
use cctop_core::session::Session;
use cctop_core::session::lookup::{Lookup, lookup};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Deserializer};
use serde_json::{Value, json};
use std::io::{BufRead, Write};

/// The MCP revisions this speaks, newest first.
///
/// What changed between them is all on the client's side of a tools-only
/// server or optional for this one — output schemas, structured content,
/// elicitation — except batches, which 2025-03-26 asks a server to accept and
/// [`answer_line`] does.
const PROTOCOL_VERSIONS: [&str; 3] = ["2025-06-18", "2025-03-26", "2024-11-05"];

/// How many sessions a listing returns before it truncates.
///
/// A machine that has been running agents for months has hundreds of dead
/// sessions, and an agent reading this pays for every one. The default is the
/// recent slice, and a caller that genuinely wants the tail asks for it.
const DEFAULT_LIMIT: usize = 25;

/// The most rows a listing will return, whatever it is asked for.
///
/// [`DEFAULT_LIMIT`] is a default, not a ceiling: a caller that wants the tail
/// asks for it. But `limit` arrives from a model, and the largest number it can
/// write is not a request for a long list — it is one call that serialises every
/// session on the machine into a single JSON-RPC response, or, for
/// `search_sessions`, one that reads every transcript instead of stopping.
const MAX_LIMIT: u64 = 200;

/// A caller's `limit`, or [`DEFAULT_LIMIT`], held inside the range a listing can
/// answer.
///
/// Both ends matter, and the floor is the one `recall` already uses: a zero
/// would answer with nothing at all, which reads to an agent as "no sessions
/// exist here" rather than "you asked for none". Anything no listing could
/// honour — a float, a negative, a string — is not a limit at all
/// ([`lenient_u64`] reads it as `None`), and falls back to the default.
fn limit_of(limit: Option<u64>) -> usize {
    match limit {
        Some(n) => n.clamp(1, MAX_LIMIT) as usize,
        None => DEFAULT_LIMIT,
    }
}

/// The longest `wait_for_session` blocks, whatever it is asked for.
///
/// A tool call holds the calling agent's turn, and most clients give up on a
/// call long before this; an agent that needs longer calls again, and learns
/// from the answer that it timed out rather than hanging on a pipe.
const MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(300);

/// What `wait_for_session` waits when not told.
const DEFAULT_WAIT: std::time::Duration = std::time::Duration::from_secs(60);

/// Serve MCP on stdin/stdout until stdin closes.
///
/// Sessions are re-read per call rather than once at startup: an MCP server
/// lives as long as the agent that spawned it, and the whole value of the
/// answers is that they describe what is happening *now*.
pub fn serve() -> anyhow::Result<()> {
    let stdin = std::io::stdin();
    let mut stdout = std::io::stdout();

    for line in stdin.lock().lines() {
        let Some(response) = answer_line(&line?) else {
            continue;
        };
        writeln!(stdout, "{response}")?;
        stdout.flush()?;
    }
    Ok(())
}

/// The reply to one line of input, or `None` when it takes none.
///
/// A line that is not JSON is answered with a parse error and a null id, as
/// JSON-RPC asks: its id cannot be read, but the client may be blocked on it,
/// and an error it can log beats a silence it can only time out on. One line
/// is one message, so the stream stays in sync either way.
///
/// A JSON array is a batch, which the 2025-03-26 revision asks a server to
/// accept: each member is answered and the answers go back as one array, or
/// nothing at all if every member was a notification.
fn answer_line(line: &str) -> Option<Value> {
    if line.trim().is_empty() {
        return None;
    }
    let request: Value = match serde_json::from_str(line) {
        Ok(v) => v,
        Err(e) => {
            return Some(error_reply(
                Value::Null,
                RpcError::new(PARSE_ERROR, format!("parse error: {e}")),
            ));
        }
    };
    match request {
        Value::Array(batch) if batch.is_empty() => Some(error_reply(
            Value::Null,
            RpcError::new(INVALID_REQUEST, "an empty batch"),
        )),
        Value::Array(batch) => {
            let replies: Vec<Value> = batch.iter().filter_map(handle).collect();
            (!replies.is_empty()).then_some(Value::Array(replies))
        }
        request => handle(&request),
    }
}

/// The JSON-RPC error codes this server answers with.
const PARSE_ERROR: i64 = -32700;
const INVALID_REQUEST: i64 = -32600;
const METHOD_NOT_FOUND: i64 = -32601;
const INVALID_PARAMS: i64 = -32602;

/// A JSON-RPC error: the request itself was wrong, as opposed to a tool that
/// ran and failed, which is a result (see [`call_tool`]).
#[derive(Debug)]
struct RpcError {
    code: i64,
    message: String,
}

impl RpcError {
    fn new(code: i64, message: impl Into<String>) -> RpcError {
        RpcError {
            code,
            message: message.into(),
        }
    }
}

fn error_reply(id: Value, error: RpcError) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {"code": error.code, "message": error.message},
    })
}

/// Register this binary as an MCP server with every agent CLI that can take
/// one, and say what happened to each.
///
/// Through each agent's own `mcp add` rather than by editing its config file.
/// Claude Code rewrites `~/.claude.json` constantly while it runs, so a second
/// writer racing it can lose either side's change; the agent's own command is
/// the one writer it does not race. A harness without such a command gets the
/// stanza to paste instead of a guess at its file format.
///
/// `scope` is `user` (every project) or `project` (this directory only —
/// Claude Code writes `.mcp.json`; Codex has no per-project servers, and says
/// so rather than being registered machine-wide by surprise).
pub fn install(scope: &str) -> Vec<String> {
    let exe = match std::env::current_exe()
        .ok()
        .and_then(|p| p.to_str().map(str::to_string))
    {
        Some(exe) => exe,
        None => return vec!["could not find cctop's own path".into()],
    };
    let mut out = Vec::new();
    let mut run = |agent: &str, args: Vec<&str>| {
        let shown = format!("{agent} {}", args.join(" "));
        match std::process::Command::new(agent).args(&args).output() {
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                out.push(format!("{agent}: not installed, skipped"));
            }
            Err(e) => out.push(format!("{agent}: could not run `{shown}`: {e}")),
            Ok(o) if o.status.success() => out.push(format!("{agent}: registered (`{shown}`)")),
            Ok(o) => {
                let said = String::from_utf8_lossy(&o.stderr).to_string()
                    + &String::from_utf8_lossy(&o.stdout);
                let said = said.trim();
                // Both CLIs refuse a name that is taken. That is the outcome
                // asked for, unless it points somewhere else — which `mcp get`
                // shows, and is the user's to decide, not this command's.
                if said.to_lowercase().contains("already exists") {
                    out.push(format!(
                        "{agent}: a server named cctop is already registered; \
                         `{agent} mcp get cctop` shows where it points"
                    ));
                } else {
                    out.push(format!("{agent}: `{shown}` failed: {said}"));
                }
            }
        }
    };
    match scope {
        "project" => {
            run(
                "claude",
                vec![
                    "mcp", "add", "--scope", "project", "cctop", "--", &exe, "--mcp",
                ],
            );
            out.push("codex: has no per-project MCP servers; run --install-mcp user for it".into());
        }
        _ => {
            run(
                "claude",
                vec![
                    "mcp", "add", "--scope", "user", "cctop", "--", &exe, "--mcp",
                ],
            );
            run("codex", vec!["mcp", "add", "cctop", "--", &exe, "--mcp"]);
        }
    }
    out.push(format!(
        "Anything else that speaks MCP: {}",
        json!({"mcpServers": {"cctop": {"command": exe, "args": ["--mcp"]}}})
    ));
    out
}

/// Answer one request, or `None` for a notification, which takes no reply.
fn handle(request: &Value) -> Option<Value> {
    // A notification is exactly a request with no id, and replying to one is a
    // protocol error rather than a harmless extra message.
    let id = request.get("id")?.clone();
    let Some(method) = request.get("method").and_then(Value::as_str) else {
        return Some(error_reply(
            id,
            RpcError::new(INVALID_REQUEST, "a request needs a method"),
        ));
    };
    let params = request.get("params");

    let result = match method {
        "initialize" => Ok(json!({
            "protocolVersion": negotiate(params),
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "cctop", "version": env!("CARGO_PKG_VERSION")},
            "instructions": INSTRUCTIONS,
        })),
        "tools/list" => Ok(json!({"tools": tool_schemas()})),
        "tools/call" => call_tool(params),
        // `ping` is the one method a client may send that means nothing.
        "ping" => Ok(json!({})),
        other => Err(RpcError::new(
            METHOD_NOT_FOUND,
            format!("unknown method '{other}'"),
        )),
    };

    Some(match result {
        Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
        Err(error) => error_reply(id, error),
    })
}

/// The protocol version to answer `initialize` with.
///
/// The client's own when this server speaks it, which is what the spec asks
/// for; otherwise the newest one here, and the client decides whether it can
/// live with that. Answering every client with one fixed revision is how a
/// newer client that would have spoken the older one is told nothing it can
/// agree to.
fn negotiate(params: Option<&Value>) -> &'static str {
    let asked = params
        .and_then(|p| p.get("protocolVersion"))
        .and_then(Value::as_str);
    PROTOCOL_VERSIONS
        .iter()
        .find(|v| Some(**v) == asked)
        .unwrap_or(&PROTOCOL_VERSIONS[0])
}

/// What the client is told the server is for, once, at `initialize`.
///
/// Clients that support it put this in the agent's own instructions, which is
/// what makes `recall` something an agent reaches for unprompted rather than a
/// tool it has to be told about. Saying so here means cctop never has to edit
/// anybody's CLAUDE.md or AGENTS.md to get the same effect.
const INSTRUCTIONS: &str = "cctop reads every coding-agent session on this machine — \
    Claude Code, Codex, OpenCode and others, past and running. Before re-deciding \
    something, re-investigating a bug, or asking the user what was settled \
    earlier, call `recall` with the question: it returns the passages of past \
    sessions, from any agent, that discussed it, and `read_passage` opens one in \
    context. Cite what you use by its session id. Use `check_conflicts` before \
    editing files another running agent may also be editing.";

/// What the server offers.
fn tool_schemas() -> Vec<Value> {
    vec![
        json!({
            "name": "list_sessions",
            "description": "List AI coding agent sessions on this machine — every harness, \
                            not just your own. Returns harness, model, working directory, \
                            git branch, token usage, estimated cost, context window \
                            occupancy, and whether the session is still running. Use this to \
                            find out what other agents are working on before you start.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "running_only": {
                        "type": "boolean",
                        "description": "Only sessions with a live process behind them.",
                    },
                    "directory": {
                        "type": "string",
                        "description": "Only sessions whose working directory is at or under \
                                        this path. Use it to ask who else is in this repo.",
                    },
                    "limit": {
                        "type": "integer",
                        "description":
                            "Maximum sessions to return, most recently active first. \
                             Defaults to 25, and is capped at 200 however large a \
                             number is passed.",
                    },
                },
            },
        }),
        json!({
            "name": "get_session_context",
            "description": "Get a context brief for one session: what it was doing, the plan \
                            it was working to, the files it changed and read, the commands it \
                            ran, and what it delegated. This is the handoff document — read it \
                            to continue another agent's work. Returns markdown.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "session_id": {
                        "type": "string",
                        "description": "Session id, or any unique prefix of one, as returned \
                                        by list_sessions.",
                    },
                },
                "required": ["session_id"],
            },
        }),
        json!({
            "name": "check_conflicts",
            "description": "Ask whether another agent is already working where you are about \
                            to. Give it your working directory and, if you know them, the files \
                            you intend to change; it answers with the running sessions in the \
                            same repository and which of your files they have already written. \
                            Two agents writing one file is not a merge conflict — nothing warns \
                            you and one edit is simply lost — so check before a batch of edits, \
                            and take a worktree if someone is there.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "Your working directory. The repository containing it is \
                                        what gets compared, so a subdirectory is fine.",
                    },
                    "files": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Paths you are about to change, absolute or relative to \
                                        `directory`. Optional: without them the answer is just \
                                        who else is in the repository.",
                    },
                },
                "required": ["directory"],
            },
        }),
        json!({
            "name": "wait_for_session",
            "description": "Block until another agent's session stops working, then say what \
                            it is doing. Use it to wait for an agent you handed work to, or \
                            one you sent a prompt to, before reading its result. Returns JSON: \
                            state (working, idle, waiting, error, ended), met, timed_out, \
                            waited_secs. On a timeout, call it again.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "session": {
                        "type": "string",
                        "description": "Session id or a unique prefix of one (as returned by \
                                        list_sessions), a cctop tab name, or a pid.",
                    },
                    "until": {
                        "type": "string",
                        "enum": ["idle", "waiting", "done", "any-stop"],
                        "description": "idle: its turn is over. waiting: it is blocked on a \
                                        question. done: it worked since this call began and \
                                        has stopped since. any-stop (default): it is not \
                                        working, now or once it stops.",
                    },
                    "timeout_secs": {
                        "type": "integer",
                        "description": "Give up after this many seconds. Defaults to 60; \
                                        at most 300.",
                    },
                },
                "required": ["session"],
            },
        }),
        json!({
            "name": "recall",
            "description": "Recall what past coding sessions on this machine said about \
                            something — decisions and their reasons, bugs and their fixes, \
                            findings, plans — from any agent (Claude Code, Codex, OpenCode \
                            and others). Returns the best-matching passages, ranked by \
                            their words and by their meaning, each with its session id, \
                            agent, directory and date. Use it before re-deriving something \
                            that may already have been worked out. This session is left out.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "A question or a few keywords, e.g. 'why is the cache \
                                        sharded' or 'vast.ai nemotron'.",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Passages to return. Defaults to 8.",
                    },
                },
                "required": ["query"],
            },
        }),
        json!({
            "name": "read_passage",
            "description": "Open one passage that `recall` returned, with the passages either \
                            side of it, to read a hit in context.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "session": {
                        "type": "string",
                        "description": "The session id from a recall result, or a unique prefix.",
                    },
                    "passage": {
                        "type": "integer",
                        "description": "The passage number from the recall result.",
                    },
                    "around": {
                        "type": "integer",
                        "description": "Neighbouring passages to include either side. Defaults to 1.",
                    },
                },
                "required": ["session", "passage"],
            },
        }),
        json!({
            "name": "search_sessions",
            "description": "Search the full text of every session transcript on this machine \
                            for a string, and return the sessions that mention it with a \
                            snippet of the match. Use it to find where something was already \
                            discussed or attempted, in any harness.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Text to look for. Case-insensitive, matched literally.",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum matching sessions to return. Defaults to 25, \
                                       and is capped at 200 however large a number is passed.",
                    },
                },
                "required": ["query"],
            },
        }),
    ]
}

/// The tools [`call_tool`] answers to, which [`tool_schemas`] must list.
const TOOLS: [&str; 7] = [
    "list_sessions",
    "get_session_context",
    "check_conflicts",
    "wait_for_session",
    "recall",
    "read_passage",
    "search_sessions",
];

/// Run one tool and wrap its output the way MCP expects.
///
/// Two kinds of failure, kept apart as the spec keeps them. A call that names
/// no tool, or one that does not exist, is the client's mistake and a JSON-RPC
/// error. A tool that ran and could not answer — no such session, an ambiguous
/// prefix, an argument it cannot use — is a *result* with `isError`: that is
/// the form the client hands to the model, which can read "matches 3 sessions"
/// and call again with more of the id. As a JSON-RPC error the same sentence
/// tends to reach only a log. Bad arguments are on the result side for the
/// same reason, as the 2025-11-25 revision moved them.
fn call_tool(params: Option<&Value>) -> Result<Value, RpcError> {
    let name = params
        .and_then(|p| p.get("name"))
        .and_then(Value::as_str)
        .ok_or_else(|| RpcError::new(INVALID_PARAMS, "tools/call needs a tool name"))?;
    if !TOOLS.contains(&name) {
        return Err(RpcError::new(
            INVALID_PARAMS,
            format!("unknown tool '{name}'"),
        ));
    }
    let args = params
        .and_then(|p| p.get("arguments"))
        .cloned()
        .unwrap_or_else(|| json!({}));
    let (text, is_error) = match run_tool(name, args) {
        Ok(text) => (text, false),
        Err(text) => (text, true),
    };
    Ok(json!({"content": [{"type": "text", "text": text}], "isError": is_error}))
}

fn run_tool(name: &str, args: Value) -> Result<String, String> {
    // Before the load below: the wait does its own, and then keeps refreshing
    // it for as long as it blocks.
    if name == "wait_for_session" {
        return wait_for_session(parse(name, args)?);
    }

    let mut loader = Loader::new();
    let sessions = loader.load(Plan::Retail);

    let text = match name {
        "list_sessions" => list_sessions(&sessions, parse(name, args)?),
        "get_session_context" => get_session_context(&sessions, &loader, parse(name, args)?)?,
        "check_conflicts" => check_conflicts(&sessions, parse(name, args)?),
        "search_sessions" => search_sessions(&sessions, parse(name, args)?)?,
        "recall" => recall(&sessions, parse(name, args)?)?,
        "read_passage" => read_passage(&sessions, parse(name, args)?)?,
        other => unreachable!("{other} is not in TOOLS"),
    };
    // Persisting is worth the write even on a read-only call: an MCP server is
    // spawned per agent, and each one that reparsed every transcript from cold
    // would undo the cache the dashboard depends on.
    loader.store().save();
    Ok(text)
}

/// A tool's arguments as its own struct, or the sentence saying which one is
/// missing or the wrong type, for the model to correct.
fn parse<T: DeserializeOwned>(tool: &str, args: Value) -> Result<T, String> {
    serde_json::from_value(args).map_err(|e| format!("{tool}: {e}"))
}

/// A number a model wrote, or `None` for anything that is not a whole
/// non-negative one — a float, a negative, a string.
///
/// Lenient rather than refused because every number here has a default, and
/// falling back to it answers the question that was asked; refusing would
/// spend a turn on a field the caller could have left out.
fn lenient_u64<'de, D: Deserializer<'de>>(d: D) -> Result<Option<u64>, D::Error> {
    Ok(Value::deserialize(d)?.as_u64())
}

#[derive(Deserialize)]
struct ListArgs {
    #[serde(default)]
    running_only: bool,
    directory: Option<String>,
    #[serde(default, deserialize_with = "lenient_u64")]
    limit: Option<u64>,
}

#[derive(Deserialize)]
struct ContextArgs {
    session_id: String,
}

#[derive(Deserialize)]
struct ConflictArgs {
    directory: String,
    #[serde(default)]
    files: Vec<String>,
}

#[derive(Deserialize)]
struct WaitArgs {
    session: String,
    until: Option<String>,
    #[serde(default, deserialize_with = "lenient_u64")]
    timeout_secs: Option<u64>,
}

#[derive(Deserialize)]
struct RecallArgs {
    query: String,
    #[serde(default, deserialize_with = "lenient_u64")]
    limit: Option<u64>,
}

#[derive(Deserialize)]
struct PassageArgs {
    session: String,
    passage: u64,
    #[serde(default, deserialize_with = "lenient_u64")]
    around: Option<u64>,
}

#[derive(Deserialize)]
struct SearchArgs {
    query: String,
    #[serde(default, deserialize_with = "lenient_u64")]
    limit: Option<u64>,
}

/// `cctop wait`, bounded by [`MAX_WAIT`], answering in the CLI's `-j` shape.
fn wait_for_session(args: WaitArgs) -> Result<String, String> {
    let target = args.session.as_str();
    let until = match args.until.as_deref() {
        None => crate::wait::Until::AnyStop,
        Some(word) => crate::wait::Until::parse(word).ok_or_else(|| {
            format!("until must be idle, waiting, done or any-stop, not '{word}'")
        })?,
    };
    let timeout = args
        .timeout_secs
        .map_or(DEFAULT_WAIT, std::time::Duration::from_secs)
        .min(MAX_WAIT);
    match crate::wait::wait(target, until, Some(timeout), false) {
        Ok(outcome) => Ok(outcome.json().to_string()),
        Err(crate::wait::Unresolved::Unknown) => Err(format!(
            "no session, tab or pid here is '{target}'; list_sessions shows them"
        )),
        Err(crate::wait::Unresolved::Ambiguous(ids)) => Err(format!(
            "'{target}' matches {} sessions; give more of the id: {}",
            ids.len(),
            ids.join(", ")
        )),
    }
}

fn list_sessions(sessions: &[Session], args: ListArgs) -> String {
    let running_only = args.running_only;
    let directory = args.directory.as_deref();
    let limit = limit_of(args.limit);

    let mut matched: Vec<&Session> = sessions
        .iter()
        .filter(|s| !running_only || s.is_running())
        .filter(|s| match directory {
            Some(dir) => s.label_source.starts_with(dir.trim_end_matches('/')),
            None => true,
        })
        .collect();
    // Most recent first: an agent reading a truncated list should get the
    // sessions that are still relevant, not whichever provider sorted first.
    matched.sort_by(|a, b| b.last_active.cmp(&a.last_active));
    let total = matched.len();
    matched.truncate(limit);

    let rows: Vec<Value> = matched
        .iter()
        .map(|s| {
            json!({
                "session_id": s.session_id,
                "harness": s.harness,
                "provider": s.provider.as_str(),
                "model": s.model,
                "title": s.title,
                // Absent for the caller's own sessions, which is all of them
                // unless cctop is reading every user's homes.
                "user": s.owner,
                "directory": s.label_source,
                "branch": cctop_core::branch::branch_of(s),
                "branch_note": "the branch checked out now, not necessarily the one it worked on",
                "running": s.is_running(),
                "started_at": s.started_at,
                "last_active": s.last_active,
                "input_tokens": s.input_tokens,
                "output_tokens": s.output_tokens,
                "estimated_cost_usd": s.total_cost,
                "context_used": s.context.map(|c| c.used),
                "context_max": s.context.map(|c| c.max),
            })
        })
        .collect();

    let payload = json!({
        "sessions": rows,
        "returned": rows.len(),
        "total_matching": total,
        // Said explicitly, because an agent handed a list of 25 out of 300 will
        // otherwise reason as though it has seen everything.
        "truncated": total > rows.len(),
        "cost_note": "Costs are estimates from published per-token rates. Flat-rate plans \
                      bill differently.",
    });
    serde_json::to_string_pretty(&payload).unwrap_or_default()
}

/// Who else is running in this repository, and which of the caller's files they
/// have already written.
///
/// The caller is very likely one of the sessions in the list — an agent asking
/// about its own repository is a session cctop can see. It is left in rather
/// than guessed out: cctop cannot tell which row is the caller (an MCP server
/// is a child process, not a session), and a wrong guess would drop the one
/// peer that mattered. `own_row_note` says so instead.
fn check_conflicts(sessions: &[Session], args: ConflictArgs) -> String {
    let directory = args.directory.as_str();
    let peers = cctop_core::collide::peers_of(sessions, directory, &args.files);
    let contested: Vec<&str> = peers
        .iter()
        .filter(|(_, shared)| !shared.is_empty())
        .flat_map(|(_, shared)| shared.iter().map(String::as_str))
        .collect();

    let rows: Vec<Value> = peers
        .iter()
        .map(|(s, shared)| {
            json!({
                "session_id": s.session_id,
                "harness": s.harness,
                "directory": s.label_source,
                "last_active": s.last_active,
                "your_files_it_has_written": shared,
                "recently_written": s.recent_writes,
            })
        })
        .collect();

    let payload = json!({
        "directory": directory,
        "agents_here": rows,
        "contested_files": contested,
        "clear": contested.is_empty(),
        "own_row_note": "One of these is probably you — an agent calling this has a session of \
                         its own. Match on session_id or working directory.",
        "advice": match contested.is_empty() {
            true if rows.len() > 1 =>
                "Another agent is in this repository but has not written your files. Nothing is \
                 lost yet; check again if you start editing widely.",
            true => "Nobody else is running here.",
            false =>
                "Another running agent has already written these files. Whichever of you saves \
                 last wins and the other edit is gone, with no warning from git. Take a worktree, \
                 or agree who finishes first.",
        },
        "limits": "Only running sessions are compared, only the files each has written recently, \
                   and a linked git worktree counts as a separate repository — which is the point \
                   of one. A file is stronger evidence than a shared repository, though: a peer \
                   that already holds one of your files is reported wherever it was launched.",
    });
    serde_json::to_string_pretty(&payload).unwrap_or_default()
}

fn get_session_context(
    sessions: &[Session],
    loader: &Loader,
    args: ContextArgs,
) -> Result<String, String> {
    let wanted = args.session_id.as_str();
    let session = match lookup(sessions, wanted) {
        Lookup::Found(only) => only,
        Lookup::None => return Err(format!("no session id starts with '{wanted}'")),
        Lookup::Ambiguous(many) => {
            return Err(format!(
                "'{wanted}' matches {} sessions: {}",
                many.len(),
                many.iter()
                    .map(|s| s.session_id.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }
    };

    let data = loader.store().session_data(session);
    Ok(cctop_core::handoff::rendered(&cctop_core::handoff::build(
        session,
        Some(&data),
    )))
}

/// Passages of past sessions, with the caller's own session left out.
///
/// The text form rather than JSON: the reader is a model, which reads prose
/// at a fraction of the tokens the same answer costs as quoted, escaped JSON.
fn recall(sessions: &[Session], args: RecallArgs) -> Result<String, String> {
    let query = args.query.as_str();
    if query.trim().is_empty() {
        return Err("recall needs a query".into());
    }
    let limit = args
        .limit
        .map_or(crate::recall::DEFAULT_LIMIT, |n| n.clamp(1, 50) as usize);
    let me = crate::recall::caller(sessions);
    let answer = crate::recall::recall(sessions, query, limit, me.as_deref());
    Ok(crate::recall::render(&answer))
}

fn read_passage(sessions: &[Session], args: PassageArgs) -> Result<String, String> {
    let around = args.around.unwrap_or(1).min(5) as usize;
    crate::recall::read(sessions, &args.session, args.passage as usize, around)
}

fn search_sessions(sessions: &[Session], args: SearchArgs) -> Result<String, String> {
    let query = args.query.as_str();
    if query.trim().is_empty() {
        return Err("search_sessions needs a non-empty query".into());
    }
    let limit = limit_of(args.limit);

    let needle = query.to_lowercase();
    let mut ordered: Vec<&Session> = sessions.iter().collect();
    ordered.sort_by(|a, b| b.last_active.cmp(&a.last_active));

    let mut hits = Vec::new();
    for session in ordered {
        if hits.len() >= limit {
            break;
        }
        let target = cctop_core::session::search::Target::of(session);
        let Some(hit) = cctop_core::session::search::find(&target, &needle) else {
            continue;
        };
        hits.push(json!({
            "session_id": session.session_id,
            "harness": session.harness,
            "provider": session.provider.as_str(),
            "directory": session.label_source,
            "title": session.title,
            "last_active": session.last_active,
            "snippet": hit.snippet,
        }));
    }

    let payload = json!({
        "query": query,
        "matches": hits,
        // Scanning stops at the limit rather than counting every match first,
        // so there is no honest total to report — say that instead of implying
        // the number returned is the number that exist.
        "note": format!(
            "Searched newest-first and stopped at {limit} matches; there may be older ones."
        ),
    });
    Ok(serde_json::to_string_pretty(&payload).unwrap_or_default())
}

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

    /// A notification carries no id, and answering one is a protocol error —
    /// the client is not waiting for a reply and will not match it to anything.
    #[test]
    fn notifications_get_no_reply() {
        let notification = json!({"jsonrpc": "2.0", "method": "notifications/initialized"});
        assert!(handle(&notification).is_none());
    }

    /// The handshake has to name a protocol version the client recognises, or
    /// the connection is dropped before any tool is ever listed.
    #[test]
    fn initialize_answers_with_the_protocol_version() {
        let request = json!({"jsonrpc": "2.0", "id": 1, "method": "initialize"});
        let response = handle(&request).expect("a reply");
        assert_eq!(response["result"]["protocolVersion"], PROTOCOL_VERSIONS[0]);
        assert_eq!(response["id"], 1);
    }

    /// The instructions are what make an agent reach for `recall` without
    /// being told to, so they must name it — and the tools they name must be
    /// ones this server actually offers.
    #[test]
    fn the_instructions_point_at_tools_that_exist() {
        let request = json!({"jsonrpc": "2.0", "id": 1, "method": "initialize"});
        let response = handle(&request).expect("a reply");
        let said = response["result"]["instructions"]
            .as_str()
            .expect("instructions");
        let offered: Vec<String> = tool_schemas()
            .iter()
            .map(|t| t["name"].as_str().unwrap_or_default().to_string())
            .collect();
        for named in ["recall", "read_passage", "check_conflicts"] {
            assert!(said.contains(named), "instructions do not mention {named}");
            assert!(offered.iter().any(|t| t == named), "{named} is not a tool");
        }
    }

    /// Every advertised tool needs a name, a description an agent can choose
    /// from, and a schema — a tool missing any of them is one no client will
    /// call correctly.
    #[test]
    fn every_tool_is_fully_described() {
        for tool in tool_schemas() {
            let name = tool["name"].as_str().expect("a name");
            assert!(
                tool["description"].as_str().is_some_and(|d| d.len() > 40),
                "{name} needs a description an agent can choose from"
            );
            assert_eq!(tool["inputSchema"]["type"], "object", "{name}");
        }
    }

    /// An unknown method is answered, not ignored: a client that sent an id is
    /// blocked waiting for something.
    #[test]
    fn an_unknown_method_still_gets_an_answer() {
        let request = json!({"jsonrpc": "2.0", "id": 7, "method": "resources/list"});
        let response = handle(&request).expect("a reply");
        assert_eq!(response["id"], 7);
        assert_eq!(response["error"]["code"], METHOD_NOT_FOUND);
        assert!(response["error"]["message"].as_str().is_some());
    }

    /// A line that is not JSON has no readable id, so it is answered against a
    /// null one — the client sees the error instead of waiting out a timeout.
    #[test]
    fn a_malformed_line_gets_a_parse_error_with_a_null_id() {
        let reply = answer_line("{\"jsonrpc\": \"2.0\", \"id\": 3,").expect("a reply");
        assert_eq!(reply["error"]["code"], PARSE_ERROR);
        assert_eq!(reply["id"], Value::Null);
        assert!(
            answer_line("   ").is_none(),
            "a blank line is not a message"
        );
    }

    /// A batch is answered member by member, in one array, leaving out the
    /// notifications; a batch of nothing but notifications gets no reply.
    #[test]
    fn a_batch_is_answered_as_one_array() {
        let line = json!([
            {"jsonrpc": "2.0", "id": 1, "method": "ping"},
            {"jsonrpc": "2.0", "method": "notifications/initialized"},
            {"jsonrpc": "2.0", "id": 2, "method": "nope"},
        ])
        .to_string();
        let reply = answer_line(&line).expect("a reply");
        let replies = reply.as_array().expect("an array");
        assert_eq!(replies.len(), 2);
        assert_eq!(replies[0]["id"], 1);
        assert_eq!(replies[1]["error"]["code"], METHOD_NOT_FOUND);
        let quiet = json!([{"jsonrpc": "2.0", "method": "notifications/initialized"}]);
        assert!(answer_line(&quiet.to_string()).is_none());
    }

    /// The client's version is echoed when this server speaks it, and the
    /// newest one here is offered when it does not.
    #[test]
    fn initialize_negotiates_the_protocol_version() {
        let init = |version: &str| {
            let request = json!({
                "jsonrpc": "2.0", "id": 1, "method": "initialize",
                "params": {"protocolVersion": version, "capabilities": {}},
            });
            handle(&request).expect("a reply")["result"]["protocolVersion"].clone()
        };
        for version in PROTOCOL_VERSIONS {
            assert_eq!(init(version), version);
        }
        assert_eq!(init("1999-01-01"), PROTOCOL_VERSIONS[0]);
    }

    /// A tool that runs and cannot answer says so as a result the model reads,
    /// with `isError`; a tool that does not exist is the client's mistake and a
    /// JSON-RPC error.
    #[test]
    fn a_tool_failure_is_a_result_and_an_unknown_tool_an_error() {
        let call = |params: Value| {
            let request =
                json!({"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": params});
            handle(&request).expect("a reply")
        };
        // Missing its required argument: refused before any session is read.
        let failed = call(json!({"name": "get_session_context", "arguments": {}}));
        assert!(failed.get("error").is_none(), "{failed}");
        assert_eq!(failed["result"]["isError"], true);
        let text = failed["result"]["content"][0]["text"].as_str().unwrap();
        assert!(text.contains("session_id"), "{text}");

        let unknown = call(json!({"name": "drop_tables"}));
        assert_eq!(unknown["error"]["code"], INVALID_PARAMS);
        let nameless = call(json!({}));
        assert_eq!(nameless["error"]["code"], INVALID_PARAMS);
    }

    /// Every name the dispatcher accepts is a tool the schema lists, and the
    /// other way round, so neither can be added without the other.
    #[test]
    fn the_dispatched_tools_are_the_listed_ones() {
        let listed: Vec<String> = tool_schemas()
            .iter()
            .map(|t| t["name"].as_str().unwrap().to_string())
            .collect();
        assert_eq!(listed, TOOLS);
    }

    fn args<T: DeserializeOwned>(value: Value) -> T {
        parse("test", value).expect("arguments that parse")
    }

    fn limit_of_json(value: Value) -> usize {
        limit_of(args::<ListArgs>(value).limit)
    }

    /// The tool's `until` enum is exactly the words the wait understands, so
    /// an agent choosing from the schema never picks one that is refused.
    #[test]
    fn the_wait_tool_offers_exactly_the_conditions_it_takes() {
        let tool = tool_schemas()
            .into_iter()
            .find(|tool| tool["name"] == "wait_for_session")
            .expect("the tool is listed");
        let words = tool["inputSchema"]["properties"]["until"]["enum"]
            .as_array()
            .expect("an enum")
            .clone();
        for word in &words {
            let word = word.as_str().expect("a word");
            assert!(crate::wait::Until::parse(word).is_some(), "{word}");
        }
        assert_eq!(words.len(), 4);
        // Refused before anything is read, rather than after a walk.
        let bad = run_tool(
            "wait_for_session",
            json!({"session": "x", "until": "never"}),
        );
        assert!(bad.unwrap_err().contains("until must be"));
        assert!(run_tool("wait_for_session", json!({})).is_err());
    }

    /// A `limit` is a model's number, and the biggest one it can write is not a
    /// request for a long list. Unclamped, `"limit": 18446744073709551615` puts
    /// every session on the machine — or every transcript, for a search — into
    /// one answer, and `0` answers "nothing exists" to an agent that asked for
    /// nothing.
    #[test]
    fn a_limit_from_the_model_is_clamped_rather_than_obeyed() {
        assert_eq!(
            limit_of_json(json!({"limit": u64::MAX})),
            MAX_LIMIT as usize
        );
        assert_eq!(
            limit_of_json(json!({"limit": MAX_LIMIT * 1_000})),
            MAX_LIMIT as usize
        );
        assert_eq!(limit_of_json(json!({"limit": 0})), 1);
        // `as_u64` refuses a negative outright, so the default is what is left.
        assert_eq!(limit_of_json(json!({"limit": -1})), DEFAULT_LIMIT);
        assert_eq!(limit_of_json(json!({"limit": i64::MIN})), DEFAULT_LIMIT);
        // Nor a float or a string, which no listing would honour either.
        assert_eq!(limit_of_json(json!({"limit": 1e30})), DEFAULT_LIMIT);
        assert_eq!(
            limit_of_json(json!({"limit": "18446744073709551615"})),
            DEFAULT_LIMIT
        );
        assert_eq!(limit_of_json(json!({})), DEFAULT_LIMIT);
        // A number inside the range is still the caller's.
        assert_eq!(limit_of_json(json!({"limit": 3})), 3);
        assert_eq!(
            limit_of_json(json!({"limit": MAX_LIMIT})),
            MAX_LIMIT as usize
        );
    }

    /// The cap has to be in the schema, not only in the handler: the schema is
    /// what a model reads *before* choosing a number, so a limit enforced
    /// silently is one it can only discover by getting fewer rows than it asked
    /// for. A description that drifted from the clamp is worse than no cap.
    #[test]
    fn the_advertised_limit_agrees_with_the_enforced_one() {
        for name in ["list_sessions", "search_sessions"] {
            let tool = tool_schemas()
                .into_iter()
                .find(|tool| tool["name"] == name)
                .expect("the tool is listed");
            let said = tool["inputSchema"]["properties"]["limit"]["description"]
                .as_str()
                .expect("a description");
            assert!(
                said.contains(&format!("{DEFAULT_LIMIT}")) && said.contains(&MAX_LIMIT.to_string()),
                "{name} advertises {said:?}, which does not name {DEFAULT_LIMIT} and {MAX_LIMIT}"
            );
        }
    }

    /// Enough rows that an unbounded limit and a clamped one cannot both be
    /// satisfied: the point of the clamp is that a huge number does not reach
    /// the whole machine.
    fn many_sessions(n: usize) -> Vec<Session> {
        (0..n)
            .map(|i| {
                let mut s = Session::new(cctop_core::pricing::Provider::Claude, format!("s{i:04}"));
                s.label_source = "/tmp/proj".into();
                s.last_active = format!("2026-09-28T00:00:{:02}.000Z", i % 60);
                s
            })
            .collect()
    }

    #[test]
    fn a_listing_stops_at_the_cap_however_large_a_limit_is() {
        let sessions = many_sessions(MAX_LIMIT as usize + 10);
        let text = list_sessions(&sessions, args(json!({"limit": u64::MAX})));
        let out: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(out["returned"], MAX_LIMIT);
        // Said out loud, because an agent handed a truncated list otherwise
        // reasons as though it has seen everything.
        assert_eq!(out["truncated"], true);
        assert_eq!(out["total_matching"], MAX_LIMIT as usize + 10);

        // The other end: a zero limit still answers with a row, rather than with
        // nothing, which would read as "this machine has no sessions".
        let text = list_sessions(&sessions, args(json!({"limit": 0})));
        let out: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(out["returned"], 1);
    }

    /// A search stops at the cap too, which is the expensive half: each row past
    /// the limit costs a transcript read, and the corpus is every transcript on
    /// the machine.
    #[test]
    fn a_search_stops_at_the_cap_however_large_a_limit_is() {
        let dir = std::env::temp_dir().join(format!("cctop-mcp-limit-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let file = dir.join("hit.jsonl");
        std::fs::write(&file, "{\"text\":\"the needle is here\"}\n").unwrap();

        let sessions: Vec<Session> = many_sessions(MAX_LIMIT as usize + 10)
            .into_iter()
            .map(|mut s| {
                s.data_file = Some(file.clone());
                s
            })
            .collect();

        let text = search_sessions(
            &sessions,
            args(json!({"query": "needle", "limit": u64::MAX})),
        )
        .expect("an answer");
        let out: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(out["matches"].as_array().unwrap().len(), MAX_LIMIT as usize);
        // The note names the limit the scan actually stopped at, which is the
        // clamped one.
        assert!(
            out["note"]
                .as_str()
                .unwrap()
                .contains(&format!("at {MAX_LIMIT} matches")),
            "note is {:?}",
            out["note"]
        );

        let text = search_sessions(&sessions, args(json!({"query": "needle", "limit": 0})))
            .expect("an answer");
        let out: Value = serde_json::from_str(&text).unwrap();
        assert_eq!(out["matches"].as_array().unwrap().len(), 1);
    }
}