diff --git a/Cargo.lock b/Cargo.lock index 679ca5e..59f9326 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -75,6 +75,7 @@ dependencies = [ "async-trait", "domain", "infrastructure", + "interprocess", "serde", "serde_json", "tauri", @@ -847,6 +848,12 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "doctest-file" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2db04e74f0a9a93103b50e90b96024c9b2bdca8bce6a632ec71b88736d3d359" + [[package]] name = "document-features" version = "0.2.12" @@ -1946,6 +1953,21 @@ dependencies = [ "libc", ] +[[package]] +name = "interprocess" +version = "2.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "069323743400cb7ab06a8fe5c1ed911d36b6919ec531661d034c89083629595b" +dependencies = [ + "doctest-file", + "futures-core", + "libc", + "recvmsg", + "tokio", + "widestring", + "windows-sys 0.61.2", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -3236,6 +3258,12 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "recvmsg" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3edd4d5d42c92f0a659926464d4cce56b562761267ecf0f469d85b7de384175" + [[package]] name = "redox_syscall" version = "0.5.18" @@ -5274,6 +5302,12 @@ dependencies = [ "windows-core 0.61.2", ] +[[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + [[package]] name = "winapi" version = "0.3.9" diff --git a/crates/app-tauri/Cargo.toml b/crates/app-tauri/Cargo.toml index b9282e4..96040c3 100644 --- a/crates/app-tauri/Cargo.toml +++ b/crates/app-tauri/Cargo.toml @@ -25,11 +25,18 @@ application = { workspace = true } infrastructure = { workspace = true } tauri = { workspace = true } tauri-plugin-dialog = { workspace = true } -tokio = { workspace = true } +# `io-std` (on top of the workspace features) gives the headless `mcp-server` +# bridge access to `tokio::io::{stdin,stdout}` without widening tokio elsewhere. +tokio = { workspace = true, features = ["io-std", "rt"] } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } uuid = { workspace = true } +# Cross-OS local IPC for the MCP loopback transport (M5a): Unix domain socket +# (Linux/macOS) + Windows named pipe behind one async API, no network port. Pulled +# in only here (the transport is an app-tauri/infra concern); its `tokio` feature +# keeps us off tokio's `net` feature workspace-wide. +interprocess = { version = "2.4", features = ["tokio"] } [features] # Passthrough toggles to enable the real embedders in an IDE build. OFF by default diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 63cbcf4..d6f65ed 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -16,6 +16,7 @@ use application::{ DetectAgentDriftInput, GetMemoryInput, GitBranchesInput, GitCheckoutInput, GitCommitInput, GitGraphInput, GitInitInput, GitLogInput, GitStagePathInput, GitStatusInput, InspectConversationInput, LaunchAgentInput, ListAgentsInput, ListLayoutsInput, LiveSessions, + McpRuntime, ListMemoriesInput, ListResumableAgentsInput, ListSkillsInput, LoadLayoutInput, MutateLayoutInput, OpenProjectInput, ReadAgentContextInput, ReadMemoryIndexInput, ReadProjectContextInput, RecallMemoryInput, @@ -1029,6 +1030,22 @@ pub async fn launch_agent( // already-live agent is refused). let node_id = request.node_id.as_deref().map(parse_node_id).transpose()?; + // Compose the MCP runtime (M5d): inject the OS/runtime facts the application + // layer cannot compute. The endpoint comes from the **single source of truth** + // (`mcp_endpoint`), the same value `ensure_mcp_server` binds the listener on + // (cadrage v5 §2). The `--project` is the hyphen-free 32-hex `simple` form the + // M5c handshake guard (`serve_peer`) compares against; the `--requester` is the + // launching agent's id. A missing executable path (should not happen) degrades + // to no runtime ⇒ apply_mcp_config writes the minimal declaration. + let mcp_runtime = std::env::current_exe().ok().map(|exe| McpRuntime { + exe: exe.to_string_lossy().into_owned(), + endpoint: crate::mcp_endpoint::mcp_endpoint(&project.id) + .as_cli_arg() + .to_owned(), + project_id: project.id.as_uuid().simple().to_string(), + requester: agent_id.to_string(), + }); + let output = state .launch_agent .execute(LaunchAgentInput { @@ -1040,6 +1057,7 @@ pub async fn launch_agent( // Resume id is a property of the hosting cell; the frontend passes the // leaf's current conversation id here. conversation_id: request.conversation_id.clone(), + mcp_runtime, }) .await .map_err(ErrorDto::from)?; diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index 072fc21..9b1eede 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -17,13 +17,49 @@ pub mod chat; pub mod commands; pub mod dto; pub mod events; +pub mod mcp_bridge; +pub mod mcp_endpoint; pub mod pty; pub mod state; +use std::process::ExitCode; + use tauri::Manager; use state::AppState; +/// The `argv[1]` subcommand token that switches the binary into the headless +/// `mcp-server` **bridge** mode (cadrage v5 §1.3) instead of launching Tauri. +pub const MCP_SERVER_SUBCOMMAND: &str = "mcp-server"; + +/// Process entry point: routes `argv` **before** anything Tauri/WebKit is touched. +/// +/// When invoked as ` mcp-server …` (an MCP CLI spawned us from the injected +/// `.mcp.json` declaration), we run the **stdio↔loopback bridge** headless and +/// **never** initialise the webview — see [`mcp_bridge::run_mcp_bridge`]. Any other +/// invocation is the normal IDE launch: [`run`] (which blocks until the window +/// closes and then exits the process itself). +/// +/// Returns the [`ExitCode`] for the bridge path; the normal path does not return. +#[must_use] +pub fn dispatch() -> ExitCode { + let mut args = std::env::args_os().skip(1); + if args + .next() + .is_some_and(|a| a == *MCP_SERVER_SUBCOMMAND) + { + // Headless bridge: bypass Tauri entirely. Forward the remaining args + // (`--endpoint`, `--project`, `--requester`) to the bridge parser. + let rest: Vec = args + .map(|a| a.to_string_lossy().into_owned()) + .collect(); + return mcp_bridge::run_mcp_bridge(rest); + } + + run(); + ExitCode::SUCCESS +} + /// Builds and runs the Tauri application. /// /// Sets up the composition root (resolving the app-data directory via the Tauri diff --git a/crates/app-tauri/src/main.rs b/crates/app-tauri/src/main.rs index 4310022..ff201c5 100644 --- a/crates/app-tauri/src/main.rs +++ b/crates/app-tauri/src/main.rs @@ -1,10 +1,13 @@ // Prevents an extra console window on Windows in release builds. #![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] -fn main() { +use std::process::ExitCode; + +fn main() -> ExitCode { // WebKitGTK's DMABUF renderer causes a blank/white window on many Linux // setups (recent Mesa/Nvidia drivers, common on Arch). Disable it before // the webview initializes, unless the user has explicitly set the variable. + // Harmless for the headless `mcp-server` bridge path, which ignores it. #[cfg(target_os = "linux")] { if std::env::var_os("WEBKIT_DISABLE_DMABUF_RENDERER").is_none() { @@ -12,5 +15,6 @@ fn main() { } } - app_tauri_lib::run(); + // Route argv before any Tauri/WebKit init: `mcp-server` ⇒ headless bridge. + app_tauri_lib::dispatch() } diff --git a/crates/app-tauri/src/mcp_bridge.rs b/crates/app-tauri/src/mcp_bridge.rs new file mode 100644 index 0000000..639830d --- /dev/null +++ b/crates/app-tauri/src/mcp_bridge.rs @@ -0,0 +1,578 @@ +//! 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" + ); + } +} diff --git a/crates/app-tauri/src/mcp_endpoint.rs b/crates/app-tauri/src/mcp_endpoint.rs new file mode 100644 index 0000000..645406b --- /dev/null +++ b/crates/app-tauri/src/mcp_endpoint.rs @@ -0,0 +1,173 @@ +//! Per-project loopback **endpoint** for the IdeA MCP transport (M5a). +//! +//! ## Where this sits in the bind (cadrage v5 §1, §2) +//! +//! The S-MCP transport is **stdio-spawn**: an MCP CLI (Claude Code, Codex) +//! reads a `{command,args}` declaration and *spawns* a thin `idea mcp-server` +//! bridge. That bridge talks JSON-RPC on its stdin/stdout to the CLI and relays +//! every line, over a **local loopback**, to the IdeA (Tauri) process where the +//! real [`OrchestratorService`](application::OrchestratorService) lives. The +//! loopback is a **Unix domain socket** (Linux/macOS) or a **Windows named +//! pipe** — never a network port, so it stays AppImage- and SSH-remote-safe and +//! needs no firewall/permission. +//! +//! This module owns the **single source of truth** for that loopback address: +//! [`mcp_endpoint`]. It is deterministic per [`ProjectId`] and is called by +//! **both** sides of the contract (cadrage §2): +//! - the side that *listens* — [`ensure_mcp_server`](crate::state::AppState::ensure_mcp_server), +//! which binds the listener at project-open and tears it down at close (M5a); +//! - the side that *writes the CLI declaration* — `apply_mcp_config` (M5d), +//! which must point the spawned bridge at the **exact** same address. +//! +//! Keeping the derivation here (not duplicated on each side) is what makes the +//! M1↔M3 coherence test possible. +//! +//! ## Transport crate choice — `interprocess` +//! +//! We use the [`interprocess`] crate's *local socket* abstraction rather than +//! `tokio::net::{UnixListener, windows::named_pipe}` directly, because: +//! - **One cross-OS API** for UDS (Unix) and named pipes (Windows): a single +//! `accept` loop in M5c, no divergent `cfg` branches with two transport types. +//! - The workspace `tokio` is built **without** the `net` feature; pulling +//! `interprocess` (with its `tokio` feature) into *only* this crate avoids +//! widening tokio's surface workspace-wide. +//! - On Unix its listener carries a *reclaim guard* that **unlinks the socket +//! file on drop** — so closing a project (dropping the handle) cleans the +//! filesystem with no leak, satisfying M5a's teardown requirement for free. +//! +//! We deliberately bind a **filesystem-path** socket on Unix (under a per-user +//! runtime dir) rather than an abstract/namespaced one, so that the existence of +//! the endpoint is observable as a real path (testable) and cleaned up on close. + +use std::path::PathBuf; + +use domain::ProjectId; + +/// The loopback address of a project's MCP endpoint — the value [`mcp_endpoint`] +/// returns. Deterministic per [`ProjectId`]; the single source of truth shared by +/// the listener (M5a/M3) and the CLI-declaration writer (M5d). +/// +/// On Unix it is a **filesystem path** to a Unix-domain socket (the file is what +/// gets bound and, on close, unlinked). On Windows it is a **named pipe** path of +/// the form `\\.\pipe\idea-mcp-`. The [`as_cli_arg`](Self::as_cli_arg) form is +/// the exact string handed to the spawned bridge via `--endpoint`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct McpEndpoint { + /// The platform-native address string (UDS path / named-pipe path). + addr: String, +} + +impl McpEndpoint { + /// The address as the bridge expects it on the command line (`--endpoint`). + /// Identical string on both sides of the contract (cadrage §2). + #[must_use] + pub fn as_cli_arg(&self) -> &str { + &self.addr + } + + /// The Unix socket **file path**, when this endpoint is a filesystem-path UDS. + /// Used by the listener to ensure the parent runtime dir exists and by tests to + /// assert creation/cleanup. `None` on platforms whose address is not a path + /// (Windows named pipes have no filesystem entry to manage). + #[must_use] + pub fn socket_path(&self) -> Option { + #[cfg(unix)] + { + Some(PathBuf::from(&self.addr)) + } + #[cfg(not(unix))] + { + None + } + } +} + +/// The directory under which per-project Unix sockets live, derived from the +/// per-user runtime dir (`$XDG_RUNTIME_DIR`, falling back to `$TMPDIR`, then +/// `/tmp`). Kept short so the full socket path stays well under the ~108-byte +/// `sockaddr_un` limit (`/run/user//idea-mcp/<32 hex>.sock` ≈ 50 bytes). +#[cfg(unix)] +fn unix_runtime_dir() -> PathBuf { + let base = std::env::var_os("XDG_RUNTIME_DIR") + .map(PathBuf::from) + .or_else(|| std::env::var_os("TMPDIR").map(PathBuf::from)) + .unwrap_or_else(|| PathBuf::from("/tmp")); + base.join("idea-mcp") +} + +/// **Single source of truth.** Computes the deterministic loopback endpoint of a +/// project's MCP transport from its [`ProjectId`]. +/// +/// Same input ⇒ same output (stable across calls and processes); distinct projects +/// ⇒ distinct endpoints (the id's 32-hex *simple* form is unique and collision-free +/// by construction). No network port is ever used. +/// +/// - **Unix**: `/idea-mcp/.sock` — a UDS file path. +/// - **Windows**: `\\.\pipe\idea-mcp-` — a named pipe. +/// +/// where `` is the project UUID in hyphen-free hex (`simple`) form. +#[must_use] +pub fn mcp_endpoint(project_id: &ProjectId) -> McpEndpoint { + // Hyphen-free, lowercase, fixed-width (32 chars): safe in both a filename and + // a `\\.\pipe\` name, and unique per project. + let id = project_id.as_uuid().simple().to_string(); + + #[cfg(unix)] + let addr = unix_runtime_dir() + .join(format!("{id}.sock")) + .to_string_lossy() + .into_owned(); + + #[cfg(windows)] + let addr = format!(r"\\.\pipe\idea-mcp-{id}"); + + #[cfg(not(any(unix, windows)))] + let addr = format!("idea-mcp-{id}"); + + McpEndpoint { addr } +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + fn pid(s: &str) -> ProjectId { + ProjectId::from_uuid(Uuid::parse_str(s).unwrap()) + } + + #[test] + fn endpoint_is_deterministic_for_the_same_project() { + let p = pid("11111111-1111-1111-1111-111111111111"); + assert_eq!(mcp_endpoint(&p), mcp_endpoint(&p)); + assert_eq!(mcp_endpoint(&p).as_cli_arg(), mcp_endpoint(&p).as_cli_arg()); + } + + #[test] + fn distinct_projects_get_distinct_endpoints() { + let a = pid("11111111-1111-1111-1111-111111111111"); + let b = pid("22222222-2222-2222-2222-222222222222"); + assert_ne!(mcp_endpoint(&a), mcp_endpoint(&b)); + assert_ne!(mcp_endpoint(&a).as_cli_arg(), mcp_endpoint(&b).as_cli_arg()); + } + + #[test] + fn address_encodes_the_project_id_without_hyphens() { + let p = pid("abcdef01-2345-6789-abcd-ef0123456789"); + let arg = mcp_endpoint(&p).as_cli_arg().to_owned(); + assert!( + arg.contains("abcdef0123456789abcdef0123456789"), + "endpoint must embed the hyphen-free id, got {arg}" + ); + } + + #[cfg(unix)] + #[test] + fn unix_endpoint_is_a_sock_path_under_the_runtime_dir() { + let p = pid("11111111-1111-1111-1111-111111111111"); + let ep = mcp_endpoint(&p); + let path = ep.socket_path().expect("unix endpoint exposes a socket path"); + assert!(path.extension().is_some_and(|e| e == "sock")); + assert_eq!(path.parent().unwrap(), unix_runtime_dir()); + } +} diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 758a9ea..483ef0d 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -50,8 +50,16 @@ use infrastructure::{ }; use crate::chat::ChatBridge; +use crate::mcp_endpoint::{mcp_endpoint, McpEndpoint}; use crate::pty::PtyBridge; +use infrastructure::StdioTransport; + +use interprocess::local_socket::tokio::Listener as LocalSocketListener; +use interprocess::local_socket::traits::tokio::Listener as _; +use interprocess::local_socket::{GenericFilePath, ListenerOptions, ToFsName}; +use tokio::io::{AsyncBufReadExt, BufReader}; + /// Everything the IPC layer needs at runtime, managed by Tauri. /// /// Use cases are stored behind `Arc` so handlers clone cheaply. The concrete @@ -878,17 +886,21 @@ impl AppState { /// lifecycle handle in [`mcp_servers`](Self::mcp_servers); dropping/stopping the /// handle tears down the supervision task. /// - /// ## Transport decision (point ouvert S-MCP) + /// ## Transport decision (S-MCP, cadrage v5 §1) & endpoint lifecycle (M5a) /// - /// At project-open time there is **no CLI connected yet**: a real `stdio` - /// transport only has a peer once an MCP-capable agent is launched with the - /// injected MCP config (M1), and the exact transport↔session bind is the open - /// **S-MCP** point of the cadrage. We therefore deliberately do **not** run a - /// blocking `McpServer::serve` here (that would have no peer to talk to and must - /// never figer the open/close of a project). Instead this hook owns the server's - /// **lifecycle** per project — creation now, clean stop on close — parked on a - /// stop signal exactly like the watcher. The per-session transport binding is - /// finalised later (S-MCP) without touching this wiring. + /// At project-open time there is **no CLI connected yet**: a peer only appears + /// once an MCP-capable agent is launched with the injected MCP config (M5d) and + /// its spawned `idea mcp-server` bridge dials this project's loopback endpoint. + /// We therefore do **not** run a blocking `McpServer::serve`/accept loop here + /// (that must never figer the open/close of a project). M5a's job is narrower: + /// **bind the project's loopback endpoint** ([`mcp_endpoint`], the single source + /// of truth) so it is ready when a bridge connects, and **hold** the listener in + /// the handle. The actual accept-and-serve-per-peer is M5b/M5c. + /// + /// Binding is **non-blocking** (create the listener, then park on the stop + /// signal) and **idempotent** (one endpoint per project: a second open returns + /// early without rebinding). On close the handle is dropped, which on Unix + /// unlinks the socket file (interprocess reclaim guard) — no leak. fn ensure_mcp_server(&self, project: &Project) { let mut servers = self .mcp_servers @@ -900,9 +912,17 @@ impl AppState { let bus = Arc::clone(&self.event_bus); let events: Arc = Arc::new(move |event| bus.publish(event)); + let endpoint = mcp_endpoint(&project.id); + let listener = bind_endpoint(&endpoint); + // The project-id string the handshake guard compares against: the same + // hyphen-free hex form the endpoint encodes, which M5d's `--project` reuses. + let project_id = project.id.as_uuid().simple().to_string(); let handle = McpServerHandle::start( McpServer::new(Arc::clone(&self.orchestrator_service), project.clone()) .with_events(events), + endpoint, + listener, + project_id, ); servers.insert(project.id, handle); } @@ -945,33 +965,196 @@ impl AppState { } } +/// Binds the project's loopback endpoint listener (M5a). Best-effort: returns the +/// bound [`LocalSocketListener`] or `None` if the bind fails (e.g. a stale socket +/// file from a crashed run). A `None` never figes open/close — the registry entry +/// is still created so the lifecycle stays idempotent; the real accept/serve (M5c) +/// will surface a hard failure if it actually needs the listener. +/// +/// On Unix this binds a filesystem-path UDS under the per-user runtime dir; the +/// parent directory is created if missing, and a corpse socket from a previous run +/// is replaced (`reclaim_name` so the file is unlinked on drop). On Windows it binds +/// the named pipe — no filesystem entry to manage. +#[must_use] +fn bind_endpoint(endpoint: &McpEndpoint) -> Option { + // Ensure the runtime dir exists (Unix path sockets need their parent dir). + if let Some(path) = endpoint.socket_path() { + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + } + let name = endpoint + .as_cli_arg() + .to_fs_name::() + .ok()?; + ListenerOptions::new() + .name(name) + // Replace a corpse socket left by a previous run; reclaim (unlink) on drop. + .reclaim_name(true) + .create_tokio() + .ok() +} + +/// Drives one accepted loopback peer (= one `idea mcp-server` bridge = one agent): +/// reads its **handshake line**, then serves the rest of the stream as JSON-RPC. +/// +/// Sequence (cadrage v5 §1.4, M5b handshake format): +/// 1. Split the duplex connection so reads and writes are independent. +/// 2. Read **one** newline-terminated handshake line +/// (`{"project":"…","requester":"…"}`) off a `BufReader` over the read half. +/// 3. If the handshake's `project` is present and **mismatches** this server's +/// project, close the connection (return) without serving — a defensive guard, it +/// logs nothing and never crashes the accept loop. +/// 4. Wrap the **same** `BufReader` (carrying any bytes already buffered past the +/// handshake) + the write half in a [`StdioTransport`] and hand it to +/// [`McpServer::serve_as`] tagged with the handshake's `requester` — so +/// `OrchestratorRequestProcessed.requester_id` becomes the real agent id. +/// +/// Generic over the stream type so it never names the `interprocess` connection +/// type and stays unit-testable over an in-memory duplex. +async fn serve_peer(server: Arc, expected_project: &str, conn: S) +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static, +{ + let (read_half, write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + // 1 handshake line. A read error or immediate EOF ⇒ no peer to serve. + let mut handshake = String::new(); + match reader.read_line(&mut handshake).await { + Ok(0) | Err(_) => return, + Ok(_) => {} + } + + let (handshake_project, requester) = parse_handshake(&handshake); + + // Defensive: a bridge dialed the wrong project's endpoint. Drop it cleanly. + if !handshake_project.is_empty() + && !expected_project.is_empty() + && handshake_project != expected_project + { + return; + } + + // The same BufReader keeps any bytes already buffered past the handshake, so no + // JSON-RPC line is lost between the handshake and the serve loop. + let mut transport = StdioTransport::from_buffered(reader, write_half); + server.serve_as(requester, &mut transport).await; +} + +/// Parses the M5b handshake line into `(project, requester)`. Tolerant: a malformed +/// or partial line yields empty strings (⇒ legacy `"mcp"` requester, no project +/// guard), never an error — the serve loop must never crash on a bad handshake. +fn parse_handshake(line: &str) -> (String, String) { + serde_json::from_str::(line.trim()) + .ok() + .map(|v| { + let project = v + .get("project") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_owned(); + let requester = v + .get("requester") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_owned(); + (project, requester) + }) + .unwrap_or_default() +} + /// Lifecycle handle for a per-project [`McpServer`] supervision task — the **twin** /// of [`OrchestratorWatchHandle`](infrastructure::OrchestratorWatchHandle). /// /// It carries the **same** stop mechanism as the watcher (a one-shot /// `mpsc::Sender<()>`): [`stop`](Self::stop) signals the task to exit, and dropping -/// the handle stops it too (the channel closes). The supervision task keeps the -/// `McpServer` alive and ready, parked on the stop signal; it spawns **no blocking -/// loop**, so it never figes the open/close of a project (see the transport -/// decision on [`AppState::ensure_mcp_server`]). +/// the handle stops it too (the channel closes). +/// +/// ## Accept-and-serve-per-peer (M5c) +/// +/// The supervision task runs an **`accept` loop** on the bound loopback `listener`, +/// arbitrated against the stop signal by `tokio::select!`: +/// - **`accept`** is async and parks on the absence of a peer, so the loop never +/// figes the open/close of a project — at project-open time no CLI is connected +/// yet (a peer only appears once an MCP-capable agent is launched, M5d). +/// - **each accepted connection** = one `idea mcp-server` bridge = one agent. The +/// task reads the **handshake line** (`{"project","requester"}`, cadrage v5 §1.4), +/// then spawns an isolated task running [`McpServer::serve_as`] over the rest of +/// the stream, so one peer's disconnection/error never affects the others or the +/// accept loop. +/// - **stop** breaks the loop, **aborts** the in-flight serve tasks (`JoinSet`), +/// and drops the listener — which on Unix unlinks the socket file via +/// interprocess' reclaim guard, so closing a project leaves no socket behind. pub struct McpServerHandle { stop: tokio::sync::mpsc::Sender<()>, + /// The loopback address this server listens on — the single source of truth + /// ([`mcp_endpoint`]) shared with the CLI-declaration writer (M5d). + endpoint: McpEndpoint, } impl McpServerHandle { - /// Spawns the supervision task that owns `server` until stopped. Must run inside - /// the ambient Tokio runtime (Tauri async commands satisfy this), exactly like - /// [`FsOrchestratorWatcher::start`](infrastructure::FsOrchestratorWatcher). + /// Spawns the supervision task that owns `server` and the bound `listener`, runs + /// the accept-and-serve-per-peer loop, and stops cleanly on signal. Must run + /// inside the ambient Tokio runtime (Tauri async commands satisfy this), exactly + /// like [`FsOrchestratorWatcher::start`](infrastructure::FsOrchestratorWatcher). #[must_use] - fn start(server: McpServer) -> Self { + fn start( + server: McpServer, + endpoint: McpEndpoint, + listener: Option, + project_id: String, + ) -> Self { let (stop_tx, mut stop_rx) = tokio::sync::mpsc::channel::<()>(1); + // The base server is shared (Arc) so each accepted peer derives its own + // requester-tagged clone (McpServer::serve_as) without contending. + let server = Arc::new(server); tokio::spawn(async move { - // Hold the server for this project's lifetime; park until stopped. When a - // per-session transport bind lands (S-MCP), it attaches to this `server`. - let _server = server; - let _ = stop_rx.recv().await; + // No listener (bind failed, e.g. stale socket): nothing to accept. Park + // on stop so the lifecycle stays idempotent and non-blocking. + let Some(listener) = listener else { + let _ = stop_rx.recv().await; + return; + }; + // In-flight per-peer serve tasks; aborted en masse on stop so no serve + // outlives the project's close. + let mut serves: tokio::task::JoinSet<()> = tokio::task::JoinSet::new(); + loop { + tokio::select! { + // Stop requested (or the handle dropped, closing the channel). + _ = stop_rx.recv() => break, + accepted = listener.accept() => { + match accepted { + Ok(conn) => { + let server = Arc::clone(&server); + let expected_project = project_id.clone(); + serves.spawn(async move { + serve_peer(server, &expected_project, conn).await; + }); + } + // A transient accept error must not kill the loop; the + // listener stays bound and keeps accepting the next peer. + Err(_) => continue, + } + } + } + } + // Stop: terminate every in-flight serve, then drop the listener (unlinks + // the Unix socket file). Pending peers die with their owning CLI anyway. + serves.abort_all(); + drop(serves); + drop(listener); }); - Self { stop: stop_tx } + Self { + stop: stop_tx, + endpoint, + } + } + + /// The loopback endpoint this project's server is bound to (M5a source of truth). + #[must_use] + pub fn endpoint(&self) -> &McpEndpoint { + &self.endpoint } /// Signals the supervision task to stop (best-effort; dropping the handle also @@ -1015,6 +1198,788 @@ pub(crate) fn build_memory_recall( } } +#[cfg(test)] +mod mcp_serve_peer_tests { + //! M5c — server side of the bind transport: [`serve_peer`] + [`parse_handshake`]. + //! + //! These drive the **private** `serve_peer` free function over an in-memory + //! `tokio::io::duplex` (no socket, no child process), the test seam the prod + //! code was made generic for (`serve_peer` over any `AsyncRead+AsyncWrite`). + //! The `OrchestratorService` is wired over the **same** in-memory fakes the + //! infrastructure MCP tests use (`infrastructure/tests/mcp_server.rs`), so MCP + //! behaviour is asserted against a real service with zero I/O. + //! + //! GARDE-FOU : every `await` that could block on a peer that never speaks is + //! bounded by `tokio::time::timeout`; a hung accept/serve fails fast instead of + //! hanging the suite. + + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use async_trait::async_trait; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use application::{ + CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, + OrchestratorService, TerminalSessions, UpdateAgentContext, + }; + use domain::agent::{AgentManifest, ManifestEntry}; + use domain::events::{DomainEvent, OrchestrationSource}; + use domain::ids::{AgentId, ProfileId, ProjectId, SkillId}; + use domain::markdown::MarkdownDoc; + use domain::ports::{ + AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream, + ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, ProfileStore, + PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec, + StoreError, + }; + use domain::profile::{AgentProfile, ContextInjection}; + use domain::project::{Project, ProjectPath}; + use domain::remote::RemoteRef; + use domain::skill::{Skill, SkillScope}; + use domain::{PtySize, SessionId}; + use serde_json::{json, Value}; + use uuid::Uuid; + + use super::serve_peer; + use infrastructure::McpServer; + + /// Test timeout for any single peer interaction. Generous but finite: a correct + /// duplex round-trip is sub-millisecond, so this only ever fires on a real hang. + const TIMEOUT: Duration = Duration::from_secs(5); + + // ----------------------------------------------------------------------- + // Fakes — mirrored from `infrastructure/tests/mcp_server.rs` (the established + // MCP harness), trimmed to exactly what `OrchestratorService::new` needs. + // ----------------------------------------------------------------------- + + #[derive(Default)] + struct ContextsInner { + manifest: AgentManifest, + contents: HashMap, + } + #[derive(Clone)] + struct FakeContexts(Arc>); + impl FakeContexts { + fn new() -> Self { + Self(Arc::new(Mutex::new(ContextsInner { + manifest: AgentManifest { + version: 1, + entries: Vec::new(), + }, + contents: HashMap::new(), + }))) + } + fn seed_agent(&self, name: &str) -> AgentId { + let id = AgentId::from_uuid(Uuid::new_v4()); + let mut inner = self.0.lock().unwrap(); + inner.manifest.entries.push(ManifestEntry { + agent_id: id, + name: name.to_owned(), + md_path: format!("agents/{name}.md"), + profile_id: ProfileId::from_uuid(Uuid::from_u128(9)), + template_id: None, + synchronized: false, + synced_template_version: None, + skills: Vec::new(), + }); + id + } + fn md_path_of(&self, agent: &AgentId) -> Option { + self.0 + .lock() + .unwrap() + .manifest + .entries + .iter() + .find(|e| &e.agent_id == agent) + .map(|e| e.md_path.clone()) + } + } + #[async_trait] + impl AgentContextStore for FakeContexts { + async fn read_context( + &self, + _project: &Project, + agent: &AgentId, + ) -> Result { + let md = self.md_path_of(agent).ok_or(StoreError::NotFound)?; + Ok(MarkdownDoc::new( + self.0 + .lock() + .unwrap() + .contents + .get(&md) + .cloned() + .unwrap_or_default(), + )) + } + async fn write_context( + &self, + _project: &Project, + agent: &AgentId, + md: &MarkdownDoc, + ) -> Result<(), StoreError> { + let path = self.md_path_of(agent).ok_or(StoreError::NotFound)?; + self.0 + .lock() + .unwrap() + .contents + .insert(path, md.as_str().to_owned()); + Ok(()) + } + async fn load_manifest(&self, _project: &Project) -> Result { + Ok(self.0.lock().unwrap().manifest.clone()) + } + async fn save_manifest( + &self, + _project: &Project, + manifest: &AgentManifest, + ) -> Result<(), StoreError> { + self.0.lock().unwrap().manifest = manifest.clone(); + Ok(()) + } + } + + #[derive(Clone)] + struct FakeProfiles(Arc>); + #[async_trait] + impl ProfileStore for FakeProfiles { + async fn list(&self) -> Result, StoreError> { + Ok((*self.0).clone()) + } + async fn save(&self, _p: &AgentProfile) -> Result<(), StoreError> { + Ok(()) + } + async fn delete(&self, _id: ProfileId) -> Result<(), StoreError> { + Ok(()) + } + async fn is_configured(&self) -> Result { + Ok(true) + } + async fn mark_configured(&self) -> Result<(), StoreError> { + Ok(()) + } + } + + #[derive(Default)] + struct FakeSkills; + #[async_trait] + impl SkillStore for FakeSkills { + async fn list( + &self, + _scope: SkillScope, + _root: &ProjectPath, + ) -> Result, StoreError> { + Ok(Vec::new()) + } + async fn get( + &self, + _scope: SkillScope, + _root: &ProjectPath, + _id: SkillId, + ) -> Result { + Err(StoreError::NotFound) + } + async fn save(&self, _skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> { + Ok(()) + } + async fn delete( + &self, + _scope: SkillScope, + _root: &ProjectPath, + _id: SkillId, + ) -> Result<(), StoreError> { + Ok(()) + } + } + + #[derive(Default)] + struct FakeRecall; + #[async_trait] + impl domain::ports::MemoryRecall for FakeRecall { + async fn recall( + &self, + _root: &ProjectPath, + _query: &domain::ports::MemoryQuery, + ) -> Result, domain::ports::MemoryError> { + Ok(Vec::new()) + } + } + + struct FakeRuntime; + impl AgentRuntime for FakeRuntime { + fn detect(&self, _p: &AgentProfile) -> Result { + Ok(true) + } + fn prepare_invocation( + &self, + profile: &AgentProfile, + _ctx: &PreparedContext, + cwd: &ProjectPath, + _session: &SessionPlan, + ) -> Result { + Ok(SpawnSpec { + command: profile.command.clone(), + args: profile.args.clone(), + cwd: cwd.clone(), + env: Vec::new(), + context_plan: Some(ContextInjectionPlan::Stdin), + }) + } + } + + #[derive(Clone, Default)] + struct FakeFs; + #[async_trait] + impl FileSystem for FakeFs { + async fn read(&self, p: &RemotePath) -> Result, FsError> { + Err(FsError::NotFound(p.as_str().to_owned())) + } + async fn write(&self, _p: &RemotePath, _d: &[u8]) -> Result<(), FsError> { + Ok(()) + } + async fn exists(&self, _p: &RemotePath) -> Result { + Ok(false) + } + async fn create_dir_all(&self, _p: &RemotePath) -> Result<(), FsError> { + Ok(()) + } + async fn list(&self, _p: &RemotePath) -> Result, FsError> { + Ok(Vec::new()) + } + async fn symlink(&self, _s: &RemotePath, _d: &RemotePath) -> Result<(), FsError> { + Ok(()) + } + } + + #[derive(Clone)] + struct FakePty; + #[async_trait] + impl PtyPort for FakePty { + async fn spawn(&self, _s: SpawnSpec, _z: PtySize) -> Result { + Ok(PtyHandle { + session_id: SessionId::from_uuid(Uuid::from_u128(777)), + }) + } + fn write(&self, _h: &PtyHandle, _d: &[u8]) -> Result<(), PtyError> { + Ok(()) + } + fn resize(&self, _h: &PtyHandle, _z: PtySize) -> Result<(), PtyError> { + Ok(()) + } + fn subscribe_output(&self, _h: &PtyHandle) -> Result { + Ok(Box::new(std::iter::empty())) + } + fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Vec::new()) + } + async fn kill(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + } + + #[derive(Default, Clone)] + struct NoopBus; + impl EventBus for NoopBus { + fn publish(&self, _e: DomainEvent) {} + fn subscribe(&self) -> EventStream { + Box::new(std::iter::empty()) + } + } + + struct SeqIds(Mutex); + impl IdGenerator for SeqIds { + fn new_uuid(&self) -> Uuid { + let mut n = self.0.lock().unwrap(); + let id = Uuid::from_u128(*n); + *n += 1; + id + } + } + + fn project() -> Project { + Project::new( + ProjectId::from_uuid(Uuid::from_u128(1000)), + "demo", + ProjectPath::new("/home/me/proj").unwrap(), + RemoteRef::local(), + 1_700_000_000_000, + ) + .unwrap() + } + + /// The hyphen-free hex project-id the handshake guard compares against — the + /// exact form `ensure_mcp_server` derives and M5d's `--project` reuses. + fn project_id_arg(p: &Project) -> String { + p.id.as_uuid().simple().to_string() + } + + /// A capturing event sink (the MCP twin of the file watcher's publish closure): + /// records every [`DomainEvent`] so a test can assert `requester_id`. + fn capturing_events() -> ( + Arc, + Arc>>, + ) { + let captured = Arc::new(Mutex::new(Vec::new())); + let sink = captured.clone(); + let publish: Arc = + Arc::new(move |e: DomainEvent| sink.lock().unwrap().push(e)); + (publish, captured) + } + + /// Builds an `OrchestratorService` over the in-memory fakes (no structured + /// registry), mirroring `infrastructure/tests/mcp_server.rs::build_service`. + fn build_service(contexts: FakeContexts) -> Arc { + let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new( + ProfileId::from_uuid(Uuid::from_u128(9)), + "Claude Code", + "claude", + Vec::new(), + ContextInjection::stdin(), + None, + "{agentRunDir}", + None, + ) + .unwrap()]))); + let sessions = Arc::new(TerminalSessions::new()); + let bus = Arc::new(NoopBus); + let create = Arc::new(CreateAgentFromScratch::new( + Arc::new(contexts.clone()), + Arc::new(SeqIds(Mutex::new(1))), + bus.clone(), + )); + let launch = Arc::new(LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(FakePty), + Arc::new(FakeSkills), + Arc::clone(&sessions), + bus.clone(), + Arc::new(SeqIds(Mutex::new(1))), + Arc::new(FakeRecall), + None, + )); + let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); + let close = Arc::new(CloseTerminal::new(Arc::new(FakePty), Arc::clone(&sessions))); + let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts))); + let create_skill = Arc::new(CreateSkill::new( + Arc::new(FakeSkills) as Arc, + Arc::new(SeqIds(Mutex::new(1))), + )); + Arc::new(OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + )) + } + + // --- duplex client helpers --------------------------------------------- + + /// Frames a `tools/call` request line (newline-terminated, as the transport + /// expects per `StdioTransport::recv`). + fn tools_call_line(id: i64, tool: &str, arguments: Value) -> String { + let mut s = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": "tools/call", + "params": { "name": tool, "arguments": arguments } + })) + .unwrap(); + s.push('\n'); + s + } + + /// A `tools/list` request line. + fn tools_list_line(id: i64) -> String { + let mut s = serde_json::to_string(&json!({ + "jsonrpc": "2.0", "id": id, "method": "tools/list" + })) + .unwrap(); + s.push('\n'); + s + } + + /// A handshake line `{"project":..,"requester":..}` followed by `\n`. + fn handshake_line(project: &str, requester: &str) -> String { + format!("{{\"project\":\"{project}\",\"requester\":\"{requester}\"}}\n") + } + + /// Reads exactly one newline-delimited JSON-RPC response off the client side of + /// the duplex, bounded by [`TIMEOUT`]. Returns `None` on EOF/timeout (the peer + /// closed without replying — used by the project-guard test). + async fn read_one_response(client: &mut R) -> Option + where + R: tokio::io::AsyncRead + Unpin, + { + let mut buf = Vec::new(); + let mut byte = [0u8; 1]; + loop { + match tokio::time::timeout(TIMEOUT, client.read(&mut byte)).await { + Ok(Ok(0)) => return None, // EOF before a full line + Ok(Ok(_)) => { + if byte[0] == b'\n' { + break; + } + buf.push(byte[0]); + } + Ok(Err(_)) => return None, + Err(_) => return None, // GARDE-FOU: timed out waiting for a reply + } + } + serde_json::from_slice(&buf).ok() + } + + /// Spawns `serve_peer` over the server half of a fresh duplex and returns the + /// client half plus the join handle. The peer is bounded by the test's reads. + fn spawn_peer( + server: Arc, + expected_project: String, + ) -> (tokio::io::DuplexStream, tokio::task::JoinHandle<()>) { + let (client, server_side) = tokio::io::duplex(64 * 1024); + let handle = tokio::spawn(async move { + serve_peer(server, &expected_project, server_side).await; + }); + (client, handle) + } + + // ----------------------------------------------------------------------- + // 1. Handshake + tools/list end-to-end over the duplex. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn handshake_then_tools_list_round_trips_over_duplex() { + let proj = project(); + let service = build_service(FakeContexts::new()); + let server = Arc::new(McpServer::new(service, proj.clone())); + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + // Write the handshake line, then a tools/list request. + client + .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) + .await + .unwrap(); + client.write_all(tools_list_line(1).as_bytes()).await.unwrap(); + client.flush().await.unwrap(); + + let resp = tokio::time::timeout(TIMEOUT, read_one_response(&mut client)) + .await + .expect("GARDE-FOU: tools/list timed out") + .expect("a tools/list response line"); + + assert_eq!(resp["id"], json!(1)); + let tools = resp["result"]["tools"].as_array().expect("tools array"); + let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect(); + for expected in [ + "idea_list_agents", + "idea_ask_agent", + "idea_launch_agent", + "idea_stop_agent", + "idea_update_context", + "idea_create_skill", + ] { + assert!(names.contains(&expected), "missing tool {expected}; got {names:?}"); + } + assert_eq!(tools.len(), 6, "exactly the six idea_* tools; got {names:?}"); + + drop(client); // EOF ⇒ serve loop ends + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer task did not finish") + .unwrap(); + } + + // ----------------------------------------------------------------------- + // 2. The handshake's requester is propagated to the processed event. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn handshake_requester_propagates_to_processed_event() { + let proj = project(); + let contexts = FakeContexts::new(); + contexts.seed_agent("architect"); + let service = build_service(contexts); + let (publish, captured) = capturing_events(); + let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + client + .write_all(handshake_line(&project_id_arg(&proj), "agent-42").as_bytes()) + .await + .unwrap(); + client + .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) + .await + .unwrap(); + client.flush().await.unwrap(); + + let resp = read_one_response(&mut client) + .await + .expect("a tools/call response"); + assert_eq!(resp["result"]["isError"], json!(false), "got {resp}"); + + drop(client); + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer did not finish") + .unwrap(); + + let events = captured.lock().unwrap(); + let processed: Vec<&DomainEvent> = events + .iter() + .filter(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })) + .collect(); + assert_eq!(processed.len(), 1, "exactly one processed event; got {events:?}"); + match processed[0] { + DomainEvent::OrchestratorRequestProcessed { + requester_id, + action, + source, + .. + } => { + assert_eq!( + requester_id, "agent-42", + "the real handshake requester must be propagated (not 'mcp')" + ); + assert_eq!(action, "idea_list_agents"); + assert_eq!(*source, OrchestrationSource::Mcp); + } + other => panic!("expected OrchestratorRequestProcessed, got {other:?}"), + } + } + + // ----------------------------------------------------------------------- + // 3. Empty requester in the handshake ⇒ legacy "mcp" label (back-compat). + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn empty_requester_handshake_falls_back_to_legacy_mcp_label() { + let proj = project(); + let contexts = FakeContexts::new(); + contexts.seed_agent("architect"); + let service = build_service(contexts); + let (publish, captured) = capturing_events(); + let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); + // Empty requester ("") in the handshake. + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + client + .write_all(handshake_line(&project_id_arg(&proj), "").as_bytes()) + .await + .unwrap(); + client + .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) + .await + .unwrap(); + client.flush().await.unwrap(); + + let _ = read_one_response(&mut client).await.expect("a response"); + drop(client); + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer did not finish") + .unwrap(); + + let events = captured.lock().unwrap(); + let processed = events + .iter() + .find_map(|e| match e { + DomainEvent::OrchestratorRequestProcessed { requester_id, .. } => { + Some(requester_id.clone()) + } + _ => None, + }) + .expect("a processed event"); + assert_eq!( + processed, "mcp", + "empty handshake requester must keep the legacy 'mcp' label" + ); + } + + // ----------------------------------------------------------------------- + // 4. JSON-RPC bytes glued onto the handshake's write are not lost + // (proves `StdioTransport::from_buffered` preserves buffered bytes). + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn jsonrpc_request_glued_to_handshake_is_served() { + let proj = project(); + let service = build_service(FakeContexts::new()); + let server = Arc::new(McpServer::new(service, proj.clone())); + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + // ONE write carrying the handshake line AND the tools/list line, back to back. + let mut glued = handshake_line(&project_id_arg(&proj), "agent-1"); + glued.push_str(&tools_list_line(7)); + client.write_all(glued.as_bytes()).await.unwrap(); + client.flush().await.unwrap(); + + let resp = read_one_response(&mut client) + .await + .expect("the glued tools/list must still be served"); + assert_eq!(resp["id"], json!(7), "the buffered request id must come through"); + assert!( + resp["result"]["tools"].is_array(), + "buffered request produced a real tools/list result; got {resp}" + ); + + drop(client); + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer did not finish") + .unwrap(); + } + + // ----------------------------------------------------------------------- + // 5a. Project guard: a mismatching handshake project ⇒ closed without serving. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn mismatched_project_handshake_is_closed_without_serving() { + let proj = project(); + let contexts = FakeContexts::new(); + contexts.seed_agent("architect"); + let service = build_service(contexts); + let (publish, captured) = capturing_events(); + let server = Arc::new(McpServer::new(service, proj.clone()).with_events(publish)); + // serve_peer is told to expect this project's id... + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + // ...but the handshake claims a DIFFERENT project. + client + .write_all(handshake_line("ffffffffffffffffffffffffffffffff", "intruder").as_bytes()) + .await + .unwrap(); + client + .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) + .await + .unwrap(); + client.flush().await.unwrap(); + + // No reply must come: the peer returned before serving. read returns EOF. + let resp = read_one_response(&mut client).await; + assert!( + resp.is_none(), + "mismatched project must be closed WITHOUT a response; got {resp:?}" + ); + + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer did not finish") + .unwrap(); + + // And nothing was dispatched ⇒ no processed event. + let events = captured.lock().unwrap(); + assert!( + !events + .iter() + .any(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })), + "a rejected peer must not dispatch anything; got {events:?}" + ); + } + + // ----------------------------------------------------------------------- + // 5b. Empty handshake project ⇒ served normally (guard does not reject). + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn empty_handshake_project_is_served_normally() { + let proj = project(); + let service = build_service(FakeContexts::new()); + let server = Arc::new(McpServer::new(service, proj.clone())); + let (mut client, peer) = spawn_peer(server, project_id_arg(&proj)); + + // Empty project in the handshake — the guard must NOT reject it. + client + .write_all(handshake_line("", "agent-1").as_bytes()) + .await + .unwrap(); + client.write_all(tools_list_line(1).as_bytes()).await.unwrap(); + client.flush().await.unwrap(); + + let resp = read_one_response(&mut client) + .await + .expect("empty-project handshake must still be served"); + assert!(resp["result"]["tools"].is_array(), "got {resp}"); + + drop(client); + tokio::time::timeout(TIMEOUT, peer) + .await + .expect("GARDE-FOU: peer did not finish") + .unwrap(); + } + + // ----------------------------------------------------------------------- + // 6. Peer isolation: one peer closing/erroring does not stop another. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn one_peer_failure_does_not_affect_a_concurrent_peer() { + let proj = project(); + let service = build_service(FakeContexts::new()); + let server = Arc::new(McpServer::new(service, proj.clone())); + + // Peer A: a broken peer that immediately closes after a partial handshake + // (no newline) — its serve_peer must end without crashing the runtime. + let (mut client_a, peer_a) = spawn_peer(Arc::clone(&server), project_id_arg(&proj)); + client_a.write_all(b"{\"project\":").await.unwrap(); // partial, no newline + client_a.flush().await.unwrap(); + drop(client_a); // abrupt close mid-handshake + + // Peer B: a healthy peer, served concurrently, must still get its reply. + let (mut client_b, peer_b) = spawn_peer(Arc::clone(&server), project_id_arg(&proj)); + client_b + .write_all(handshake_line(&project_id_arg(&proj), "agent-b").as_bytes()) + .await + .unwrap(); + client_b.write_all(tools_list_line(2).as_bytes()).await.unwrap(); + client_b.flush().await.unwrap(); + + let resp = read_one_response(&mut client_b) + .await + .expect("healthy peer B must be served despite peer A failing"); + assert_eq!(resp["id"], json!(2)); + assert!(resp["result"]["tools"].is_array(), "got {resp}"); + + drop(client_b); + // Both peer tasks must terminate cleanly within the bound. + tokio::time::timeout(TIMEOUT, peer_a) + .await + .expect("GARDE-FOU: peer A did not finish") + .unwrap(); + tokio::time::timeout(TIMEOUT, peer_b) + .await + .expect("GARDE-FOU: peer B did not finish") + .unwrap(); + } + + // ----------------------------------------------------------------------- + // Bonus — parse_handshake unit behaviour (tolerant parsing contract). + // ----------------------------------------------------------------------- + + #[test] + fn parse_handshake_is_tolerant() { + use super::parse_handshake; + // Well-formed. + assert_eq!( + parse_handshake("{\"project\":\"p1\",\"requester\":\"a1\"}\n"), + ("p1".to_owned(), "a1".to_owned()) + ); + // Malformed JSON ⇒ empty strings, never a panic. + assert_eq!(parse_handshake("{not json"), (String::new(), String::new())); + // Missing fields ⇒ empty strings. + assert_eq!(parse_handshake("{}"), (String::new(), String::new())); + } +} + #[cfg(test)] mod tests { //! Wiring contract for [`build_memory_recall`] (Pièce 1, §14.5.5). @@ -1142,3 +2107,887 @@ mod tests { } } } + +// =========================================================================== +// M5e — Smoke end-to-end over the REAL loopback (no real CLI, no network). +// =========================================================================== + +#[cfg(test)] +mod mcp_e2e_loopback_tests { + //! M5e — **smoke end-to-end** of the bind transport over a **real** + //! `interprocess` loopback (cadrage v5 §6 row M5e, §2 contract). + //! + //! ## Chosen e2e level — the **real accept loop**, justified + //! + //! Unlike M5c (which drove the *private* `serve_peer` over an in-memory + //! `tokio::io::duplex`), M5e proves the chain across an **actual** local socket. + //! Two options were on the table (cadrage M5e): + //! + //! 1. the **full `McpServerHandle::start` accept loop** on a real + //! `mcp_endpoint`, dialed by a real `interprocess` client, or + //! 2. a single `serve_peer` over one accepted real socket. + //! + //! We pick **(1) the real accept loop**. It is the *most faithful* slice — it + //! exercises exactly the production path a launched CLI's `idea mcp-server` + //! bridge takes: `bind_endpoint` → `McpServerHandle::start` → `listener.accept()` + //! → `serve_peer` (handshake + project guard) → `StdioTransport` → + //! `McpServer::serve_as` → the **real `dispatch`** (over the same in-memory fakes + //! as M2/M5c). And it is **safely bounded**: the accept loop is already async and + //! parks on the absence of a peer, while the *client* side is a plain + //! request/response exchange we wrap in `tokio::time::timeout`. Option (2) would + //! skip the bind/accept/lifecycle wiring for no extra safety, since the risk + //! (a peer that never speaks) is bounded identically by the client-side timeout. + //! + //! No production test seam was added: `McpServerHandle::start`, `bind_endpoint` + //! and `mcp_endpoint` are all reachable from this in-crate module as-is. The + //! client uses the same `interprocess` `Stream` the real bridge uses. + //! + //! GARDE-FOU : the listener bind, every client connect, and every response read + //! is wrapped in `tokio::time::timeout`; a hung accept/serve/handshake fails the + //! test fast instead of hanging the suite. + + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use async_trait::async_trait; + use interprocess::local_socket::tokio::Stream as LocalSocketStream; + use interprocess::local_socket::traits::tokio::Stream as _; + use tokio::io::{AsyncWriteExt, BufReader, AsyncBufReadExt}; + use uuid::Uuid; + + use application::{ + CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, + OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext, + }; + use domain::agent::{AgentManifest, ManifestEntry}; + use domain::events::{DomainEvent, OrchestrationSource}; + use domain::ids::{AgentId, NodeId, ProfileId, ProjectId, SkillId}; + use domain::markdown::MarkdownDoc; + use domain::ports::{ + AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, ContextInjectionPlan, + DirEntry, EventBus, EventStream, ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, + PreparedContext, ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, ReplyEvent, + ReplyStream, RuntimeError, SessionPlan, SkillStore, SpawnSpec, StoreError, + }; + use domain::profile::{AgentProfile, ContextInjection}; + use domain::project::{Project, ProjectPath}; + use domain::remote::RemoteRef; + use domain::skill::{Skill, SkillScope}; + use domain::terminal::{SessionKind, TerminalSession}; + use domain::{PtySize, SessionId}; + use serde_json::{json, Value}; + + use super::{bind_endpoint, mcp_endpoint, McpServerHandle}; + use crate::mcp_endpoint::McpEndpoint; + use infrastructure::McpServer; + + /// Test timeout for any single loopback interaction. Generous but finite: a + /// correct round-trip is sub-millisecond, so this only ever fires on a real hang. + const TIMEOUT: Duration = Duration::from_secs(5); + + // ----------------------------------------------------------------------- + // Fakes — mirrored from `infrastructure/tests/mcp_server.rs` (the established + // MCP harness). Includes a `FakeSession` so `idea_ask_agent` resolves inline. + // ----------------------------------------------------------------------- + + #[derive(Default)] + struct ContextsInner { + manifest: AgentManifest, + contents: HashMap, + } + #[derive(Clone)] + struct FakeContexts(Arc>); + impl FakeContexts { + fn new() -> Self { + Self(Arc::new(Mutex::new(ContextsInner { + manifest: AgentManifest { + version: 1, + entries: Vec::new(), + }, + contents: HashMap::new(), + }))) + } + fn seed_agent(&self, name: &str) -> AgentId { + let id = AgentId::from_uuid(Uuid::new_v4()); + let mut inner = self.0.lock().unwrap(); + inner.manifest.entries.push(ManifestEntry { + agent_id: id, + name: name.to_owned(), + md_path: format!("agents/{name}.md"), + profile_id: ProfileId::from_uuid(Uuid::from_u128(9)), + template_id: None, + synchronized: false, + synced_template_version: None, + skills: Vec::new(), + }); + id + } + fn md_path_of(&self, agent: &AgentId) -> Option { + self.0 + .lock() + .unwrap() + .manifest + .entries + .iter() + .find(|e| &e.agent_id == agent) + .map(|e| e.md_path.clone()) + } + } + #[async_trait] + impl AgentContextStore for FakeContexts { + async fn read_context( + &self, + _project: &Project, + agent: &AgentId, + ) -> Result { + let md = self.md_path_of(agent).ok_or(StoreError::NotFound)?; + Ok(MarkdownDoc::new( + self.0 + .lock() + .unwrap() + .contents + .get(&md) + .cloned() + .unwrap_or_default(), + )) + } + async fn write_context( + &self, + _project: &Project, + agent: &AgentId, + md: &MarkdownDoc, + ) -> Result<(), StoreError> { + let path = self.md_path_of(agent).ok_or(StoreError::NotFound)?; + self.0 + .lock() + .unwrap() + .contents + .insert(path, md.as_str().to_owned()); + Ok(()) + } + async fn load_manifest(&self, _project: &Project) -> Result { + Ok(self.0.lock().unwrap().manifest.clone()) + } + async fn save_manifest( + &self, + _project: &Project, + manifest: &AgentManifest, + ) -> Result<(), StoreError> { + self.0.lock().unwrap().manifest = manifest.clone(); + Ok(()) + } + } + + #[derive(Clone)] + struct FakeProfiles(Arc>); + #[async_trait] + impl ProfileStore for FakeProfiles { + async fn list(&self) -> Result, StoreError> { + Ok((*self.0).clone()) + } + async fn save(&self, _p: &AgentProfile) -> Result<(), StoreError> { + Ok(()) + } + async fn delete(&self, _id: ProfileId) -> Result<(), StoreError> { + Ok(()) + } + async fn is_configured(&self) -> Result { + Ok(true) + } + async fn mark_configured(&self) -> Result<(), StoreError> { + Ok(()) + } + } + + #[derive(Default)] + struct FakeSkills; + #[async_trait] + impl SkillStore for FakeSkills { + async fn list( + &self, + _scope: SkillScope, + _root: &ProjectPath, + ) -> Result, StoreError> { + Ok(Vec::new()) + } + async fn get( + &self, + _scope: SkillScope, + _root: &ProjectPath, + _id: SkillId, + ) -> Result { + Err(StoreError::NotFound) + } + async fn save(&self, _skill: &Skill, _root: &ProjectPath) -> Result<(), StoreError> { + Ok(()) + } + async fn delete( + &self, + _scope: SkillScope, + _root: &ProjectPath, + _id: SkillId, + ) -> Result<(), StoreError> { + Ok(()) + } + } + + #[derive(Default)] + struct FakeRecall; + #[async_trait] + impl domain::ports::MemoryRecall for FakeRecall { + async fn recall( + &self, + _root: &ProjectPath, + _query: &domain::ports::MemoryQuery, + ) -> Result, domain::ports::MemoryError> { + Ok(Vec::new()) + } + } + + struct FakeRuntime; + impl AgentRuntime for FakeRuntime { + fn detect(&self, _p: &AgentProfile) -> Result { + Ok(true) + } + fn prepare_invocation( + &self, + profile: &AgentProfile, + _ctx: &PreparedContext, + cwd: &ProjectPath, + _session: &SessionPlan, + ) -> Result { + Ok(SpawnSpec { + command: profile.command.clone(), + args: profile.args.clone(), + cwd: cwd.clone(), + env: Vec::new(), + context_plan: Some(ContextInjectionPlan::Stdin), + }) + } + } + + #[derive(Clone, Default)] + struct FakeFs; + #[async_trait] + impl FileSystem for FakeFs { + async fn read(&self, p: &RemotePath) -> Result, FsError> { + Err(FsError::NotFound(p.as_str().to_owned())) + } + async fn write(&self, _p: &RemotePath, _d: &[u8]) -> Result<(), FsError> { + Ok(()) + } + async fn exists(&self, _p: &RemotePath) -> Result { + Ok(false) + } + async fn create_dir_all(&self, _p: &RemotePath) -> Result<(), FsError> { + Ok(()) + } + async fn list(&self, _p: &RemotePath) -> Result, FsError> { + Ok(Vec::new()) + } + async fn symlink(&self, _s: &RemotePath, _d: &RemotePath) -> Result<(), FsError> { + Ok(()) + } + } + + #[derive(Clone)] + struct FakePty; + #[async_trait] + impl PtyPort for FakePty { + async fn spawn(&self, _s: SpawnSpec, _z: PtySize) -> Result { + Ok(PtyHandle { + session_id: SessionId::from_uuid(Uuid::from_u128(777)), + }) + } + fn write(&self, _h: &PtyHandle, _d: &[u8]) -> Result<(), PtyError> { + Ok(()) + } + fn resize(&self, _h: &PtyHandle, _z: PtySize) -> Result<(), PtyError> { + Ok(()) + } + fn subscribe_output(&self, _h: &PtyHandle) -> Result { + Ok(Box::new(std::iter::empty())) + } + fn scrollback(&self, _h: &PtyHandle) -> Result, PtyError> { + Ok(Vec::new()) + } + async fn kill(&self, _h: &PtyHandle) -> Result { + Ok(ExitStatus { code: Some(0) }) + } + } + + /// A structured session whose turn deterministically ends on a `Final` carrying a + /// fixed reply — lets us exercise `idea_ask_agent` without a real CLI. + struct FakeSession { + id: SessionId, + reply: String, + } + #[async_trait] + impl AgentSession for FakeSession { + fn id(&self) -> SessionId { + self.id + } + fn conversation_id(&self) -> Option { + None + } + async fn send(&self, _prompt: &str) -> Result { + let events = vec![ + ReplyEvent::TextDelta { + text: "thinking…".to_owned(), + }, + ReplyEvent::Final { + content: self.reply.clone(), + }, + ]; + Ok(Box::new(events.into_iter())) + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } + } + + #[derive(Default, Clone)] + struct NoopBus; + impl EventBus for NoopBus { + fn publish(&self, _e: DomainEvent) {} + fn subscribe(&self) -> EventStream { + Box::new(std::iter::empty()) + } + } + + struct SeqIds(Mutex); + impl IdGenerator for SeqIds { + fn new_uuid(&self) -> Uuid { + let mut n = self.0.lock().unwrap(); + let id = Uuid::from_u128(*n); + *n += 1; + id + } + } + + fn project() -> Project { + // A fresh project id per call ⇒ a distinct endpoint per test (no socket-file + // collision when tests run in parallel). + Project::new( + ProjectId::from_uuid(Uuid::new_v4()), + "demo", + ProjectPath::new("/home/me/proj").unwrap(), + RemoteRef::local(), + 1_700_000_000_000, + ) + .unwrap() + } + + fn project_id_arg(p: &Project) -> String { + p.id.as_uuid().simple().to_string() + } + + fn capturing_events() -> ( + Arc, + Arc>>, + ) { + let captured = Arc::new(Mutex::new(Vec::new())); + let sink = captured.clone(); + let publish: Arc = + Arc::new(move |e: DomainEvent| sink.lock().unwrap().push(e)); + (publish, captured) + } + + /// Builds an `OrchestratorService` over the in-memory fakes, returning the + /// shared `TerminalSessions` (so a test can pre-bind a live **PTY** session for a + /// raw-PTY target) and `StructuredSessions` (so a test can pre-insert a replying + /// **structured** session for `idea_ask_agent`). Mirrors the established harness; + /// no use case is re-invented. + fn build_service( + contexts: FakeContexts, + ) -> ( + Arc, + Arc, + Arc, + ) { + let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new( + ProfileId::from_uuid(Uuid::from_u128(9)), + "Claude Code", + "claude", + Vec::new(), + ContextInjection::stdin(), + None, + "{agentRunDir}", + None, + ) + .unwrap()]))); + let sessions = Arc::new(TerminalSessions::new()); + let structured = Arc::new(StructuredSessions::new()); + let bus = Arc::new(NoopBus); + let create = Arc::new(CreateAgentFromScratch::new( + Arc::new(contexts.clone()), + Arc::new(SeqIds(Mutex::new(1))), + bus.clone(), + )); + let launch = Arc::new(LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(FakePty), + Arc::new(FakeSkills), + Arc::clone(&sessions), + bus.clone(), + Arc::new(SeqIds(Mutex::new(1))), + Arc::new(FakeRecall), + None, + )); + let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); + let close = Arc::new(CloseTerminal::new(Arc::new(FakePty), Arc::clone(&sessions))); + let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts))); + let create_skill = Arc::new(CreateSkill::new( + Arc::new(FakeSkills) as Arc, + Arc::new(SeqIds(Mutex::new(1))), + )); + let service = OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + ) + .with_structured(Arc::clone(&structured)); + (Arc::new(service), sessions, structured) + } + + // --- real loopback harness --------------------------------------------- + + /// Stands up the **real** production accept-and-serve path on a real + /// `interprocess` loopback for `project`: binds the endpoint via the production + /// `bind_endpoint`, starts `McpServerHandle::start` (the M5c accept loop), and + /// returns the live handle plus the endpoint to dial. + /// + /// Asserts the bind actually succeeded (a `None` listener would make the e2e + /// vacuous) — if the per-user runtime dir is unwritable the test fails loudly + /// rather than silently testing nothing. + fn start_real_server( + service: Arc, + project: &Project, + events: Option>, + ) -> (McpServerHandle, McpEndpoint) { + let endpoint = mcp_endpoint(&project.id); + let listener = bind_endpoint(&endpoint); + assert!( + listener.is_some(), + "M5e needs a real bound listener; bind_endpoint returned None for {:?}", + endpoint.as_cli_arg() + ); + let mut server = McpServer::new(service, project.clone()); + if let Some(publish) = events { + server = server.with_events(publish); + } + let handle = McpServerHandle::start( + server, + endpoint.clone(), + listener, + project_id_arg(project), + ); + (handle, endpoint) + } + + /// Connects a **real** `interprocess` client to `endpoint`, bounded by [`TIMEOUT`] + /// so an endpoint that never accepts fails fast (GARDE-FOU). + async fn connect_client(endpoint: &McpEndpoint) -> LocalSocketStream { + use interprocess::local_socket::{GenericFilePath, ToFsName as _}; + let name = endpoint + .as_cli_arg() + .to_fs_name::() + .expect("valid endpoint name"); + tokio::time::timeout(TIMEOUT, LocalSocketStream::connect(name)) + .await + .expect("GARDE-FOU: connect to endpoint timed out") + .expect("connect to a live endpoint") + } + + fn handshake_line(project: &str, requester: &str) -> String { + format!("{{\"project\":\"{project}\",\"requester\":\"{requester}\"}}\n") + } + + fn tools_call_line(id: i64, tool: &str, arguments: Value) -> String { + let mut s = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": "tools/call", + "params": { "name": tool, "arguments": arguments } + })) + .unwrap(); + s.push('\n'); + s + } + + /// Reads exactly one newline-delimited JSON-RPC response off the client side of + /// the real socket, bounded by [`TIMEOUT`]. `None` on EOF/timeout. + async fn read_one_response(reader: &mut R) -> Option + where + R: AsyncBufReadExt + Unpin, + { + let mut line = String::new(); + match tokio::time::timeout(TIMEOUT, reader.read_line(&mut line)).await { + Ok(Ok(0)) => None, // EOF before a full line + Ok(Ok(_)) => serde_json::from_str(line.trim_end()).ok(), + Ok(Err(_)) => None, + Err(_) => None, // GARDE-FOU: timed out waiting for a reply + } + } + + // ----------------------------------------------------------------------- + // 1. idea_list_agents e2e over the real loopback ⇒ JSON array of agents. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn list_agents_round_trips_over_real_loopback() { + let contexts = FakeContexts::new(); + contexts.seed_agent("architect"); + contexts.seed_agent("dev-backend"); + let (service, _pty, _structured) = build_service(contexts); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, None); + + let conn = connect_client(&endpoint).await; + let (read_half, mut write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + // Handshake (real project id) + a tools/call idea_list_agents. + write_half + .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) + .await + .unwrap(); + write_half + .write_all(tools_call_line(1, "idea_list_agents", json!({})).as_bytes()) + .await + .unwrap(); + write_half.flush().await.unwrap(); + + let resp = read_one_response(&mut reader) + .await + .expect("a list_agents response line"); + assert_eq!(resp["id"], json!(1)); + assert!(resp["error"].is_null(), "transport error: {resp}"); + let result = &resp["result"]; + assert_eq!(result["isError"], json!(false), "got {result}"); + + // The inline text is the JSON array of the project's agents. + let text = result["content"][0]["text"].as_str().expect("text block"); + let agents: Value = serde_json::from_str(text).expect("reply must be a JSON array"); + let arr = agents.as_array().expect("array"); + assert_eq!(arr.len(), 2, "two seeded agents expected; got {text}"); + let names: Vec<&str> = arr.iter().map(|a| a["name"].as_str().unwrap()).collect(); + assert!(names.contains(&"architect"), "got {names:?}"); + assert!(names.contains(&"dev-backend"), "got {names:?}"); + + drop(write_half); // EOF ⇒ serve loop ends cleanly + handle.stop(); + } + + // ----------------------------------------------------------------------- + // 2. idea_ask_agent e2e: structured target ⇒ reply returned INLINE. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn ask_structured_agent_returns_reply_inline_over_real_loopback() { + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("architect"); + let (service, _pty, structured) = build_service(contexts); + structured.insert( + Arc::new(FakeSession { + id: SessionId::from_uuid(Uuid::from_u128(4242)), + reply: "the answer is 42".to_owned(), + }), + agent_id, + NodeId::from_uuid(Uuid::from_u128(7)), + ); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, None); + + let conn = connect_client(&endpoint).await; + let (read_half, mut write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + write_half + .write_all(handshake_line(&project_id_arg(&proj), "agent-asker").as_bytes()) + .await + .unwrap(); + write_half + .write_all( + tools_call_line( + 7, + "idea_ask_agent", + json!({ "target": "architect", "task": "What is the answer?" }), + ) + .as_bytes(), + ) + .await + .unwrap(); + write_half.flush().await.unwrap(); + + let resp = read_one_response(&mut reader) + .await + .expect("an ask response line"); + assert_eq!(resp["id"], json!(7)); + assert!(resp["error"].is_null(), "transport error: {resp}"); + let result = &resp["result"]; + assert_eq!(result["isError"], json!(false), "got {result}"); + assert_eq!( + result["content"][0]["text"].as_str().expect("text block"), + "the answer is 42", + "ask reply must be returned inline over the real loopback; got {result}" + ); + + drop(write_half); + handle.stop(); + } + + // ----------------------------------------------------------------------- + // 3. idea_ask_agent against a raw-PTY target ⇒ typed tool error (isError:true), + // no panic, connection stays healthy for a follow-up call. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn ask_raw_pty_target_is_typed_error_over_real_loopback() { + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("dev"); + let (service, pty, _structured) = build_service(contexts); + // Pre-bind a live **PTY** session for the agent (no structured session): the + // ask path must reject it with AppError::Invalid ⇒ tool result isError:true. + let session_id = SessionId::from_uuid(Uuid::from_u128(555)); + pty.insert( + PtyHandle { session_id }, + TerminalSession::starting( + session_id, + NodeId::from_uuid(Uuid::from_u128(8)), + ProjectPath::new("/home/me/proj").unwrap(), + SessionKind::Agent { agent_id }, + PtySize { rows: 24, cols: 80 }, + ), + ); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, None); + + let conn = connect_client(&endpoint).await; + let (read_half, mut write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + write_half + .write_all(handshake_line(&project_id_arg(&proj), "agent-asker").as_bytes()) + .await + .unwrap(); + write_half + .write_all( + tools_call_line(3, "idea_ask_agent", json!({ "target": "dev", "task": "hi" })) + .as_bytes(), + ) + .await + .unwrap(); + write_half.flush().await.unwrap(); + + let resp = read_one_response(&mut reader) + .await + .expect("an ask response line"); + assert_eq!(resp["id"], json!(3)); + // Healthy connection: NO JSON-RPC transport error... + assert!( + resp["error"].is_null(), + "a raw-PTY target must be a tool error, not a transport error: {resp}" + ); + // ...but the tool result is flagged as an error. + let result = &resp["result"]; + assert_eq!(result["isError"], json!(true), "got {result}"); + assert!( + !result["content"][0]["text"] + .as_str() + .unwrap_or_default() + .is_empty(), + "error text should describe the failure; got {result}" + ); + + // The connection is still healthy: a follow-up tools/list still answers. + write_half + .write_all( + serde_json::to_string(&json!({ + "jsonrpc": "2.0", "id": 4, "method": "tools/list" + })) + .map(|mut s| { + s.push('\n'); + s + }) + .unwrap() + .as_bytes(), + ) + .await + .unwrap(); + write_half.flush().await.unwrap(); + let again = read_one_response(&mut reader) + .await + .expect("server still serves after a tool error"); + assert_eq!(again["id"], json!(4)); + assert!(again["result"]["tools"].is_array(), "got {again}"); + + drop(write_half); + handle.stop(); + } + + // ----------------------------------------------------------------------- + // 4. Malformed JSON-RPC after the handshake ⇒ JSON-RPC error, never a panic, + // the server stays alive for a subsequent valid call. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn malformed_jsonrpc_after_handshake_errors_and_server_survives() { + let contexts = FakeContexts::new(); + contexts.seed_agent("architect"); + let (service, _pty, _structured) = build_service(contexts); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, None); + + let conn = connect_client(&endpoint).await; + let (read_half, mut write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + write_half + .write_all(handshake_line(&project_id_arg(&proj), "agent-1").as_bytes()) + .await + .unwrap(); + // A malformed JSON-RPC line (not valid JSON) — must NOT crash the serve loop. + write_half + .write_all(b"{ this is not json\n") + .await + .unwrap(); + write_half.flush().await.unwrap(); + + let resp = read_one_response(&mut reader) + .await + .expect("a parse error still owes a response"); + let error = &resp["error"]; + assert!(!error.is_null(), "malformed input must yield a JSON-RPC error: {resp}"); + // JSON-RPC mandates a null id when the request could not be correlated. + assert_eq!(resp["id"], Value::Null, "got {resp}"); + assert!(resp["result"].is_null(), "no result on parse error; got {resp}"); + + // The server survived: a subsequent valid call still answers. + write_half + .write_all( + serde_json::to_string(&json!({ + "jsonrpc": "2.0", "id": 2, "method": "tools/list" + })) + .map(|mut s| { + s.push('\n'); + s + }) + .unwrap() + .as_bytes(), + ) + .await + .unwrap(); + write_half.flush().await.unwrap(); + let again = read_one_response(&mut reader) + .await + .expect("server must survive malformed input"); + assert_eq!(again["id"], json!(2)); + assert!(again["result"]["tools"].is_array(), "got {again}"); + + drop(write_half); + handle.stop(); + } + + // ----------------------------------------------------------------------- + // 5. Bonus — the handshake requester crosses the REAL loopback and tags the + // OrchestratorRequestProcessed event with the real agent id (not "mcp"). + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn handshake_requester_propagates_over_real_loopback() { + let contexts = FakeContexts::new(); + let (service, _pty, _structured) = build_service(contexts); + let (publish, captured) = capturing_events(); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, Some(publish)); + + let conn = connect_client(&endpoint).await; + let (read_half, mut write_half) = tokio::io::split(conn); + let mut reader = BufReader::new(read_half); + + // Handshake carries the real requesting agent id. + write_half + .write_all(handshake_line(&project_id_arg(&proj), "agent-42").as_bytes()) + .await + .unwrap(); + // A successful launch (creates + launches an agent from scratch) ⇒ a + // processed beacon is emitted, tagged with the handshake requester. + write_half + .write_all( + tools_call_line( + 1, + "idea_launch_agent", + json!({ "target": "dev-backend", "profile": "claude-code" }), + ) + .as_bytes(), + ) + .await + .unwrap(); + write_half.flush().await.unwrap(); + + let resp = read_one_response(&mut reader) + .await + .expect("a launch response line"); + assert!(resp["error"].is_null(), "transport error: {resp}"); + assert_eq!(resp["result"]["isError"], json!(false), "got {resp}"); + + // Drain the connection so the serve task has surely published the event. + drop(write_half); + // Give the spawned serve task a beat to flush its publish (bounded). + let _ = tokio::time::timeout(TIMEOUT, async { + loop { + if captured + .lock() + .unwrap() + .iter() + .any(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + + let events = captured.lock().unwrap(); + let processed: Vec<&DomainEvent> = events + .iter() + .filter(|e| matches!(e, DomainEvent::OrchestratorRequestProcessed { .. })) + .collect(); + assert_eq!( + processed.len(), + 1, + "expected exactly one processed event; got {events:?}" + ); + match processed[0] { + DomainEvent::OrchestratorRequestProcessed { + requester_id, + action, + ok, + source, + } => { + assert_eq!(*source, OrchestrationSource::Mcp, "MCP door must tag Mcp"); + assert_eq!(action, "idea_launch_agent"); + assert!(*ok, "the launch succeeded"); + assert_eq!( + requester_id, "agent-42", + "the real handshake requester must cross the loopback (not 'mcp')" + ); + } + other => panic!("expected OrchestratorRequestProcessed, got {other:?}"), + } + drop(events); + + handle.stop(); + } +} diff --git a/crates/app-tauri/tests/orchestrator_wiring.rs b/crates/app-tauri/tests/orchestrator_wiring.rs index f58dcf7..bbc871a 100644 --- a/crates/app-tauri/tests/orchestrator_wiring.rs +++ b/crates/app-tauri/tests/orchestrator_wiring.rs @@ -10,7 +10,9 @@ //! stopping unregisters. use std::path::PathBuf; +use std::time::Duration; +use app_tauri_lib::mcp_endpoint::mcp_endpoint; use app_tauri_lib::state::AppState; use domain::ports::IdGenerator; use domain::project::{Project, ProjectPath}; @@ -211,3 +213,120 @@ async fn mcp_servers_are_isolated_per_project() { assert!(has_mcp(&state, &b.id)); assert_eq!(mcp_count(&state), 1); } + +// --- M5a: per-project loopback MCP endpoint lifecycle --- +// +// `mcp_endpoint(project_id)` is the single source of truth for the loopback +// address (cadrage v5 §2). `ensure_mcp_server` binds it at open; dropping the +// handle on close unlinks it (Unix). On Unix the endpoint is a UDS *file* whose +// existence is directly observable; these tests assert bind → idempotence → +// cleanup → determinism/no-collision → coexistence with the file watcher. + +/// Polls until `cond()` holds or the bound elapses (cleanup is async: the handle's +/// supervision task drops the listener — and unlinks the socket — only after the +/// stop signal propagates). Bounded so a regression fails fast, never hangs. +#[cfg(unix)] +async fn wait_until(mut cond: impl FnMut() -> bool) -> bool { + for _ in 0..100 { + if cond() { + return true; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + cond() +} + +#[cfg(unix)] +fn socket_exists(project: &Project) -> bool { + mcp_endpoint(&project.id) + .socket_path() + .map(|p| p.exists()) + .unwrap_or(false) +} + +#[cfg(unix)] +#[tokio::test] +async fn open_binds_the_project_loopback_endpoint() { + let state = AppState::build(temp_path("appdata")); + let project = make_project(); + + assert!(!socket_exists(&project), "no socket before open"); + + state.ensure_orchestrator_watch(&project); + assert!( + wait_until(|| socket_exists(&project)).await, + "the project's loopback socket is bound on open" + ); + + state.stop_orchestrator_watch(&project.id); +} + +#[cfg(unix)] +#[tokio::test] +async fn double_open_keeps_a_single_endpoint_no_address_in_use() { + let state = AppState::build(temp_path("appdata")); + let project = make_project(); + + state.ensure_orchestrator_watch(&project); + assert!(wait_until(|| socket_exists(&project)).await); + + // A second open must NOT rebind (which would fail "address in use" on a live + // socket) — it returns early. One endpoint, still bound, no panic. + state.ensure_orchestrator_watch(&project); + assert_eq!(mcp_count(&state), 1, "one endpoint per project"); + assert!(socket_exists(&project), "endpoint still bound after re-open"); + + state.stop_orchestrator_watch(&project.id); +} + +#[cfg(unix)] +#[tokio::test] +async fn close_cleans_up_the_endpoint_socket_file() { + let state = AppState::build(temp_path("appdata")); + let project = make_project(); + + state.ensure_orchestrator_watch(&project); + assert!(wait_until(|| socket_exists(&project)).await); + + state.stop_orchestrator_watch(&project.id); + assert!( + wait_until(|| !socket_exists(&project)).await, + "the socket file is unlinked on close — no leak" + ); +} + +#[test] +fn endpoint_is_deterministic_and_collision_free_across_projects() { + let p1 = make_project(); + let p2 = make_project(); + + // Stable for the same project across calls. + assert_eq!(mcp_endpoint(&p1.id), mcp_endpoint(&p1.id)); + // Distinct projects ⇒ distinct endpoints (no collision). + assert_ne!(mcp_endpoint(&p1.id), mcp_endpoint(&p2.id)); + assert_ne!( + mcp_endpoint(&p1.id).as_cli_arg(), + mcp_endpoint(&p2.id).as_cli_arg() + ); +} + +#[cfg(unix)] +#[tokio::test] +async fn file_watcher_and_loopback_endpoint_live_together() { + let state = AppState::build(temp_path("appdata")); + let project = make_project(); + + state.ensure_orchestrator_watch(&project); + + // R0/M3 invariant intact: the file watcher and the MCP server are both live... + assert!(has_watcher(&state, &project.id), "watcher live"); + assert!(has_mcp(&state, &project.id), "mcp server live"); + // ...and on Unix the loopback endpoint is actually bound beside the watcher. + assert!( + wait_until(|| socket_exists(&project)).await, + "endpoint bound alongside the live file watcher" + ); + + state.stop_orchestrator_watch(&project.id); + assert!(wait_until(|| !socket_exists(&project)).await); +} diff --git a/crates/application/src/agent/lifecycle.rs b/crates/application/src/agent/lifecycle.rs index 2715d53..521df47 100644 --- a/crates/application/src/agent/lifecycle.rs +++ b/crates/application/src/agent/lifecycle.rs @@ -512,6 +512,11 @@ impl ChangeAgentProfile { // Conversation id discarded: the previous one belonged to the old // engine; the new profile starts (or assigns) a fresh one. conversation_id: None, + // Internal relaunch (no app-tauri composition root in scope) ⇒ the + // MCP runtime is not injected here; `apply_mcp_config` falls back to + // the minimal declaration. A profile-hot-swap that needs the real + // endpoint is re-driven through the app-tauri launch path. + mcp_runtime: None, }) .await?; Ok(Some(output.session)) @@ -637,6 +642,50 @@ pub struct LaunchAgentInput { /// when the profile supports it). The caller (which owns the layout) reads this /// from the leaf's [`domain::layout::LeafCell::conversation_id`]. pub conversation_id: Option, + /// Runtime facts needed to write the **real** IdeA MCP server declaration + /// (M5d). These are **OS/runtime data** (the IdeA executable path, the + /// project's loopback endpoint) that live in `app-tauri` — they are *injected + /// as data* from the composition root, never computed in `application` (which + /// must not depend on `app-tauri`, cadrage v5 §0.3 / §7). + /// + /// `None` ⇒ no runtime injected (launches issued from inside `application`: + /// the orchestrator's `spawn_agent`/`ask_agent`, a profile hot-swap relaunch, + /// or tests). In that case [`apply_mcp_config`](LaunchAgent::apply_mcp_config) + /// still honours the profile's `McpConfigStrategy`, but falls back to a + /// **coherent minimal** declaration (the `idea mcp-server` command without the + /// endpoint/project/requester args) rather than a project-bound one — see + /// [`mcp_server_declaration`]. + pub mcp_runtime: Option, +} + +/// OS/runtime facts injected by the composition root (`app-tauri`) to materialise +/// the **real** IdeA MCP server declaration in an agent's `.mcp.json` (cadrage v5 +/// §2). Carrying these as **plain data** on [`LaunchAgentInput`] keeps +/// `application` free of any `app-tauri` / `current_exe` / `mcp_endpoint` +/// dependency: the endpoint stays computed by the *single source of truth* +/// (`app-tauri::mcp_endpoint`) and only its **string** crosses the layer boundary. +/// +/// All four fields are the exact strings the spawned bridge expects on its command +/// line (` mcp-server --endpoint --project --requester +/// `); the `project_id` must already be in the **hyphen-free 32-hex +/// `simple` form** consumed by the M5c handshake guard (`serve_peer`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct McpRuntime { + /// Absolute path to the IdeA executable (`std::env::current_exe()`), used as the + /// declaration's `command` — the CLI spawns *this* binary in its `mcp-server` + /// bridge mode. + pub exe: String, + /// The project's loopback endpoint string (`mcp_endpoint(project).as_cli_arg()`), + /// passed as `--endpoint`. **Same source of truth** as the listener bound by + /// `ensure_mcp_server` (cadrage v5 §2 coherence invariant). + pub endpoint: String, + /// The project id in the **hyphen-free 32-hex `simple` form** consumed by the + /// M5c handshake guard, passed as `--project`. + pub project_id: String, + /// The launching agent's id, passed as `--requester` so the server tags + /// `OrchestratorRequestProcessed.requester_id` with the real agent (cadrage v5 + /// §1.4) instead of the frozen `"mcp"` placeholder. + pub requester: String, } /// Descripteur d'une session **structurée** (IA, cellule chat) démarrée par @@ -1042,7 +1091,8 @@ impl LaunchAgent { // IdeA matérialise SA config MCP au format de CETTE CLI, dans le **même** // run dir isolé que le convention file et le seed de permissions. `None` // ⇒ aucun write/flag/env (chemin actuel inchangé, zéro régression). - self.apply_mcp_config(&profile, &run_dir, &mut spec).await; + self.apply_mcp_config(&profile, &run_dir, input.mcp_runtime.as_ref(), &mut spec) + .await; // 5b. ── POINT DE ROUTAGE §17.4 : IA structuré vs terminal brut ── // Le convention file (CLAUDE.md / AGENTS.md) vient d'être écrit dans le @@ -1331,14 +1381,18 @@ impl LaunchAgent { /// `profile.mcp == None` ⇒ no-op: no write, no flag, no env (current path, zero /// regression). /// - /// Note (M1): the embedded server declaration (command + transport) is a coherent - /// **placeholder** — the real IdeA MCP server and its exact launch command land in - /// M2/M3. The strategy plumbing here is stable; only the declaration content will - /// be finalised then. + /// MCP runtime (M5d): when the composition root injects a [`McpRuntime`], the + /// declaration written for a `ConfigFile` strategy is the **real** one — it points + /// the spawned `idea mcp-server` bridge at the project's exact loopback endpoint + /// (same source of truth as `ensure_mcp_server`, cadrage v5 §2). When no runtime is + /// injected (launches issued from inside `application`), a coherent **minimal** + /// declaration is written instead (see [`mcp_server_declaration`]). `Flag`/`Env` + /// are unaffected by the runtime: they keep passing the run-dir config path. async fn apply_mcp_config( &self, profile: &AgentProfile, run_dir: &ProjectPath, + runtime: Option<&McpRuntime>, spec: &mut SpawnSpec, ) { let Some(mcp) = &profile.mcp else { @@ -1351,7 +1405,7 @@ impl LaunchAgent { match self.fs.exists(&path).await { Ok(true) => {} Ok(false) => { - let declaration = mcp_server_declaration(mcp.transport); + let declaration = mcp_server_declaration(mcp.transport, runtime); // Best-effort: a write failure must not fail the launch. let _ = self.fs.write(&path, declaration.as_bytes()).await; } @@ -1448,24 +1502,67 @@ fn reattach_decision( } /// Builds the IdeA MCP server declaration written into a CLI's `ConfigFile` MCP -/// config (cadrage v3, Décision 3). Coherent **placeholder** for M1: it describes a -/// stdio/socket server launched by the `idea` binary — the exact command and the -/// real server land in **M2/M3**. Kept pure (no I/O) so it is unit-testable. +/// config (cadrage v3 D3 ; cadrage v5 §2). Kept pure (no I/O) so it is unit-testable. +/// +/// Two shapes, by whether the composition root injected a [`McpRuntime`]: +/// +/// - **`Some(runtime)` — the real declaration (M5d, end of the placeholder)**: +/// `command = runtime.exe` (the IdeA executable), and `args` carry the +/// `mcp-server` subcommand plus the project's exact loopback `--endpoint`, its +/// `--project` (hyphen-free 32-hex form, consumed by the M5c guard) and the +/// launching agent as `--requester`. The endpoint string comes verbatim from the +/// **single source of truth** (`app-tauri::mcp_endpoint`), so the bridge connects +/// to exactly what `ensure_mcp_server` listens on. +/// +/// - **`None` — coherent minimal declaration**: launches issued from inside +/// `application` (orchestrator/hot-swap/tests) have no OS/runtime facts to inject; +/// rather than fabricate a project-bound endpoint, we write the `idea mcp-server` +/// command without the endpoint/project/requester args. It stays a valid, +/// self-consistent `mcpServers/idea` entry (the bridge would resolve the endpoint +/// from its own project context), surfacing `transport` for the future socket path. #[must_use] -fn mcp_server_declaration(transport: domain::profile::McpTransport) -> String { +fn mcp_server_declaration( + transport: domain::profile::McpTransport, + runtime: Option<&McpRuntime>, +) -> String { // Common `.mcp.json`-style shape (Claude Code et CLIs apparentées). The transport - // is surfaced so a socket-based CLI can be wired in M2 without changing M1's seam. + // is surfaced so a socket-based CLI can be wired later without changing this seam. let transport_label = match transport { domain::profile::McpTransport::Stdio => "stdio", domain::profile::McpTransport::Socket => "socket", }; + + // `command` + extra args depend on whether OS/runtime facts were injected. + // Each `args` entry is emitted as an escaped JSON string so an exe path or + // endpoint with spaces/backslashes/quotes stays valid JSON. + let (command, extra_args) = match runtime { + Some(rt) => { + let arg = |label: &str, value: &str| { + format!( + ",\n {},\n {}", + json_string(label), + json_string(value) + ) + }; + let extra = format!( + "{}{}{}", + arg("--endpoint", &rt.endpoint), + arg("--project", &rt.project_id), + arg("--requester", &rt.requester), + ); + (rt.exe.as_str(), extra) + } + None => ("idea", String::new()), + }; + let command = json_string(command); + format!( r#"{{ "mcpServers": {{ "idea": {{ - "command": "idea", + "command": {command}, "args": [ - "mcp-server" + "mcp-server"{extra_args} ], "transport": "{transport_label}" }} @@ -1475,6 +1572,13 @@ fn mcp_server_declaration(transport: domain::profile::McpTransport) -> String { ) } +/// Wraps a string as a JSON string literal (quotes + escaping), reusing the path +/// escaper so an exe path / endpoint with spaces, backslashes or quotes stays valid +/// JSON in the generated `.mcp.json`. +fn json_string(s: &str) -> String { + format!("\"{}\"", json_escape(s)) +} + /// Builds an absolute path string by joining a [`ProjectPath`] with a relative /// segment using a POSIX separator. fn join(base: &ProjectPath, rel: &str) -> String { diff --git a/crates/application/src/agent/mod.rs b/crates/application/src/agent/mod.rs index 16973ac..beb8419 100644 --- a/crates/application/src/agent/mod.rs +++ b/crates/application/src/agent/mod.rs @@ -26,7 +26,7 @@ pub use resume::{ pub use lifecycle::{ ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, CreateAgentFromScratch, CreateAgentInput, CreateAgentOutput, DeleteAgent, DeleteAgentInput, LaunchAgent, - LaunchAgentInput, LaunchAgentOutput, ListAgents, ListAgentsInput, ListAgentsOutput, + LaunchAgentInput, LaunchAgentOutput, ListAgents, ListAgentsInput, ListAgentsOutput, McpRuntime, ReadAgentContext, ReadAgentContextInput, ReadAgentContextOutput, StructuredSessionDescriptor, UpdateAgentContext, UpdateAgentContextInput, AGENT_MEMORY_RECALL_BUDGET, diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 24a258f..a1783c7 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -34,7 +34,7 @@ pub use agent::{ DetectProfilesInput, DetectProfilesOutput, FirstRunState, FirstRunStateOutput, InspectConversation, InspectConversationInput, InspectConversationOutput, LaunchAgent, LaunchAgentInput, LaunchAgentOutput, ListAgents, ListAgentsInput, ListAgentsOutput, - ListProfiles, ListProfilesOutput, ListResumableAgents, ListResumableAgentsInput, + ListProfiles, ListProfilesOutput, ListResumableAgents, ListResumableAgentsInput, McpRuntime, ListResumableAgentsOutput, ProfileAvailability, ReadAgentContext, ReadAgentContextInput, ReadAgentContextOutput, ReferenceProfiles, ReferenceProfilesOutput, ResumableAgent, SaveProfile, SaveProfileInput, SaveProfileOutput, StructuredSessionDescriptor, UpdateAgentContext, diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index f2850e4..d902a63 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -308,6 +308,10 @@ impl OrchestratorService { cols: DEFAULT_COLS, node_id, conversation_id: None, + // Orchestrator-driven launch (inside `application`): no OS/runtime + // facts to inject here; the real MCP declaration is written when the + // agent is (re)launched through the app-tauri composition root. + mcp_runtime: None, }) .await?; @@ -413,6 +417,9 @@ impl OrchestratorService { cols: DEFAULT_COLS, node_id: None, conversation_id: None, + // ask_agent revival launch (inside `application`): MCP runtime not + // injected here (see the `spawn_agent` note above). + mcp_runtime: None, }) .await?; diff --git a/crates/application/tests/agent_lifecycle.rs b/crates/application/tests/agent_lifecycle.rs index 431bd57..fcf37bc 100644 --- a/crates/application/tests/agent_lifecycle.rs +++ b/crates/application/tests/agent_lifecycle.rs @@ -812,6 +812,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput { cols: 80, node_id: None, conversation_id: None, + mcp_runtime: None, } } @@ -1726,6 +1727,258 @@ async fn launch_mcp_env_appends_var_and_path_to_env() { ); } +// --------------------------------------------------------------------------- +// LaunchAgent — MCP runtime declaration (M5d, cadrage v5 §2) +// --------------------------------------------------------------------------- + +/// A `LaunchAgentInput` carrying an injected [`McpRuntime`], otherwise identical +/// to [`launch_input`]. Lets the M5d tests drive the real declaration path while +/// reusing the M1 harness (`launch_fixture_with_profile`, `FakeFs` write trace). +fn launch_input_with_runtime( + agent_id: AgentId, + runtime: application::McpRuntime, +) -> LaunchAgentInput { + LaunchAgentInput { + mcp_runtime: Some(runtime), + ..launch_input(agent_id) + } +} + +/// Reads the single `.mcp.json` written by a launch, parsing it as JSON. Panics +/// with a clear message if zero or several were written, so an unexpected write +/// shape is surfaced rather than silently picked. +fn read_single_mcp_json(fs: &FakeFs) -> serde_json::Value { + let mcp = writes_ending_with(fs, "/.mcp.json"); + assert_eq!(mcp.len(), 1, "exactly one .mcp.json must be written"); + let body = String::from_utf8(mcp[0].1.clone()).unwrap(); + serde_json::from_str(&body) + .unwrap_or_else(|e| panic!("written .mcp.json must be valid JSON ({e}): {body}")) +} + +#[tokio::test] +async fn launch_mcp_runtime_writes_real_declaration_with_ordered_args() { + // M5d-1 — ConfigFile{".mcp.json"} + an injected McpRuntime ⇒ the written + // declaration is the REAL one: command == exe, and args == the ordered + // ["mcp-server","--endpoint",endpoint,"--project",project_id,"--requester", + // requester]. Structure-asserted via parsed JSON (not a fragile string match). + let (launch, agent, fs, _pty, _bus, _sessions, _tr, _session) = + launch_fixture_with_profile( + profile_with_mcp(pid(9), McpConfigStrategy::config_file(".mcp.json").unwrap()), + Some(ContextInjectionPlan::File { + target: "CLAUDE.md".to_owned(), + }), + ); + + let exe = "/abs/idea"; + let endpoint = "/run/idea-mcp/abc.sock"; + // project_id in the hyphen-free 32-hex `simple` form consumed by the M5c guard + // (`as_uuid().simple()` — see the coherence note / M5e smoke e2e below). + let project_id = "0123456789abcdef0123456789abcdef"; + let requester = "agent-7"; + let runtime = application::McpRuntime { + exe: exe.to_owned(), + endpoint: endpoint.to_owned(), + project_id: project_id.to_owned(), + requester: requester.to_owned(), + }; + + launch + .execute(launch_input_with_runtime(agent.id, runtime)) + .await + .unwrap(); + + let doc = read_single_mcp_json(&fs); + let server = &doc["mcpServers"]["idea"]; + assert_eq!( + server["command"].as_str(), + Some(exe), + "command must be the injected exe, got: {doc}" + ); + let args: Vec<&str> = server["args"] + .as_array() + .expect("args must be a JSON array") + .iter() + .map(|v| v.as_str().expect("each arg is a JSON string")) + .collect(); + assert_eq!( + args, + vec![ + "mcp-server", + "--endpoint", + endpoint, + "--project", + project_id, + "--requester", + requester, + ], + "args must carry the ordered mcp-server flags, got: {args:?}" + ); +} + +#[tokio::test] +async fn launch_mcp_runtime_special_chars_stay_valid_json() { + // M5d-2 — an exe/endpoint with a space and a backslash ⇒ the .mcp.json stays + // valid JSON (parse OK, asserted by read_single_mcp_json) and the values are + // faithfully preserved (correct escaping, no corruption). + let (launch, agent, fs, _pty, _bus, _sessions, _tr, _session) = + launch_fixture_with_profile( + profile_with_mcp(pid(9), McpConfigStrategy::config_file(".mcp.json").unwrap()), + Some(ContextInjectionPlan::File { + target: "CLAUDE.md".to_owned(), + }), + ); + + let exe = r"C:\Program Files\IdeA\idea.exe"; + let endpoint = r"/tmp/idea mcp/a b\c.sock"; + let runtime = application::McpRuntime { + exe: exe.to_owned(), + endpoint: endpoint.to_owned(), + project_id: "0123456789abcdef0123456789abcdef".to_owned(), + requester: "agent X".to_owned(), + }; + + launch + .execute(launch_input_with_runtime(agent.id, runtime)) + .await + .unwrap(); + + // Parsing already proves the JSON is valid despite the special chars. + let doc = read_single_mcp_json(&fs); + let server = &doc["mcpServers"]["idea"]; + assert_eq!( + server["command"].as_str(), + Some(exe), + "exe with backslashes/spaces must round-trip faithfully" + ); + let args: Vec<&str> = server["args"] + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap()) + .collect(); + assert_eq!( + args.get(2).copied(), + Some(endpoint), + "endpoint with a space and a backslash must round-trip faithfully, got: {args:?}" + ); +} + +#[tokio::test] +async fn launch_mcp_runtime_none_writes_minimal_declaration() { + // M5d-3 — mcp_runtime = None + an MCP profile ⇒ the minimal declaration is + // written: command == "idea", args == ["mcp-server"]. Still valid JSON. + let (launch, agent, fs, _pty, _bus, _sessions, _tr, _session) = + launch_fixture_with_profile( + profile_with_mcp(pid(9), McpConfigStrategy::config_file(".mcp.json").unwrap()), + Some(ContextInjectionPlan::File { + target: "CLAUDE.md".to_owned(), + }), + ); + + // launch_input has mcp_runtime: None. + launch.execute(launch_input(agent.id)).await.unwrap(); + + let doc = read_single_mcp_json(&fs); + let server = &doc["mcpServers"]["idea"]; + assert_eq!( + server["command"].as_str(), + Some("idea"), + "the minimal declaration must use the `idea` command, got: {doc}" + ); + let args: Vec<&str> = server["args"] + .as_array() + .expect("args must be a JSON array") + .iter() + .map(|v| v.as_str().unwrap()) + .collect(); + assert_eq!( + args, + vec!["mcp-server"], + "the minimal declaration must carry only the mcp-server subcommand, got: {args:?}" + ); +} + +#[tokio::test] +async fn launch_mcp_runtime_with_no_mcp_profile_writes_nothing() { + // M5d-4 — a profile WITHOUT MCP ⇒ no .mcp.json is written, even though a + // runtime is supplied (unchanged from M1: mcp == None short-circuits). + let (launch, agent, fs, _pty, _bus, _sessions, _tr, _session) = launch_fixture( + ContextInjection::convention_file("CLAUDE.md").unwrap(), + Some(ContextInjectionPlan::File { + target: "CLAUDE.md".to_owned(), + }), + ); + + let runtime = application::McpRuntime { + exe: "/abs/idea".to_owned(), + endpoint: "/run/idea-mcp/abc.sock".to_owned(), + project_id: "0123456789abcdef0123456789abcdef".to_owned(), + requester: "agent-7".to_owned(), + }; + + launch + .execute(launch_input_with_runtime(agent.id, runtime)) + .await + .unwrap(); + + assert!( + writes_ending_with(&fs, "/.mcp.json").is_empty(), + "no MCP config file when the profile carries no McpCapability" + ); +} + +#[tokio::test] +async fn launch_mcp_runtime_is_non_clobbering() { + // M5d-5 — a pre-existing .mcp.json is NOT overwritten, even when a real runtime + // is injected (unchanged non-clobbering contract from M1). + let (launch, agent, fs, _pty, _bus, _sessions, _tr, _session) = + launch_fixture_with_profile( + profile_with_mcp(pid(9), McpConfigStrategy::config_file(".mcp.json").unwrap()), + Some(ContextInjectionPlan::File { + target: "CLAUDE.md".to_owned(), + }), + ); + let run_dir = format!("/home/me/proj/.ideai/run/{}", agent.id); + fs.mark_existing(&format!("{run_dir}/.mcp.json")); + + let runtime = application::McpRuntime { + exe: "/abs/idea".to_owned(), + endpoint: "/run/idea-mcp/abc.sock".to_owned(), + project_id: "0123456789abcdef0123456789abcdef".to_owned(), + requester: "agent-7".to_owned(), + }; + + launch + .execute(launch_input_with_runtime(agent.id, runtime)) + .await + .unwrap(); + + assert!( + writes_ending_with(&fs, "/.mcp.json").is_empty(), + "an existing .mcp.json must not be clobbered even with an injected runtime" + ); +} + +#[test] +fn mcp_runtime_project_id_uses_simple_uuid_form() { + // M5d-6 (coherence, pure) — app-tauri builds the McpRuntime.project_id as + // `project.id.as_uuid().simple().to_string()` (commands.rs::launch_agent), + // i.e. the hyphen-free 32-hex `simple` form the M5c handshake guard + // (`serve_peer`) compares against. The McpRuntime construction is inlined in + // the Tauri command (not an extractable pure fn), so we cannot unit-test the + // wiring here without an AppState. Instead we PIN the format contract the M5d + // tests above rely on, and flag that the end-to-end injection coherence + // (--endpoint == mcp_endpoint(..).as_cli_arg() and --project == this form) is + // exercised by the M5e smoke e2e. + let uuid = Uuid::from_u128(0x0123_4567_89ab_cdef_0123_4567_89ab_cdef); + let simple = uuid.as_simple().to_string(); + assert_eq!(simple, "0123456789abcdef0123456789abcdef"); + assert!( + !simple.contains('-') && simple.len() == 32, + "the `simple` form is hyphen-free 32-hex, matching the project_id used in the M5d tests" + ); +} + // --------------------------------------------------------------------------- // LaunchAgent — session resume (T4) // --------------------------------------------------------------------------- diff --git a/crates/application/tests/structured_launch_d3.rs b/crates/application/tests/structured_launch_d3.rs index b874338..6bc938a 100644 --- a/crates/application/tests/structured_launch_d3.rs +++ b/crates/application/tests/structured_launch_d3.rs @@ -611,6 +611,7 @@ fn launch_input(agent_id: AgentId) -> LaunchAgentInput { cols: 80, node_id: None, conversation_id: None, + mcp_runtime: None, } } diff --git a/crates/infrastructure/src/orchestrator/mcp/server.rs b/crates/infrastructure/src/orchestrator/mcp/server.rs index 8ca1f3f..19c2720 100644 --- a/crates/infrastructure/src/orchestrator/mcp/server.rs +++ b/crates/infrastructure/src/orchestrator/mcp/server.rs @@ -47,6 +47,12 @@ pub struct McpServer { /// delegation arrived through the MCP door. `None` ⇒ no publication (the server /// still works), so existing call sites stay valid. events: Option>, + /// Identity of the connected peer — the real agent id carried in the loopback + /// handshake (cadrage v5 §1.4). When non-empty it becomes + /// [`DomainEvent::OrchestratorRequestProcessed`]'s `requester_id` instead of the + /// frozen `"mcp"` placeholder; empty ⇒ the legacy `"mcp"` label (back-compat for + /// M2 callers and connections that arrive without a requester). + requester: String, } impl McpServer { @@ -59,6 +65,7 @@ impl McpServer { service, project, events: None, + requester: String::new(), } } @@ -72,6 +79,35 @@ impl McpServer { self } + /// Returns a per-connection clone of this server tagged with the connected + /// peer's `requester` id (the loopback handshake's `requester`, cadrage v5 §1.4). + /// + /// The base server (one per project, M3) is shared by every connection; each + /// accepted peer derives its own contextualized server so concurrent peers never + /// share a requester. An empty `requester` keeps the legacy `"mcp"` label. + /// + /// Cheap: only `Arc`s, a `Project` clone, and the requester `String` are copied. + #[must_use] + pub fn for_requester(&self, requester: impl Into) -> Self { + Self { + service: Arc::clone(&self.service), + project: self.project.clone(), + events: self.events.clone(), + requester: requester.into(), + } + } + + /// Serves JSON-RPC messages from `transport`, tagging every processed + /// `tools/call` with `requester` as the delegating agent (cadrage v5 §1.4). + /// + /// A thin wrapper over [`serve`](Self::serve) on a per-connection + /// [`for_requester`](Self::for_requester) clone: the caller (the per-peer accept + /// loop) need not mutate the shared project server. An empty `requester` yields + /// the legacy `"mcp"` label. + pub async fn serve_as(&self, requester: impl Into, transport: &mut T) { + self.for_requester(requester).serve(transport).await; + } + /// Serves JSON-RPC messages from `transport` until it closes (clean EOF). /// /// Every inbound line is handled in isolation: a malformed line or an unknown @@ -211,12 +247,18 @@ impl McpServer { /// Publishes [`DomainEvent::OrchestratorRequestProcessed`] for a handled /// `tools/call`, tagged [`OrchestrationSource::Mcp`]. The requester id is the - /// stable `"mcp"` label until the per-session transport bind (open point S-MCP) - /// carries the connected agent's identity. No-op without an event sink. + /// connected peer's real agent id (carried in the loopback handshake, cadrage v5 + /// §1.4), falling back to the legacy `"mcp"` label when no identity was supplied. + /// No-op without an event sink. fn publish_processed(&self, action: &str, ok: bool) { if let Some(publish) = &self.events { + let requester_id = if self.requester.is_empty() { + "mcp".to_owned() + } else { + self.requester.clone() + }; publish(DomainEvent::OrchestratorRequestProcessed { - requester_id: "mcp".to_owned(), + requester_id, action: action.to_owned(), ok, source: OrchestrationSource::Mcp, diff --git a/crates/infrastructure/src/orchestrator/mcp/transport.rs b/crates/infrastructure/src/orchestrator/mcp/transport.rs index fe58b3d..4626f38 100644 --- a/crates/infrastructure/src/orchestrator/mcp/transport.rs +++ b/crates/infrastructure/src/orchestrator/mcp/transport.rs @@ -42,6 +42,18 @@ where writer, } } + + /// Wraps an **already-buffered** reader plus a writer as a line-delimited JSON + /// transport. + /// + /// Used by the per-peer accept loop (M5c): the handshake line (cadrage v5 §1.4) + /// is read off a `BufReader` over the loopback's read half *before* serving, and + /// that same `BufReader` — carrying any bytes already buffered past the handshake + /// — is handed here so no JSON-RPC byte is lost between the handshake and the + /// serve loop. + pub fn from_buffered(reader: BufReader, writer: W) -> Self { + Self { reader, writer } + } } #[async_trait::async_trait]