diff --git a/Cargo.lock b/Cargo.lock index 5e1db82c7..57596ad82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2852,6 +2852,7 @@ dependencies = [ "tempfile", "tokio", "toml 0.8.23", + "tracing", ] [[package]] diff --git a/lib/apps/fabro-cli/src/main.rs b/lib/apps/fabro-cli/src/main.rs index 0d0aa1c79..f01e1cc65 100644 --- a/lib/apps/fabro-cli/src/main.rs +++ b/lib/apps/fabro-cli/src/main.rs @@ -120,8 +120,12 @@ async fn main() { "{:?}", miette::Report::new(CliDiagnostic::new(err, !json_mode)) ); - std::process::exit(exit_code); } + + // Exit rather than returning. A command can leave a blocked worker thread + // behind — `mcp start` parks Tokio's stdin reader on a read only the MCP + // host can end — and dropping the runtime would wait on it forever. + std::process::exit(exit_code); } fn install_miette_hook() { diff --git a/lib/apps/fabro-cli/tests/it/cmd/mcp.rs b/lib/apps/fabro-cli/tests/it/cmd/mcp.rs index dc1052ac0..e810bd4d9 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/mcp.rs @@ -8,16 +8,20 @@ )] use std::collections::HashMap; +#[cfg(unix)] +use std::fs; use std::io::{BufRead as _, Write as _}; use std::path::{Path, PathBuf}; -use std::process::Stdio; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::thread; +use std::time::{Duration, Instant}; use chrono::{DateTime, Duration as ChronoDuration, Utc}; use fabro_client::{AuthEntry, AuthStore, DevTokenEntry, OAuthEntry, StoredSubject}; use fabro_mcp::client::McpClient; use fabro_mcp::config::{McpServerSettings, McpTransport}; use fabro_test::{fabro_json_snapshot, fabro_snapshot, test_context}; -use fabro_types::RunId; +use fabro_types::{Graph, RunId, WorkflowSettings, test_support}; use httpmock::Method::{GET, POST}; use httpmock::MockServer; @@ -515,50 +519,69 @@ async fn stdio_server_initializes_and_lists_run_tools() { fn stdio_start_writes_only_json_rpc_to_stdout() { let context = test_context!(); let fixture = mcp_stdio_fixture(&context, &[]); - let mut cmd = std::process::Command::new(&fixture.command[0]); - cmd.args(&fixture.command[1..]) - .env_clear() - .envs(&fixture.env) - .current_dir(&fixture.current_dir) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()); - let mut child = cmd.spawn().unwrap(); - let mut stdin = child.stdin.take().unwrap(); - writeln!( - stdin, - r#"{{"jsonrpc":"2.0","id":1,"method":"initialize","params":{{"protocolVersion":"2025-06-18","capabilities":{{}},"clientInfo":{{"name":"fabro-test","version":"0.0.0"}}}}}}"# - ) - .unwrap(); + let (mut child, _stdin, response) = + spawn_stdio_server(&fixture, Path::new(&fixture.command[0])); - let stdout = child.stdout.take().unwrap(); - let (tx, rx) = std::sync::mpsc::channel(); - std::thread::spawn(move || { - let mut line = String::new(); - let result = std::io::BufReader::new(stdout).read_line(&mut line); - let _ = tx.send(result.map(|_| line)); - }); - - let line = rx - .recv_timeout(std::time::Duration::from_secs(5)) - .expect("initialize response should arrive") - .expect("stdout should be readable"); - let value: serde_json::Value = serde_json::from_str(line.trim()).unwrap(); - assert_eq!(value["jsonrpc"], "2.0"); + assert_eq!(response["jsonrpc"], "2.0"); + assert_eq!(response["result"]["serverInfo"]["name"], "fabro"); + assert_eq!( + response["result"]["serverInfo"]["version"], + env!("CARGO_PKG_VERSION") + ); let _ = child.kill(); let _ = child.wait(); } +#[cfg(unix)] +#[test] +fn stdio_server_exits_when_executable_is_replaced() { + let context = test_context!(); + let fixture = mcp_stdio_fixture(&context, &[]); + let directory = tempfile::tempdir().expect("replacement directory should exist"); + // A symlink stands in for a Homebrew install: the server follows it to the + // real binary, so replacing the link changes the identity it watches without + // copying a multi-hundred-megabyte executable. + let executable = directory.path().join("fabro"); + std::os::unix::fs::symlink(&fixture.command[0], &executable) + .expect("Fabro executable should be linked"); + + // `_stdin` holds the pipe open so replacement, not EOF, stops the server. + let (mut child, _stdin, response) = spawn_stdio_server(&fixture, &executable); + assert_eq!(response["result"]["serverInfo"]["name"], "fabro"); + + let replacement = directory.path().join("fabro-replacement"); + fs::write(&replacement, b"replacement").expect("replacement file should be written"); + fs::rename(replacement, &executable).expect("Fabro executable should be replaced"); + + // The server takes up to 1s to notice the replacement and then bounds its + // own shutdown at 5s, so 6s is the worst case. Allow more so a loaded runner + // cannot fail a healthy server; a passing run exits in about a second. + let deadline = Instant::now() + Duration::from_secs(20); + let status = loop { + if let Some(status) = child.try_wait().expect("MCP server should be polled") { + break status; + } + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("MCP server did not exit after its executable was replaced"); + } + thread::sleep(Duration::from_millis(50)); + }; + + assert!(status.success(), "MCP server should exit successfully"); +} + #[tokio::test(flavor = "multi_thread")] async fn stdio_startup_and_list_tools_is_fast() { let context = test_context!(); - let start = std::time::Instant::now(); + let start = Instant::now(); let client = spawn_mcp_client(&context, &[]).await; let tools = client.list_tools().await.unwrap(); assert_eq!(tools.len(), MCP_RUN_TOOL_NAMES.len()); - assert!(start.elapsed() < std::time::Duration::from_secs(2)); + assert!(start.elapsed() < Duration::from_secs(2)); client .shutdown() .await @@ -1602,12 +1625,21 @@ async fn mcp_get_resolves_selector_and_returns_summary_projection_and_questions( let projection = server.mock(|when, then| { when.method(GET) .path(format!("/api/v1/runs/{run_id}/state")); + let mut body = run_projection_json(&run_id, &serde_json::json!({ "kind": "running" })); + body["spec"]["settings"]["run"]["model"] = serde_json::json!({ + "provider": "openai", + "name": "gpt-5.6-sol", + "fallbacks": { + "gpt-5.6-sol": ["gpt-5.6-terra"] + }, + "controls": { + "reasoning_effort": null, + "speed": null + } + }); then.status(200) .header("Content-Type", "application/json") - .json_body(run_projection_json( - &run_id, - &serde_json::json!({ "kind": "running" }), - )); + .json_body(body); }); let questions = server.mock(|when, then| { when.method(GET) @@ -1644,6 +1676,10 @@ async fn mcp_get_resolves_selector_and_returns_summary_projection_and_questions( assert_eq!(get["summary"]["workflow_name"], "Simple"); assert_eq!(get["summary"]["workflow_slug"], "simple"); assert_eq!(get["projection"]["status"]["kind"], "running"); + assert_eq!( + get["projection"]["spec"]["settings"]["run"]["model"]["fallbacks"]["gpt-5.6-sol"][0], + "gpt-5.6-terra" + ); assert_eq!(get["questions"][0]["id"], "q-1"); resolve.assert(); retrieve.assert(); @@ -2001,6 +2037,82 @@ async fn mcp_events_filters_find_matches_beyond_first_page() { .expect("MCP client should shut down"); } +#[tokio::test(flavor = "multi_thread")] +async fn mcp_events_decodes_run_created_with_model_keyed_fallbacks() { + let context = test_context!(); + let server = MockServer::start(); + let target_url = format!("{}/api/v1", server.base_url()); + let target: fabro_client::ServerTarget = target_url.parse().unwrap(); + seed_dev_token_auth(&context.home_dir, &target, TEST_DEV_TOKEN); + let run_id = unique_run_id(); + let resolve = mock_resolved_run(&server, "nightly", &run_id); + let mut settings = serde_json::to_value(WorkflowSettings::default()) + .expect("workflow settings should serialize"); + settings["run"]["model"] = serde_json::json!({ + "provider": "openai", + "name": "gpt-5.6-sol", + "fallbacks": { + "gpt-5.6-sol": ["gpt-5.6-terra"] + }, + "controls": { + "reasoning_effort": null, + "speed": null + } + }); + let event = serde_json::json!({ + "seq": 1, + "id": "evt-created", + "ts": "2026-04-05T12:00:00Z", + "run_id": run_id, + "event": "run.created", + "properties": { + "settings": settings, + "graph": Graph::new("Remote Workflow"), + "labels": {}, + "run_dir": "/tmp/run", + "source_directory": "/srv/repo", + "provenance": test_support::test_run_provenance() + }, + "actor": null + }); + let events = server.mock(|when, then| { + when.method(GET) + .path(format!("/api/v1/runs/{run_id}/events")) + .query_param_missing("limit"); + then.status(200) + .header("Content-Type", "application/json") + .json_body(serde_json::json!({ + "data": [event], + "meta": { "has_more": false } + })); + }); + let client = spawn_mcp_client(&context, &["--server", &target_url]).await; + + let result = call_tool_json( + &client, + "fabro_run_events", + serde_json::json!({ + "run_id": "nightly", + "action": "search", + "query": "gpt-5.6-terra", + "first": 1 + }), + ) + .await; + + assert_eq!( + result["events"][0]["event"]["properties"]["settings"]["run"]["model"]["fallbacks"]["gpt-5.6-sol"] + [0], + "gpt-5.6-terra" + ); + resolve.assert(); + events.assert(); + client + .shutdown() + .await + .expect("MCP client should shut down"); +} + #[tokio::test(flavor = "multi_thread")] async fn mcp_events_requires_action_specific_inputs_before_auth() { let context = test_context!(); @@ -2379,6 +2491,45 @@ fn mcp_stdio_fixture(context: &fabro_test::TestContext, extra_args: &[&str]) -> } } +const MCP_INITIALIZE_REQUEST: &str = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"fabro-test","version":"0.0.0"}}}"#; + +/// Starts `fabro mcp start` as a raw child process, sends `initialize`, and +/// returns the decoded response. Unlike `spawn_mcp_client`, the caller keeps +/// the `Child` and its stdin, so it can observe how and when the server exits. +fn spawn_stdio_server( + fixture: &McpStdioFixture, + program: &Path, +) -> (Child, ChildStdin, serde_json::Value) { + let mut child = Command::new(program) + .args(&fixture.command[1..]) + .env_clear() + .envs(&fixture.env) + .current_dir(&fixture.current_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("MCP server should start"); + + let mut stdin = child.stdin.take().expect("MCP stdin should be available"); + writeln!(stdin, "{MCP_INITIALIZE_REQUEST}").expect("initialize request should be written"); + + let stdout = child.stdout.take().expect("MCP stdout should be available"); + let (tx, rx) = std::sync::mpsc::channel(); + thread::spawn(move || { + let mut line = String::new(); + let result = std::io::BufReader::new(stdout).read_line(&mut line); + let _ = tx.send(result.map(|_| line)); + }); + let line = rx + .recv_timeout(Duration::from_secs(5)) + .expect("initialize response should arrive") + .expect("MCP stdout should be readable"); + + let response = serde_json::from_str(line.trim()).expect("response should be JSON"); + (child, stdin, response) +} + fn write_mcp_server_settings( context: &mut fabro_test::TestContext, storage_dir: &Path, diff --git a/lib/apps/fabro-mcp-server/Cargo.toml b/lib/apps/fabro-mcp-server/Cargo.toml index cdbf6b65f..758a11fd3 100644 --- a/lib/apps/fabro-mcp-server/Cargo.toml +++ b/lib/apps/fabro-mcp-server/Cargo.toml @@ -32,7 +32,8 @@ serde_json.workspace = true strum.workspace = true tokio.workspace = true toml.workspace = true +tracing.workspace = true [dev-dependencies] httpmock = "0.8" -tempfile = "3" \ No newline at end of file +tempfile = "3" diff --git a/lib/apps/fabro-mcp-server/src/config.rs b/lib/apps/fabro-mcp-server/src/config.rs index 9eb8a67c2..ec7154348 100644 --- a/lib/apps/fabro-mcp-server/src/config.rs +++ b/lib/apps/fabro-mcp-server/src/config.rs @@ -9,9 +9,7 @@ use anyhow::{Context as _, Result, anyhow}; use serde_json::map::Entry; use serde_json::{Map, Value, json}; -use crate::{McpAgent, McpConfigSettings, McpInitSettings}; - -const SERVER_NAME: &str = "fabro"; +use crate::{McpAgent, McpConfigSettings, McpInitSettings, SERVER_NAME}; pub fn config_json(settings: &McpConfigSettings) -> Result { serde_json::to_string_pretty(&generic_config(settings)) diff --git a/lib/apps/fabro-mcp-server/src/executable_monitor.rs b/lib/apps/fabro-mcp-server/src/executable_monitor.rs new file mode 100644 index 000000000..ae0eb6b30 --- /dev/null +++ b/lib/apps/fabro-mcp-server/src/executable_monitor.rs @@ -0,0 +1,139 @@ +//! Detects when an upgrade replaces the executable that launched this MCP +//! server. +//! +//! MCP hosts can keep stdio servers alive for days. Without this check, an old +//! process keeps its old API response decoder after the `fabro` file on disk is +//! upgraded. + +use std::fs::Metadata; +use std::path::{self, PathBuf}; +use std::time::Duration; +use std::{env, fs, io}; + +use tokio::time::{self, Instant, MissedTickBehavior}; + +const CHECK_INTERVAL: Duration = Duration::from_secs(1); + +pub(crate) struct ExecutableMonitor { + path: PathBuf, + identity: Identity, +} + +impl ExecutableMonitor { + pub(crate) fn current() -> io::Result { + Self::new(invoked_executable_path()?) + } + + fn new(path: PathBuf) -> io::Result { + let identity = identity(&fs::metadata(&path)?); + Ok(Self { path, identity }) + } + + pub(crate) async fn wait_until_replaced(self) { + let mut interval = time::interval_at(Instant::now() + CHECK_INTERVAL, CHECK_INTERVAL); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + interval.tick().await; + if self.was_replaced() { + return; + } + } + } + + /// Reads the identity synchronously. This is a `stat` of a page-cached + /// inode once per second, so handing it to Tokio's blocking pool would + /// cost more than the call itself and would keep a pool thread resident + /// for the life of the server. + fn was_replaced(&self) -> bool { + !fs::metadata(&self.path).is_ok_and(|metadata| identity(&metadata) == self.identity) + } +} + +/// Identifies the file behind an executable path. Upgrades always swap a new +/// file into place — `fabro upgrade` renames over the old one and Homebrew +/// repoints a symlink — so the identity changes even though the path does not. +#[cfg(unix)] +type Identity = (u64, u64); + +#[cfg(unix)] +fn identity(metadata: &Metadata) -> Identity { + use std::os::unix::fs::MetadataExt as _; + + (metadata.dev(), metadata.ino()) +} + +#[cfg(not(unix))] +type Identity = (u64, Option); + +#[cfg(not(unix))] +fn identity(metadata: &Metadata) -> Identity { + (metadata.len(), metadata.modified().ok()) +} + +/// Resolves the executable path to watch. +/// +/// `argv[0]` wins when it carries a directory, because it names the path the +/// host actually launched, symlink included. MCP hosts normally launch a bare +/// `fabro` found on `PATH`, which leaves `current_exe`: it reports the symlink +/// on macOS, and the Homebrew symlink's own target on Linux. +fn invoked_executable_path() -> io::Result { + match env::args_os().next().map(PathBuf::from) { + Some(invoked) if invoked.components().count() > 1 => path::absolute(invoked), + _ => env::current_exe(), + } +} + +#[cfg(test)] +mod tests { + use tokio::fs as async_fs; + + use super::*; + + #[tokio::test] + async fn unchanged_executable_is_current() { + let directory = tempfile::tempdir().expect("temp directory should exist"); + let executable = directory.path().join("fabro"); + async_fs::write(&executable, b"current") + .await + .expect("fixture executable should be written"); + let monitor = ExecutableMonitor::new(executable).unwrap(); + + assert!(!monitor.was_replaced()); + } + + #[tokio::test] + async fn atomic_executable_replacement_is_detected() { + let directory = tempfile::tempdir().expect("temp directory should exist"); + let executable = directory.path().join("fabro"); + let replacement = directory.path().join("fabro-new"); + async_fs::write(&executable, b"old") + .await + .expect("old fixture executable should be written"); + async_fs::write(&replacement, b"new executable") + .await + .expect("new fixture executable should be written"); + let monitor = ExecutableMonitor::new(executable.clone()).unwrap(); + + async_fs::rename(&replacement, &executable) + .await + .expect("fixture executable should be replaced"); + + assert!(monitor.was_replaced()); + } + + #[tokio::test] + async fn removed_executable_is_detected() { + let directory = tempfile::tempdir().expect("temp directory should exist"); + let executable = directory.path().join("fabro"); + async_fs::write(&executable, b"current") + .await + .expect("fixture executable should be written"); + let monitor = ExecutableMonitor::new(executable.clone()).unwrap(); + + async_fs::remove_file(executable) + .await + .expect("fixture executable should be removed"); + + assert!(monitor.was_replaced()); + } +} diff --git a/lib/apps/fabro-mcp-server/src/lib.rs b/lib/apps/fabro-mcp-server/src/lib.rs index ad6264dcb..068f99c14 100644 --- a/lib/apps/fabro-mcp-server/src/lib.rs +++ b/lib/apps/fabro-mcp-server/src/lib.rs @@ -1,4 +1,5 @@ mod config; +mod executable_monitor; mod manifest_builder; mod server; @@ -12,6 +13,10 @@ pub use config::{config_json, init_agent}; use fabro_client::Client; pub use server::start; +/// The name this MCP server reports over the wire and registers under in agent +/// config files. +pub(crate) const SERVER_NAME: &str = "fabro"; + pub type FabroClientFuture = Pin> + Send>>; pub type FabroClientFactory = Arc FabroClientFuture + Send + Sync>; diff --git a/lib/apps/fabro-mcp-server/src/server.rs b/lib/apps/fabro-mcp-server/src/server.rs index c1b932308..6a9c9941d 100644 --- a/lib/apps/fabro-mcp-server/src/server.rs +++ b/lib/apps/fabro-mcp-server/src/server.rs @@ -1,19 +1,24 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use anyhow::Result; use fabro_tool::fabro_client::ClientBackend; use fabro_tool::{self as run_tools, FabroToolBackend}; +use fabro_util::version::FABRO_VERSION; use rmcp::handler::server::router::tool::ToolRouter; use rmcp::handler::server::wrapper::Parameters; -use rmcp::model::{CallToolResult, Content, ServerCapabilities, ServerInfo}; +use rmcp::model::{CallToolResult, Content, Implementation, ServerCapabilities, ServerInfo}; use rmcp::transport::stdio; use rmcp::{ErrorData, ServerHandler, serve_server, tool, tool_handler, tool_router}; use serde::Serialize; use tokio::sync::OnceCell; +use tokio::time; +use tracing::warn; -use crate::FabroMcpServerSettings; +use crate::executable_monitor::ExecutableMonitor; use crate::manifest_builder::McpRunManifestBuilder; +use crate::{FabroMcpServerSettings, SERVER_NAME}; #[derive(Clone)] pub(crate) struct FabroMcpServer { @@ -23,10 +28,44 @@ pub(crate) struct FabroMcpServer { tool_router: ToolRouter, } +/// How long to wait for the MCP service to stop after an upgrade is detected. +/// The wait is bounded because the transport closes by writing to stdout, which +/// blocks if the host has stopped reading. +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); + pub async fn start(settings: FabroMcpServerSettings) -> Result<()> { - let server = FabroMcpServer::new(Arc::new(settings)); - let service = serve_server(server, stdio()).await?; - service.waiting().await?; + let monitor = match ExecutableMonitor::current() { + Ok(monitor) => Some(monitor), + Err(error) => { + warn!( + %error, + "Upgrade detection is unavailable; this MCP server will keep running after an \ + upgrade replaces it" + ); + None + } + }; + let service = serve_server(FabroMcpServer::new(Arc::new(settings)), stdio()).await?; + let Some(monitor) = monitor else { + service.waiting().await?; + return Ok(()); + }; + + let cancellation = service.cancellation_token(); + let mut service_wait = Box::pin(service.waiting()); + tokio::select! { + result = &mut service_wait => { + result?; + } + () = monitor.wait_until_replaced() => { + // An upgrade replaced the executable, so stop serving and let the + // host reconnect to the new one. The CLI exits the process rather + // than returning, because Tokio's stdin worker stays blocked on a + // read that only the host can end. + cancellation.cancel(); + let _ = time::timeout(SHUTDOWN_TIMEOUT, service_wait).await; + } + } Ok(()) } @@ -34,6 +73,7 @@ pub async fn start(settings: FabroMcpServerSettings) -> Result<()> { impl ServerHandler for FabroMcpServer { fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + .with_server_info(Implementation::new(SERVER_NAME, FABRO_VERSION).with_title("Fabro")) .with_instructions("Use these tools to create, inspect, control, wait for, and read events from Fabro workflow runs.") } } @@ -251,6 +291,22 @@ mod tests { use super::*; use crate::FabroMcpServerSettings; + #[test] + fn server_info_reports_fabro_version() { + let settings = FabroMcpServerSettings { + cwd: PathBuf::from("."), + config_path: PathBuf::from("fabro.toml"), + client_factory: Arc::new(|| { + Box::pin(async { panic!("client should not be constructed while reading info") }) + }), + }; + let info = FabroMcpServer::new(Arc::new(settings)).get_info(); + + assert_eq!(info.server_info.name, "fabro"); + assert_eq!(info.server_info.title.as_deref(), Some("Fabro")); + assert_eq!(info.server_info.version, FABRO_VERSION); + } + #[test] fn fabro_run_pair_tool_is_registered_with_stage_based_schema() { let settings = FabroMcpServerSettings {