Merge pull request #703 from fabro-sh/fix/mcp-restart-after-upgrade-697

Fix stale MCP servers after upgrades

Cargo.lock

@@ -2852,6 +2852,7 @@ dependencies = [
  "tempfile",
  "tokio",
  "toml 0.8.23",
+ "tracing",
 ]
 
 [[package]]

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() {

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,

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"
+tempfile = "3"

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<String> {
     serde_json::to_string_pretty(&generic_config(settings))

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> {
+        Self::new(invoked_executable_path()?)
+    }
+
+    fn new(path: PathBuf) -> io::Result<Self> {
+        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<std::time::SystemTime>);
+
+#[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<PathBuf> {
+    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());
+    }
+}

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<Box<dyn Future<Output = Result<Client>> + Send>>;
 
 pub type FabroClientFactory = Arc<dyn Fn() -> FabroClientFuture + Send + Sync>;

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<Self>,
 }
 
+/// 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 {