feat(agent): bind transport S-MCP — outils idea_* vivants de bout en bout (M5a-e) — §14.3.1
Dernier kilomètre de l'orchestration native : une CLI MCP réellement lancée
joint le serveur MCP du projet et ses outils idea_* aboutissent au vrai dispatch.
- M5a endpoint loopback par projet (interprocess UDS/named pipe, source unique
mcp_endpoint, cleanup au close) — zéro port réseau, AppImage/SSH-safe.
- M5b sous-commande `idea mcp-server` : pont stdio↔loopback headless (avant init
Tauri), handshake {"project","requester"}, endpoint-absent borné, EOF propre.
- M5c McpServerHandle boucle accept + serve_peer par pair ; requester réel
propagé jusqu'à OrchestratorRequestProcessed (fin du "mcp" figé) ; isolation
des pairs ; terminaison propre.
- M5d apply_mcp_config écrit la déclaration réelle (current_exe + --endpoint
mcp_endpoint + --project simple-uuid + --requester) ; McpRuntime injecté comme
donnée (application ne dépend pas de app-tauri).
- M5e smoke e2e sur vrai loopback : list/ask inline, cible PTY → erreur typée,
JSON malformé → pas de panic, requester propagé.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
578
crates/app-tauri/src/mcp_bridge.rs
Normal file
578
crates/app-tauri/src/mcp_bridge.rs
Normal file
@ -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* `<exe IdeA> 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":"<project-id>","requester":"<requester-id>"}\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<I, S>(args: I) -> Result<Self, String>
|
||||
where
|
||||
I: IntoIterator<Item = S>,
|
||||
S: Into<String>,
|
||||
{
|
||||
let mut endpoint: Option<String> = 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<u8> {
|
||||
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<I, S>(args: I) -> ExitCode
|
||||
where
|
||||
I: IntoIterator<Item = S>,
|
||||
S: Into<String>,
|
||||
{
|
||||
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<LocalSocketStream, String> {
|
||||
let name = endpoint
|
||||
.to_fs_name::<GenericFilePath>()
|
||||
.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<CIn, COut, LIn, LOut>(
|
||||
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<u8> = 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<u8> = 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::<GenericFilePath>().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<u8> = 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user