From 69e4df079492e87360063a8436c8193728621819 Mon Sep 17 00:00:00 2001 From: Max Loffgren Date: Mon, 5 Oct 2026 21:38:33 -0500 Subject: [PATCH 1/4] feat: adopt MCP servers through a proxy tool and a generated skill Nothing the model reads mentioned MCP: the frozen pointer named no surface for it and no --spec contract covered it, so a request to add an MCP server left the agent to improvise a JSON-RPC client for each server. openmax --mcp-list and --mcp-call are a one-shot MCP stdio client outside the agent loop. Each run starts the server named after --, performs the initialize handshake with protocol-version negotiation, pages through tools/list or makes one tools/call, bounds every wait with --mcp-timeout, passes the server's stderr through, and stops the server (stdin closed, then SIGTERM, then SIGKILL). openmax --spec mcp prints the recipe built on it: one proxy tool per server with a {tool, arguments} schema and the server's credentials granted through env, plus one skill whose body --mcp-list generates, so tool schemas cost nothing until the skill is read. Each call spawns the server; servers that need a persistent session are not supported yet. The frozen pointer now names mcp among its surfaces: 4 more bytes on every request (3,416, under the unchanged 3,500 cap). --- crates/core/src/prompt.rs | 11 +- crates/core/src/spec.rs | 117 +++++- crates/tui/Cargo.toml | 14 +- crates/tui/src/main.rs | 177 +++++++- crates/tui/src/mcp.rs | 737 +++++++++++++++++++++++++++++++++ crates/tui/tests/mcp_bridge.rs | 556 +++++++++++++++++++++++++ docs/extending.md | 38 +- docs/usage.md | 5 + 8 files changed, 1628 insertions(+), 27 deletions(-) create mode 100644 crates/tui/src/mcp.rs create mode 100644 crates/tui/tests/mcp_bridge.rs diff --git a/crates/core/src/prompt.rs b/crates/core/src/prompt.rs index ca1000f1..366b1518 100644 --- a/crates/core/src/prompt.rs +++ b/crates/core/src/prompt.rs @@ -199,7 +199,7 @@ pub fn system_prompt_with_breakdown(project_root: &Path, registry: &Registry) -> /// the prefix carries the index and never the manual. Every line here is /// resent on every request for the life of the session. const SELF_EXTENSION: &str = "\n\nWhen the user asks for a reusable capability (tool, skill, prompt template, hook, permission rule, provider, or memory), read its contract first with bash: openmax --spec . It gives the file path, format, approval, and activation rules; verify with openmax --check.\n\ -Surfaces: tools|skills|prompts|hooks|permissions|providers|memory|stdio.\n\ +Surfaces: tools|skills|prompts|hooks|permissions|providers|memory|stdio|mcp.\n\ Past sessions and memories: openmax --recall \"\"; capability-file history and restore: openmax --ledger. Isolated or parallel work: a child openmax -p or openmax --stdio process, in tmux when it must outlive the turn."; /// One line per skill: name, description, and the SKILL.md path the model @@ -734,13 +734,17 @@ mod tests { // authoring surfaces, and the moments that need them carry their own // pointer (--check settings rows print "openmax --spec settings"; the // guide's memory line names --recall). A new --spec surface must - // either join the pointer or this list, consciously. + // either join the pointer or this list, consciously. `mcp` adds no + // file kind of its own, but it is named: it is the recipe that turns + // an MCP server into a tool and a skill, and without it a request + // for a server leaves the agent writing its own protocol client. let pointer: Vec<&str> = SELF_EXTENSION.lines() .find_map(|line| line.strip_prefix("Surfaces: ")) .expect("the guide lists its authoring contracts") .trim_end_matches('.') .split('|') .collect(); + assert!(pointer.contains(&"mcp"), "the pointer names the MCP recipe: {pointer:?}"); for s in &pointer { assert!(crate::spec::SURFACES.contains(s), "the pointer names a surface --spec lacks: {s}"); assert!(crate::spec::render(s).is_some(), "the named contract must render: {s}"); @@ -762,7 +766,8 @@ mod tests { /// The path-free base rules, extension pointer, and builtin schemas must /// fit in 3,500 bytes (the payload measured 3,412 bytes, 770 tokens on a - /// current tokenizer, when the cap was set). Grounding sections have + /// current tokenizer, when the cap was set, and 3,416 bytes, 773 tokens, + /// once the pointer named `mcp`). Grounding sections have /// separate caps. Measure the payload with /// dump_frozen_prompt_payload_for_tokenizer and a real tokenizer before /// changing this budget; provider framing is not included. diff --git a/crates/core/src/spec.rs b/crates/core/src/spec.rs index 5bfe4d63..f1e08793 100644 --- a/crates/core/src/spec.rs +++ b/crates/core/src/spec.rs @@ -10,7 +10,7 @@ //! parsers in tests, so the printed contract cannot drift from the loop. /// Every surface `render` accepts, in the order the help text lists them. -pub const SURFACES: [&str; 11] = [ +pub const SURFACES: [&str; 12] = [ "tools", "skills", "prompts", @@ -21,6 +21,7 @@ pub const SURFACES: [&str; 11] = [ "memory", "recall", "stdio", + "mcp", "usage", ]; @@ -37,6 +38,7 @@ pub fn render(surface: &str) -> Option<&'static str> { "memory" => Some(MEMORY), "recall" => Some(RECALL), "stdio" => Some(STDIO), + "mcp" => Some(MCP), _ => None, } } @@ -989,6 +991,83 @@ Validate a stream against the contract: `openmax --check --stdio` reads JSONL on stdin, reports each line, and exits nonzero on any violation. "#; +const MCP: &str = r#"# MCP servers + +The agent loop hosts no MCP client. A server joins through two files the loop +already loads: one proxy tool, and one skill whose body lists the server's +tools. Every request pays only for the proxy's small schema and the skill's +index line; the server's own tool schemas are read when the skill is. This +binary is the bridge between them, so the host needs nothing installed +besides the server itself. + +Each call starts the server, makes one request, and stops it, so the bridge +serves stateless use. A server that needs a persistent session (state kept +between calls, a login done once per process) is not supported yet. + +## Commands + +- `openmax --mcp-list -- ` prints the server's tools as a + skill body: a header, then one line per tool, + `- name(arg: type, opt?: type): description`, required arguments first and + `?` marking an optional one. With `--json` it prints the full tool + definitions instead. +- `openmax --mcp-call -- ` reads + {"tool": "", "arguments": {...}} on stdin, calls that tool, and prints + the result's text. A result the tool marks as an error, an error reply, a + malformed reply, and a server that exits or stops answering all exit 1 with + the reason on stderr. Input it cannot read exits 2 before the server starts. + +Everything after `--` is the server's argv, run without a shell. Each wait for +a reply is bounded by `--mcp-timeout ` (default 30, at most 3600); raise +it for a server that is slow to start, such as a first `npx` download. The +server's stderr passes through. When the run ends, the server's stdin is +closed; a server still running 2 seconds later gets SIGTERM, then SIGKILL. +The wire is JSON-RPC 2.0 over stdio with the initialize handshake, protocol +revisions 2024-11-05 through 2025-11-25. A server that speaks only a later, +handshake-free revision refuses initialize, and its error is reported. + +## Recipe + +For a server started as `notes-server --stdio` that reads `NOTES_TOKEN`: + +1. Generate the skill with bash, so its tool list costs no tokens to write: + +```sh +tools=$(openmax --mcp-list -- notes-server --stdio) && mkdir -p .agents/skills/notes && +printf '%s\n' '---' 'name: notes' 'description: Tools of the notes MCP server; call them with the notes tool' '---' "$tools" > .agents/skills/notes/SKILL.md +``` + +2. Write the proxy tool, `.openmax/tools/notes.toml`: + +```toml +name = "notes" +description = "Call a notes MCP server tool; the notes skill lists names and arguments" +command = "sh" +args = ["-c", "exec \"$OPENMAX_BIN\" --mcp-call -- \"$@\"", "mcp", "notes-server", "--stdio"] +env = ["NOTES_TOKEN"] +timeout_secs = 120 +mutating = true + +[params] +type = "object" +required = ["tool", "arguments"] +[params.properties.tool] +type = "string" +[params.properties.arguments] +type = "object" +``` + +The words after `"mcp"` are the server command. `$OPENMAX_BIN` names the +binary hosting the session (see `openmax --spec tools`), and the call's JSON +reaches it on stdin. A tool's environment is scrubbed and the server inherits +exactly what the tool receives, so name every credential the server reads in +`env`. Keep `mutating = true` unless every tool the server offers is +read-only. `timeout_secs` must cover starting the server and the call. + +3. Verify both files with `openmax --check`. They activate like any other tool +and skill (`openmax --spec tools`, `openmax --spec skills`). +"#; + #[cfg(test)] mod tests { use super::*; @@ -1257,11 +1336,45 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + /// The MCP recipe's proxy tool is a tool manifest like any other, so it + /// must load through the real parser, and its schema must be the + /// documented `{tool, arguments}` pair: the model sees nothing else of + /// the server until it reads the skill. The pass-through is the binary + /// itself, named by `$OPENMAX_BIN`, and credentials reach the server only + /// through the manifest's `env` grant. + #[test] + fn the_mcp_recipe_tool_loads_as_a_tool_and_arguments_proxy() { + let text = render("mcp").expect("--spec mcp renders"); + let start = text.find("```toml\n").expect("the recipe shows the proxy tool") + "```toml\n".len(); + let end = text[start..].find("```").expect("the block closes") + start; + let spec = crate::registry::parse_tool_file_from_text_for_tests(&text[start..end]) + .expect("the recipe's proxy tool parses"); + assert_eq!( + spec.parameters, + serde_json::json!({ + "type": "object", + "required": ["tool", "arguments"], + "properties": {"tool": {"type": "string"}, "arguments": {"type": "object"}}, + }) + ); + assert!(spec.mutating, "a server's tools may write, so the proxy is gated as mutating"); + let crate::registry::ToolKind::External(tool) = spec.kind else { panic!("external") }; + assert_eq!(tool.env, ["NOTES_TOKEN"], "the server's credential is granted through env"); + assert_eq!(tool.command, "sh"); + assert_eq!(tool.args[1], "exec \"$OPENMAX_BIN\" --mcp-call -- \"$@\""); + // The words after $0 are the server command, run as its own argv. + assert_eq!(tool.args[3..], ["notes-server", "--stdio"]); + let flat = text.split_whitespace().collect::>().join(" "); + assert!(flat.contains("Each call starts the server"), "the spec says calls are stateless"); + assert!(flat.contains("needs a persistent session"), "the spec names what is unsupported"); + assert!(flat.contains("is not supported yet"), "the spec names what is unsupported"); + } + /// Every file surface tells the agent how to verify what it wrote, and /// every spec states when the file takes effect. #[test] fn specs_name_verification_and_activation() { - for name in ["tools", "skills", "prompts", "hooks", "permissions", "providers", "memory"] { + for name in ["tools", "skills", "prompts", "hooks", "permissions", "providers", "memory", "mcp"] { let text = render(name).unwrap(); assert!(text.contains("openmax --check"), "{name} spec must point at --check"); } diff --git a/crates/tui/Cargo.toml b/crates/tui/Cargo.toml index 9df2b107..18dbd2a1 100644 --- a/crates/tui/Cargo.toml +++ b/crates/tui/Cargo.toml @@ -34,8 +34,14 @@ lexopt = "0.3" unicode-width = "0.2" unicode-segmentation = "1.13" ignore.workspace = true - -[dev-dependencies] -# The CLI tests give a child a pseudo-terminal on stdin, which is how an -# interactive shell runs openmax. +# SIGTERM for an MCP server still running after `--mcp-call` or `--mcp-list` +# closed its stdin. The CLI tests also use it to give a child a +# pseudo-terminal on stdin, which is how an interactive shell runs openmax. libc.workspace = true + +# The MCP bridge tests run their own binary again as a fixture MCP server. A +# libtest binary prints its banner on stdout before any test runs, and stdout +# is that server's protocol channel, so this target supplies its own main. +[[test]] +name = "mcp_bridge" +harness = false diff --git a/crates/tui/src/main.rs b/crates/tui/src/main.rs index a72e6d3a..f984ec15 100644 --- a/crates/tui/src/main.rs +++ b/crates/tui/src/main.rs @@ -3,6 +3,7 @@ mod clipboard; mod completion; mod headless; mod input; +mod mcp; mod stdio; mod theme; mod ui; @@ -67,7 +68,20 @@ options: stdin against the openmax-stdio contract instead --spec print the authoring contract for one surface and exit (tools, skills, prompts, hooks, permissions, - providers, settings, memory, recall, stdio, usage) + providers, settings, memory, recall, stdio, mcp, + usage) + --mcp-list -- + start an MCP stdio server, print its tools as a + skill body (name, arguments, description), and stop + it; with --json, the full tool definitions + --mcp-call -- + call one tool of an MCP stdio server: reads + {\"tool\":\"\",\"arguments\":{...}} on stdin + and prints the result's text; a tool or protocol + error exits 1. Recipe: --spec mcp + --mcp-timeout + with --mcp-list or --mcp-call, the longest wait for + each server reply (default 30, at most 3600) --trust-project trust this exact project root in auto mode (a project already trusted keeps its mode), then run -V, --version print the version @@ -103,6 +117,15 @@ struct CliArgs { ledger_repair: bool, /// Surface name whose authoring contract should be printed (`--spec`). spec: Option, + /// `--mcp-list`: print an MCP server's tools as a skill body. + mcp_list: bool, + /// `--mcp-call`: call one tool of an MCP server, read on stdin. + mcp_call: bool, + /// The MCP server's own command line: every word after `--`, kept as + /// separate argv words (prompt words are joined with spaces). + mcp_server: Option>, + /// `--mcp-timeout`: seconds each wait for an MCP server reply may take. + mcp_timeout: Option, trust_project: bool, /// One prompt string per headless turn (tokens between repeated -p flags /// are joined with spaces into a single turn). @@ -134,13 +157,28 @@ where recall: None, ledger_repair: false, spec: None, + mcp_list: false, + mcp_call: false, + mcp_server: None, + mcp_timeout: None, trust_project: false, prompts: Vec::new(), }; // Tokens for the current -p group; flushed into prompts on the next -p or end. let mut current: Vec = Vec::new(); let mut parser = lexopt::Parser::from_args(args); - while let Some(arg) = parser.next()? { + loop { + // Once an MCP operation is named, a `--` starts the server's command + // line, which is the server's to parse: `npx -y pkg` must not reach + // this parser as a `-y` option. + if (out.mcp_list || out.mcp_call) && out.mcp_server.is_none() { + if let Some(mut raw) = parser.try_raw_args() { + if raw.next_if(|word| word == "--").is_some() { + out.mcp_server = Some(raw.collect()); + } + } + } + let Some(arg) = parser.next()? else { break }; match arg { Short('c') | Long("continue") => out.continue_session = true, Short('m') | Long("model") => out.model = Some(parser.value()?.string()?), @@ -162,6 +200,9 @@ where Long("recall") => set_once(&mut out.recall, "--recall", &mut parser)?, Long("ledger-repair") => out.ledger_repair = true, Long("spec") => set_once(&mut out.spec, "--spec", &mut parser)?, + Long("mcp-list") => out.mcp_list = true, + Long("mcp-call") => out.mcp_call = true, + Long("mcp-timeout") => out.mcp_timeout = Some(parser.value()?.parse()?), Long("trust-project") => out.trust_project = true, Short('V') | Long("version") => { println!("openmax {}", env!("CARGO_PKG_VERSION")); @@ -172,6 +213,12 @@ where std::process::exit(0); } Value(v) => current.push(v.string()?), + _ if out.mcp_list || out.mcp_call => { + return Err(lexopt::Error::from(format!( + "{}; an MCP server's command and its options go after --", + arg.unexpected() + ))); + } _ => return Err(arg.unexpected()), } } @@ -291,6 +338,10 @@ fn refusal(cli: &CliArgs) -> Option { recall, ledger_repair, spec, + mcp_list, + mcp_call, + mcp_server, + mcp_timeout, trust_project, prompts, } = cli; @@ -298,6 +349,8 @@ fn refusal(cli: &CliArgs) -> Option { // of its own: validating a protocol stream on stdin. let operations: Vec<&str> = [ (spec.is_some(), "--spec"), + (*mcp_list, "--mcp-list"), + (*mcp_call, "--mcp-call"), (recall.is_some(), "--recall"), (*ledger, "--ledger"), (*ledger_repair, "--ledger-repair"), @@ -317,13 +370,15 @@ fn refusal(cli: &CliArgs) -> Option { let operation = operations.first().copied(); // Each option and the operations that read it. const SESSIONS: &[Option<&str>] = &[None, Some("--stdio"), Some("--print")]; - let options: [(bool, &str, &[Option<&str>]); 6] = [ + const MCP: &[Option<&str>] = &[Some("--mcp-list"), Some("--mcp-call")]; + let options: [(bool, &str, &[Option<&str>]); 7] = [ (*trust_project, "--trust-project", SESSIONS), (*continue_session, "--continue", SESSIONS), (model.is_some(), "--model", SESSIONS), (provider.is_some(), "--provider", SESSIONS), - (*json, "--json", &[Some("--print"), Some("--check"), Some("--recall")]), + (*json, "--json", &[Some("--print"), Some("--check"), Some("--recall"), Some("--mcp-list")]), (*run_examples, "--run-examples", &[Some("--check")]), + (mcp_timeout.is_some(), "--mcp-timeout", MCP), ]; for (on, flag, readers) in options { if !on || readers.contains(&operation) { @@ -351,15 +406,27 @@ fn refusal(cli: &CliArgs) -> Option { return Some(match operation { None => "unexpected arguments (use --print for headless)".to_string(), Some(operation) => { - let stdin = match operation { + let hint = match operation { "--stdio" => "; it reads commands on stdin", "--check --stdio" => "; it reads the stream on stdin", + "--mcp-list" => "; the server's command goes after --", + "--mcp-call" => "; the server's command goes after --, and the call on stdin", _ => "", }; - format!("{operation} takes no other arguments (got '{words}'){stdin}") + format!("{operation} takes no other arguments (got '{words}'){hint}") } }); } + if let Some(operation @ ("--mcp-list" | "--mcp-call")) = operation { + if mcp_server.as_ref().is_none_or(|words| words.is_empty()) { + return Some(format!( + "{operation} needs the MCP server's command after --: openmax {operation} -- [args...]" + )); + } + } + if mcp_timeout.is_some_and(|secs| !(1..=mcp::MAX_TIMEOUT_SECS).contains(&secs)) { + return Some(format!("--mcp-timeout takes 1 to {} seconds", mcp::MAX_TIMEOUT_SECS)); + } None } @@ -404,6 +471,21 @@ async fn main() -> std::io::Result<()> { } } + if cli.mcp_list || cli.mcp_call { + // A one-shot protocol client, outside the agent loop: no session, no + // endpoint, no state dir, no trust. What it starts is the command the + // caller named, run exactly as the caller could run it. + let server = cli.mcp_server.as_deref().unwrap_or_default(); + let timeout = std::time::Duration::from_secs( + cli.mcp_timeout.unwrap_or(mcp::DEFAULT_TIMEOUT_SECS), + ); + let code = match cli.mcp_list { + true => mcp::list(server, timeout, cli.json), + false => mcp::call(server, timeout), + }; + std::process::exit(code); + } + if let Some(query) = &cli.recall { // Read-only introspection, like --ledger: no session, no endpoint, // no trust gate. Recall only ever surfaces this project's own history @@ -2529,9 +2611,10 @@ mod tests { /// Every operation a command line can select, as its args and the name a /// refusal gives it. `-p` comes last in any line built from these: the - /// tokens after it are its prompt. An empty name is the interactive - /// session. - const OPERATIONS: [(&[&str], &str); 11] = [ + /// tokens after it are its prompt. An MCP operation's server command + /// comes after `--`, so `refusal_of` closes such a line with one. An + /// empty name is the interactive session. + const OPERATIONS: [(&[&str], &str); 13] = [ (&[], ""), (&["--spec", "tools"], "--spec"), (&["--recall", "q"], "--recall"), @@ -2543,6 +2626,8 @@ mod tests { (&["--check", "--stdio"], "--check --stdio"), (&["--stdio"], "--stdio"), (&["-p", "x"], "--print"), + (&["--mcp-list"], "--mcp-list"), + (&["--mcp-call"], "--mcp-call"), ]; /// Whether a refusal names `flag` as a whole flag, so `--ledger` is not @@ -2554,7 +2639,9 @@ mod tests { } fn refusal_of(args: &[&str]) -> Option { - refusal(&parse_args_from(args.iter().copied()).unwrap()) + let mcp = args.iter().any(|a| a.starts_with("--mcp-") && *a != "--mcp-timeout"); + let server: &[&str] = if mcp { &["--", "srv"] } else { &[] }; + refusal(&parse_args_from(args.iter().chain(server).copied()).unwrap()) } /// Every pair of operations is refused before either runs, naming both: @@ -2594,13 +2681,14 @@ mod tests { #[test] fn an_option_the_operation_does_not_read_is_refused_naming_both() { let sessions: &[&str] = &["", "--stdio", "--print"]; - let options: [(&[&str], &str, &[&str]); 7] = [ + let options: [(&[&str], &str, &[&str]); 8] = [ (&["--trust-project"], "--trust-project", sessions), (&["--continue"], "--continue", sessions), (&["-m", "m"], "--model", sessions), (&["--provider", "p"], "--provider", sessions), - (&["--json"], "--json", &["--print", "--check", "--recall"]), + (&["--json"], "--json", &["--print", "--check", "--recall", "--mcp-list"]), (&["--run-examples"], "--run-examples", &["--check"]), + (&["--mcp-timeout", "5"], "--mcp-timeout", &["--mcp-list", "--mcp-call"]), // A bare word is a prompt, which only --print reads. (&["extra"], "", &["--print"]), ]; @@ -2626,11 +2714,76 @@ mod tests { &["--trust-project", "--continue", "-m", "m", "--provider", "p", "-p", "--json", "x"], &["--trust-project", "--continue", "--stdio"], &["--recall", "q", "--json"], + &["--mcp-timeout", "60", "--mcp-list", "--json"], + &["--mcp-call", "--mcp-timeout", "60"], ] { assert_eq!(refusal_of(args), None, "{args:?} is a valid command line"); } } + /// An MCP server's command line belongs to the server: every word after + /// `--` reaches it as its own argv word, options included, while `--` + /// keeps its ordinary meaning on a line with no MCP operation. + #[test] + fn the_mcp_server_command_is_every_word_after_the_separator() { + let words = ["--mcp-timeout", "5", "--mcp-call", "--", "npx", "-y", "@scope/server", "--port", "3"]; + let cli = parse_args_from(words).unwrap(); + assert!(cli.mcp_call && !cli.mcp_list && cli.prompts.is_empty()); + assert_eq!(cli.mcp_timeout, Some(5)); + let server: Vec<&str> = + cli.mcp_server.iter().flatten().map(|w| w.to_str().unwrap()).collect(); + assert_eq!(server, ["npx", "-y", "@scope/server", "--port", "3"]); + assert_eq!(refusal(&cli), None); + // One `--`: a second belongs to the server too. + let cli = parse_args_from(["--mcp-list", "--json", "--", "srv", "--", "x"]).unwrap(); + assert_eq!(cli.mcp_server.unwrap(), ["srv", "--", "x"]); + // Without an MCP operation, `--` still ends openmax's own options. + let cli = parse_args_from(["-p", "--", "--literal"]).unwrap(); + assert_eq!(cli.prompts, ["--literal"]); + assert!(cli.mcp_server.is_none()); + } + + /// Each way an MCP command line goes wrong is refused before anything + /// runs, and the refusal says where the server's command belongs. + #[test] + fn an_mcp_operation_without_a_server_command_is_refused() { + for args in [&["--mcp-list"][..], &["--mcp-call", "--"]] { + let reason = refusal(&parse_args_from(args.iter().copied()).unwrap()).unwrap(); + assert!(reason.contains("needs the MCP server's command after --"), "{args:?}: {reason}"); + } + let reason = refusal(&parse_args_from(["--mcp-list", "srv"]).unwrap()).unwrap(); + assert!(reason.contains("'srv'") && reason.contains("after --"), "{reason}"); + let err = parse_args_from(["--mcp-list", "npx", "-y", "pkg"]).err().unwrap().to_string(); + assert!(err.contains("go after --"), "{err}"); + for secs in ["0", "3601"] { + let cli = parse_args_from(["--mcp-timeout", secs, "--mcp-list", "--", "srv"]).unwrap(); + assert_eq!(refusal(&cli).as_deref(), Some("--mcp-timeout takes 1 to 3600 seconds")); + } + assert!(parse_args_from(["--mcp-timeout", "soon", "--mcp-list", "--", "srv"]).is_err()); + } + + /// `--spec mcp` is what the agent copies from, so every command it shows + /// must be a command line this binary runs as written. + #[test] + fn every_command_the_mcp_spec_shows_is_accepted() { + let spec = open_max_core::spec::render("mcp").expect("--spec mcp renders"); + let mut shown = Vec::new(); + for (at, _) in spec.match_indices("openmax --") { + // A command ends at its closing backtick, a shell operator, or + // the end of its line. + let rest = &spec[at..]; + let end = rest.find(['`', ')', '>', '&', '|', '\n']).unwrap_or(rest.len()); + let words: Vec<&str> = rest[..end].split_whitespace().skip(1).collect(); + let cli = parse_args_from(words.iter().copied()) + .unwrap_or_else(|e| panic!("{words:?} does not parse: {e}")); + assert_eq!(refusal(&cli), None, "{words:?}"); + shown.push(words.join(" ")); + } + for operation in ["--mcp-list --", "--mcp-call --", "--mcp-list -- notes-server --stdio"] { + assert!(shown.iter().any(|c| c.starts_with(operation)), "{operation} is shown: {shown:?}"); + } + } + /// With no operation named, a stray word is most likely a prompt missing /// its -p, so that refusal points at --print. Beside a named operation it /// is most likely meant for that operation (`--check x.toml`, diff --git a/crates/tui/src/mcp.rs b/crates/tui/src/mcp.rs new file mode 100644 index 00000000..9f4f98c2 --- /dev/null +++ b/crates/tui/src/mcp.rs @@ -0,0 +1,737 @@ +//! `openmax --mcp-list` and `openmax --mcp-call`: a one-shot MCP client over +//! stdio, run as its own process and never inside the agent loop. +//! +//! An MCP server becomes a proxy tool plus a skill (`openmax --spec mcp`); +//! this module is the protocol half of that recipe, so the proxy needs no +//! interpreter on the host. Each run starts the server, performs the +//! initialize handshake, makes one request (all pages of tools/list, or one +//! tools/call), and stops the server. Nothing stays resident and no schema +//! reaches the frozen prompt. + +use std::ffi::OsString; +use std::io::{BufRead, BufReader, IsTerminal, Read, Write}; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +use serde_json::{json, Value}; + +/// The default bound on each wait for a server reply, in seconds. +pub const DEFAULT_TIMEOUT_SECS: u64 = 30; +/// The largest `--mcp-timeout` accepted, in seconds. +pub const MAX_TIMEOUT_SECS: u64 = 3600; + +/// The initialize-based protocol revisions this client speaks, newest first; +/// the first is the one it asks for. +const PROTOCOL_VERSIONS: [&str; 4] = ["2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05"]; +/// One message, or the call read on stdin. A server that writes a longer +/// line without ending it would otherwise grow this process without bound. +const MAX_MESSAGE_BYTES: u64 = 64 << 20; +/// tools/list pages followed before the server is judged to be looping. +const MAX_PAGES: usize = 1000; +/// How long a server gets to exit after its stdin closes, and again after +/// SIGTERM, before the next, harder step. +const STOP_GRACE: Duration = Duration::from_secs(2); +/// Tool descriptions are cut to this many characters in the list; the full +/// text is in `--mcp-list --json`. +const MAX_DESCRIPTION_CHARS: usize = 300; + +/// `--mcp-list`: print the server's tools as a skill body, or with `json` the +/// full definitions. Returns the exit code. +pub fn list(server: &[OsString], timeout: Duration, json: bool) -> i32 { + let run = || -> Result { + let mut session = Session::start(server, timeout)?; + let init = session.initialize()?; + let tools = session.list_tools()?; + if json { + return serde_json::to_string_pretty(&tools).map_err(|e| e.to_string()); + } + render_list(&init, &tools) + }; + match run() { + Ok(text) => { + println!("{}", text.trim_end()); + 0 + } + Err(reason) => { + eprintln!("openmax: {reason}"); + 1 + } + } +} + +/// `--mcp-call`: read `{"tool", "arguments"}` on stdin, call that tool, and +/// print the result's text. Returns the exit code: 2 when the input cannot +/// be read (no server is started), 1 when the server or the tool fails. +pub fn call(server: &[OsString], timeout: Duration) -> i32 { + let stdin = std::io::stdin(); + if stdin.is_terminal() { + eprintln!(r#"openmax: --mcp-call reads the call on stdin, as {{"tool": "", "arguments": {{...}}}}"#); + return 2; + } + let mut input = String::new(); + if let Err(e) = stdin.lock().take(MAX_MESSAGE_BYTES).read_to_string(&mut input) { + eprintln!("openmax: --mcp-call could not read stdin: {e}"); + return 2; + } + let (tool, arguments) = match parse_call(&input) { + Ok(call) => call, + Err(reason) => { + eprintln!("openmax: {reason}"); + return 2; + } + }; + let run = || -> Result { + let mut session = Session::start(server, timeout)?; + session.initialize()?; + session.request("tools/call", json!({"name": tool, "arguments": arguments})) + }; + match run() { + Ok(result) => { + let text = render_content(&result); + if result.get("isError").and_then(Value::as_bool) == Some(true) { + eprintln!( + "openmax: the server's {} tool reported an error: {text}", + open_max_core::text::one_line(&tool) + ); + return 1; + } + if !text.is_empty() { + println!("{}", text.trim_end_matches('\n')); + } + 0 + } + Err(reason) => { + eprintln!("openmax: {reason}"); + 1 + } + } +} + +/// The call the proxy tool writes: a tool name and an arguments object, and +/// nothing else. An unknown key is refused rather than dropped, because the +/// likeliest one is a misspelled `arguments`, which would otherwise call the +/// tool with none. +fn parse_call(input: &str) -> Result<(String, Value), String> { + const SHAPE: &str = r#"--mcp-call expects {"tool": "", "arguments": {...}} on stdin"#; + let value: Value = serde_json::from_str(input).map_err(|e| format!("{SHAPE}: {e}"))?; + let Value::Object(mut fields) = value else { + return Err(format!("{SHAPE}, not {}", kind(&value))); + }; + let tool = match fields.remove("tool") { + Some(Value::String(tool)) if !tool.trim().is_empty() => tool, + Some(other) => return Err(format!("{SHAPE}: tool must be a non-empty string, not {}", kind(&other))), + None => return Err(format!("{SHAPE}: tool is missing")), + }; + let arguments = match fields.remove("arguments") { + None | Some(Value::Null) => json!({}), + Some(arguments @ Value::Object(_)) => arguments, + Some(other) => return Err(format!("{SHAPE}: arguments must be an object, not {}", kind(&other))), + }; + if let Some(key) = fields.keys().next() { + return Err(format!("{SHAPE}: unknown key '{}'", open_max_core::text::one_line(key))); + } + Ok((tool, arguments)) +} + +fn kind(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "a boolean", + Value::Number(_) => "a number", + Value::String(_) => "a string", + Value::Array(_) => "an array", + Value::Object(_) => "an object", + } +} + +/// One running server: its stdin for requests, and a reader thread turning +/// its stdout into lines, so every wait can be bounded by a timeout. +struct Session { + child: Child, + stdin: Option, + lines: mpsc::Receiver, String>>, + timeout: Duration, + next_id: u64, + stopped: bool, +} + +impl Session { + fn start(server: &[OsString], timeout: Duration) -> Result { + let Some((program, args)) = server.split_first() else { + return Err("no MCP server command was given after --".into()); + }; + // stderr is inherited: a server's diagnostics are the caller's to read. + let mut child = Command::new(program) + .args(args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|e| { + format!( + "cannot start the MCP server {}: {e}", + open_max_core::text::one_line(&program.to_string_lossy()) + ) + })?; + let stdout = child.stdout.take().expect("stdout is piped"); + let stdin = child.stdin.take(); + let (tx, lines) = mpsc::channel(); + // Detached: a server can leave a descendant holding its stdout open, + // and the run must not wait on that to end. + std::thread::spawn(move || read_lines(stdout, tx)); + Ok(Self { child, stdin, lines, timeout, next_id: 1, stopped: false }) + } + + /// The handshake: ask for the newest revision this client speaks, accept + /// any revision it speaks that the server answers with, then confirm. + /// Returns the initialize result (server info and instructions). + fn initialize(&mut self) -> Result { + let result = self.request( + "initialize", + json!({ + "protocolVersion": PROTOCOL_VERSIONS[0], + "capabilities": {}, + "clientInfo": {"name": "openmax", "version": env!("CARGO_PKG_VERSION")}, + }), + )?; + let Some(version) = result.get("protocolVersion").and_then(Value::as_str) else { + return Err("the server's initialize result has no protocolVersion".into()); + }; + if !PROTOCOL_VERSIONS.contains(&version) { + return Err(format!( + "the server speaks MCP protocol revision {}; this client speaks {}", + open_max_core::text::one_line(version), + PROTOCOL_VERSIONS.join(", ") + )); + } + self.send(&json!({"jsonrpc": "2.0", "method": "notifications/initialized"}), "notifications/initialized")?; + Ok(result) + } + + /// Every page of tools/list, following `nextCursor` until the server + /// stops returning one. + fn list_tools(&mut self) -> Result, String> { + let mut tools = Vec::new(); + let mut cursors = std::collections::HashSet::new(); + let mut params = json!({}); + loop { + let page = self.request("tools/list", params)?; + let Some(items) = page.get("tools").and_then(Value::as_array) else { + return Err("a tools/list result has no tools array".into()); + }; + tools.extend(items.iter().cloned()); + // An empty cursor ends the list too: clients that test the + // cursor for truth stop there, and servers rely on it. + let cursor = match page.get("nextCursor") { + None | Some(Value::Null) => break, + Some(Value::String(cursor)) if cursor.is_empty() => break, + Some(Value::String(cursor)) => cursor.clone(), + Some(other) => return Err(format!("a tools/list nextCursor is {}, not a string", kind(other))), + }; + if !cursors.insert(cursor.clone()) || cursors.len() > MAX_PAGES { + return Err(format!( + "tools/list is still paging after {} pages; the server repeats or never ends its cursors", + cursors.len() + )); + } + params = json!({"cursor": cursor}); + } + Ok(tools) + } + + /// Send one request and wait, within the timeout, for its response. + /// Notifications are ignored and requests from the server are answered + /// while waiting, so neither side blocks on the other. + fn request(&mut self, method: &str, params: Value) -> Result { + let id = self.next_id; + self.next_id += 1; + self.send(&json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}), method)?; + let deadline = Instant::now() + self.timeout; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(self.silent(method)); + } + let line = match self.lines.recv_timeout(remaining) { + Ok(Ok(line)) => line, + Ok(Err(reason)) => return Err(reason), + Err(mpsc::RecvTimeoutError::Timeout) => return Err(self.silent(method)), + Err(mpsc::RecvTimeoutError::Disconnected) => return Err(self.closed(method)), + }; + let text = String::from_utf8_lossy(&line); + let text = text.trim(); + if text.is_empty() { + continue; + } + let value: Value = serde_json::from_str(text) + .map_err(|e| malformed(method, text, &format!("not JSON ({e})")))?; + // A batch is one line holding several messages; answer each. + let messages = match value { + Value::Array(messages) => messages, + message => vec![message], + }; + for message in messages { + if let Some(outcome) = self.handle(method, id, text, message)? { + return outcome; + } + } + } + } + + /// One message read while waiting on request `id`: Some(outcome) when it + /// is that request's response. + fn handle( + &mut self, + method: &str, + id: u64, + line: &str, + message: Value, + ) -> Result>, String> { + let Value::Object(message) = message else { + return Err(malformed(method, line, "not a JSON-RPC object")); + }; + if message.get("jsonrpc").and_then(Value::as_str) != Some("2.0") { + return Err(malformed(method, line, "not JSON-RPC 2.0")); + } + if let Some(request) = message.get("method").and_then(Value::as_str) { + // A request from the server carries an id and needs an answer; + // a notification (log lines, progress) needs none. This client + // declares no capabilities, so ping is the one request it serves. + if let Some(request_id) = message.get("id") { + let reply = match request { + "ping" => json!({"jsonrpc": "2.0", "id": request_id, "result": {}}), + _ => json!({"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32601, + "message": format!("method not found: {request} (this client declares no capabilities)"), + }}), + }; + self.send(&reply, method)?; + } + return Ok(None); + } + // An error about a request the server could not read carries a null + // id, and only one request is ever open, so it is that request's. + let unread = message.get("id") == Some(&Value::Null) && message.contains_key("error"); + if message.get("id") != Some(&json!(id)) && !unread { + // A response to no request this client has open. + return Ok(None); + } + if let Some(error) = message.get("error") { + let text = error.get("message").and_then(Value::as_str).unwrap_or("no message"); + let code = error.get("code").map(Value::to_string).unwrap_or_else(|| "none".into()); + return Ok(Some(Err(format!( + "the server answered {method} with an error: {} (code {})", + open_max_core::text::one_line(text), + open_max_core::text::one_line(&code) + )))); + } + match message.get("result") { + Some(result) => Ok(Some(Ok(result.clone()))), + None => Err(malformed(method, line, "a response with neither result nor error")), + } + } + + fn send(&mut self, message: &Value, method: &str) -> Result<(), String> { + let Some(stdin) = self.stdin.as_mut() else { + return Err(format!("the server's stdin is closed before {method}")); + }; + let mut line = message.to_string(); + line.push('\n'); + if stdin.write_all(line.as_bytes()).and_then(|()| stdin.flush()).is_err() { + return Err(self.closed(method)); + } + Ok(()) + } + + fn silent(&self, method: &str) -> String { + format!( + "the MCP server did not answer {method} within {}s (raise --mcp-timeout for a server that is slow to start)", + self.timeout.as_secs() + ) + } + + /// The server closed its output or its input: name how it ended, when + /// it has. + fn closed(&mut self, method: &str) -> String { + let deadline = Instant::now() + Duration::from_millis(500); + let status = loop { + match self.child.try_wait() { + Ok(Some(status)) => break Some(status), + Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(10)), + _ => break None, + } + }; + match status { + Some(status) => format!("the MCP server exited ({status}) before answering {method}"), + None => format!("the MCP server closed its stdio before answering {method}"), + } + } + + /// Stop the server the way the stdio transport defines: close its stdin, + /// then SIGTERM, then SIGKILL, each after a grace period, and reap it. + fn stop(&mut self) { + if std::mem::replace(&mut self.stopped, true) { + return; + } + drop(self.stdin.take()); + if self.exited_within(STOP_GRACE) { + return; + } + #[cfg(unix)] + { + if let Ok(pid) = libc::pid_t::try_from(self.child.id()) { + // SAFETY: the pid is this process's own unreaped child, so it + // cannot have been reused by another process. + unsafe { libc::kill(pid, libc::SIGTERM) }; + if self.exited_within(STOP_GRACE) { + return; + } + } + } + let _ = self.child.kill(); + let _ = self.child.wait(); + } + + fn exited_within(&mut self, grace: Duration) -> bool { + let deadline = Instant::now() + grace; + loop { + match self.child.try_wait() { + Ok(Some(_)) => return true, + Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(10)), + _ => return false, + } + } + } +} + +impl Drop for Session { + fn drop(&mut self) { + self.stop(); + } +} + +/// Forward the server's stdout one line at a time, ending at EOF, at a read +/// error, or at a line longer than any message may be. +fn read_lines(stdout: impl Read, tx: mpsc::Sender, String>>) { + let mut reader = BufReader::new(stdout); + loop { + let mut line = Vec::new(); + match (&mut reader).take(MAX_MESSAGE_BYTES + 1).read_until(b'\n', &mut line) { + Ok(0) => return, + Ok(_) if line.len() as u64 > MAX_MESSAGE_BYTES => { + let _ = tx.send(Err(format!( + "the MCP server wrote a line longer than {} MiB", + MAX_MESSAGE_BYTES >> 20 + ))); + return; + } + Ok(_) => { + if tx.send(Ok(line)).is_err() { + return; + } + } + Err(e) => { + let _ = tx.send(Err(format!("reading the MCP server's stdout failed: {e}"))); + return; + } + } + } +} + +/// A reply that is not the protocol: quote the start of the line, flattened, +/// so the reason fits one line and shows what the server actually wrote. +fn malformed(method: &str, line: &str, what: &str) -> String { + let shown = truncated(&open_max_core::text::one_line(line), 200); + format!("the MCP server sent a malformed reply to {method}, {what}: {shown}") +} + +/// The skill body: a header naming the server and how to call it, the +/// server's own instructions when it gives any, and one line per tool. +fn render_list(init: &Value, tools: &[Value]) -> Result { + let info = &init["serverInfo"]; + let server = [info["name"].as_str(), info["version"].as_str()] + .into_iter() + .flatten() + .map(open_max_core::text::one_line) + .collect::>() + .join(" "); + let version = init["protocolVersion"].as_str().unwrap_or_default(); + let mut out = format!( + "MCP server {}(protocol {version}): {} tool{}. Call one through this server's proxy tool \ + with {{\"tool\": \"\", \"arguments\": {{...}}}}. In each line `?` marks an optional \ + argument; `openmax --mcp-list --json -- ` prints the full schemas.\n", + if server.is_empty() { String::new() } else { format!("{server} ") }, + tools.len(), + if tools.len() == 1 { "" } else { "s" }, + ); + if let Some(instructions) = init["instructions"].as_str().map(str::trim).filter(|s| !s.is_empty()) { + // Quoted line by line: instructions are usually markdown, and their + // own `- ` bullets would otherwise read as tool lines. + out.push_str("\nServer instructions:\n"); + for line in instructions.lines() { + out.push_str(format!("> {}", open_max_core::text::one_line(line)).trim_end()); + out.push('\n'); + } + } + out.push('\n'); + for tool in tools { + let Some(name) = tool.get("name").and_then(Value::as_str) else { + return Err(format!("a tools/list entry has no name: {}", truncated(&tool.to_string(), 200))); + }; + out.push_str(&format!( + "- {}({})", + open_max_core::text::one_line(name), + argument_summary(tool.get("inputSchema")) + )); + let description = compact(tool.get("description").and_then(Value::as_str).unwrap_or_default()); + if !description.is_empty() { + out.push_str(": "); + out.push_str(&truncated(&description, MAX_DESCRIPTION_CHARS)); + } + out.push('\n'); + } + Ok(out) +} + +/// `name: type` per argument, required ones first, each group in name order, +/// with `?` after an optional name. Nested objects stay `object`; the full +/// schema is one `--json` away. +fn argument_summary(schema: Option<&Value>) -> String { + let Some(schema) = schema else { return String::new() }; + let Some(properties) = schema.get("properties").and_then(Value::as_object) else { + return String::new(); + }; + let required: Vec<&str> = schema + .get("required") + .and_then(Value::as_array) + .map(|names| names.iter().filter_map(Value::as_str).collect()) + .unwrap_or_default(); + let (first, rest): (Vec<_>, Vec<_>) = + properties.iter().partition(|(name, _)| required.contains(&name.as_str())); + first + .iter() + .map(|(name, schema)| format!("{}: {}", open_max_core::text::one_line(name), type_of(schema, 0))) + .chain(rest.iter().map(|(name, schema)| { + format!("{}?: {}", open_max_core::text::one_line(name), type_of(schema, 0)) + })) + .collect::>() + .join(", ") +} + +/// A compact type for one argument schema: the declared type, `T[]` for an +/// array, a `|` union for alternatives, and the literal values of an enum. +fn type_of(schema: &Value, depth: usize) -> String { + if depth > 4 { + return "any".into(); + } + if let Some(values) = schema.get("enum").and_then(Value::as_array) { + let mut shown: Vec = + values.iter().take(8).map(|v| open_max_core::text::one_line(&v.to_string())).collect(); + if values.len() > 8 { + shown.push("…".into()); + } + return shown.join("|"); + } + if let Some(value) = schema.get("const") { + return open_max_core::text::one_line(&value.to_string()); + } + for key in ["anyOf", "oneOf"] { + if let Some(alternatives) = schema.get(key).and_then(Value::as_array) { + let mut types: Vec = Vec::new(); + for alternative in alternatives { + let ty = type_of(alternative, depth + 1); + if !types.contains(&ty) { + types.push(ty); + } + } + return types.join("|"); + } + } + match schema.get("type") { + Some(Value::String(ty)) if ty == "array" => match schema.get("items") { + Some(items) => { + let item = type_of(items, depth + 1); + if item.contains('|') { format!("({item})[]") } else { format!("{item}[]") } + } + None => "array".into(), + }, + Some(Value::String(ty)) => open_max_core::text::one_line(ty), + Some(Value::Array(types)) => types + .iter() + .filter_map(Value::as_str) + .map(open_max_core::text::one_line) + .collect::>() + .join("|"), + _ if schema.get("properties").is_some() => "object".into(), + _ => "any".into(), + } +} + +/// One line with runs of whitespace collapsed. +fn compact(text: &str) -> String { + open_max_core::text::one_line(text).split_whitespace().collect::>().join(" ") +} + +fn truncated(text: &str, max_chars: usize) -> String { + match text.char_indices().nth(max_chars) { + Some((at, _)) => format!("{}…", &text[..at]), + None => text.to_string(), + } +} + +/// A tools/call result as text: each content item on its own line, text as +/// written, and binary items described rather than dumped. A result with no +/// content falls back to its structured content. +fn render_content(result: &Value) -> String { + let items = result.get("content").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default(); + if items.is_empty() { + return match result.get("structuredContent") { + Some(structured) if !structured.is_null() => structured.to_string(), + _ => String::new(), + }; + } + items + .iter() + .map(|item| { + let field = |key: &str| item.get(key).and_then(Value::as_str).unwrap_or_default(); + match field("type") { + "text" => field("text").to_string(), + media @ ("image" | "audio") => { + format!("[{media} {}, {} bytes]", field("mimeType"), base64_len(field("data"))) + } + "resource" => { + let resource = &item["resource"]; + let uri = resource["uri"].as_str().unwrap_or_default(); + match (resource["text"].as_str(), resource["blob"].as_str()) { + (Some(text), _) => format!("[resource {uri}]\n{text}"), + (None, Some(blob)) => format!( + "[resource {uri} {}, {} bytes]", + resource["mimeType"].as_str().unwrap_or_default(), + base64_len(blob) + ), + (None, None) => format!("[resource {uri}]"), + } + } + "resource_link" => format!("[resource link {} {}]", field("uri"), field("name")), + _ => item.to_string(), + } + }) + .collect::>() + .join("\n") +} + +/// The decoded size of base64 text, without decoding it. +fn base64_len(data: &str) -> usize { + let data = data.trim_end(); + let padding = data.bytes().rev().take_while(|b| *b == b'=').count(); + (data.len() / 4 * 3).saturating_sub(padding) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The proxy hands over exactly what the model wrote; a misspelled key + /// would otherwise call the tool with no arguments at all. + #[test] + fn a_call_is_a_tool_name_and_an_arguments_object() { + assert_eq!( + parse_call(r#"{"tool":"search","arguments":{"q":"x"}}"#).unwrap(), + ("search".to_string(), json!({"q": "x"})) + ); + assert_eq!(parse_call(r#"{"tool":"now"}"#).unwrap(), ("now".to_string(), json!({}))); + for (input, reason) in [ + (r#"{"tool":"search","args":{"q":"x"}}"#, "unknown key 'args'"), + (r#"{"arguments":{}}"#, "tool is missing"), + (r#"{"tool":"","arguments":{}}"#, "tool must be a non-empty string"), + (r#"{"tool":"search","arguments":"{\"q\":1}"}"#, "arguments must be an object, not a string"), + (r#"["search"]"#, "not an array"), + ("search", "expects"), + ] { + let err = parse_call(input).unwrap_err(); + assert!(err.contains(reason), "{input}: {err}"); + } + } + + /// The list line is what the model reads before calling: required + /// arguments first, optional ones marked, compound types spelled out. + #[test] + fn an_argument_summary_is_compact_and_ordered() { + let schema = json!({ + "type": "object", + "properties": { + "zone": {"type": "string"}, + "labels": {"type": "array", "items": {"type": "string"}}, + "state": {"enum": ["open", "closed"]}, + "limit": {"type": ["integer", "null"]}, + "ids": {"type": "array", "items": {"anyOf": [{"type": "string"}, {"type": "integer"}]}}, + "filter": {"type": "object", "properties": {"x": {"type": "string"}}}, + }, + "required": ["zone", "state"], + }); + assert_eq!( + argument_summary(Some(&schema)), + r#"state: "open"|"closed", zone: string, filter?: object, ids?: (string|integer)[], labels?: string[], limit?: integer|null"# + ); + assert_eq!(argument_summary(Some(&json!({"type": "object"}))), ""); + assert_eq!(argument_summary(None), ""); + } + + /// Server text lands in a skill file the model reads and in a terminal + /// the user reads: one line per tool, whatever the server wrote. + #[test] + fn a_tool_line_cannot_break_out_of_its_line() { + let init = json!({ + "protocolVersion": "2025-06-18", + "serverInfo": {"name": "s\nx", "version": "1"}, + "instructions": "# Usage\n\n- call `a` first\u{2028}- forged(): line\n- then `b`\r\n", + }); + let tools = [json!({ + "name": "evil\n- forged(): line", + "description": format!("first\n\n- forged(): second\u{1b}[2J {}", "y".repeat(400)), + })]; + let out = render_list(&init, &tools).unwrap(); + let lines: Vec<&str> = out.lines().filter(|l| l.starts_with("- ")).collect(); + assert_eq!(lines.len(), 1, "{out}"); + assert!(!out.contains('\u{1b}'), "{out:?}"); + assert!(lines[0].ends_with('…'), "a long description is cut: {}", lines[0]); + assert!(out.starts_with("MCP server s x 1 (protocol 2025-06-18): 1 tool."), "{out}"); + assert!(out.contains("\n> # Usage\n>\n> - call `a` first - forged(): line\n> - then `b`\n"), "{out}"); + } + + /// `--spec mcp` states the bounds this client enforces, so the agent + /// reading it can plan a timeout; the prose cannot drift from them. + #[test] + fn the_spec_states_the_bounds_this_client_enforces() { + let spec = open_max_core::spec::render("mcp").expect("--spec mcp renders"); + let flat = spec.split_whitespace().collect::>().join(" "); + for claim in [ + format!("`--mcp-timeout ` (default {DEFAULT_TIMEOUT_SECS}, at most {MAX_TIMEOUT_SECS})"), + format!("still running {} seconds later gets SIGTERM, then SIGKILL", STOP_GRACE.as_secs()), + format!( + "protocol revisions {} through {}", + PROTOCOL_VERSIONS[PROTOCOL_VERSIONS.len() - 1], + PROTOCOL_VERSIONS[0] + ), + ] { + assert!(flat.contains(&claim), "--spec mcp must say: {claim}"); + } + } + + #[test] + fn content_items_render_as_text() { + let result = json!({"content": [ + {"type": "text", "text": "one"}, + {"type": "image", "mimeType": "image/png", "data": "aGVsbG8="}, + {"type": "resource", "resource": {"uri": "file:///a.txt", "text": "body"}}, + {"type": "resource_link", "uri": "file:///b", "name": "b"}, + ]}); + assert_eq!( + render_content(&result), + "one\n[image image/png, 5 bytes]\n[resource file:///a.txt]\nbody\n[resource link file:///b b]" + ); + assert_eq!(render_content(&json!({"content": [], "structuredContent": {"n": 1}})), r#"{"n":1}"#); + assert_eq!(render_content(&json!({})), ""); + } +} diff --git a/crates/tui/tests/mcp_bridge.rs b/crates/tui/tests/mcp_bridge.rs new file mode 100644 index 00000000..4f5bbbeb --- /dev/null +++ b/crates/tui/tests/mcp_bridge.rs @@ -0,0 +1,556 @@ +//! The MCP bridge end to end: `openmax --mcp-list` and `openmax --mcp-call` +//! against a fixture MCP stdio server, and a proxy tool written exactly as +//! `openmax --spec mcp` prints it, called by a model turn through the harness. +//! +//! The fixture server is this test binary run again as +//! ` --mcp-fixture `. A libtest binary prints its banner on +//! stdout before any test runs, and stdout is the server's protocol channel, +//! so this target has no libtest harness (`harness = false`): `main` either +//! serves or runs the tests listed in `TESTS`. + +use std::io::{BufRead, Read, Write}; +use std::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use serde_json::{json, Value}; + +const FIXTURE: &str = "--mcp-fixture"; + +const TESTS: [(&str, fn()); 6] = [ + ("a_proxy_tool_written_from_the_spec_lists_and_calls_through_openmax", proxy_tool_from_the_spec), + ("an_is_error_result_exits_nonzero_with_its_text", is_error_result), + ("a_malformed_reply_is_reported_and_exits_nonzero", malformed_reply), + ("a_server_that_never_answers_times_out_and_is_stopped", silent_server), + ("every_page_of_tools_is_listed", paginated_list), + ("a_server_request_is_answered_and_a_notification_ignored", server_traffic_mid_call), +]; + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) == Some(FIXTURE) { + serve(&args[1..]); + return; + } + // Enough of libtest's command line for `cargo test `, `--exact`, + // `--list`, and `--ignored` (this target has no ignored tests) to mean + // what they mean everywhere else. + if args.iter().any(|a| a == "--list") { + for (name, _) in TESTS { + println!("{name}: test"); + } + return; + } + let exact = args.iter().any(|a| a == "--exact"); + let filters: Vec<&str> = + args.iter().filter(|a| !a.starts_with('-')).map(String::as_str).collect(); + let selected: Vec<(&str, fn())> = TESTS + .into_iter() + .filter(|_| !args.iter().any(|a| a == "--ignored")) + .filter(|(name, _)| { + filters.is_empty() + || filters.iter().any(|f| if exact { name == f } else { name.contains(f) }) + }) + .collect(); + println!("\nrunning {} tests", selected.len()); + let mut failed = Vec::new(); + for (name, test) in &selected { + let ok = std::panic::catch_unwind(*test).is_ok(); + println!("test {name} ... {}", if ok { "ok" } else { "FAILED" }); + if !ok { + failed.push(*name); + } + } + println!( + "\ntest result: {}. {} passed; {} failed\n", + if failed.is_empty() { "ok" } else { "FAILED" }, + selected.len() - failed.len(), + failed.len() + ); + if !failed.is_empty() { + std::process::exit(101); + } +} + +// ---------------------------------------------------------------- fixture + +/// One fixture MCP server. Modes: `basic` (two tools, `echo` and `fail`), +/// `paged` (four tools over three tools/list pages), `malformed` (answers +/// initialize with a line that is not JSON-RPC), `unsupported` (answers with +/// a protocol version no client knows), and `silent ` (never reads +/// or writes, and ignores its closed stdin). +fn serve(args: &[String]) { + let mode = args.first().map(String::as_str).unwrap_or("basic"); + // A line on stderr, which the bridge must pass through to its caller. + eprintln!("fixture {mode}: serving"); + if mode == "silent" { + if let Some(pidfile) = args.get(1) { + std::fs::write(pidfile, std::process::id().to_string()).unwrap(); + } + std::thread::sleep(Duration::from_secs(120)); + return; + } + let stdin = std::io::stdin(); + let mut lines = stdin.lock().lines(); + let mut out = std::io::stdout().lock(); + let mut initialized = false; + while let Some(Ok(line)) = lines.next() { + let msg: Value = serde_json::from_str(&line).expect("the client writes one JSON message per line"); + assert_eq!(msg["jsonrpc"], "2.0", "{line}"); + let id = msg["id"].clone(); + let result = |result: Value| json!({"jsonrpc": "2.0", "id": id, "result": result}); + let error = |code: i64, message: String| { + json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}) + }; + let method = msg["method"].as_str().unwrap_or_default(); + let reply = match method { + "initialize" => { + let requested = msg["params"]["protocolVersion"].as_str().unwrap_or_default(); + let version = if mode == "unsupported" { "2099-01-01" } else { requested }; + if mode == "malformed" { + writeln!(out, "this line is not JSON-RPC").unwrap(); + out.flush().unwrap(); + continue; + } + result(json!({ + "protocolVersion": version, + "capabilities": {"tools": {}}, + "serverInfo": {"name": "fixture", "version": "1.0.0"}, + })) + } + "notifications/initialized" => { + initialized = true; + continue; + } + _ if !initialized => error(-32600, format!("{method} before notifications/initialized")), + "tools/list" => match (mode, msg["params"]["cursor"].as_str()) { + ("paged", None) => result(json!({"tools": [tool("alpha")], "nextCursor": "page-2"})), + ("paged", Some("page-2")) => { + result(json!({"tools": [tool("beta"), tool("gamma")], "nextCursor": "page-3"})) + } + ("paged", Some("page-3")) => result(json!({"tools": [tool("delta")]})), + (_, None) => result(json!({"tools": [ + { + "name": "echo", + "description": "Echo the text back\nwith the server's view of its environment", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string"}, "repeat": {"type": "integer"}}, + "required": ["text"], + }, + }, + {"name": "fail", "description": "Always reports an error", "inputSchema": {"type": "object"}}, + ]})), + (_, Some(cursor)) => error(-32602, format!("unknown cursor {cursor}")), + }, + "tools/call" => { + // Server traffic in the middle of a call: a notification the + // client must ignore, and a request it must answer before the + // call's own reply can arrive. + send(&mut out, &json!({"jsonrpc": "2.0", "method": "notifications/message", "params": {"level": "info", "data": "working"}})); + send(&mut out, &json!({"jsonrpc": "2.0", "id": "fixture-ping", "method": "ping"})); + let pong: Value = serde_json::from_str(&lines.next().unwrap().unwrap()).unwrap(); + if pong != json!({"jsonrpc": "2.0", "id": "fixture-ping", "result": {}}) { + send(&mut out, &error(-32603, format!("the client answered ping with {pong}"))); + continue; + } + let arguments = &msg["params"]["arguments"]; + let env = |name: &str| std::env::var(name).unwrap_or_else(|_| "unset".into()); + match msg["params"]["name"].as_str().unwrap_or_default() { + "echo" => result(json!({"content": [ + {"type": "text", "text": format!( + "echo: {}; NOTES_TOKEN={}; NOT_GRANTED={}", + arguments["text"].as_str().unwrap_or_default(), + env("NOTES_TOKEN"), + env("NOT_GRANTED"), + )}, + {"type": "image", "mimeType": "image/png", "data": "aGk="}, + ]})), + "fail" => result(json!({ + "content": [{"type": "text", "text": "the fail tool always fails"}], + "isError": true, + })), + other => error(-32602, format!("unknown tool: {other}")), + } + } + other => error(-32601, format!("method not found: {other}")), + }; + send(&mut out, &reply); + } +} + +fn tool(name: &str) -> Value { + json!({"name": name, "description": format!("The {name} tool"), "inputSchema": {"type": "object"}}) +} + +fn send(out: &mut impl Write, message: &Value) { + writeln!(out, "{message}").unwrap(); + out.flush().unwrap(); +} + +// ---------------------------------------------------------------- helpers + +fn openmax_bin() -> &'static str { + env!("CARGO_BIN_EXE_openmax") +} + +fn fixture() -> String { + std::env::current_exe().unwrap().display().to_string() +} + +/// A fresh project dir plus a fresh HOME, so trust and settings never leak +/// between tests or into the developer's real ~/.openmax. +fn fresh_dirs(tag: &str) -> (PathBuf, PathBuf) { + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let base = std::env::temp_dir().join(format!("openmax-mcp-{tag}-{}-{nonce}", std::process::id())); + let project = base.join("project"); + let home = base.join("home"); + std::fs::create_dir_all(&project).unwrap(); + std::fs::create_dir_all(&home).unwrap(); + (project, home) +} + +fn openmax(project: &Path, home: &Path) -> Command { + let mut c = Command::new(openmax_bin()); + c.current_dir(project); + c.env("HOME", home); + c.env_remove("OPENMAX_API_KEY"); + // A developer may run cargo test from inside a session; the harness + // marks such children and trust would refuse. + c.env_remove("OPENMAX_SESSION"); + // Tests are human-run automation with no terminal. + c.env("OPENMAX_HUMAN_ATTEST", "1"); + // The fixture reports these two; only a test that sets them sees them. + c.env_remove("NOTES_TOKEN"); + c.env_remove("NOT_GRANTED"); + c +} + +struct Run { + code: Option, + stdout: String, + stderr: String, + elapsed: Duration, +} + +/// `openmax -- --mcp-fixture `, with +/// `stdin` written to the bridge, bounded so a hang fails the test instead of +/// stalling the suite. +fn bridge(flags: &[&str], fixture_args: &[&str], stdin: &str) -> Run { + let (project, home) = fresh_dirs("bridge"); + let mut command = openmax(&project, &home); + command.args(flags).arg("--").arg(fixture()).arg(FIXTURE).args(fixture_args); + let run = finish(command, stdin); + let _ = std::fs::remove_dir_all(project.parent().unwrap()); + run +} + +fn finish(mut command: Command, stdin: &str) -> Run { + let start = Instant::now(); + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + // A run refused before it reads stdin closes the pipe; that is its + // exit status to report, not a test failure here. + let _ = child.stdin.take().unwrap().write_all(stdin.as_bytes()); + let mut stdout = child.stdout.take().unwrap(); + let mut stderr = child.stderr.take().unwrap(); + let out = std::thread::spawn(move || { + let mut text = String::new(); + stdout.read_to_string(&mut text).unwrap(); + text + }); + let err = std::thread::spawn(move || { + let mut text = String::new(); + stderr.read_to_string(&mut text).unwrap(); + text + }); + let deadline = start + Duration::from_secs(60); + let status = loop { + if let Some(status) = child.try_wait().unwrap() { + break status; + } + if Instant::now() >= deadline { + let _ = child.kill(); + panic!("openmax did not finish: {}", err.join().unwrap()); + } + std::thread::sleep(Duration::from_millis(10)); + }; + Run { + code: status.code(), + stdout: out.join().unwrap(), + stderr: err.join().unwrap(), + elapsed: start.elapsed(), + } +} + +/// The body of the first fenced block in `text` opened with "```". +fn fenced(text: &str, lang: &str) -> String { + let open = format!("```{lang}\n"); + let start = text.find(&open).unwrap_or_else(|| panic!("no ```{lang} block in:\n{text}")) + open.len(); + let end = text[start..].find("```").expect("the block closes") + start; + text[start..end].to_string() +} + +fn shell_quote(word: &str) -> String { + format!("'{}'", word.replace('\'', r"'\''")) +} + +fn toml_string(word: &str) -> String { + format!("\"{}\"", word.replace('\\', r"\\").replace('"', "\\\"")) +} + +/// One scripted SSE completion per request; every request body is appended +/// to `record` so the test can read what the model was sent afterwards. +fn spawn_recording_server(bodies: Vec, record: PathBuf) -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + for sse in bodies { + let Ok((mut stream, _)) = listener.accept() else { return }; + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") { + match stream.read(&mut byte) { + Ok(1) => head.push(byte[0]), + _ => return, + } + } + let length: usize = String::from_utf8_lossy(&head) + .lines() + .find_map(|l| { + let (k, v) = l.split_once(':')?; + k.eq_ignore_ascii_case("content-length").then(|| v.trim().parse().ok())? + }) + .unwrap_or(0); + let mut body = vec![0u8; length]; + if stream.read_exact(&mut body).is_err() { + return; + } + let mut log = std::fs::OpenOptions::new().create(true).append(true).open(&record).unwrap(); + log.write_all(&body).unwrap(); + log.write_all(b"\n").unwrap(); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{sse}", + sse.len(), + ); + let _ = stream.write_all(response.as_bytes()); + } + }); + format!("http://{addr}/v1") +} + +fn sse(chunks: &[Value]) -> String { + let mut out = String::new(); + for chunk in chunks { + out.push_str(&format!("data: {chunk}\n\n")); + } + out.push_str("data: [DONE]\n\n"); + out +} + +// ------------------------------------------------------------------ tests + +/// The whole recipe, as written: the skill comes from the spec's own shell +/// lines and the proxy tool from its own TOML, with only the server command +/// swapped for the fixture. A model turn then calls a server tool through +/// that proxy, so the call crosses the harness, the tool's scrubbed +/// environment, `$OPENMAX_BIN --mcp-call`, and the server, and its result +/// comes back to the model. +fn proxy_tool_from_the_spec() { + let (project, home) = fresh_dirs("proxy"); + let spec = openmax(&project, &home).args(["--spec", "mcp"]).output().unwrap(); + assert!(spec.status.success(), "{}", String::from_utf8_lossy(&spec.stderr)); + let spec = String::from_utf8(spec.stdout).unwrap(); + + // The skill, generated by the spec's shell lines with `openmax` on PATH + // naming the binary under test. + let lines = fenced(&spec, "sh"); + let server = format!("{} {FIXTURE} basic", shell_quote(&fixture())); + let script = lines.replace("notes-server --stdio", &server); + assert_ne!(script, lines, "the recipe's shell lines start `notes-server --stdio`:\n{lines}"); + let bin = project.parent().unwrap().join("bin"); + std::fs::create_dir_all(&bin).unwrap(); + std::os::unix::fs::symlink(openmax_bin(), bin.join("openmax")).unwrap(); + let path = format!("{}:{}", bin.display(), std::env::var("PATH").unwrap_or_default()); + let generated = Command::new("sh") + .arg("-c") + .arg(&script) + .current_dir(&project) + .env("PATH", path) + .env("HOME", &home) + .env_remove("OPENMAX_SESSION") + .output() + .unwrap(); + assert!(generated.status.success(), "{}", String::from_utf8_lossy(&generated.stderr)); + let skill = std::fs::read_to_string(project.join(".agents/skills/notes/SKILL.md")).unwrap(); + assert!(skill.starts_with("---\nname: notes\ndescription: "), "{skill}"); + assert!( + skill.contains("\n- echo(text: string, repeat?: integer): Echo the text back with the server's view of its environment\n"), + "{skill}" + ); + assert!(skill.contains("\n- fail(): Always reports an error\n"), "{skill}"); + assert!(!skill.contains("fixture basic: serving"), "server stderr is not part of the list: {skill}"); + + // The proxy tool, byte for byte the spec's except for the server command. + let manifest = fenced(&spec, "toml"); + let words = [fixture(), FIXTURE.into(), "basic".into()].map(|w| toml_string(&w)).join(", "); + let rewritten = manifest.replace("\"notes-server\", \"--stdio\"", &words); + assert_ne!(rewritten, manifest, "the recipe's tool runs `notes-server --stdio`:\n{manifest}"); + std::fs::create_dir_all(project.join(".openmax/tools")).unwrap(); + std::fs::write(project.join(".openmax/tools/notes.toml"), rewritten).unwrap(); + + let check = openmax(&project, &home).arg("--check").output().unwrap(); + let report = String::from_utf8_lossy(&check.stdout); + assert!(check.status.success(), "{report}"); + assert!(report.lines().any(|l| l.starts_with("ok skill") && l.contains("notes")), "{report}"); + assert!(!report.lines().any(|l| l.starts_with("err")), "{report}"); + + let record = project.parent().unwrap().join("requests.jsonl"); + let call = json!({"tool": "echo", "arguments": {"text": "hello from the model"}}); + let base_url = spawn_recording_server( + vec![ + sse(&[ + json!({"choices": [{"delta": {"tool_calls": [{"index": 0, "id": "call_0", "type": "function", + "function": {"name": "notes", "arguments": call.to_string()}}]}, "finish_reason": null}]}), + json!({"choices": [{"delta": {}, "finish_reason": "tool_calls"}]}), + ]), + sse(&[ + json!({"choices": [{"delta": {"content": "done"}, "finish_reason": null}]}), + json!({"choices": [{"delta": {}, "finish_reason": "stop"}]}), + ]), + ], + record.clone(), + ); + std::fs::create_dir_all(home.join(".openmax")).unwrap(); + std::fs::write( + home.join(".openmax/settings.json"), + format!( + r#"{{"base_url":"{base_url}","model":"stub-model","approval_mode":"auto","context_tokens":16384}}"# + ), + ) + .unwrap(); + let mut turn = openmax(&project, &home); + turn.args(["--trust-project", "-p", "use the notes server"]) + .env("NOTES_TOKEN", "granted-token") + .env("NOT_GRANTED", "leaked"); + let run = finish(turn, ""); + assert_eq!(run.code, Some(0), "stdout: {}\nstderr: {}", run.stdout, run.stderr); + + let sent = std::fs::read_to_string(&record).unwrap(); + let second: Value = serde_json::from_str(sent.lines().nth(1).expect("a second request")).unwrap(); + let result = second["messages"] + .as_array() + .unwrap() + .iter() + .find(|m| m["role"] == "tool") + .expect("the tool result goes back to the model")["content"] + .as_str() + .unwrap() + .to_string(); + assert!( + result.contains("echo: hello from the model; NOTES_TOKEN=granted-token; NOT_GRANTED=unset"), + "the call reaches the server with exactly the granted credentials: {result}" + ); + assert!(result.contains("[image image/png, 2 bytes]"), "{result}"); + assert!(result.contains("fixture basic: serving"), "the server's stderr passes through: {result}"); + assert!(!result.contains("exit code"), "{result}"); + let _ = std::fs::remove_dir_all(project.parent().unwrap()); +} + +/// A result the tool marks `isError` is a failed call, and the model must be +/// able to read why; an error reply is the same. Input the bridge cannot +/// read is the caller's mistake, refused before any server starts. +fn is_error_result() { + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"fail","arguments":{}}"#); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("the fail tool always fails"), "{}", run.stderr); + assert!(run.stdout.is_empty(), "{}", run.stdout); + + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"nope","arguments":{}}"#); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("unknown tool: nope"), "{}", run.stderr); + + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"echo","args":{"text":"x"}}"#); + assert_eq!(run.code, Some(2), "{}", run.stderr); + assert!(run.stderr.contains("'args'"), "the unknown key is named: {}", run.stderr); + assert!(!run.stderr.contains("serving"), "no server starts for unreadable input: {}", run.stderr); + + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"echo","arguments":{"text":"plain"}}"#); + assert_eq!(run.code, Some(0), "{}", run.stderr); + assert!(run.stdout.starts_with("echo: plain; "), "{}", run.stdout); +} + +/// A reply that is not JSON-RPC, and a protocol version the bridge does not +/// speak, both end the run with the reason instead of a guess. +fn malformed_reply() { + let run = bridge(&["--mcp-list"], &["malformed"], ""); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("this line is not JSON-RPC"), "the bad line is quoted: {}", run.stderr); + assert!(run.stdout.is_empty(), "{}", run.stdout); + + let run = bridge(&["--mcp-list"], &["unsupported"], ""); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!( + run.stderr.contains("2099-01-01") && run.stderr.contains("2025-11-25"), + "both sides' versions are named: {}", + run.stderr + ); +} + +/// A server that never answers costs one bounded wait, and is stopped even +/// though it ignores its closed stdin. +fn silent_server() { + let (dir, _) = fresh_dirs("silent"); + let pidfile = dir.join("pid"); + let run = bridge(&["--mcp-timeout", "1", "--mcp-list"], &["silent", pidfile.to_str().unwrap()], ""); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("did not answer initialize within 1s"), "{}", run.stderr); + assert!(run.elapsed < Duration::from_secs(15), "{:?}", run.elapsed); + let pid: i32 = std::fs::read_to_string(&pidfile).unwrap().trim().parse().unwrap(); + // SAFETY: signal 0 only asks whether the process exists. + assert_ne!(unsafe { libc::kill(pid, 0) }, 0, "the silent server outlived the run"); + let _ = std::fs::remove_dir_all(dir.parent().unwrap()); +} + +/// tools/list is paginated: every page is fetched with the cursor the last +/// one returned, and the list and JSON forms both carry every tool, once. +fn paginated_list() { + let run = bridge(&["--mcp-list"], &["paged"], ""); + assert_eq!(run.code, Some(0), "{}", run.stderr); + let lines: Vec<&str> = run.stdout.lines().filter(|l| l.starts_with("- ")).collect(); + assert_eq!( + lines, + ["- alpha(): The alpha tool", "- beta(): The beta tool", "- gamma(): The gamma tool", "- delta(): The delta tool"], + "{}", + run.stdout + ); + + let run = bridge(&["--mcp-list", "--json"], &["paged"], ""); + assert_eq!(run.code, Some(0), "{}", run.stderr); + let tools: Value = serde_json::from_str(&run.stdout).unwrap(); + let names: Vec<&str> = tools.as_array().unwrap().iter().map(|t| t["name"].as_str().unwrap()).collect(); + assert_eq!(names, ["alpha", "beta", "gamma", "delta"]); +} + +/// The fixture sends a notification and a ping in the middle of a call and +/// refuses to answer the call until the ping is answered: a client that +/// treated either as the call's reply, or left the ping unanswered, fails. +fn server_traffic_mid_call() { + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"echo","arguments":{"text":"x"}}"#); + assert_eq!(run.code, Some(0), "{}", run.stderr); + assert_eq!( + run.stdout, + "echo: x; NOTES_TOKEN=unset; NOT_GRANTED=unset\n[image image/png, 2 bytes]\n", + "{}", + run.stderr + ); + assert!(run.stderr.contains("fixture basic: serving"), "{}", run.stderr); +} diff --git a/docs/extending.md b/docs/extending.md index 65951388..6a77a2c0 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -52,6 +52,32 @@ description: How to cut a release of this project Full instructions, checklists, commands... ``` +## MCP servers + +There is no MCP client in the agent loop. A server is adopted through the two +surfaces above: one proxy tool per server with a `{tool, arguments}` schema, +and one skill whose body lists the server's tools. Every request pays only +for that schema and the skill's index line; the server's own tool schemas are +read when the skill is. The `openmax` binary is the bridge between them, so a +host needs nothing installed besides the server. + +```sh +openmax --mcp-list -- npx -y @modelcontextprotocol/server-filesystem ~/notes +echo '{"tool":"list_allowed_directories","arguments":{}}' | + openmax --mcp-call -- npx -y @modelcontextprotocol/server-filesystem ~/notes +``` + +`--mcp-list` prints the tools as a skill body, one line each with a compact +argument summary (`--json` prints the full definitions). `--mcp-call` reads +the call on stdin and prints the result's text; a tool error or a protocol +error exits 1 with the reason. Each run starts the server, performs the +initialize handshake, makes one request, and stops it, with every wait +bounded by `--mcp-timeout` (default 30 seconds). A server that needs a +persistent session is not supported yet. `openmax --spec mcp` prints the +recipe the agent follows: the proxy tool's TOML, with the server's +credentials granted through `env`, and the shell lines that generate the +skill. + ## Memory One durable fact per markdown file under `.openmax/memory/`, written with the @@ -432,12 +458,12 @@ loop never loads it and so never fails closed on it. ## Self-description `openmax --spec ` prints the complete authoring contract for one -surface (`tools`, `skills`, `prompts`, `hooks`, `permissions`, `providers`, or -`stdio`): file grammar, field caps and defaults, hook stdin payload shapes, and -activation timing. The frozen prompt carries only a one-line pointer to it, so -the full contract costs zero tokens until the agent reads it, and the printed -examples are parsed by the same validation code in tests so the text cannot -drift from the binary. +surface (`tools`, `skills`, `prompts`, `hooks`, `permissions`, `providers`, +`stdio`, or `mcp`): file grammar, field caps and defaults, hook stdin payload +shapes, and activation timing. The frozen prompt carries only a one-line +pointer to it, so the full contract costs zero tokens until the agent reads +it, and the printed examples are parsed by the same validation code in tests +so the text cannot drift from the binary. ## Freezing and re-freezing diff --git a/docs/usage.md b/docs/usage.md index 2ca2ef68..a3f95d65 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -23,8 +23,13 @@ openmax --check # validate extension files and the session openmax --spec hooks # print an extension surface's contract openmax --recall "deploy port" # search past sessions and memories openmax --stdio # full session over JSONL pipes +openmax --mcp-list -- # list an MCP server's tools ``` +`openmax --mcp-list` and `openmax --mcp-call` are a one-shot MCP stdio +client, used by the proxy tool an MCP server is adopted through; see +[extending](extending.md#mcp-servers) and `openmax --spec mcp`. + `openmax --check --json` prints the same findings as one JSON array of `{surface, path, status, message}` objects (status `ok`, `warn`, or `err`), with the same exit code, so the agent can parse its own verification. From 20aadebd593de2dae87928da05302e1253e60c2d Mon Sep 17 00:00:00 2001 From: Max Loffgren Date: Mon, 5 Oct 2026 21:58:35 -0500 Subject: [PATCH 2/4] fix(tui): keep a server that stops reading from stalling the MCP bridge openmax --mcp-call wrote each request to the server's stdin with a blocking write from the thread that waits for replies. A server that answered initialize and then stopped reading held any request larger than the pipe buffer before --mcp-timeout started counting, so the proxy tool hung until the server exited on its own. Writes now go through a writer thread fed by a queue, so the timeout bounds every wait. Stopping the server closes the queue, and the SIGTERM that follows frees a writer still blocked on a full pipe. The stop sequence moves into Drop, removing a guard that could never fire. The bridge test target's own main read the value of libtest options such as --test-threads 1, and the pattern after --skip, as name filters, so those command lines ran none of its tests, or only the skipped one, and still reported ok. It now parses them as libtest does. --spec mcp also names the quoted server instructions that --mcp-list prints. --- crates/core/src/spec.rs | 3 +- crates/tui/src/mcp.rs | 77 +++++++++++--------- crates/tui/tests/mcp_bridge.rs | 127 +++++++++++++++++++++++++++------ 3 files changed, 152 insertions(+), 55 deletions(-) diff --git a/crates/core/src/spec.rs b/crates/core/src/spec.rs index f1e08793..cf48e835 100644 --- a/crates/core/src/spec.rs +++ b/crates/core/src/spec.rs @@ -1007,7 +1007,8 @@ between calls, a login done once per process) is not supported yet. ## Commands - `openmax --mcp-list -- ` prints the server's tools as a - skill body: a header, then one line per tool, + skill body: a header, the server's instructions (when it gives any) as a + quoted block, then one line per tool, `- name(arg: type, opt?: type): description`, required arguments first and `?` marking an optional one. With `--json` it prints the full tool definitions instead. diff --git a/crates/tui/src/mcp.rs b/crates/tui/src/mcp.rs index 9f4f98c2..2f5fb7e2 100644 --- a/crates/tui/src/mcp.rs +++ b/crates/tui/src/mcp.rs @@ -145,15 +145,19 @@ fn kind(value: &Value) -> &'static str { } } -/// One running server: its stdin for requests, and a reader thread turning -/// its stdout into lines, so every wait can be bounded by a timeout. +/// One running server, with a writer thread feeding its stdin and a reader +/// thread turning its stdout into lines. The thread that waits touches +/// neither pipe, so each wait is bounded by the timeout whatever the server +/// does: one that stops reading cannot hold a request larger than the pipe +/// buffer forever. struct Session { child: Child, - stdin: Option, + /// The writer thread's queue; dropping it closes the server's stdin once + /// the queue drains. + writes: Option>>, lines: mpsc::Receiver, String>>, timeout: Duration, next_id: u64, - stopped: bool, } impl Session { @@ -175,12 +179,15 @@ impl Session { ) })?; let stdout = child.stdout.take().expect("stdout is piped"); - let stdin = child.stdin.take(); + let stdin = child.stdin.take().expect("stdin is piped"); let (tx, lines) = mpsc::channel(); - // Detached: a server can leave a descendant holding its stdout open, - // and the run must not wait on that to end. + let (writes, rx) = mpsc::channel(); + // Both detached: a server can leave a descendant holding its stdout + // open, or stop reading its stdin, and the run must not wait on + // either to end. std::thread::spawn(move || read_lines(stdout, tx)); - Ok(Self { child, stdin, lines, timeout, next_id: 1, stopped: false }) + std::thread::spawn(move || write_lines(stdin, rx)); + Ok(Self { child, writes: Some(writes), lines, timeout, next_id: 1 }) } /// The handshake: ask for the newest revision this client speaks, accept @@ -332,16 +339,15 @@ impl Session { } } + /// Queue one message for the writer thread. It fails only once the + /// writer has stopped, which a failed write to the server causes. fn send(&mut self, message: &Value, method: &str) -> Result<(), String> { - let Some(stdin) = self.stdin.as_mut() else { - return Err(format!("the server's stdin is closed before {method}")); - }; let mut line = message.to_string(); line.push('\n'); - if stdin.write_all(line.as_bytes()).and_then(|()| stdin.flush()).is_err() { - return Err(self.closed(method)); + match &self.writes { + Some(writes) if writes.send(line.into_bytes()).is_ok() => Ok(()), + _ => Err(self.closed(method)), } - Ok(()) } fn silent(&self, method: &str) -> String { @@ -368,13 +374,25 @@ impl Session { } } + fn exited_within(&mut self, grace: Duration) -> bool { + let deadline = Instant::now() + grace; + loop { + match self.child.try_wait() { + Ok(Some(_)) => return true, + Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(10)), + _ => return false, + } + } + } +} + +impl Drop for Session { /// Stop the server the way the stdio transport defines: close its stdin, /// then SIGTERM, then SIGKILL, each after a grace period, and reap it. - fn stop(&mut self) { - if std::mem::replace(&mut self.stopped, true) { - return; - } - drop(self.stdin.take()); + /// Closing the queue closes stdin once the writer drains it; a writer + /// stuck on a server that stopped reading is freed when the server dies. + fn drop(&mut self) { + drop(self.writes.take()); if self.exited_within(STOP_GRACE) { return; } @@ -392,22 +410,15 @@ impl Session { let _ = self.child.kill(); let _ = self.child.wait(); } - - fn exited_within(&mut self, grace: Duration) -> bool { - let deadline = Instant::now() + grace; - loop { - match self.child.try_wait() { - Ok(Some(_)) => return true, - Ok(None) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(10)), - _ => return false, - } - } - } } -impl Drop for Session { - fn drop(&mut self) { - self.stop(); +/// Write each queued message to the server's stdin, ending when the queue +/// closes (which closes stdin) or a write fails. +fn write_lines(mut stdin: ChildStdin, rx: mpsc::Receiver>) { + for line in rx { + if stdin.write_all(&line).and_then(|()| stdin.flush()).is_err() { + return; + } } } diff --git a/crates/tui/tests/mcp_bridge.rs b/crates/tui/tests/mcp_bridge.rs index 4f5bbbeb..5dfbe440 100644 --- a/crates/tui/tests/mcp_bridge.rs +++ b/crates/tui/tests/mcp_bridge.rs @@ -18,13 +18,15 @@ use serde_json::{json, Value}; const FIXTURE: &str = "--mcp-fixture"; -const TESTS: [(&str, fn()); 6] = [ +const TESTS: [(&str, fn()); 8] = [ ("a_proxy_tool_written_from_the_spec_lists_and_calls_through_openmax", proxy_tool_from_the_spec), ("an_is_error_result_exits_nonzero_with_its_text", is_error_result), ("a_malformed_reply_is_reported_and_exits_nonzero", malformed_reply), ("a_server_that_never_answers_times_out_and_is_stopped", silent_server), + ("a_server_that_stops_reading_cannot_stall_a_large_call", stalled_server), ("every_page_of_tools_is_listed", paginated_list), ("a_server_request_is_answered_and_a_notification_ignored", server_traffic_mid_call), + ("this_harness_reads_libtest_options_as_libtest_does", harness_options), ]; fn main() { @@ -33,26 +35,13 @@ fn main() { serve(&args[1..]); return; } - // Enough of libtest's command line for `cargo test `, `--exact`, - // `--list`, and `--ignored` (this target has no ignored tests) to mean - // what they mean everywhere else. + let selected = select(&args); if args.iter().any(|a| a == "--list") { - for (name, _) in TESTS { + for (name, _) in &selected { println!("{name}: test"); } return; } - let exact = args.iter().any(|a| a == "--exact"); - let filters: Vec<&str> = - args.iter().filter(|a| !a.starts_with('-')).map(String::as_str).collect(); - let selected: Vec<(&str, fn())> = TESTS - .into_iter() - .filter(|_| !args.iter().any(|a| a == "--ignored")) - .filter(|(name, _)| { - filters.is_empty() - || filters.iter().any(|f| if exact { name == f } else { name.contains(f) }) - }) - .collect(); println!("\nrunning {} tests", selected.len()); let mut failed = Vec::new(); for (name, test) in &selected { @@ -73,21 +62,55 @@ fn main() { } } +/// The tests a libtest command line selects, so `cargo test `, +/// `--exact`, `--skip`, `--list`, and `--ignored` (this target has no ignored +/// tests) mean what they mean everywhere else. The value after a libtest +/// option that takes one is that option's, never a name filter. +fn select(args: &[String]) -> Vec<(&'static str, fn())> { + let (mut filters, mut skips) = (Vec::new(), Vec::new()); + let (mut exact, mut ignored) = (false, false); + let mut words = args.iter().map(String::as_str); + while let Some(word) = words.next() { + match word { + "--exact" => exact = true, + "--ignored" => ignored = true, + "--skip" => skips.extend(words.next()), + "--test-threads" | "--color" | "--format" | "--logfile" | "--shuffle-seed" | "-Z" => { + words.next(); + } + _ if word.starts_with("--skip=") => skips.push(&word["--skip=".len()..]), + _ if word.starts_with('-') => {} + filter => filters.push(filter), + } + } + let matches = |name: &str, filter: &str| if exact { name == filter } else { name.contains(filter) }; + TESTS + .into_iter() + .filter(|_| !ignored) + .filter(|(name, _)| filters.is_empty() || filters.iter().any(|f| matches(name, f))) + .filter(|(name, _)| !skips.iter().any(|f| matches(name, f))) + .collect() +} + // ---------------------------------------------------------------- fixture /// One fixture MCP server. Modes: `basic` (two tools, `echo` and `fail`), /// `paged` (four tools over three tools/list pages), `malformed` (answers /// initialize with a line that is not JSON-RPC), `unsupported` (answers with -/// a protocol version no client knows), and `silent ` (never reads -/// or writes, and ignores its closed stdin). +/// a protocol version no client knows), `silent ` (never reads or +/// writes, and ignores its closed stdin), and `stalled ` (answers +/// initialize, then never reads again). fn serve(args: &[String]) { let mode = args.first().map(String::as_str).unwrap_or("basic"); // A line on stderr, which the bridge must pass through to its caller. eprintln!("fixture {mode}: serving"); - if mode == "silent" { + let write_pid = || { if let Some(pidfile) = args.get(1) { std::fs::write(pidfile, std::process::id().to_string()).unwrap(); } + }; + if mode == "silent" { + write_pid(); std::thread::sleep(Duration::from_secs(120)); return; } @@ -113,11 +136,22 @@ fn serve(args: &[String]) { out.flush().unwrap(); continue; } - result(json!({ + let reply = result(json!({ "protocolVersion": version, "capabilities": {"tools": {}}, "serverInfo": {"name": "fixture", "version": "1.0.0"}, - })) + })); + if mode == "stalled" { + // Alive, but its stdin fills up and stays full. The 30 + // seconds stay under `finish`'s bound, so a bridge that + // waits for this exit fails its assertions instead of + // hanging the suite. + send(&mut out, &reply); + write_pid(); + std::thread::sleep(Duration::from_secs(30)); + return; + } + reply } "notifications/initialized" => { initialized = true; @@ -520,6 +554,23 @@ fn silent_server() { let _ = std::fs::remove_dir_all(dir.parent().unwrap()); } +/// A server that stops reading its stdin cannot hold the bridge on a write: +/// a call larger than any pipe buffer still ends at the timeout, and the +/// server is stopped. +fn stalled_server() { + let (dir, _) = fresh_dirs("stalled"); + let pidfile = dir.join("pid"); + let call = json!({"tool": "echo", "arguments": {"text": "x".repeat(1 << 20)}}).to_string(); + let run = bridge(&["--mcp-timeout", "1", "--mcp-call"], &["stalled", pidfile.to_str().unwrap()], &call); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("did not answer tools/call within 1s"), "{}", run.stderr); + assert!(run.elapsed < Duration::from_secs(15), "{:?}", run.elapsed); + let pid: i32 = std::fs::read_to_string(&pidfile).unwrap().trim().parse().unwrap(); + // SAFETY: signal 0 only asks whether the process exists. + assert_ne!(unsafe { libc::kill(pid, 0) }, 0, "the stalled server outlived the run"); + let _ = std::fs::remove_dir_all(dir.parent().unwrap()); +} + /// tools/list is paginated: every page is fetched with the cursor the last /// one returned, and the list and JSON forms both carry every tool, once. fn paginated_list() { @@ -554,3 +605,37 @@ fn server_traffic_mid_call() { ); assert!(run.stderr.contains("fixture basic: serving"), "{}", run.stderr); } + +/// `cargo test -- ` reaches this target too. An option's +/// value read as a name filter, or `--skip` read as a selection, would drop +/// this whole suite while the run still reports ok. +fn harness_options() { + let names = |args: &[&str]| -> Vec<&str> { + let args: Vec = args.iter().map(|a| a.to_string()).collect(); + select(&args).into_iter().map(|(name, _)| name).collect() + }; + let all: Vec<&str> = TESTS.iter().map(|(name, _)| *name).collect(); + let silent = "a_server_that_never_answers_times_out_and_is_stopped"; + let others: Vec<&str> = all.iter().copied().filter(|name| *name != silent).collect(); + assert_eq!(names(&[]), all); + for values in [ + &["--test-threads", "1"][..], + &["--color", "always"], + &["--format", "pretty"], + &["--logfile", "out.txt"], + &["--shuffle-seed", "7"], + &["-Z", "unstable-options"], + &["--test-threads=1", "--nocapture", "-q"], + ] { + assert_eq!(names(values), all, "{values:?}"); + } + assert_eq!(names(&["--skip", "never_answers"]), others); + assert_eq!(names(&["--skip=never_answers"]), others); + assert_eq!(names(&["--skip", "never_answers", "--exact"]), all, "--exact applies to --skip"); + assert_eq!(names(&["--exact", "--skip", silent]), others); + assert_eq!(names(&["never_answers", "--test-threads", "1"]), [silent]); + assert_eq!(names(&["never_answers", "--exact"]), Vec::<&str>::new()); + assert_eq!(names(&[silent, "--exact"]), [silent]); + assert_eq!(names(&["--ignored"]), Vec::<&str>::new()); + assert_eq!(names(&["--include-ignored"]), all); +} From ed3f4885988abc947cc59925bb1b836be9cb815e Mon Sep 17 00:00:00 2001 From: Max Loffgren Date: Mon, 5 Oct 2026 22:11:41 -0500 Subject: [PATCH 3/4] docs: say that the MCP tool listing reads every page The extending guide said each bridge run makes one request, but `--mcp-list` against a paginating server sends one tools/list request per page and follows nextCursor until the last one. The sentence now names what a run sends: every page of tools/list, or one tools/call. --- docs/extending.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/extending.md b/docs/extending.md index 6a77a2c0..c8c3acf2 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -71,12 +71,12 @@ echo '{"tool":"list_allowed_directories","arguments":{}}' | argument summary (`--json` prints the full definitions). `--mcp-call` reads the call on stdin and prints the result's text; a tool error or a protocol error exits 1 with the reason. Each run starts the server, performs the -initialize handshake, makes one request, and stops it, with every wait -bounded by `--mcp-timeout` (default 30 seconds). A server that needs a -persistent session is not supported yet. `openmax --spec mcp` prints the -recipe the agent follows: the proxy tool's TOML, with the server's -credentials granted through `env`, and the shell lines that generate the -skill. +initialize handshake, makes one request (every page of tools/list, or one +tools/call), and stops it, with every wait bounded by `--mcp-timeout` +(default 30 seconds). A server that needs a persistent session is not +supported yet. `openmax --spec mcp` prints the recipe the agent follows: the +proxy tool's TOML, with the server's credentials granted through `env`, and +the shell lines that generate the skill. ## Memory From 456c037758699a1f6bca710aa5e6380f79e5d74a Mon Sep 17 00:00:00 2001 From: Max Loffgren Date: Mon, 5 Oct 2026 22:32:08 -0500 Subject: [PATCH 4/4] fix(tui): hold back a flooding MCP server and refuse cut calls and malformed results The bridge queued every line a server wrote in an unbounded channel, so a server looping on notifications could grow the bridge without limit while a call waited or while the server was being stopped. The reader now hands over one line at a time through a zero-capacity channel, and a flooding server blocks on its own stdout instead. --mcp-call read at most 64 MiB of stdin and parsed what it got, so a call cut at the bound could still parse and run without its tail. It now reads one byte past the bound and refuses larger input with exit 2 before any server starts. A tools/call result whose content was not an array printed nothing and exited 0, reporting a malformed reply as an empty, completed call; it is now reported as malformed and exits 1. --- crates/tui/src/mcp.rs | 114 ++++++++++++++++++++++++++++----- crates/tui/tests/mcp_bridge.rs | 22 ++++++- 2 files changed, 119 insertions(+), 17 deletions(-) diff --git a/crates/tui/src/mcp.rs b/crates/tui/src/mcp.rs index 2f5fb7e2..de457b9d 100644 --- a/crates/tui/src/mcp.rs +++ b/crates/tui/src/mcp.rs @@ -69,11 +69,24 @@ pub fn call(server: &[OsString], timeout: Duration) -> i32 { eprintln!(r#"openmax: --mcp-call reads the call on stdin, as {{"tool": "", "arguments": {{...}}}}"#); return 2; } - let mut input = String::new(); - if let Err(e) = stdin.lock().take(MAX_MESSAGE_BYTES).read_to_string(&mut input) { + // One byte past the bound tells a call that was cut from one that fits: + // a cut call can still parse, and would run without its tail. + let mut input = Vec::new(); + if let Err(e) = stdin.lock().take(MAX_MESSAGE_BYTES + 1).read_to_end(&mut input) { eprintln!("openmax: --mcp-call could not read stdin: {e}"); return 2; } + if input.len() as u64 > MAX_MESSAGE_BYTES { + eprintln!("openmax: --mcp-call reads at most {} MiB on stdin", MAX_MESSAGE_BYTES >> 20); + return 2; + } + let input = match String::from_utf8(input) { + Ok(input) => input, + Err(e) => { + eprintln!("openmax: --mcp-call could not read stdin: {e}"); + return 2; + } + }; let (tool, arguments) = match parse_call(&input) { Ok(call) => call, Err(reason) => { @@ -88,7 +101,13 @@ pub fn call(server: &[OsString], timeout: Duration) -> i32 { }; match run() { Ok(result) => { - let text = render_content(&result); + let text = match render_content(&result) { + Ok(text) => text, + Err(reason) => { + eprintln!("openmax: {reason}"); + return 1; + } + }; if result.get("isError").and_then(Value::as_bool) == Some(true) { eprintln!( "openmax: the server's {} tool reported an error: {text}", @@ -180,12 +199,11 @@ impl Session { })?; let stdout = child.stdout.take().expect("stdout is piped"); let stdin = child.stdin.take().expect("stdin is piped"); - let (tx, lines) = mpsc::channel(); let (writes, rx) = mpsc::channel(); // Both detached: a server can leave a descendant holding its stdout // open, or stop reading its stdin, and the run must not wait on // either to end. - std::thread::spawn(move || read_lines(stdout, tx)); + let lines = spawn_reader(stdout); std::thread::spawn(move || write_lines(stdin, rx)); Ok(Self { child, writes: Some(writes), lines, timeout, next_id: 1 }) } @@ -422,9 +440,21 @@ fn write_lines(mut stdin: ChildStdin, rx: mpsc::Receiver>) { } } +/// Start the thread that reads the server's stdout, and return its lines. +/// The queue holds nothing: the reader hands over one line and reads the +/// next only once that one is taken. Each line is capped but their number is +/// not, so a queue that buffered ahead would let a server that floods its +/// output (a notification loop) grow this process while the session is busy +/// or stopping it; held back, the server blocks on its own stdout instead. +fn spawn_reader(stdout: impl Read + Send + 'static) -> mpsc::Receiver, String>> { + let (tx, lines) = mpsc::sync_channel(0); + std::thread::spawn(move || read_lines(stdout, tx)); + lines +} + /// Forward the server's stdout one line at a time, ending at EOF, at a read /// error, or at a line longer than any message may be. -fn read_lines(stdout: impl Read, tx: mpsc::Sender, String>>) { +fn read_lines(stdout: impl Read, tx: mpsc::SyncSender, String>>) { let mut reader = BufReader::new(stdout); loop { let mut line = Vec::new(); @@ -593,16 +623,26 @@ fn truncated(text: &str, max_chars: usize) -> String { /// A tools/call result as text: each content item on its own line, text as /// written, and binary items described rather than dumped. A result with no -/// content falls back to its structured content. -fn render_content(result: &Value) -> String { - let items = result.get("content").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default(); +/// content falls back to its structured content; beside content it is not +/// repeated, because the protocol asks a tool that returns structured content +/// to return it as text too, and printing both would double such a result. +/// Content that is not a list is a malformed reply, not an empty result. +fn render_content(result: &Value) -> Result { + let items = match result.get("content") { + None => &[][..], + Some(Value::Array(items)) => items.as_slice(), + Some(other) => { + let what = format!("content is {}, not an array", kind(other)); + return Err(malformed("tools/call", &other.to_string(), &what)); + } + }; if items.is_empty() { - return match result.get("structuredContent") { + return Ok(match result.get("structuredContent") { Some(structured) if !structured.is_null() => structured.to_string(), _ => String::new(), - }; + }); } - items + Ok(items .iter() .map(|item| { let field = |key: &str| item.get(key).and_then(Value::as_str).unwrap_or_default(); @@ -629,7 +669,7 @@ fn render_content(result: &Value) -> String { } }) .collect::>() - .join("\n") + .join("\n")) } /// The decoded size of base64 text, without decoding it. @@ -730,6 +770,48 @@ mod tests { } } + /// Each line is capped, but a server can write any number of them: a + /// notification loop must stall the server on its own stdout, not queue + /// up in this process while the session is busy or stopping it. + #[test] + fn the_reader_reads_no_further_than_one_line_ahead() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// A server stdout that yields one notification per read, counting. + struct Flood { + left: usize, + read: Arc, + } + impl Read for Flood { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + const LINE: &[u8] = b"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\"}\n"; + if self.left == 0 { + return Ok(0); + } + buf[..LINE.len()].copy_from_slice(LINE); + self.left -= 1; + self.read.fetch_add(1, Ordering::SeqCst); + Ok(LINE.len()) + } + } + + const TOTAL: usize = 10_000; + let read = Arc::new(AtomicUsize::new(0)); + let lines = spawn_reader(Flood { left: TOTAL, read: read.clone() }); + std::thread::sleep(Duration::from_millis(200)); + let ahead = read.load(Ordering::SeqCst); + assert!(ahead <= 1, "the reader ran {ahead} lines ahead of a session that took none"); + for taken in 1..=3 { + lines.recv().unwrap().unwrap(); + std::thread::sleep(Duration::from_millis(20)); + let ahead = read.load(Ordering::SeqCst); + assert!(ahead <= taken + 1, "the reader ran {ahead} lines ahead of a session that took {taken}"); + } + // Holding back loses nothing: every line still arrives. + assert_eq!(lines.iter().filter(Result::is_ok).count() + 3, TOTAL); + } + #[test] fn content_items_render_as_text() { let result = json!({"content": [ @@ -739,10 +821,10 @@ mod tests { {"type": "resource_link", "uri": "file:///b", "name": "b"}, ]}); assert_eq!( - render_content(&result), + render_content(&result).unwrap(), "one\n[image image/png, 5 bytes]\n[resource file:///a.txt]\nbody\n[resource link file:///b b]" ); - assert_eq!(render_content(&json!({"content": [], "structuredContent": {"n": 1}})), r#"{"n":1}"#); - assert_eq!(render_content(&json!({})), ""); + assert_eq!(render_content(&json!({"content": [], "structuredContent": {"n": 1}})).unwrap(), r#"{"n":1}"#); + assert_eq!(render_content(&json!({})).unwrap(), ""); } } diff --git a/crates/tui/tests/mcp_bridge.rs b/crates/tui/tests/mcp_bridge.rs index 5dfbe440..dcac39f9 100644 --- a/crates/tui/tests/mcp_bridge.rs +++ b/crates/tui/tests/mcp_bridge.rs @@ -96,7 +96,8 @@ fn select(args: &[String]) -> Vec<(&'static str, fn())> { /// One fixture MCP server. Modes: `basic` (two tools, `echo` and `fail`), /// `paged` (four tools over three tools/list pages), `malformed` (answers -/// initialize with a line that is not JSON-RPC), `unsupported` (answers with +/// initialize with a line that is not JSON-RPC), `numeric-content` (answers +/// every tools/call with a number for content), `unsupported` (answers with /// a protocol version no client knows), `silent ` (never reads or /// writes, and ignores its closed stdin), and `stalled ` (answers /// initialize, then never reads again). @@ -192,6 +193,7 @@ fn serve(args: &[String]) { let arguments = &msg["params"]["arguments"]; let env = |name: &str| std::env::var(name).unwrap_or_else(|_| "unset".into()); match msg["params"]["name"].as_str().unwrap_or_default() { + _ if mode == "numeric-content" => result(json!({"content": 123})), "echo" => result(json!({"content": [ {"type": "text", "text": format!( "echo: {}; NOTES_TOKEN={}; NOT_GRANTED={}", @@ -517,6 +519,17 @@ fn is_error_result() { assert!(run.stderr.contains("'args'"), "the unknown key is named: {}", run.stderr); assert!(!run.stderr.contains("serving"), "no server starts for unreadable input: {}", run.stderr); + // Input past the 64 MiB bound: the bound falls just after a whole call + // padded with whitespace, so the part read parses. The byte after it is + // part of the input too, and it makes the input invalid JSON, so nothing + // may run. + let call = r#"{"tool":"echo","arguments":{"text":"cut"}}"#; + let input = format!("{call}{}}}", " ".repeat((64 << 20) - call.len())); + let run = bridge(&["--mcp-call"], &["basic"], &input); + assert_eq!(run.code, Some(2), "stdout: {}\nstderr: {}", run.stdout, run.stderr); + assert!(run.stderr.contains("64 MiB"), "the bound is named: {}", run.stderr); + assert!(!run.stderr.contains("serving"), "no server starts for oversized input: {}", run.stderr); + let run = bridge(&["--mcp-call"], &["basic"], r#"{"tool":"echo","arguments":{"text":"plain"}}"#); assert_eq!(run.code, Some(0), "{}", run.stderr); assert!(run.stdout.starts_with("echo: plain; "), "{}", run.stdout); @@ -537,6 +550,13 @@ fn malformed_reply() { "both sides' versions are named: {}", run.stderr ); + + // A result's content is a list of items; anything else is not an empty + // result to report as a completed call. + let run = bridge(&["--mcp-call"], &["numeric-content"], r#"{"tool":"echo","arguments":{"text":"x"}}"#); + assert_eq!(run.code, Some(1), "{}", run.stderr); + assert!(run.stderr.contains("content is a number, not an array: 123"), "{}", run.stderr); + assert!(run.stdout.is_empty(), "{}", run.stdout); } /// A server that never answers costs one bounded wait, and is stopped even