From c144f3ca2c229fdf7d450a649d27ae9d81217b4a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 11 May 2026 08:23:48 -0400 Subject: [PATCH] feat(cli): add mcp run control tools --- Cargo.lock | 1 + lib/crates/fabro-cli/tests/it/cmd/mcp.rs | 257 ++++++++++++++ lib/crates/fabro-mcp-server/Cargo.toml | 1 + lib/crates/fabro-mcp-server/src/run_tools.rs | 355 +++++++++++++++++-- 4 files changed, 589 insertions(+), 25 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c946b7520..1a1024a5b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2031,6 +2031,7 @@ dependencies = [ "fabro-client", "fabro-http", "fabro-types", + "fabro-util", "rmcp", "schemars 1.2.1", "serde", diff --git a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs index 4635a515a..9ede4bc43 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs @@ -484,6 +484,204 @@ async fn mcp_create_and_search_manage_real_runs_with_cli_auth() { harness.shutdown().await; } +#[tokio::test(flavor = "multi_thread")] +async fn mcp_lifecycle_tools_manage_real_run() { + let context = test_context!(); + let harness = + RealAuthHarness::start_with_dev_token(fabro_test::GitHubAppState::default()).await; + let target_url = harness.api_target(); + let target: fabro_client::ServerTarget = target_url.parse().unwrap(); + seed_dev_token_auth(&context.home_dir, &target, TEST_DEV_TOKEN); + let workflow = context.install_fixture("simple.fabro"); + let client = spawn_mcp_client(&context, &["--server", &target_url]).await; + let run_id = create_mcp_run(&client, workflow, true).await; + let cancel = call_tool_json( + &client, + "fabro_run_interact", + serde_json::json!({ "run_id": run_id, "action": "cancel" }), + ) + .await; + + let gather = call_tool_json( + &client, + "fabro_run_gather", + serde_json::json!({ + "run_ids": [run_id], + "timeout_seconds": 20, + "poll_interval_seconds": 5 + }), + ) + .await; + let run_id = gather["runs"][0]["run_id"].as_str().unwrap().to_string(); + let get = call_tool_json( + &client, + "fabro_run_interact", + serde_json::json!({ "run_id": run_id, "action": "get" }), + ) + .await; + let events = call_tool_json( + &client, + "fabro_run_events", + serde_json::json!({ "run_id": run_id, "action": "list", "first": 5 }), + ) + .await; + let archive = call_tool_json( + &client, + "fabro_run_interact", + serde_json::json!({ "run_id": run_id, "action": "archive" }), + ) + .await; + let unarchive = call_tool_json( + &client, + "fabro_run_interact", + serde_json::json!({ "run_id": run_id, "action": "unarchive" }), + ) + .await; + let search = call_tool_json( + &client, + "fabro_run_search", + serde_json::json!({ "run_ids": [run_id], "archived": false }), + ) + .await; + + fabro_json_snapshot!( + context, + serde_json::json!({ + "gather": normalize_gather(gather), + "cancel_action": cancel["action"], + "get_status": get["result"]["summary"]["status"], + "events_nonempty": events["events"].as_array().is_some_and(|events| !events.is_empty()), + "archive_action": archive["action"], + "unarchive_action": unarchive["action"], + "unarchived_search_count": search["runs"].as_array().unwrap().len(), + }), + @r#" + { + "gather": { + "runs": [ + { + "run_id": "[RUN_ID]", + "workflow_name": "Simple", + "workflow_slug": "simple", + "status": "failed", + "archived": false, + "created_at": "[TIMESTAMP]", + "started_at": null, + "completed_at": "[TIMESTAMP]", + "labels": { + "source": "mcp-test" + }, + "source_directory": "[SOURCE_DIRECTORY]", + "repo_origin_url": null, + "goal": "Run the Fabro workflow." + } + ], + "timed_out": false, + "elapsed_seconds": "[ELAPSED]" + }, + "cancel_action": "cancel", + "get_status": "failed", + "events_nonempty": true, + "archive_action": "archive", + "unarchive_action": "unarchive", + "unarchived_search_count": 1 + } + "# + ); + + harness.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_gather_rejects_too_many_runs() { + let context = test_context!(); + let client = spawn_mcp_client(&context, &["--server", "http://127.0.0.1:9"]).await; + let run_ids = (0..51) + .map(|index| format!("run_{index}")) + .collect::>(); + + let error = call_tool_error_text( + &client, + "fabro_run_gather", + serde_json::json!({ "run_ids": run_ids }), + ) + .await; + + assert!(error.contains("run_ids"), "{error}"); + assert_eq!(client.list_tools().await.unwrap().len(), 5); +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_gather_returns_timeout_result() { + let context = test_context!(); + let harness = + RealAuthHarness::start_with_dev_token(fabro_test::GitHubAppState::default()).await; + let target_url = harness.api_target(); + let target: fabro_client::ServerTarget = target_url.parse().unwrap(); + seed_dev_token_auth(&context.home_dir, &target, TEST_DEV_TOKEN); + let workflow = context.install_fixture("simple.fabro"); + let client = spawn_mcp_client(&context, &["--server", &target_url]).await; + let run_id = create_mcp_run(&client, workflow, false).await; + + let start = std::time::Instant::now(); + let gather = call_tool_json( + &client, + "fabro_run_gather", + serde_json::json!({ + "run_ids": [run_id], + "timeout_seconds": 1, + "poll_interval_seconds": 5 + }), + ) + .await; + + assert_eq!(gather["timed_out"], true); + assert!(start.elapsed() < std::time::Duration::from_secs(4)); + assert_eq!(gather["runs"][0]["status"], "submitted"); + + harness.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_interact_error_does_not_stop_server() { + let context = test_context!(); + let client = spawn_mcp_client(&context, &["--server", "http://127.0.0.1:9"]).await; + + let error = call_tool_error_text( + &client, + "fabro_run_interact", + serde_json::json!({ "run_id": "run_123", "action": "message" }), + ) + .await; + + assert!(error.contains("message"), "{error}"); + assert_eq!(client.list_tools().await.unwrap().len(), 5); +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_tool_auth_error_mentions_login() { + let context = test_context!(); + let harness = + RealAuthHarness::start_with_dev_token(fabro_test::GitHubAppState::default()).await; + let target_url = harness.api_target(); + let client = spawn_mcp_client(&context, &["--server", &target_url]).await; + + let error = call_tool_error_text( + &client, + "fabro_run_search", + serde_json::json!({ "first": 1 }), + ) + .await; + + assert!( + error.contains("Run `fabro auth login` to authenticate."), + "{error}" + ); + assert_eq!(client.list_tools().await.unwrap().len(), 5); + + harness.shutdown().await; +} + fn expected_claude_config_path(home_dir: &Path) -> PathBuf { #[cfg(target_os = "macos")] { @@ -582,6 +780,45 @@ async fn call_tool_json( .expect("tool result should include structured content") } +async fn call_tool_error_text( + client: &McpClient, + name: &str, + arguments: serde_json::Value, +) -> String { + let result = client + .call_tool(name, arguments, std::time::Duration::from_secs(30)) + .await + .expect("tool call should complete"); + assert_eq!(result.is_error, Some(true), "tool should return error"); + result + .content + .first() + .and_then(|content| serde_json::to_value(content).ok()) + .and_then(|content| content["text"].as_str().map(ToOwned::to_owned)) + .expect("tool error should include text") +} + +async fn create_mcp_run(client: &McpClient, workflow: PathBuf, start: bool) -> String { + let create = call_tool_json( + client, + "fabro_run_create", + serde_json::json!({ + "runs": [{ + "workflow": workflow, + "dry_run": true, + "auto_approve": true, + "labels": { "source": "mcp-test" }, + "start": start + }] + }), + ) + .await; + create["runs"][0]["run_id"] + .as_str() + .expect("create result should include run id") + .to_string() +} + fn normalize_run_search(mut value: serde_json::Value) -> serde_json::Value { if let Some(runs) = value["runs"].as_array_mut() { for run in runs { @@ -600,3 +837,23 @@ fn normalize_run_search(mut value: serde_json::Value) -> serde_json::Value { } value } + +fn normalize_gather(mut value: serde_json::Value) -> serde_json::Value { + value["elapsed_seconds"] = serde_json::json!("[ELAPSED]"); + if let Some(runs) = value["runs"].as_array_mut() { + for run in runs { + run["run_id"] = serde_json::json!("[RUN_ID]"); + run["created_at"] = serde_json::json!("[TIMESTAMP]"); + if run["started_at"].is_string() { + run["started_at"] = serde_json::json!("[TIMESTAMP]"); + } + if run["completed_at"].is_string() { + run["completed_at"] = serde_json::json!("[TIMESTAMP]"); + } + if run["source_directory"].is_string() { + run["source_directory"] = serde_json::json!("[SOURCE_DIRECTORY]"); + } + } + } + value +} diff --git a/lib/crates/fabro-mcp-server/Cargo.toml b/lib/crates/fabro-mcp-server/Cargo.toml index b391116d4..4c47e35d4 100644 --- a/lib/crates/fabro-mcp-server/Cargo.toml +++ b/lib/crates/fabro-mcp-server/Cargo.toml @@ -20,6 +20,7 @@ fabro-api = { path = "../fabro-api" } fabro-client = { path = "../fabro-client" } fabro-http.workspace = true fabro-types = { path = "../fabro-types" } +fabro-util = { path = "../fabro-util" } rmcp = { workspace = true, features = ["server", "macros", "schemars", "transport-io"] } schemars = "1.2.1" serde.workspace = true diff --git a/lib/crates/fabro-mcp-server/src/run_tools.rs b/lib/crates/fabro-mcp-server/src/run_tools.rs index 97684c20e..ca9e30281 100644 --- a/lib/crates/fabro-mcp-server/src/run_tools.rs +++ b/lib/crates/fabro-mcp-server/src/run_tools.rs @@ -6,17 +6,18 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::{Duration, Instant}; use chrono::{DateTime, NaiveDate, Utc}; use fabro_api::types; use fabro_client::Client; -use fabro_types::{Run, RunId, RunStatus}; +use fabro_types::{EventEnvelope, Run, RunId, RunStatus}; +use fabro_util::exit::{self, ExitClass}; use rmcp::model::{CallToolResult, Content}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use serde_json::Value; -use tokio::fs; -use tokio::task::yield_now; +use serde_json::{Value, json}; +use tokio::{fs, time}; #[derive(Debug)] pub(crate) struct ToolError { @@ -154,7 +155,7 @@ pub(crate) struct RunSummaryResult { pub(crate) goal: String, } -#[derive(Debug, Deserialize, Serialize, JsonSchema)] +#[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub(crate) enum RunInteractAction { Get, @@ -252,7 +253,7 @@ pub(crate) struct GatherRunsResult { pub(crate) elapsed_seconds: u64, } -#[derive(Debug, Deserialize, Serialize, JsonSchema)] +#[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema)] #[serde(rename_all = "snake_case")] pub(crate) enum RunEventsAction { List, @@ -294,6 +295,17 @@ impl TryFrom for ValidatedRunEvents { if first > 200 { return Err(ToolError::message("first must be <= 200")); } + if let Some(direction) = params.direction.as_deref() { + if !matches!(direction, "asc" | "desc") { + return Err(ToolError::message("direction must be `asc` or `desc`")); + } + } + if let Some(created_after) = params.created_after.as_deref() { + parse_datetime_filter("created_after", created_after)?; + } + if let Some(created_before) = params.created_before.as_deref() { + parse_datetime_filter("created_before", created_before)?; + } Ok(Self { raw: params }) } } @@ -417,33 +429,169 @@ pub(crate) async fn search_runs( } pub(crate) async fn interact_run( - _client: Arc, - _params: ValidatedInteractRun, + client: Arc, + params: ValidatedInteractRun, ) -> ToolResult { - yield_now().await; - Err(ToolError::message( - "fabro_run_interact is not implemented yet", - )) + let raw = params.raw; + let run_id = client + .resolve_run(&raw.run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + .id; + let result = match raw.action { + RunInteractAction::Get => interact_get(&client, &run_id).await?, + RunInteractAction::Start => { + client + .start_run(&run_id, false) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "summary": run_summary_result(&retrieve_run(&client, &run_id).await?) }) + } + RunInteractAction::Message => { + let message = raw + .message + .expect("validated message action has a message") + .trim() + .to_string(); + client + .steer_run(&run_id, message.clone(), raw.interrupt.unwrap_or(false)) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "message": message, "interrupt": raw.interrupt.unwrap_or(false) }) + } + RunInteractAction::Cancel => { + client + .cancel_run(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "summary": run_summary_result(&retrieve_run(&client, &run_id).await?) }) + } + RunInteractAction::Archive => { + client + .archive_run(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "summary": run_summary_result(&retrieve_run(&client, &run_id).await?) }) + } + RunInteractAction::Unarchive => { + client + .unarchive_run(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "summary": run_summary_result(&retrieve_run(&client, &run_id).await?) }) + } + RunInteractAction::GetQuestions => { + let questions = client + .list_run_questions(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "questions": questions }) + } + RunInteractAction::Answer => { + let question_id = raw + .question_id + .expect("validated answer action has a question_id"); + let body = answer_to_submit_request( + raw.answer.expect("validated answer action has an answer"), + )?; + client + .submit_run_answer(&run_id, &question_id, body) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + json!({ "question_id": question_id, "submitted": true }) + } + }; + + Ok(InteractRunResult { + run_id: run_id.to_string(), + action: raw.action, + result, + }) } pub(crate) async fn gather_runs( - _client: Arc, - _params: ValidatedGatherRuns, + client: Arc, + params: ValidatedGatherRuns, ) -> ToolResult { - yield_now().await; - Err(ToolError::message( - "fabro_run_gather is not implemented yet", - )) + let start = Instant::now(); + let deadline = start + Duration::from_secs(params.timeout_seconds); + let mut run_ids = Vec::with_capacity(params.run_ids.len()); + for selector in params.run_ids { + run_ids.push( + client + .resolve_run(&selector) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + .id, + ); + } + + loop { + let mut summaries = Vec::with_capacity(run_ids.len()); + for run_id in &run_ids { + summaries.push(retrieve_run(&client, run_id).await?); + } + if summaries + .iter() + .all(|run| run.lifecycle.status.is_terminal()) + { + return Ok(GatherRunsResult { + runs: summaries.iter().map(run_summary_result).collect(), + timed_out: false, + elapsed_seconds: start.elapsed().as_secs(), + }); + } + let now = Instant::now(); + if now >= deadline { + return Ok(GatherRunsResult { + runs: summaries.iter().map(run_summary_result).collect(), + timed_out: true, + elapsed_seconds: start.elapsed().as_secs(), + }); + } + let sleep_for = Duration::from_secs(params.poll_interval_seconds).min(deadline - now); + time::sleep(sleep_for).await; + } } pub(crate) async fn run_events( - _client: Arc, - _params: ValidatedRunEvents, + client: Arc, + params: ValidatedRunEvents, ) -> ToolResult { - yield_now().await; - Err(ToolError::message( - "fabro_run_events is not implemented yet", - )) + let raw = params.raw; + let run_id = client + .resolve_run(&raw.run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))? + .id; + let mut events = client + .list_run_events(&run_id, raw.after, Some(event_fetch_limit(&raw))) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + filter_events(&mut events, &raw)?; + if raw.direction.as_deref() == Some("desc") { + events.reverse(); + } + let offset = raw.offset.unwrap_or(0); + let first = raw.first.or(raw.limit).unwrap_or(50).min(200); + let page = events + .into_iter() + .skip(offset) + .take(first) + .collect::>(); + let max_content_length = raw.max_content_length.unwrap_or(20_000); + let results = page + .iter() + .map(|event| run_event_result(event, max_content_length)) + .collect::>>()?; + let next_cursor = page.last().map(|event| event.seq.saturating_add(1)); + + Ok(RunEventsResult { + run_id: run_id.to_string(), + action: raw.action, + events: results, + next_cursor, + }) } pub(crate) fn success_result( @@ -511,7 +659,131 @@ fn validate_len(name: &str, len: usize, min: usize, max: usize) -> ToolResult<() } fn format_tool_error(err: &anyhow::Error) -> String { - format!("{err:#}") + let mut rendered = format!("{err:#}"); + if exit::exit_class_for(err) == Some(ExitClass::AuthRequired) + && !rendered.contains("fabro auth login") + { + rendered.push_str("\nRun `fabro auth login` to authenticate."); + } + rendered +} + +async fn retrieve_run(client: &Client, run_id: &RunId) -> ToolResult { + client + .retrieve_run(run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err)) +} + +async fn interact_get(client: &Client, run_id: &RunId) -> ToolResult { + let summary = retrieve_run(client, run_id).await?; + let projection = client + .get_run_state(run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + Ok(json!({ + "summary": run_summary_result(&summary), + "projection": projection, + })) +} + +fn answer_to_submit_request(answer: Value) -> ToolResult { + let payload = match answer { + Value::Bool(true) => json!({ "kind": "yes" }), + Value::Bool(false) => json!({ "kind": "no" }), + Value::String(text) => json!({ "kind": "text", "text": text }), + Value::Object(mut object) => { + if let Some(option) = object.remove("option") { + json!({ "kind": "selected", "option_key": option }) + } else if let Some(options) = object.remove("options") { + json!({ "kind": "multi_selected", "option_keys": options }) + } else if let Some(text) = object.remove("text") { + json!({ "kind": "text", "text": text }) + } else { + return Err(ToolError::message( + "answer object must contain one of: option, options, text", + )); + } + } + other => { + return Err(ToolError::message(format!( + "unsupported answer value: {other}; expected boolean, string, or object", + ))); + } + }; + serde_json::from_value(payload) + .map_err(|err| ToolError::message(format!("failed to build submit-answer request: {err}"))) +} + +fn event_fetch_limit(params: &FabroRunEventsParams) -> usize { + params + .first + .or(params.limit) + .unwrap_or(50) + .saturating_add(params.offset.unwrap_or(0)) + .clamp(1, 200) +} + +fn filter_events(events: &mut Vec, params: &FabroRunEventsParams) -> ToolResult<()> { + if let Some(event_ids) = params.event_ids.as_ref() { + events.retain(|event| event_ids.contains(&event.event.id)); + } + if let Some(event_types) = params.event_types.as_ref() { + events.retain(|event| { + event_types + .iter() + .any(|event_type| event_type == event.event.event_name()) + }); + } + if let Some(categories) = params.categories.as_ref() { + events.retain(|event| { + let category = event + .event + .event_name() + .split('.') + .next() + .unwrap_or_default(); + categories.iter().any(|candidate| candidate == category) + }); + } + if let Some(created_after) = params.created_after.as_deref() { + let cutoff = parse_datetime_filter("created_after", created_after)?; + events.retain(|event| event.event.ts >= cutoff); + } + if let Some(created_before) = params.created_before.as_deref() { + let cutoff = parse_datetime_filter("created_before", created_before)?; + events.retain(|event| event.event.ts <= cutoff); + } + if matches!(params.action, RunEventsAction::Search) { + if let Some(query) = params.query.as_deref() { + events.retain(|event| { + serde_json::to_string(event).is_ok_and(|serialized| serialized.contains(query)) + }); + } + } + Ok(()) +} + +fn run_event_result( + event: &EventEnvelope, + max_content_length: usize, +) -> ToolResult { + let mut serialized = serde_json::to_string(event) + .map_err(|err| ToolError::message(format!("failed to serialize event: {err}")))?; + let truncated = serialized.len() > max_content_length; + let event_value = if truncated { + serialized.truncate(max_content_length); + Value::String(serialized) + } else { + serde_json::to_value(event) + .map_err(|err| ToolError::message(format!("failed to serialize event: {err}")))? + }; + Ok(RunEventResult { + event_id: event.event.id.clone(), + sequence: event.seq, + event: event_value, + truncated, + }) } async fn build_run_manifest(spec: &CreateRunSpec, cwd: &Path) -> ToolResult { @@ -748,4 +1020,37 @@ mod tests { assert!(err.as_str().contains("goal")); assert!(err.as_str().contains("null")); } + + #[test] + fn answer_payloads_map_to_submit_answer_wire_json() { + let cases = [ + (json!(true), json!({ "kind": "yes" })), + (json!(false), json!({ "kind": "no" })), + (json!("hello"), json!({ "kind": "text", "text": "hello" })), + ( + json!({ "option": "a" }), + json!({ "kind": "selected", "option_key": "a" }), + ), + ( + json!({ "options": ["a", "b"] }), + json!({ "kind": "multi_selected", "option_keys": ["a", "b"] }), + ), + ( + json!({ "text": "hello" }), + json!({ "kind": "text", "text": "hello" }), + ), + ]; + + for (answer, expected) in cases { + let request = answer_to_submit_request(answer).unwrap(); + assert_eq!(serde_json::to_value(request).unwrap(), expected); + } + } + + #[test] + fn unsupported_answer_object_is_rejected() { + let err = answer_to_submit_request(json!({ "value": "yes" })).unwrap_err(); + + assert!(err.as_str().contains("option, options, text")); + } }