use std::collections::HashMap; use std::sync::LazyLock; use async_trait::async_trait; use chrono::{DateTime, NaiveDate, Utc}; use fabro_api::types; use fabro_types::{ PairId, PairMessageRecord, PairMessageRequest, PairRecord, PairTranscriptResponse, Run, RunId, RunPairStatusResponse, StageId, }; use fabro_util::exit::{self, ExitClass}; use schemars::JsonSchema; use serde::Serialize; use serde_json::Value; #[derive(Debug)] pub struct ToolError { message: String, } impl ToolError { pub fn message(message: impl Into) -> Self { Self { message: message.into(), } } pub fn from_anyhow(err: &anyhow::Error) -> Self { Self::message(format_tool_error(err)) } #[must_use] pub fn as_str(&self) -> &str { &self.message } } impl std::fmt::Display for ToolError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str(&self.message) } } impl std::error::Error for ToolError {} pub type ToolResult = Result; #[async_trait] pub trait FabroToolBackend: Send + Sync { async fn create_workflow_version( &self, _source: crate::ValidatedWorkflowVersionCreate, ) -> anyhow::Result { Err(workflow_version_tool_unavailable_error()) } async fn create_run_from_intent(&self, intent: fabro_types::RunIntent) -> anyhow::Result; async fn resolve_run(&self, selector: &str) -> anyhow::Result; async fn retrieve_run(&self, run_id: &RunId) -> anyhow::Result; async fn start_run(&self, run_id: &RunId, resume: bool) -> anyhow::Result; async fn approve_run(&self, run_id: &RunId) -> anyhow::Result; async fn deny_run(&self, run_id: &RunId, reason: Option) -> anyhow::Result; async fn cancel_run(&self, run_id: &RunId) -> anyhow::Result; async fn interrupt_run(&self, run_id: &RunId) -> anyhow::Result<()>; async fn steer_run(&self, run_id: &RunId, text: String, interrupt: bool) -> anyhow::Result<()>; async fn archive_run(&self, run_id: &RunId) -> anyhow::Result; async fn unarchive_run(&self, run_id: &RunId) -> anyhow::Result; async fn list_store_runs(&self) -> anyhow::Result>; async fn list_store_runs_by_parent(&self, parent_id: RunId) -> anyhow::Result>; async fn link_run_parent(&self, child_id: &RunId, parent_id: &RunId) -> anyhow::Result; async fn unlink_run_parent(&self, child_id: &RunId) -> anyhow::Result; async fn get_run_state(&self, run_id: &RunId) -> anyhow::Result; /// The run's stream past `after` (the last `stream_seq` seen; `0` from /// the start), at most `limit` items when a limit is given. async fn list_run_stream( &self, run_id: &RunId, after: u64, limit: Option, ) -> anyhow::Result>; async fn list_run_questions(&self, run_id: &RunId) -> anyhow::Result>; async fn submit_run_answer( &self, run_id: &RunId, question_id: &str, body: types::SubmitAnswerRequest, ) -> anyhow::Result<()>; async fn get_run_pair_status(&self, _run_id: &RunId) -> anyhow::Result { Err(pair_tool_unavailable_error()) } async fn start_run_pair( &self, _run_id: &RunId, _stage_id: StageId, ) -> anyhow::Result { Err(pair_tool_unavailable_error()) } async fn get_run_pair(&self, _run_id: &RunId, _pair_id: &PairId) -> anyhow::Result { Err(pair_tool_unavailable_error()) } async fn end_run_pair(&self, _run_id: &RunId, _pair_id: &PairId) -> anyhow::Result { Err(pair_tool_unavailable_error()) } async fn send_run_pair_message( &self, _run_id: &RunId, _pair_id: &PairId, _request: PairMessageRequest, ) -> anyhow::Result { Err(pair_tool_unavailable_error()) } async fn get_run_pair_transcript( &self, _run_id: &RunId, _pair_id: &PairId, _since_seq: Option, _limit: Option, ) -> anyhow::Result { Err(pair_tool_unavailable_error()) } } fn pair_tool_unavailable_error() -> anyhow::Error { ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into() } pub(crate) fn workflow_version_tool_unavailable_error() -> anyhow::Error { ToolError::message(format!( "{FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME} is not available" )) .into() } #[derive(Debug, Serialize, JsonSchema)] pub struct RunSummaryResult { pub run_id: String, pub parent_id: Option, pub children_count: u64, pub workflow_name: Option, pub workflow_graph_name: Option, pub workflow_slug: Option, pub status: String, pub archived: bool, pub created_at: String, pub started_at: Option, pub completed_at: Option, pub labels: HashMap, pub source_directory: Option, pub repo_origin_url: Option, pub goal: String, } #[derive(Debug, Clone, PartialEq, Serialize)] pub struct ToolDefinition { pub name: &'static str, pub description: &'static str, pub parameters: Value, } pub const FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME: &str = "fabro_workflow_version_create"; pub const FABRO_RUN_CREATE_TOOL_NAME: &str = "fabro_run_create"; pub const FABRO_RUN_SEARCH_TOOL_NAME: &str = "fabro_run_search"; pub const FABRO_RUN_GET_TOOL_NAME: &str = "fabro_run_get"; pub const FABRO_RUN_INTERACT_TOOL_NAME: &str = "fabro_run_interact"; pub const FABRO_RUN_GATHER_TOOL_NAME: &str = "fabro_run_gather"; pub const FABRO_RUN_EVENTS_TOOL_NAME: &str = "fabro_run_events"; pub const FABRO_RUN_PAIR_TOOL_NAME: &str = "fabro_run_pair"; static TOOL_DEFINITIONS: LazyLock> = LazyLock::new(|| { vec![ tool_definition::( FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, "Register supplied workflow file contents and all local dependencies as a reusable immutable workflow version ID. Obtain files with shell/read tools first; this does not create or start a run.", ), tool_definition::( FABRO_RUN_CREATE_TOOL_NAME, "Create runs from registered workflow_version_id values and canonical RunIntent settings. Register contents with fabro_workflow_version_create first. Standalone calls require an explicit target; native workers may inherit the parent target. Starts runs by default.", ), tool_definition::( FABRO_RUN_SEARCH_TOOL_NAME, "Search Fabro workflow runs by id, parent, workflow, labels, status, archival state, and creation time.", ), tool_definition::( FABRO_RUN_GET_TOOL_NAME, "Read-only inspection of a Fabro run: returns its summary, projection, and pending questions without mutating state.", ), tool_definition::( FABRO_RUN_INTERACT_TOOL_NAME, "Control a Fabro run: start, approve, deny, message, interrupt, cancel, archive, unarchive, link or unlink a parent, inspect or answer questions. Use fabro_run_get for read-only inspection.", ), tool_definition::( FABRO_RUN_GATHER_TOOL_NAME, "Wait for Fabro runs to reach terminal states, returning current state on timeout.", ), tool_definition::( FABRO_RUN_PAIR_TOOL_NAME, "Inspect, start, message, end, or read transcript for a live Fabro run pairing session.", ), tool_definition::( FABRO_RUN_EVENTS_TOOL_NAME, "List, inspect, or search stored events for a Fabro workflow run.", ), ] }); #[must_use] pub fn tool_definitions() -> &'static [ToolDefinition] { TOOL_DEFINITIONS.as_slice() } fn tool_definition(name: &'static str, description: &'static str) -> ToolDefinition where T: JsonSchema, { ToolDefinition { name, description, parameters: serde_json::to_value(schemars::schema_for!(T)) .expect("tool parameter schema should serialize"), } } pub(super) fn validate_len(name: &str, len: usize, min: usize, max: usize) -> ToolResult<()> { if len < min { return Err(ToolError::message(format!( "{name} must contain at least {min} item(s)" ))); } if len > max { return Err(ToolError::message(format!( "{name} must contain no more than {max} item(s)" ))); } Ok(()) } pub(super) async fn retrieve_run( backend: &dyn FabroToolBackend, run_id: &RunId, ) -> ToolResult { backend .retrieve_run(run_id) .await .map_err(|err| ToolError::from_anyhow(&err)) } pub(crate) fn run_summary_result(run: &Run) -> RunSummaryResult { RunSummaryResult { run_id: run.id.to_string(), parent_id: run.parent_id.map(|parent_id| parent_id.to_string()), children_count: run.children_count, workflow_name: run.workflow.name.clone(), workflow_graph_name: run.workflow.graph_name.clone(), workflow_slug: run.workflow.slug.clone(), status: run.lifecycle.status.kind().to_string(), archived: run.lifecycle.archived, created_at: run.timestamps.created_at.to_rfc3339(), started_at: run .timestamps .started_at .map(|timestamp| timestamp.to_rfc3339()), completed_at: run .timestamps .completed_at .map(|timestamp| timestamp.to_rfc3339()), labels: run.labels.clone(), source_directory: run.source_directory.clone(), repo_origin_url: run .repository .as_ref() .and_then(|repository| repository.origin_url.clone()), goal: run.goal.clone(), } } pub(crate) fn parse_datetime_filter(name: &str, raw: &str) -> ToolResult> { if let Ok(timestamp) = DateTime::parse_from_rfc3339(raw) { return Ok(timestamp.with_timezone(&Utc)); } let date = NaiveDate::parse_from_str(raw, "%Y-%m-%d").map_err(|err| { ToolError::message(format!("{name} must be RFC3339 or YYYY-MM-DD: {err}")) })?; let datetime = date .and_hms_opt(0, 0, 0) .ok_or_else(|| ToolError::message(format!("{name} contains an invalid date")))?; Ok(DateTime::from_naive_utc_and_offset(datetime, Utc)) } fn format_tool_error(err: &anyhow::Error) -> String { 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 } #[cfg(test)] mod tests { use chrono::{TimeZone, Utc}; use fabro_api::types::Usage; use fabro_types::{ RunLifecycle, RunLinks, RunOrigin, RunStatus, RunTimestamps, WorkflowRef, test_support, }; use super::*; fn shared_tool_names() -> Vec<&'static str> { tool_definitions() .iter() .map(|definition| definition.name) .collect() } #[test] fn shared_tool_definitions_include_run_management_catalog() { assert_eq!(shared_tool_names(), vec![ FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME, FABRO_RUN_CREATE_TOOL_NAME, FABRO_RUN_SEARCH_TOOL_NAME, FABRO_RUN_GET_TOOL_NAME, FABRO_RUN_INTERACT_TOOL_NAME, FABRO_RUN_GATHER_TOOL_NAME, FABRO_RUN_PAIR_TOOL_NAME, FABRO_RUN_EVENTS_TOOL_NAME, ]); } #[test] fn workflow_version_create_has_strict_content_schema() { let definition = tool_definitions() .iter() .find(|definition| definition.name == FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME) .expect("workflow version creation should be in the shared catalog"); let schema = &definition.parameters; assert_eq!(schema["additionalProperties"], false); assert_eq!(schema["properties"].as_object().unwrap().len(), 2); assert_eq!( schema["required"], serde_json::json!(["entrypoint", "files"]) ); } #[test] fn pair_tool_definition_exposes_pair_schema() { let definition = tool_definitions() .iter() .find(|definition| definition.name == FABRO_RUN_PAIR_TOOL_NAME) .expect("pair tool should be in the shared catalog"); let schema = &definition.parameters; let schema_text = schema.to_string(); assert_eq!( definition.description, "Inspect, start, message, end, or read transcript for a live Fabro run pairing session." ); for field in [ "action", "run_id", "pair_id", "stage_id", "text", "client_message_id", "since_seq", "limit", ] { assert!( schema.pointer(&format!("/properties/{field}")).is_some(), "pair schema should expose {field}: {schema}" ); } for action in ["status", "start", "get", "message", "end", "transcript"] { assert!( schema_text.contains(&format!("\"{action}\"")), "pair schema should expose action {action}: {schema}" ); } } #[test] fn interact_tool_definition_exposes_approval_schema() { let definition = tool_definitions() .iter() .find(|definition| definition.name == FABRO_RUN_INTERACT_TOOL_NAME) .expect("interact tool should be in the shared catalog"); let schema = &definition.parameters; let schema_text = schema.to_string(); assert!( definition.description.contains("approve") && definition.description.contains("deny"), "interact description should include approval actions: {}", definition.description ); for field in ["action", "run_id", "reason"] { assert!( schema.pointer(&format!("/properties/{field}")).is_some(), "interact schema should expose {field}: {schema}" ); } for action in ["approve", "deny"] { assert!( schema_text.contains(&format!("\"{action}\"")), "interact schema should expose action {action}: {schema}" ); } } #[test] fn run_summary_result_includes_parent_metadata() { let parent_id = run_id("01KRBZW4DW0000000000000002"); let run = Run { id: run_id("01KRBZW5C00000000000000001"), parent_id: Some(parent_id), children_count: 3, title: "test".to_string(), goal: "test".to_string(), workflow: WorkflowRef { slug: Some("simple".to_string()), name: Some("Simple".to_string()), graph_name: Some("GraphName".to_string()), node_count: 0, edge_count: 0, }, automation: None, repository: None, created_by: test_support::test_principal(), origin: RunOrigin::default(), labels: HashMap::new(), lifecycle: RunLifecycle { status: RunStatus::Submitted, approval: None, pending_control: None, queue_position: None, error: None, archived: false, archived_at: None, }, sandbox: None, models: Vec::new(), source_directory: None, timestamps: RunTimestamps { created_at: Utc.with_ymd_and_hms(2026, 5, 11, 12, 0, 0).unwrap(), started_at: None, last_event_at: None, completed_at: None, }, timing: None, usage: Usage::default(), size: fabro_types::RunSize::default(), ask_fabro: fabro_types::AskFabro::default(), diff: None, pull_request: None, current_question: None, superseded_by: None, retried_from: None, links: RunLinks { web: None }, }; let summary = run_summary_result(&run); assert_eq!(summary.parent_id, Some(parent_id.to_string())); assert_eq!(summary.children_count, 3); assert_eq!(summary.workflow_name.as_deref(), Some("Simple")); assert_eq!(summary.workflow_graph_name.as_deref(), Some("GraphName")); } fn run_id(raw: &str) -> RunId { raw.parse().expect("test run id should parse") } }