//! The `idea mcp-server` **bridge** — a transparent stdio↔loopback tube (M5b). //! //! ## Where this sits in the bind (cadrage v5 §1.3, §2) //! //! An MCP CLI (Claude Code, Codex) reads a `{command,args}` declaration and //! *spawns* ` mcp-server --endpoint <…> --project <…> --requester <…>`. //! That spawned process is **this bridge**: it must **bypass Tauri/WebKit** and run //! headless. It speaks JSON-RPC (JSON Lines) to the CLI on **stdin/stdout**, and //! relays every line, byte-for-byte, over the project's **loopback** (the Unix //! socket / Windows named pipe bound by [`crate::state::AppState::ensure_mcp_server`] //! at [`crate::mcp_endpoint::mcp_endpoint`]). The real `McpServer` (which holds the //! `OrchestratorService`) lives in the Tauri process and answers in M5c. //! //! **Zero business logic.** The bridge knows only *stdio + loopback + JSON lines*: //! no `OrchestratorService`, no use case (hexagonal boundary, cadrage §5). It never //! parses the JSON-RPC payloads it carries. //! //! ## Handshake format (consumed by M5c) //! //! Right after connecting to the loopback, **before** any JSON-RPC traffic, the //! bridge writes a **single newline-terminated JSON line** carrying the caller's //! identity: //! //! ```text //! {"project":"","requester":""}\n //! ``` //! //! - It is a **distinct line** from the JSON-RPC stream that follows: the server //! (M5c) reads exactly one line off a fresh connection, parses it as this //! handshake, then hands the rest of the stream to `McpServer::serve`. //! - `requester` is the **real agent id** (cadrage §1.4) — it lets the server tag //! `OrchestratorRequestProcessed.requester_id` with the actual agent instead of //! the frozen `"mcp"` placeholder. When `--requester` is omitted the field is the //! empty string. //! - Both values are JSON strings, so any future id shape stays escape-safe. //! //! ## Failure posture (never hang) //! //! - **Endpoint absent / unreachable** ⇒ connect with a **bounded timeout**; on //! timeout or connect error, return a **non-zero** exit code immediately. Never //! block forever waiting for a listener that will never appear. //! - **stdin EOF** (the CLI closed) ⇒ **clean exit, code 0**; the loopback //! connection is dropped (closed) on the way out. //! - **Missing required args** ⇒ a clear error on stderr and a non-zero code. use std::process::ExitCode; use std::time::Duration; use interprocess::local_socket::tokio::Stream as LocalSocketStream; use interprocess::local_socket::traits::tokio::Stream as _; use interprocess::local_socket::{GenericFilePath, ToFsName as _}; use tokio::io::{ AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, }; /// How long the bridge waits to connect to the project loopback before giving up. /// Bounded so an absent/unreachable endpoint fails fast instead of hanging. const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); /// Parsed `mcp-server` invocation arguments. /// /// `--endpoint` is **required** (without it the bridge has nowhere to relay). /// `--project` and `--requester` are optional identity hints forwarded verbatim in /// the handshake; a missing one becomes the empty string. #[derive(Debug, Clone, PartialEq, Eq)] pub struct BridgeArgs { /// The loopback address to connect to (the `--endpoint` value, i.e. /// [`crate::mcp_endpoint::McpEndpoint::as_cli_arg`]). pub endpoint: String, /// The project id (`--project`), forwarded in the handshake. Empty if absent. pub project: String, /// The requesting agent id (`--requester`), forwarded in the handshake. Empty /// if absent. pub requester: String, } impl BridgeArgs { /// Parses `mcp-server`'s own arguments (i.e. `argv` **after** the `mcp-server` /// subcommand token). Recognises `--endpoint`, `--project`, `--requester`, each /// taking the following token as its value. /// /// # Errors /// Returns a human-readable message when `--endpoint` is missing, when a flag is /// given without a value, or when an unknown flag is encountered. pub fn parse(args: I) -> Result where I: IntoIterator, S: Into, { let mut endpoint: Option = None; let mut project = String::new(); let mut requester = String::new(); let mut it = args.into_iter().map(Into::into); while let Some(flag) = it.next() { match flag.as_str() { "--endpoint" => { endpoint = Some( it.next() .ok_or_else(|| "--endpoint requires a value".to_string())?, ); } "--project" => { project = it .next() .ok_or_else(|| "--project requires a value".to_string())?; } "--requester" => { requester = it .next() .ok_or_else(|| "--requester requires a value".to_string())?; } other => { return Err(format!("unknown argument: {other}")); } } } let endpoint = endpoint.ok_or_else(|| "--endpoint is required".to_string())?; Ok(Self { endpoint, project, requester, }) } /// The first handshake line the bridge writes on the loopback (newline /// included). See the module docs for the format consumed by M5c. fn handshake_line(&self) -> Vec { let line = serde_json::json!({ "project": self.project, "requester": self.requester, }) .to_string(); let mut bytes = line.into_bytes(); bytes.push(b'\n'); bytes } } /// Synchronous entry point called from `main`/`lib::run` when `argv[1] == /// "mcp-server"`. Parses the args, stands up a Tokio runtime, and runs the bridge. /// /// Returns the process [`ExitCode`]: `0` on a clean stdin-EOF shutdown, non-zero on /// any argument/connection/relay failure. Never blocks indefinitely (connect is /// time-bounded). #[must_use] pub fn run_mcp_bridge(args: I) -> ExitCode where I: IntoIterator, S: Into, { let args = match BridgeArgs::parse(args) { Ok(a) => a, Err(e) => { eprintln!("idea mcp-server: {e}"); return ExitCode::FAILURE; } }; // A current-thread runtime is enough: the bridge is two coupled I/O loops, no // CPU work. Keeps the headless process lean (no Tauri, no webview). let runtime = match tokio::runtime::Builder::new_current_thread() .enable_io() .enable_time() .build() { Ok(rt) => rt, Err(e) => { eprintln!("idea mcp-server: failed to start runtime: {e}"); return ExitCode::FAILURE; } }; runtime.block_on(async move { match bridge_over_loopback(&args).await { Ok(()) => ExitCode::SUCCESS, Err(e) => { eprintln!("idea mcp-server: {e}"); ExitCode::FAILURE } } }) } /// Connects to the project loopback at `args.endpoint` (time-bounded) and relays /// between the **real stdio** of this process and that connection. async fn bridge_over_loopback(args: &BridgeArgs) -> Result<(), String> { let conn = connect_loopback(&args.endpoint).await?; // Split the duplex loopback into independent owned halves so the two relay // directions can run without sharing a lock. let (lp_read, lp_write) = tokio::io::split(conn); relay( args, tokio::io::stdin(), tokio::io::stdout(), lp_read, lp_write, ) .await } /// Connects to the loopback endpoint with a bounded timeout. An absent or /// unreachable endpoint surfaces as an `Err` (⇒ non-zero exit), **never a hang**. async fn connect_loopback(endpoint: &str) -> Result { let name = endpoint .to_fs_name::() .map_err(|e| format!("invalid endpoint {endpoint:?}: {e}"))?; match tokio::time::timeout(CONNECT_TIMEOUT, LocalSocketStream::connect(name)).await { Ok(Ok(stream)) => Ok(stream), Ok(Err(e)) => Err(format!("cannot connect to endpoint {endpoint:?}: {e}")), Err(_) => Err(format!( "timed out connecting to endpoint {endpoint:?} after {}s", CONNECT_TIMEOUT.as_secs() )), } } /// The transparent relay, generic over its four streams so tests drive it with /// in-memory pipes (no real stdio, no real socket). /// /// Sequence: /// 1. Write the **handshake line** (`project`+`requester`) to the loopback. /// 2. Loop: read one **line** from the CLI (`cli_in`) → write it to the loopback; /// read one **line** from the loopback → write it to the CLI (`cli_out`). /// 3. **CLI stdin EOF** ⇒ stop, return `Ok(())` (clean code 0). Dropping the /// loopback writer closes that direction of the connection. /// /// The loop is request/response paced (one CLI line in, one loopback line back), /// matching the JSON Lines `StdioTransport` the server speaks. A loopback that /// closes mid-exchange surfaces as an error (non-zero exit). async fn relay( args: &BridgeArgs, cli_in: CIn, mut cli_out: COut, lp_read: LIn, mut lp_write: LOut, ) -> Result<(), String> where CIn: AsyncRead + Unpin, COut: AsyncWrite + Unpin, LIn: AsyncRead + Unpin, LOut: AsyncWrite + Unpin, { // 1. Handshake (identity) before any JSON-RPC byte. lp_write .write_all(&args.handshake_line()) .await .map_err(|e| format!("handshake write failed: {e}"))?; lp_write .flush() .await .map_err(|e| format!("handshake flush failed: {e}"))?; let mut cli_reader = BufReader::new(cli_in); let mut lp_reader = BufReader::new(lp_read); let mut cli_line = String::new(); let mut lp_line = String::new(); loop { // 2a. CLI → loopback (one line). EOF ⇒ clean shutdown. cli_line.clear(); let n = cli_reader .read_line(&mut cli_line) .await .map_err(|e| format!("stdin read failed: {e}"))?; if n == 0 { // CLI closed stdin: nothing more to forward. Clean exit. return Ok(()); } lp_write .write_all(cli_line.as_bytes()) .await .map_err(|e| format!("loopback write failed: {e}"))?; lp_write .flush() .await .map_err(|e| format!("loopback flush failed: {e}"))?; // 2b. loopback → CLI (one line). EOF here is a server hangup mid-exchange. lp_line.clear(); let m = lp_reader .read_line(&mut lp_line) .await .map_err(|e| format!("loopback read failed: {e}"))?; if m == 0 { return Err("loopback closed before answering".to_string()); } cli_out .write_all(lp_line.as_bytes()) .await .map_err(|e| format!("stdout write failed: {e}"))?; cli_out .flush() .await .map_err(|e| format!("stdout flush failed: {e}"))?; } } #[cfg(test)] mod tests { use super::*; use std::sync::Arc; use interprocess::local_socket::tokio::Listener as LocalSocketListener; use interprocess::local_socket::traits::tokio::Listener as _; use interprocess::local_socket::{GenericFilePath, ListenerOptions}; use tokio::io::AsyncReadExt as _; // ---- argument parsing ------------------------------------------------- #[test] fn parses_all_three_flags() { let a = BridgeArgs::parse([ "--endpoint", "/tmp/x.sock", "--project", "p1", "--requester", "agent-7", ]) .unwrap(); assert_eq!(a.endpoint, "/tmp/x.sock"); assert_eq!(a.project, "p1"); assert_eq!(a.requester, "agent-7"); } #[test] fn endpoint_is_required() { let err = BridgeArgs::parse(["--project", "p1"]).unwrap_err(); assert!(err.contains("--endpoint"), "got: {err}"); } #[test] fn project_and_requester_default_to_empty() { let a = BridgeArgs::parse(["--endpoint", "/tmp/x.sock"]).unwrap(); assert_eq!(a.project, ""); assert_eq!(a.requester, ""); } #[test] fn flag_without_value_is_an_error() { assert!(BridgeArgs::parse(["--endpoint"]).is_err()); assert!(BridgeArgs::parse(["--endpoint", "/x", "--project"]).is_err()); } #[test] fn unknown_flag_is_an_error() { let err = BridgeArgs::parse(["--endpoint", "/x", "--bogus"]).unwrap_err(); assert!(err.contains("unknown"), "got: {err}"); } #[test] fn handshake_line_carries_identity_as_json() { let a = BridgeArgs { endpoint: "/tmp/x.sock".into(), project: "proj".into(), requester: "rq".into(), }; let line = a.handshake_line(); assert_eq!(*line.last().unwrap(), b'\n'); let v: serde_json::Value = serde_json::from_slice(&line[..line.len() - 1]).unwrap(); assert_eq!(v["project"], "proj"); assert_eq!(v["requester"], "rq"); } // ---- relay (in-memory streams, no socket) ----------------------------- /// Nominal relay over in-memory duplex pipes: handshake reaches the fake /// server, a scripted request is forwarded, and the canned response comes back /// on the CLI stdout. #[tokio::test] async fn relay_forwards_handshake_request_and_response() { // CLI stdin: one JSON-RPC request line, then EOF. let cli_in = b"{\"method\":\"tools/call\",\"id\":1}\n".to_vec(); let mut cli_out: Vec = Vec::new(); // The "loopback" is a duplex pipe: the bridge writes to `lp_write`/reads // from `lp_read`; the fake server reads from `srv_read`/writes to // `srv_write`. let (lp_read, srv_write) = tokio::io::duplex(4096); // server → bridge let (srv_read, lp_write) = tokio::io::duplex(4096); // bridge → server let args = BridgeArgs { endpoint: "unused".into(), project: "proj-1".into(), requester: "agent-9".into(), }; // Fake server: read handshake line, then read the request line, then answer. let server = tokio::spawn(async move { let mut reader = BufReader::new(srv_read); let mut handshake = String::new(); reader.read_line(&mut handshake).await.unwrap(); let mut request = String::new(); reader.read_line(&mut request).await.unwrap(); let mut w = srv_write; w.write_all(b"{\"result\":\"ok\",\"id\":1}\n").await.unwrap(); w.flush().await.unwrap(); (handshake, request) }); relay( &args, &cli_in[..], &mut cli_out, lp_read, lp_write, ) .await .unwrap(); let (handshake, request) = server.await.unwrap(); let hs: serde_json::Value = serde_json::from_str(handshake.trim_end()).unwrap(); assert_eq!(hs["project"], "proj-1"); assert_eq!(hs["requester"], "agent-9"); assert_eq!(request.trim_end(), "{\"method\":\"tools/call\",\"id\":1}"); assert_eq!( String::from_utf8(cli_out).unwrap(), "{\"result\":\"ok\",\"id\":1}\n" ); } /// stdin EOF with no request ⇒ relay returns `Ok` (code 0), having still sent /// the handshake. #[tokio::test] async fn relay_clean_exit_on_immediate_eof() { let cli_in: &[u8] = b""; // immediate EOF let mut cli_out: Vec = Vec::new(); let (lp_read, srv_write) = tokio::io::duplex(4096); let (srv_read, lp_write) = tokio::io::duplex(4096); let args = BridgeArgs { endpoint: "unused".into(), project: "p".into(), requester: String::new(), }; let server = tokio::spawn(async move { let mut reader = BufReader::new(srv_read); let mut handshake = String::new(); reader.read_line(&mut handshake).await.unwrap(); // Keep srv_write alive until the bridge drops its writer (EOF). let mut buf = Vec::new(); let _ = reader.read_to_end(&mut buf).await; drop(srv_write); handshake }); relay(&args, cli_in, &mut cli_out, lp_read, lp_write) .await .expect("clean EOF exit"); let handshake = server.await.unwrap(); assert!(handshake.contains("\"project\":\"p\"")); assert!(cli_out.is_empty(), "no response expected on immediate EOF"); } // ---- end-to-end over a real loopback ---------------------------------- fn temp_endpoint(tag: &str) -> String { let dir = std::env::temp_dir(); let pid = std::process::id(); let nanos = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_nanos(); dir.join(format!("idea-mcp-bridge-test-{tag}-{pid}-{nanos}.sock")) .to_string_lossy() .into_owned() } fn bind(endpoint: &str) -> LocalSocketListener { let name = endpoint.to_fs_name::().unwrap(); ListenerOptions::new() .name(name) .reclaim_name(true) .create_tokio() .expect("bind test listener") } /// Endpoint absent ⇒ `connect_loopback` fails fast (no hang), so the bridge /// would exit non-zero. Bounded by a test timeout well under CONNECT_TIMEOUT /// for the "connect error" path (a non-existent socket errors immediately). #[tokio::test] async fn connect_to_absent_endpoint_errors_fast() { let endpoint = temp_endpoint("absent"); let res = tokio::time::timeout( Duration::from_secs(10), connect_loopback(&endpoint), ) .await .expect("connect_loopback must not hang"); assert!(res.is_err(), "absent endpoint must yield an error"); } /// `run_mcp_bridge` with a missing `--endpoint` returns a non-zero code. #[test] fn run_bridge_missing_endpoint_is_non_zero() { let code = run_mcp_bridge(["--project", "p"]); assert_eq!(code, ExitCode::FAILURE); } /// Full end-to-end over a **real** interprocess loopback: a test listener plays /// the server (reads the handshake, echoes a canned response); the relay drives /// it over an actual connection. Verifies the handshake crosses the real socket /// and the response returns on the CLI stdout. #[tokio::test] async fn end_to_end_over_real_loopback() { let endpoint = temp_endpoint("e2e"); let listener = bind(&endpoint); let captured = Arc::new(tokio::sync::Mutex::new((String::new(), String::new()))); let captured_srv = Arc::clone(&captured); // Fake server accepts one connection, reads handshake + request, answers. let server = tokio::spawn(async move { let conn = listener.accept().await.unwrap(); let (r, w) = tokio::io::split(conn); let mut reader = BufReader::new(r); let mut handshake = String::new(); reader.read_line(&mut handshake).await.unwrap(); let mut request = String::new(); reader.read_line(&mut request).await.unwrap(); *captured_srv.lock().await = (handshake, request); let mut w = w; w.write_all(b"{\"result\":\"pong\",\"id\":42}\n") .await .unwrap(); w.flush().await.unwrap(); // Hold until the bridge finishes (its stdin EOF will end it). let mut rest = Vec::new(); let _ = reader.read_to_end(&mut rest).await; }); let args = BridgeArgs { endpoint: endpoint.clone(), project: "proj-e2e".into(), requester: "agent-e2e".into(), }; let cli_in = b"{\"method\":\"ping\",\"id\":42}\n".to_vec(); let mut cli_out: Vec = Vec::new(); // Drive the whole bridge_over_loopback path (real connect + split + relay), // but feed our own stdio streams via `relay` to avoid touching process std. let conn = tokio::time::timeout( Duration::from_secs(10), connect_loopback(&endpoint), ) .await .expect("no hang") .expect("connect to live endpoint"); let (lp_read, lp_write) = tokio::io::split(conn); tokio::time::timeout( Duration::from_secs(10), relay(&args, &cli_in[..], &mut cli_out, lp_read, lp_write), ) .await .expect("relay must not hang") .expect("relay ok"); server.await.unwrap(); let (handshake, request) = captured.lock().await.clone(); let hs: serde_json::Value = serde_json::from_str(handshake.trim_end()).unwrap(); assert_eq!(hs["project"], "proj-e2e"); assert_eq!(hs["requester"], "agent-e2e"); assert_eq!(request.trim_end(), "{\"method\":\"ping\",\"id\":42}"); assert_eq!( String::from_utf8(cli_out).unwrap(), "{\"result\":\"pong\",\"id\":42}\n" ); } }