From 32b88d78337793867b19ac807cfae5cf0fac04fb Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 19 Apr 2026 19:31:03 -0400 Subject: [PATCH] fix(async-fs): move remaining filesystem work off Tokio paths Replace the remaining blocking filesystem touches in shared async code with Tokio-native I/O or explicit blocking boundaries. This keeps provider file loading, workflow metadata rebuilds, and related export paths compatible with the stricter clippy async-fs rules without changing their external behavior. --- lib/crates/fabro-agent/src/cli.rs | 15 +- lib/crates/fabro-devcontainer/src/compose.rs | 90 +++---- lib/crates/fabro-devcontainer/src/lib.rs | 77 +++--- .../fabro-llm/src/providers/anthropic.rs | 92 ++++---- lib/crates/fabro-llm/src/providers/common.rs | 13 +- lib/crates/fabro-llm/src/providers/gemini.rs | 222 +++++++++--------- lib/crates/fabro-llm/src/providers/openai.rs | 196 ++++++++-------- lib/crates/fabro-workflow/src/artifact.rs | 13 +- .../fabro-workflow/src/artifact_snapshot.rs | 2 +- .../src/handler/manager_loop.rs | 3 +- .../fabro-workflow/src/operations/create.rs | 49 ++-- .../src/operations/rebuild_meta.rs | 64 +++-- lib/crates/fabro-workflow/src/run_dump.rs | 2 +- lib/crates/fabro-workflow/src/run_lookup.rs | 3 +- 14 files changed, 449 insertions(+), 392 deletions(-) diff --git a/lib/crates/fabro-agent/src/cli.rs b/lib/crates/fabro-agent/src/cli.rs index 5515986a3..c5c5c2aa0 100644 --- a/lib/crates/fabro-agent/src/cli.rs +++ b/lib/crates/fabro-agent/src/cli.rs @@ -16,6 +16,7 @@ use fabro_llm::types::{Request, Response}; use fabro_mcp::config::McpServerSettings; use fabro_model::{Catalog, ModelHandle, Provider}; use fabro_util::terminal::Styles; +use tokio::io::{AsyncWriteExt, stdout}; use tokio::signal; use tokio::sync::Mutex as AsyncMutex; @@ -552,18 +553,12 @@ pub async fn run_with_args_and_client( tokio::spawn(async move { match output_format { OutputFormat::Json => { + let mut stdout = stdout(); while let Ok(event) = rx.recv().await { if let Ok(json) = serde_json::to_string(&event) { - #[expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: blocking stdout inside tokio::spawn. Acceptable \ - today (low-volume event stream, CLI output), but should \ - migrate to tokio::io::stdout to avoid worker stalls under \ - pipe backpressure." - )] - let mut stdout = std::io::stdout().lock(); - let _ = writeln!(stdout, "{json}"); - let _ = stdout.flush(); + let _ = stdout.write_all(json.as_bytes()).await; + let _ = stdout.write_all(b"\n").await; + let _ = stdout.flush().await; } } } diff --git a/lib/crates/fabro-devcontainer/src/compose.rs b/lib/crates/fabro-devcontainer/src/compose.rs index 59b05119c..8b4a5a95f 100644 --- a/lib/crates/fabro-devcontainer/src/compose.rs +++ b/lib/crates/fabro-devcontainer/src/compose.rs @@ -1,11 +1,8 @@ -#![expect( - clippy::disallowed_methods, - reason = "sync parse of docker-compose.yml files during devcontainer resolution; not on a Tokio path" -)] - use std::collections::HashMap; use std::path::{Path, PathBuf}; +use tokio::fs; + /// Extracted configuration from a Docker Compose service. #[derive(Debug, Clone, Default)] pub(crate) struct ComposeServiceSpec { @@ -24,11 +21,11 @@ pub(crate) struct ComposeBuild { } /// Parse a Docker Compose file and extract config for the named service. -pub(crate) fn parse_compose( +pub(crate) async fn parse_compose( compose_path: &Path, service_name: &str, ) -> Result { - let contents = std::fs::read_to_string(compose_path).map_err(|e| { + let contents = fs::read_to_string(compose_path).await.map_err(|e| { format!( "failed to read compose file {}: {e}", compose_path.display() @@ -164,7 +161,7 @@ fn parse_environment(service: &serde_yaml::Value) -> HashMap { /// Parse multiple Docker Compose files and merge config for the named service. /// Later files override earlier files for image/build/user; ports accumulate /// (deduped); environment keys from later files override earlier ones. -pub(crate) fn parse_compose_multi( +pub(crate) async fn parse_compose_multi( compose_paths: &[PathBuf], service_name: &str, ) -> Result { @@ -172,7 +169,8 @@ pub(crate) fn parse_compose_multi( let mut found_service = false; for path in compose_paths { - let contents = std::fs::read_to_string(path) + let contents = fs::read_to_string(path) + .await .map_err(|e| format!("failed to read compose file {}: {e}", path.display()))?; let doc: serde_yaml::Value = serde_yaml::from_str(&contents) @@ -231,8 +229,8 @@ mod tests { f } - #[test] - fn service_with_image_only() { + #[tokio::test] + async fn service_with_image_only() { let f = write_compose( r" services: @@ -240,7 +238,7 @@ services: image: nginx:latest ", ); - let cfg = parse_compose(f.path(), "web").unwrap(); + let cfg = parse_compose(f.path(), "web").await.unwrap(); assert_eq!(cfg.image.as_deref(), Some("nginx:latest")); assert!(cfg.build.is_none()); assert!(cfg.ports.is_empty()); @@ -248,8 +246,8 @@ services: assert!(cfg.user.is_none()); } - #[test] - fn service_with_build_string() { + #[tokio::test] + async fn service_with_build_string() { let f = write_compose( r" services: @@ -257,14 +255,14 @@ services: build: ./src ", ); - let cfg = parse_compose(f.path(), "app").unwrap(); + let cfg = parse_compose(f.path(), "app").await.unwrap(); let build = cfg.build.unwrap(); assert_eq!(build.context, "./src"); assert!(build.dockerfile.is_none()); } - #[test] - fn service_with_build_object() { + #[tokio::test] + async fn service_with_build_object() { let f = write_compose( r" services: @@ -274,14 +272,14 @@ services: dockerfile: Dockerfile.dev ", ); - let cfg = parse_compose(f.path(), "app").unwrap(); + let cfg = parse_compose(f.path(), "app").await.unwrap(); let build = cfg.build.unwrap(); assert_eq!(build.context, "./app"); assert_eq!(build.dockerfile.as_deref(), Some("Dockerfile.dev")); } - #[test] - fn ports_various_formats() { + #[tokio::test] + async fn ports_various_formats() { let f = write_compose( r#" services: @@ -294,12 +292,12 @@ services: - "9090:9090/tcp" "#, ); - let cfg = parse_compose(f.path(), "web").unwrap(); + let cfg = parse_compose(f.path(), "web").await.unwrap(); assert_eq!(cfg.ports, vec![80, 3000, 5432, 9090]); } - #[test] - fn environment_as_array() { + #[tokio::test] + async fn environment_as_array() { let f = write_compose( r#" services: @@ -310,14 +308,14 @@ services: - "DEBUG=true" "#, ); - let cfg = parse_compose(f.path(), "app").unwrap(); + let cfg = parse_compose(f.path(), "app").await.unwrap(); assert_eq!(cfg.environment.len(), 2); assert_eq!(cfg.environment["DATABASE_URL"], "postgres://localhost/db"); assert_eq!(cfg.environment["DEBUG"], "true"); } - #[test] - fn environment_as_object() { + #[tokio::test] + async fn environment_as_object() { let f = write_compose( r" services: @@ -328,14 +326,14 @@ services: PORT: 3000 ", ); - let cfg = parse_compose(f.path(), "app").unwrap(); + let cfg = parse_compose(f.path(), "app").await.unwrap(); assert_eq!(cfg.environment.len(), 2); assert_eq!(cfg.environment["RAILS_ENV"], "production"); assert_eq!(cfg.environment["PORT"], "3000"); } - #[test] - fn service_not_found() { + #[tokio::test] + async fn service_not_found() { let f = write_compose( r" services: @@ -343,18 +341,20 @@ services: image: nginx ", ); - let err = parse_compose(f.path(), "missing").unwrap_err(); + let err = parse_compose(f.path(), "missing").await.unwrap_err(); assert!(err.contains("service 'missing' not found")); } - #[test] - fn file_not_found() { - let err = parse_compose(Path::new("/nonexistent/docker-compose.yml"), "web").unwrap_err(); + #[tokio::test] + async fn file_not_found() { + let err = parse_compose(Path::new("/nonexistent/docker-compose.yml"), "web") + .await + .unwrap_err(); assert!(err.contains("failed to read compose file")); } - #[test] - fn service_with_user() { + #[tokio::test] + async fn service_with_user() { let f = write_compose( r#" services: @@ -363,12 +363,12 @@ services: user: "1000:1000" "#, ); - let cfg = parse_compose(f.path(), "app").unwrap(); + let cfg = parse_compose(f.path(), "app").await.unwrap(); assert_eq!(cfg.user.as_deref(), Some("1000:1000")); } - #[test] - fn multi_compose_merge() { + #[tokio::test] + async fn multi_compose_merge() { let base = write_compose( r#" services: @@ -393,15 +393,15 @@ services: "#, ); let paths = vec![base.path().to_path_buf(), over.path().to_path_buf()]; - let cfg = parse_compose_multi(&paths, "app").unwrap(); + let cfg = parse_compose_multi(&paths, "app").await.unwrap(); assert_eq!(cfg.image.as_deref(), Some("node:22")); assert_eq!(cfg.ports, vec![3000, 9229]); assert_eq!(cfg.environment["NODE_ENV"], "development"); assert_eq!(cfg.environment["DEBUG"], "true"); } - #[test] - fn multi_compose_service_not_found() { + #[tokio::test] + async fn multi_compose_service_not_found() { let f = write_compose( r" services: @@ -410,12 +410,12 @@ services: ", ); let paths = vec![f.path().to_path_buf()]; - let err = parse_compose_multi(&paths, "missing").unwrap_err(); + let err = parse_compose_multi(&paths, "missing").await.unwrap_err(); assert!(err.contains("service 'missing' not found")); } - #[test] - fn multi_compose_skips_file_without_service() { + #[tokio::test] + async fn multi_compose_skips_file_without_service() { let base = write_compose( r" services: @@ -431,7 +431,7 @@ services: ", ); let paths = vec![base.path().to_path_buf(), over.path().to_path_buf()]; - let cfg = parse_compose_multi(&paths, "app").unwrap(); + let cfg = parse_compose_multi(&paths, "app").await.unwrap(); assert_eq!(cfg.image.as_deref(), Some("node:22")); } } diff --git a/lib/crates/fabro-devcontainer/src/lib.rs b/lib/crates/fabro-devcontainer/src/lib.rs index b1e4af4ba..81507f6d4 100644 --- a/lib/crates/fabro-devcontainer/src/lib.rs +++ b/lib/crates/fabro-devcontainer/src/lib.rs @@ -11,6 +11,7 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use fabro_util::env::SystemEnv; +use tokio::fs; pub use types::DevcontainerJson; /// Lifecycle command — string, array, or object (parallel) form. @@ -161,14 +162,8 @@ pub struct DevcontainerResolver; impl DevcontainerResolver { /// path: repo root (or explicit .devcontainer/ path) - #[expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: DevcontainerResolver::resolve does sync std::fs::read_to_string / \ - read_dir across several devcontainer.json lookups. One-shot per workflow run \ - (not per-request); acceptable today but should migrate to tokio::fs." - )] pub async fn resolve(path: &Path) -> Result { - let (json_path, devcontainer) = Self::find_and_parse(path)?; + let (json_path, devcontainer) = Self::find_and_parse(path).await?; let repo_root = Self::repo_root_from_json_path(&json_path, path); let base_dir = json_path.parent().unwrap_or(path); @@ -219,6 +214,7 @@ impl DevcontainerResolver { .clone(); let compose_config = compose::parse_compose_multi(&compose_paths, &service_name) + .await .map_err(DevcontainerError::Compose)?; let mut environment = HashMap::new(); @@ -241,9 +237,11 @@ impl DevcontainerResolver { let df_path = compose_base_dir .join(&build.context) .join(build.dockerfile.as_deref().unwrap_or("Dockerfile")); - std::fs::read_to_string(&df_path).map_err(|source| DevcontainerError::ReadFile { - path: df_path, - source, + fs::read_to_string(&df_path).await.map_err(|source| { + DevcontainerError::ReadFile { + path: df_path, + source, + } })? } else { format!( @@ -307,7 +305,7 @@ impl DevcontainerResolver { build.dockerfile.as_deref().unwrap_or("Dockerfile"), &vars, )); - let content = std::fs::read_to_string(&df_path).map_err(|source| { + let content = fs::read_to_string(&df_path).await.map_err(|source| { DevcontainerError::ReadFile { path: df_path, source, @@ -415,12 +413,7 @@ impl DevcontainerResolver { }) } - #[expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: sync std::fs helpers for devcontainer.json lookup; called once at \ - workflow startup via resolve(). Should migrate to tokio::fs with resolve()." - )] - fn find_and_parse(path: &Path) -> Result<(PathBuf, DevcontainerJson)> { + async fn find_and_parse(path: &Path) -> Result<(PathBuf, DevcontainerJson)> { // Check standard locations let candidates = [ path.join(".devcontainer/devcontainer.json"), @@ -429,7 +422,7 @@ impl DevcontainerResolver { for candidate in &candidates { if candidate.exists() { - let raw = std::fs::read_to_string(candidate).map_err(|source| { + let raw = fs::read_to_string(candidate).await.map_err(|source| { DevcontainerError::ReadFile { path: candidate.clone(), source, @@ -444,10 +437,12 @@ impl DevcontainerResolver { // Check if path itself is a devcontainer.json if path.is_file() && path.file_name().is_some_and(|n| n == "devcontainer.json") { let raw = - std::fs::read_to_string(path).map_err(|source| DevcontainerError::ReadFile { - path: path.to_path_buf(), - source, - })?; + fs::read_to_string(path) + .await + .map_err(|source| DevcontainerError::ReadFile { + path: path.to_path_buf(), + source, + })?; let stripped = jsonc::strip_jsonc(&raw); let parsed: DevcontainerJson = serde_json::from_str(&stripped)?; return Ok((path.to_path_buf(), parsed)); @@ -457,23 +452,43 @@ impl DevcontainerResolver { // devcontainer.json let devcontainer_dir = path.join(".devcontainer"); if devcontainer_dir.is_dir() { - let mut subdirs: Vec = std::fs::read_dir(&devcontainer_dir) - .map_err(|source| DevcontainerError::ReadFile { + let mut entries = fs::read_dir(&devcontainer_dir).await.map_err(|source| { + DevcontainerError::ReadFile { path: devcontainer_dir.clone(), source, - })? - .filter_map(std::result::Result::ok) - .filter(|entry| entry.path().is_dir()) - .map(|entry| entry.path()) - .filter(|dir| dir.join("devcontainer.json").exists()) - .collect(); + } + })?; + let mut subdirs = Vec::new(); + + while let Some(entry) = + entries + .next_entry() + .await + .map_err(|source| DevcontainerError::ReadFile { + path: devcontainer_dir.clone(), + source, + })? + { + let entry_path = entry.path(); + let file_type = + entry + .file_type() + .await + .map_err(|source| DevcontainerError::ReadFile { + path: entry_path.clone(), + source, + })?; + if file_type.is_dir() && entry_path.join("devcontainer.json").exists() { + subdirs.push(entry_path); + } + } // Sort alphabetically to get deterministic first pick subdirs.sort(); if let Some(subdir) = subdirs.first() { let candidate = subdir.join("devcontainer.json"); - let raw = std::fs::read_to_string(&candidate).map_err(|source| { + let raw = fs::read_to_string(&candidate).await.map_err(|source| { DevcontainerError::ReadFile { path: candidate.clone(), source, diff --git a/lib/crates/fabro-llm/src/providers/anthropic.rs b/lib/crates/fabro-llm/src/providers/anthropic.rs index bfc1b7014..9705e2a92 100644 --- a/lib/crates/fabro-llm/src/providers/anthropic.rs +++ b/lib/crates/fabro-llm/src/providers/anthropic.rs @@ -232,7 +232,7 @@ fn parse_content_block(block: &serde_json::Value) -> Option { } /// Translate a unified `ContentPart` to an Anthropic content block JSON value. -fn content_part_to_api(part: &ContentPart) -> Option { +async fn content_part_to_api(part: &ContentPart) -> Option { match part { ContentPart::Text(text) => Some(serde_json::json!({"type": "text", "text": text})), ContentPart::ToolCall(tc) => Some(serde_json::json!({ @@ -270,7 +270,7 @@ fn content_part_to_api(part: &ContentPart) -> Option { ContentPart::Image(img) => { if let Some(url) = &img.url { if common::is_file_path(url) { - return match common::load_file_as_base64(url) { + return match common::load_file_as_base64(url).await { Ok((b64, mime)) => Some(serde_json::json!({ "type": "image", "source": {"type": "base64", "media_type": mime, "data": b64} @@ -290,7 +290,7 @@ fn content_part_to_api(part: &ContentPart) -> Option { ContentPart::Document(doc) => { if let Some(url) = &doc.url { if common::is_file_path(url) { - return match common::load_file_as_base64(url) { + return match common::load_file_as_base64(url).await { Ok((b64, mime)) => Some(serde_json::json!({ "type": "document", "source": {"type": "base64", "media_type": mime, "data": b64} @@ -318,7 +318,7 @@ fn content_part_to_api(part: &ContentPart) -> Option { /// /// Handles: role mapping, content block translation, strict alternation /// (merging consecutive same-role messages), and tool results in user messages. -fn translate_messages(messages: &[&Message]) -> Vec { +async fn translate_messages(messages: &[&Message]) -> Vec { let mut api_messages: Vec = Vec::new(); for msg in messages { @@ -330,8 +330,12 @@ fn translate_messages(messages: &[&Message]) -> Vec { Role::System | Role::Developer => continue, }; - let content: Vec = - msg.content.iter().filter_map(content_part_to_api).collect(); + let mut content = Vec::new(); + for part in &msg.content { + if let Some(block) = content_part_to_api(part).await { + content.push(block); + } + } if content.is_empty() { continue; @@ -1076,13 +1080,13 @@ fn merge_provider_options( /// Build an Anthropic API request and HTTP request builder for the given /// unified request. -fn build_api_request( +async fn build_api_request( adapter: &Adapter, request: &Request, stream: bool, ) -> (ApiRequest, fabro_http::RequestBuilder) { let (system, other_messages) = extract_system_prompt(&request.messages); - let mut api_messages = translate_messages(&other_messages); + let mut api_messages = translate_messages(&other_messages).await; let mut omit_tools = false; let tool_choice_json = request.tool_choice.as_ref().and_then(|tc| { @@ -1243,7 +1247,7 @@ impl ProviderAdapter for Adapter { return self.complete_via_stream(request).await; } - let (_api_request, req_builder) = build_api_request(self, request, false); + let (_api_request, req_builder) = build_api_request(self, request, false).await; let mut req = req_builder; if let Some(t) = self.http.request_timeout { @@ -1311,7 +1315,7 @@ impl ProviderAdapter for Adapter { if let Some(tc) = &request.tool_choice { validate_tool_choice(self, tc)?; } - let (_api_request, req_builder) = build_api_request(self, request, true); + let (_api_request, req_builder) = build_api_request(self, request, true).await; let http_resp = req_builder .send() @@ -1706,15 +1710,15 @@ mod tests { assert_eq!(arr[0]["cache_control"]["type"], "ephemeral"); } - #[test] - fn build_api_request_omits_whitespace_only_system_prompt() { + #[tokio::test] + async fn build_api_request_omits_whitespace_only_system_prompt() { let adapter = Adapter::new("test-key"); let request = Request { messages: vec![Message::system(" \n\t"), Message::user("Hello")], ..make_base_request() }; - let (api_request, _req_builder) = build_api_request(&adapter, &request, false); + let (api_request, _req_builder) = build_api_request(&adapter, &request, false).await; assert!( api_request.system.is_none(), "whitespace-only system prompts should be omitted" @@ -2000,44 +2004,50 @@ mod tests { } } - #[test] - fn document_url_translates_to_url_source() { + #[tokio::test] + async fn document_url_translates_to_url_source() { let part = ContentPart::Document(DocumentData { url: Some("https://example.com/doc.pdf".to_string()), data: None, media_type: None, file_name: None, }); - let result = content_part_to_api(&part).expect("should produce JSON"); + let result = content_part_to_api(&part) + .await + .expect("should produce JSON"); assert_eq!(result["type"], "document"); assert_eq!(result["source"]["type"], "url"); assert_eq!(result["source"]["url"], "https://example.com/doc.pdf"); } - #[test] - fn document_base64_data_translates_to_base64_source() { + #[tokio::test] + async fn document_base64_data_translates_to_base64_source() { let part = ContentPart::Document(DocumentData { url: None, data: Some(vec![0x25, 0x50, 0x44, 0x46]), media_type: Some("application/pdf".to_string()), file_name: Some("test.pdf".to_string()), }); - let result = content_part_to_api(&part).expect("should produce JSON"); + let result = content_part_to_api(&part) + .await + .expect("should produce JSON"); assert_eq!(result["type"], "document"); assert_eq!(result["source"]["type"], "base64"); assert_eq!(result["source"]["media_type"], "application/pdf"); assert!(result["source"]["data"].as_str().is_some()); } - #[test] - fn document_base64_defaults_to_pdf_mime() { + #[tokio::test] + async fn document_base64_defaults_to_pdf_mime() { let part = ContentPart::Document(DocumentData { url: None, data: Some(vec![1, 2, 3]), media_type: None, file_name: None, }); - let result = content_part_to_api(&part).expect("should produce JSON"); + let result = content_part_to_api(&part) + .await + .expect("should produce JSON"); assert_eq!(result["source"]["media_type"], "application/pdf"); } @@ -2148,14 +2158,16 @@ mod tests { assert_eq!(body["top_k"], 40); } - #[test] - fn audio_produces_text_fallback() { + #[tokio::test] + async fn audio_produces_text_fallback() { let part = ContentPart::Audio(AudioData { url: Some("https://example.com/audio.wav".to_string()), data: None, media_type: None, }); - let result = content_part_to_api(&part).expect("should produce JSON"); + let result = content_part_to_api(&part) + .await + .expect("should produce JSON"); assert_eq!(result["type"], "text"); assert_eq!( result["text"], @@ -2163,51 +2175,51 @@ mod tests { ); } - #[test] - fn build_api_request_maps_reasoning_effort_to_output_config() { + #[tokio::test] + async fn build_api_request_maps_reasoning_effort_to_output_config() { let adapter = Adapter::new("test-key"); let request = Request { reasoning_effort: Some(ReasoningEffort::Medium), ..make_base_request() }; - let (api_request, _req_builder) = build_api_request(&adapter, &request, false); + let (api_request, _req_builder) = build_api_request(&adapter, &request, false).await; assert_eq!( api_request.output_config, Some(serde_json::json!({"effort": "medium"})) ); } - #[test] - fn build_api_request_omits_output_config_when_no_reasoning_effort() { + #[tokio::test] + async fn build_api_request_omits_output_config_when_no_reasoning_effort() { let adapter = Adapter::new("test-key"); let request = make_base_request(); - let (api_request, _req_builder) = build_api_request(&adapter, &request, false); + let (api_request, _req_builder) = build_api_request(&adapter, &request, false).await; assert!(api_request.output_config.is_none()); } - #[test] - fn build_api_request_sets_speed() { + #[tokio::test] + async fn build_api_request_sets_speed() { let adapter = Adapter::new("test-key"); let request = Request { speed: Some("fast".to_string()), ..make_base_request() }; - let (api_request, _req_builder) = build_api_request(&adapter, &request, false); + let (api_request, _req_builder) = build_api_request(&adapter, &request, false).await; assert_eq!(api_request.speed, Some("fast".to_string())); } - #[test] - fn build_api_request_injects_fast_mode_beta_header() { + #[tokio::test] + async fn build_api_request_injects_fast_mode_beta_header() { let adapter = Adapter::new("test-key"); let request = Request { speed: Some("fast".to_string()), ..make_base_request() }; - let (_api_request, req_builder) = build_api_request(&adapter, &request, false); + let (_api_request, req_builder) = build_api_request(&adapter, &request, false).await; let built = req_builder.build().expect("should build request"); let beta = built .headers() @@ -2244,8 +2256,8 @@ mod tests { assert_eq!(effort_to_budget_tokens("max", 16_000), 16_000); } - #[test] - fn build_api_request_falls_back_to_thinking_budget_for_non_effort_model() { + #[tokio::test] + async fn build_api_request_falls_back_to_thinking_budget_for_non_effort_model() { let adapter = Adapter::new("test-key"); let request = Request { model: "claude-sonnet-4-5".to_string(), @@ -2254,7 +2266,7 @@ mod tests { ..make_base_request() }; - let (api_request, _req_builder) = build_api_request(&adapter, &request, false); + let (api_request, _req_builder) = build_api_request(&adapter, &request, false).await; assert!( api_request.output_config.is_none(), "non-effort models must not receive output_config" diff --git a/lib/crates/fabro-llm/src/providers/common.rs b/lib/crates/fabro-llm/src/providers/common.rs index 893def7a5..9b592bfbf 100644 --- a/lib/crates/fabro-llm/src/providers/common.rs +++ b/lib/crates/fabro-llm/src/providers/common.rs @@ -1,7 +1,7 @@ use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use fabro_http::HeaderMap; -use tokio::time; +use tokio::{fs, time}; use tracing::warn; use crate::error::{Error, error_from_status_code}; @@ -92,14 +92,7 @@ pub fn mime_from_extension(path: &str) -> &str { /// /// # Errors /// Returns an error if the file cannot be read. -#[expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: sync std::fs::read for file:// attachments, invoked from sync \ - translators (translate_input/translate_messages) across all providers. Pre-existing; \ - 7 call sites in sync translators would need restructuring to wrap in spawn_blocking \ - at each async chokepoint. file:// URLs are rare in practice; revisit if usage grows." -)] -pub fn load_file_as_base64(path: &str) -> Result<(String, String), std::io::Error> { +pub async fn load_file_as_base64(path: &str) -> Result<(String, String), std::io::Error> { let expanded = path.strip_prefix("~/").map_or_else( || path.to_string(), |rest| { @@ -107,7 +100,7 @@ pub fn load_file_as_base64(path: &str) -> Result<(String, String), std::io::Erro format!("{home}/{rest}") }, ); - let data = std::fs::read(&expanded).map_err(|err| { + let data = fs::read(&expanded).await.map_err(|err| { std::io::Error::new(err.kind(), format!("read attachment {expanded}: {err}")) })?; let mime = mime_from_extension(&expanded).to_string(); diff --git a/lib/crates/fabro-llm/src/providers/gemini.rs b/lib/crates/fabro-llm/src/providers/gemini.rs index 03020c129..487b0eaa3 100644 --- a/lib/crates/fabro-llm/src/providers/gemini.rs +++ b/lib/crates/fabro-llm/src/providers/gemini.rs @@ -213,7 +213,7 @@ fn build_tool_call_id_to_name(messages: &[&Message]) -> std::collections::HashMa } /// Translate unified messages to Gemini content format. -fn translate_messages(messages: &[&Message]) -> Vec { +async fn translate_messages(messages: &[&Message]) -> Vec { let id_to_name = build_tool_call_id_to_name(messages); let mut contents: Vec = Vec::new(); @@ -224,10 +224,9 @@ fn translate_messages(messages: &[&Message]) -> Vec { Role::System | Role::Developer => continue, }; - let parts: Vec = msg - .content - .iter() - .filter_map(|part| match part { + let mut parts = Vec::new(); + for part in &msg.content { + let maybe_part = match part { ContentPart::Text(text) => Some(serde_json::json!({"text": text})), ContentPart::ToolCall(tc) => { let mut part_json = serde_json::json!({ @@ -246,72 +245,72 @@ fn translate_messages(messages: &[&Message]) -> Vec { } Some(part_json) } - ContentPart::Image(img) => { - img.url.as_ref().map_or_else( - || { - img.data.as_ref().map(|data| { - let mime = img.media_type.as_deref().unwrap_or("image/png"); - let b64 = BASE64_STANDARD.encode(data); - serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) - }) - }, - |url| { - if common::is_file_path(url) { - match common::load_file_as_base64(url) { - Ok((b64, mime)) => Some(serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}})), - Err(_) => None, - } - } else { - let mime = img.media_type.as_deref().unwrap_or("image/png"); - Some(serde_json::json!({"fileData": {"mimeType": mime, "fileUri": url}})) + ContentPart::Image(img) => match &img.url { + Some(url) => { + if common::is_file_path(url) { + match common::load_file_as_base64(url).await { + Ok((b64, mime)) => Some(serde_json::json!({ + "inlineData": {"mimeType": mime, "data": b64} + })), + Err(_) => None, } - }, - ) - } - ContentPart::Audio(audio) => { - audio.url.as_ref().map_or_else( - || { - audio.data.as_ref().map(|data| { - let mime = audio.media_type.as_deref().unwrap_or("audio/wav"); - let b64 = BASE64_STANDARD.encode(data); - serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) - }) - }, - |url| { - if common::is_file_path(url) { - match common::load_file_as_base64(url) { - Ok((b64, mime)) => Some(serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}})), - Err(_) => None, - } - } else { - let mime = audio.media_type.as_deref().unwrap_or("audio/wav"); - Some(serde_json::json!({"fileData": {"mimeType": mime, "fileUri": url}})) + } else { + let mime = img.media_type.as_deref().unwrap_or("image/png"); + Some(serde_json::json!({ + "fileData": {"mimeType": mime, "fileUri": url} + })) + } + } + None => img.data.as_ref().map(|data| { + let mime = img.media_type.as_deref().unwrap_or("image/png"); + let b64 = BASE64_STANDARD.encode(data); + serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) + }), + }, + ContentPart::Audio(audio) => match &audio.url { + Some(url) => { + if common::is_file_path(url) { + match common::load_file_as_base64(url).await { + Ok((b64, mime)) => Some(serde_json::json!({ + "inlineData": {"mimeType": mime, "data": b64} + })), + Err(_) => None, } - }, - ) - } - ContentPart::Document(doc) => { - doc.url.as_ref().map_or_else( - || { - doc.data.as_ref().map(|data| { - let mime = doc.media_type.as_deref().unwrap_or("application/pdf"); - let b64 = BASE64_STANDARD.encode(data); - serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) - }) - }, - |url| { - if common::is_file_path(url) { - match common::load_file_as_base64(url) { - Ok((b64, mime)) => Some(serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}})), - Err(_) => None, - } - } else { - let mime = doc.media_type.as_deref().unwrap_or("application/pdf"); - Some(serde_json::json!({"fileData": {"mimeType": mime, "fileUri": url}})) + } else { + let mime = audio.media_type.as_deref().unwrap_or("audio/wav"); + Some(serde_json::json!({ + "fileData": {"mimeType": mime, "fileUri": url} + })) + } + } + None => audio.data.as_ref().map(|data| { + let mime = audio.media_type.as_deref().unwrap_or("audio/wav"); + let b64 = BASE64_STANDARD.encode(data); + serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) + }), + }, + ContentPart::Document(doc) => match &doc.url { + Some(url) => { + if common::is_file_path(url) { + match common::load_file_as_base64(url).await { + Ok((b64, mime)) => Some(serde_json::json!({ + "inlineData": {"mimeType": mime, "data": b64} + })), + Err(_) => None, } - }, - ) - } + } else { + let mime = doc.media_type.as_deref().unwrap_or("application/pdf"); + Some(serde_json::json!({ + "fileData": {"mimeType": mime, "fileUri": url} + })) + } + } + None => doc.data.as_ref().map(|data| { + let mime = doc.media_type.as_deref().unwrap_or("application/pdf"); + let b64 = BASE64_STANDARD.encode(data); + serde_json::json!({"inlineData": {"mimeType": mime, "data": b64}}) + }), + }, ContentPart::ToolResult(tr) => { // Gemini's functionResponse uses the function *name*, not the call ID. // Look up the original function name from the tool call mapping. @@ -337,8 +336,11 @@ fn translate_messages(messages: &[&Message]) -> Vec { })) } _ => None, - }) - .collect(); + }; + if let Some(part_json) = maybe_part { + parts.push(part_json); + } + } if parts.is_empty() { continue; @@ -408,14 +410,14 @@ fn translate_response_format( /// /// Returns a `serde_json::Value` so that `provider_options.gemini` fields can /// be merged into the request before sending. -fn build_api_request(request: &Request) -> serde_json::Value { +async fn build_api_request(request: &Request) -> serde_json::Value { let (system_text, other_messages) = extract_system_prompt(&request.messages); let system_instruction = system_text.map(|text| SystemInstruction { parts: vec![serde_json::json!({"text": text})], }); - let contents = translate_messages(&other_messages); + let contents = translate_messages(&other_messages).await; let (response_mime_type, response_schema) = request .response_format @@ -895,7 +897,7 @@ impl ProviderAdapter for Adapter { if let Some(tc) = &request.tool_choice { validate_tool_choice(self, tc)?; } - let api_body = build_api_request(request); + let api_body = build_api_request(request).await; let url = format!( "{}/models/{}:generateContent", @@ -965,7 +967,7 @@ impl ProviderAdapter for Adapter { if let Some(tc) = &request.tool_choice { validate_tool_choice(self, tc)?; } - let api_body = build_api_request(request); + let api_body = build_api_request(request).await; let url = format!( "{}/models/{}:streamGenerateContent?alt=sse", @@ -1016,16 +1018,16 @@ mod tests { } } - #[test] - fn provider_options_none_produces_standard_body() { + #[tokio::test] + async fn provider_options_none_produces_standard_body() { let request = minimal_request(); - let body = build_api_request(&request); + let body = build_api_request(&request).await; assert!(body.get("safetySettings").is_none()); assert!(body.get("cachedContent").is_none()); } - #[test] - fn provider_options_gemini_safety_settings_merged() { + #[tokio::test] + async fn provider_options_gemini_safety_settings_merged() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "gemini": { @@ -1035,7 +1037,7 @@ mod tests { } })); - let body = build_api_request(&request); + let body = build_api_request(&request).await; let safety = body .get("safetySettings") .expect("safetySettings should be present"); @@ -1044,8 +1046,8 @@ mod tests { assert_eq!(arr[0]["category"], "HARM_CATEGORY_HARASSMENT"); } - #[test] - fn provider_options_gemini_cached_content_merged() { + #[tokio::test] + async fn provider_options_gemini_cached_content_merged() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "gemini": { @@ -1053,7 +1055,7 @@ mod tests { } })); - let body = build_api_request(&request); + let body = build_api_request(&request).await; assert_eq!( body.get("cachedContent") .and_then(serde_json::Value::as_str), @@ -1061,8 +1063,8 @@ mod tests { ); } - #[test] - fn provider_options_gemini_multiple_fields_merged() { + #[tokio::test] + async fn provider_options_gemini_multiple_fields_merged() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "gemini": { @@ -1072,7 +1074,7 @@ mod tests { } })); - let body = build_api_request(&request); + let body = build_api_request(&request).await; assert!(body.get("safetySettings").is_some()); assert_eq!( body.get("cachedContent") @@ -1085,8 +1087,8 @@ mod tests { ); } - #[test] - fn provider_options_other_provider_ignored() { + #[tokio::test] + async fn provider_options_other_provider_ignored() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "anthropic": { @@ -1094,12 +1096,12 @@ mod tests { } })); - let body = build_api_request(&request); + let body = build_api_request(&request).await; assert!(body.get("auto_cache").is_none()); } - #[test] - fn provider_options_gemini_preserves_standard_fields() { + #[tokio::test] + async fn provider_options_gemini_preserves_standard_fields() { let mut request = minimal_request(); request.temperature = Some(0.5); request.max_tokens = Some(100); @@ -1109,7 +1111,7 @@ mod tests { } })); - let body = build_api_request(&request); + let body = build_api_request(&request).await; let gen_config = body .get("generationConfig") .expect("generationConfig should exist"); @@ -1141,8 +1143,8 @@ mod tests { assert!(body.get("contents").is_some()); } - #[test] - fn audio_url_translates_to_file_data() { + #[tokio::test] + async fn audio_url_translates_to_file_data() { let msg = Message { role: Role::User, content: vec![ContentPart::Audio(AudioData { @@ -1153,15 +1155,15 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; assert_eq!(contents.len(), 1); let part = &contents[0].parts[0]; assert_eq!(part["fileData"]["mimeType"], "audio/wav"); assert_eq!(part["fileData"]["fileUri"], "https://example.com/audio.wav"); } - #[test] - fn audio_base64_translates_to_inline_data() { + #[tokio::test] + async fn audio_base64_translates_to_inline_data() { let msg = Message { role: Role::User, content: vec![ContentPart::Audio(AudioData { @@ -1172,14 +1174,14 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; let part = &contents[0].parts[0]; assert_eq!(part["inlineData"]["mimeType"], "audio/wav"); assert!(part["inlineData"]["data"].as_str().is_some()); } - #[test] - fn document_url_translates_to_file_data() { + #[tokio::test] + async fn document_url_translates_to_file_data() { let msg = Message { role: Role::User, content: vec![ContentPart::Document(DocumentData { @@ -1191,14 +1193,14 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; let part = &contents[0].parts[0]; assert_eq!(part["fileData"]["mimeType"], "application/pdf"); assert_eq!(part["fileData"]["fileUri"], "https://example.com/doc.pdf"); } - #[test] - fn document_base64_translates_to_inline_data() { + #[tokio::test] + async fn document_base64_translates_to_inline_data() { let msg = Message { role: Role::User, content: vec![ContentPart::Document(DocumentData { @@ -1210,7 +1212,7 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; let part = &contents[0].parts[0]; assert_eq!(part["inlineData"]["mimeType"], "application/pdf"); assert!(part["inlineData"]["data"].as_str().is_some()); @@ -1372,8 +1374,8 @@ mod tests { } } - #[test] - fn translate_messages_function_call_includes_thought_signature() { + #[tokio::test] + async fn translate_messages_function_call_includes_thought_signature() { let mut tc = ToolCall::new( "call-1", "get_weather", @@ -1387,7 +1389,7 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; assert_eq!(contents.len(), 1); let part = &contents[0].parts[0]; @@ -1395,8 +1397,8 @@ mod tests { assert_eq!(part["thoughtSignature"], "sig456"); } - #[test] - fn translate_messages_function_call_without_thought_signature() { + #[tokio::test] + async fn translate_messages_function_call_without_thought_signature() { let tc = ToolCall::new( "call-1", "get_weather", @@ -1409,7 +1411,7 @@ mod tests { name: None, tool_call_id: None, }; - let contents = translate_messages(&[&msg]); + let contents = translate_messages(&[&msg]).await; assert_eq!(contents.len(), 1); let part = &contents[0].parts[0]; diff --git a/lib/crates/fabro-llm/src/providers/openai.rs b/lib/crates/fabro-llm/src/providers/openai.rs index 23d5d020d..5ba18de98 100644 --- a/lib/crates/fabro-llm/src/providers/openai.rs +++ b/lib/crates/fabro-llm/src/providers/openai.rs @@ -193,7 +193,7 @@ fn map_finish_reason(status: Option<&str>, has_tool_calls: bool) -> FinishReason } /// Translate unified messages to Responses API `input` array format. -fn translate_input(messages: &[Message]) -> (Option, Vec) { +async fn translate_input(messages: &[Message]) -> (Option, Vec) { let mut instructions_parts: Vec = Vec::new(); let mut input: Vec = Vec::new(); @@ -203,37 +203,40 @@ fn translate_input(messages: &[Message]) -> (Option, Vec { - let content: Vec = msg - .content - .iter() - .filter_map(|part| match part { + let mut content = Vec::new(); + for part in &msg.content { + let maybe_content = match part { ContentPart::Text(text) => { Some(serde_json::json!({"type": "input_text", "text": text})) } - ContentPart::Image(img) => { - img.url.as_ref().map_or_else( - || { - img.data.as_ref().map(|data| { - let mime = img.media_type.as_deref().unwrap_or("image/png"); - let b64 = BASE64_STANDARD.encode(data); - serde_json::json!({"type": "input_image", "image_url": format!("data:{mime};base64,{b64}")}) - }) - }, - |url| { - if common::is_file_path(url) { - match common::load_file_as_base64(url) { - Ok((b64, mime)) => Some(serde_json::json!({"type": "input_image", "image_url": format!("data:{mime};base64,{b64}")})), - Err(_) => None, - } - } else { - Some(serde_json::json!({"type": "input_image", "image_url": url})) + ContentPart::Image(img) => match &img.url { + Some(url) => { + if common::is_file_path(url) { + match common::load_file_as_base64(url).await { + Ok((b64, mime)) => Some(serde_json::json!({ + "type": "input_image", + "image_url": format!("data:{mime};base64,{b64}"), + })), + Err(_) => None, } - }, - ) - } - ContentPart::Audio(_) => { - Some(serde_json::json!({"type": "input_text", "text": "[Audio content not supported by this provider]"})) - } + } else { + Some( + serde_json::json!({"type": "input_image", "image_url": url}), + ) + } + } + None => img.data.as_ref().map(|data| { + let mime = img.media_type.as_deref().unwrap_or("image/png"); + let b64 = BASE64_STANDARD.encode(data); + serde_json::json!({ + "type": "input_image", + "image_url": format!("data:{mime};base64,{b64}"), + }) + }), + }, + ContentPart::Audio(_) => Some( + serde_json::json!({"type": "input_text", "text": "[Audio content not supported by this provider]"}), + ), ContentPart::Document(doc) => { let desc = doc.file_name.as_ref().map_or_else( || "[Document content not supported by this provider]".to_string(), @@ -242,8 +245,11 @@ fn translate_input(messages: &[Message]) -> (Option, Vec None, - }) - .collect(); + }; + if let Some(content_part) = maybe_content { + content.push(content_part); + } + } if !content.is_empty() { input.push(serde_json::json!({ "type": "message", @@ -383,8 +389,8 @@ fn translate_response_format(format: &ResponseFormat) -> Option ApiRequest { - let (instructions, input) = translate_input(&request.messages); +async fn build_api_request(request: &Request, stream: bool, codex_mode: bool) -> ApiRequest { + let (instructions, input) = translate_input(&request.messages).await; let api_tools = request.tools.as_ref().map(|t| translate_tools(t)); let tool_choice = request.tool_choice.as_ref().map(translate_tool_choice); let reasoning = request @@ -437,8 +443,12 @@ fn build_api_request(request: &Request, stream: bool, codex_mode: bool) -> ApiRe /// Serialize an `ApiRequest` to JSON and merge any `provider_options.openai` /// keys into it. -fn build_request_body(request: &Request, stream: bool, codex_mode: bool) -> serde_json::Value { - let api_request = build_api_request(request, stream, codex_mode); +async fn build_request_body( + request: &Request, + stream: bool, + codex_mode: bool, +) -> serde_json::Value { + let api_request = build_api_request(request, stream, codex_mode).await; let mut body = serde_json::to_value(&api_request).unwrap_or_else(|_| serde_json::json!({})); if let Some(openai_opts) = request @@ -952,7 +962,7 @@ impl ProviderAdapter for Adapter { if let Some(tc) = &request.tool_choice { validate_tool_choice(self, tc)?; } - let request_body = build_request_body(request, false, false); + let request_body = build_request_body(request, false, false).await; let url = format!("{}/responses", self.http.base_url); let mut req = self.build_request(&url).json(&request_body); @@ -1011,7 +1021,7 @@ impl ProviderAdapter for Adapter { if let Some(tc) = &request.tool_choice { validate_tool_choice(self, tc)?; } - let request_body = build_request_body(request, true, self.codex_mode); + let request_body = build_request_body(request, true, self.codex_mode).await; let url = format!("{}/responses", self.http.base_url); let http_resp = self @@ -1103,8 +1113,8 @@ mod tests { } } - #[test] - fn build_request_body_includes_metadata() { + #[tokio::test] + async fn build_request_body_includes_metadata() { let mut metadata = HashMap::new(); metadata.insert("user_id".to_string(), "u123".to_string()); metadata.insert("session".to_string(), "s456".to_string()); @@ -1112,21 +1122,21 @@ mod tests { let mut request = minimal_request(); request.metadata = Some(metadata); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; let meta = body.get("metadata").expect("metadata should be present"); assert_eq!(meta["user_id"], "u123"); assert_eq!(meta["session"], "s456"); } - #[test] - fn build_request_body_omits_metadata_when_none() { + #[tokio::test] + async fn build_request_body_omits_metadata_when_none() { let request = minimal_request(); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; assert!(body.get("metadata").is_none()); } - #[test] - fn build_request_body_merges_provider_options_openai() { + #[tokio::test] + async fn build_request_body_merges_provider_options_openai() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "openai": { @@ -1135,13 +1145,13 @@ mod tests { } })); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; assert_eq!(body["store"], true); assert_eq!(body["previous_response_id"], "resp_abc123"); } - #[test] - fn build_request_body_provider_options_override_fields() { + #[tokio::test] + async fn build_request_body_provider_options_override_fields() { let mut request = minimal_request(); request.temperature = Some(0.5); request.provider_options = Some(serde_json::json!({ @@ -1150,13 +1160,13 @@ mod tests { } })); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; // provider_options should override the base field assert_eq!(body["temperature"], 0.9); } - #[test] - fn build_request_body_ignores_non_openai_provider_options() { + #[tokio::test] + async fn build_request_body_ignores_non_openai_provider_options() { let mut request = minimal_request(); request.provider_options = Some(serde_json::json!({ "anthropic": { @@ -1164,29 +1174,29 @@ mod tests { } })); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; // anthropic options should not leak into the OpenAI request assert!(body.get("thinking").is_none()); } - #[test] - fn build_request_body_no_provider_options() { + #[tokio::test] + async fn build_request_body_no_provider_options() { let request = minimal_request(); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; assert_eq!(body["model"], "gpt-4o"); // stream field is omitted when false (skip_serializing_if) assert!(body.get("stream").is_none()); } - #[test] - fn build_request_body_stream_flag() { + #[tokio::test] + async fn build_request_body_stream_flag() { let request = minimal_request(); - let body = build_request_body(&request, true, false); + let body = build_request_body(&request, true, false).await; assert!(body["stream"].as_bool().unwrap_or(false)); } - #[test] - fn build_request_body_metadata_and_provider_options_together() { + #[tokio::test] + async fn build_request_body_metadata_and_provider_options_together() { let mut metadata = HashMap::new(); metadata.insert("trace_id".to_string(), "t789".to_string()); @@ -1198,7 +1208,7 @@ mod tests { } })); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; assert_eq!(body["metadata"]["trace_id"], "t789"); assert_eq!(body["store"], true); } @@ -1237,8 +1247,8 @@ mod tests { assert!(adapter.project_id.is_none()); assert!(adapter.http.default_headers.is_empty()); } - #[test] - fn audio_content_produces_text_fallback() { + #[tokio::test] + async fn audio_content_produces_text_fallback() { let msg = Message { role: Role::User, content: vec![ContentPart::Audio(AudioData { @@ -1249,7 +1259,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let content = input[0]["content"] .as_array() .expect("content should be array"); @@ -1260,8 +1270,8 @@ mod tests { ); } - #[test] - fn document_content_produces_text_fallback_with_filename() { + #[tokio::test] + async fn document_content_produces_text_fallback_with_filename() { let msg = Message { role: Role::User, content: vec![ContentPart::Document(DocumentData { @@ -1273,7 +1283,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let content = input[0]["content"] .as_array() .expect("content should be array"); @@ -1284,8 +1294,8 @@ mod tests { ); } - #[test] - fn document_content_produces_text_fallback_without_filename() { + #[tokio::test] + async fn document_content_produces_text_fallback_without_filename() { let msg = Message { role: Role::User, content: vec![ContentPart::Document(DocumentData { @@ -1297,7 +1307,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let content = input[0]["content"] .as_array() .expect("content should be array"); @@ -1335,8 +1345,8 @@ mod tests { } } - #[test] - fn translate_input_uses_item_id_for_id_field() { + #[tokio::test] + async fn translate_input_uses_item_id_for_id_field() { let mut tc = ToolCall::new( "call_xyz789", "get_weather", @@ -1350,7 +1360,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let fc = &input[0]; assert_eq!(fc["type"], "function_call"); // id field uses the fc_ prefixed item ID @@ -1359,8 +1369,8 @@ mod tests { assert_eq!(fc["call_id"], "call_xyz789"); } - #[test] - fn translate_input_falls_back_to_tc_id_without_metadata() { + #[tokio::test] + async fn translate_input_falls_back_to_tc_id_without_metadata() { let tc = ToolCall::new("call_xyz789", "get_weather", serde_json::json!({})); let msg = Message { @@ -1369,7 +1379,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let fc = &input[0]; // Without provider_metadata, both fields use tc.id assert_eq!(fc["id"], "call_xyz789"); @@ -1445,8 +1455,8 @@ mod tests { assert!(matches!(&parts[3], ContentPart::ToolCall(_))); } - #[test] - fn reasoning_items_round_trip_through_translate_input() { + #[tokio::test] + async fn reasoning_items_round_trip_through_translate_input() { let reasoning = serde_json::json!({ "type": "reasoning", "id": "rs_abc123", @@ -1467,7 +1477,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; assert_eq!(input.len(), 2); // Reasoning item is emitted first assert_eq!(input[0]["type"], "reasoning"); @@ -1478,8 +1488,8 @@ mod tests { assert_eq!(input[1]["call_id"], "call_789"); } - #[test] - fn reasoning_message_function_call_round_trip() { + #[tokio::test] + async fn reasoning_message_function_call_round_trip() { // Simulates an assistant turn with reasoning + text + tool call. // The opaque message item (with id/status) must be used instead of // constructing a new one from Text, so the reasoning item can find @@ -1516,7 +1526,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; assert_eq!(input.len(), 3); // Reasoning first assert_eq!(input[0]["type"], "reasoning"); @@ -1530,8 +1540,8 @@ mod tests { assert_eq!(input[2]["id"], "fc_def456"); } - #[test] - fn text_without_opaque_message_still_constructs_message() { + #[tokio::test] + async fn text_without_opaque_message_still_constructs_message() { // For non-OpenAI turns or turns without preserved message items, // Text parts should still produce a constructed message. let msg = Message { @@ -1540,7 +1550,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; assert_eq!(input.len(), 1); assert_eq!(input[0]["type"], "message"); assert_eq!(input[0]["role"], "assistant"); @@ -1548,8 +1558,8 @@ mod tests { assert!(input[0].get("id").is_none()); } - #[test] - fn parse_output_round_trips_function_call_ids() { + #[tokio::test] + async fn parse_output_round_trips_function_call_ids() { // Simulate a response from the Responses API let output = vec![serde_json::json!({ "type": "function_call", @@ -1567,7 +1577,7 @@ mod tests { name: None, tool_call_id: None, }; - let (_, input) = translate_input(&[msg]); + let (_, input) = translate_input(&[msg]).await; let fc = &input[0]; // The round-tripped function call should have correct IDs @@ -1575,12 +1585,12 @@ mod tests { assert_eq!(fc["call_id"], "call_001"); } - #[test] - fn build_request_body_includes_stop_sequences() { + #[tokio::test] + async fn build_request_body_includes_stop_sequences() { let mut request = minimal_request(); request.stop_sequences = Some(vec!["END".to_string(), "STOP".to_string()]); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; let stop = body.get("stop").expect("stop should be present"); let arr = stop.as_array().expect("stop should be an array"); assert_eq!(arr.len(), 2); @@ -1588,10 +1598,10 @@ mod tests { assert_eq!(arr[1], "STOP"); } - #[test] - fn build_request_body_omits_stop_when_none() { + #[tokio::test] + async fn build_request_body_omits_stop_when_none() { let request = minimal_request(); - let body = build_request_body(&request, false, false); + let body = build_request_body(&request, false, false).await; assert!(body.get("stop").is_none()); } diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index eb1e370ed..089f8d1eb 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/lib/crates/fabro-workflow/src/artifact.rs @@ -1,8 +1,3 @@ -#![expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: mixed async/sync workflow artifact lifecycle. Sync std::fs::write remains at per-stage persistence points; the Tokio-path hot reads were migrated to tokio::fs in commit 9d1c0d98c. Remaining writes should follow." -)] - use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -288,14 +283,14 @@ async fn materialize_blob_ref( let path = local_materialized_blob_path(run_dir, blob_id); if !path.exists() { if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent).map_err(|err| { + fs::create_dir_all(parent).await.map_err(|err| { Error::Io(format!( "creating artifact blob directory {}: {err}", parent.display() )) })?; } - std::fs::write(&path, &bytes).map_err(|err| { + fs::write(&path, &bytes).await.map_err(|err| { Error::Io(format!("writing artifact blob {}: {err}", path.display())) })?; } @@ -367,6 +362,10 @@ fn local_materialized_blob_path(run_dir: &Path, blob_id: &RunBlobId) -> PathBuf } #[cfg(test)] +#[expect( + clippy::disallowed_methods, + reason = "tests write artifact fixtures to disk" +)] mod tests { use std::hash::{Hash, Hasher}; use std::sync::Arc; diff --git a/lib/crates/fabro-workflow/src/artifact_snapshot.rs b/lib/crates/fabro-workflow/src/artifact_snapshot.rs index 1ef500fdb..78f8ac5df 100644 --- a/lib/crates/fabro-workflow/src/artifact_snapshot.rs +++ b/lib/crates/fabro-workflow/src/artifact_snapshot.rs @@ -328,7 +328,7 @@ pub async fn collect_artifacts( error = e.as_str(), "Asset hash failed" ); - let _ = std::fs::remove_file(&dest); + let _ = fs::remove_file(&dest).await; hash_errors += 1; } }, diff --git a/lib/crates/fabro-workflow/src/handler/manager_loop.rs b/lib/crates/fabro-workflow/src/handler/manager_loop.rs index 35b30629e..075319dc2 100644 --- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs @@ -9,6 +9,7 @@ use fabro_graphviz::graph::{AttrValue, Graph, Node}; use fabro_store::{ArtifactStore, Database}; use fabro_types::settings::SettingsLayer; use object_store::memory::InMemory; +use tokio::fs; use tokio::time::{sleep, timeout}; use super::{EngineServices, Handler}; @@ -195,7 +196,7 @@ impl Handler for SubWorkflowHandler { // Build child RunOptions let visit = visit_from_context(context) as u64; let child_logs = run_dir.join(format!("stages/{}@{visit}/child", node.id)); - let _ = std::fs::create_dir_all(&child_logs); + let _ = fs::create_dir_all(&child_logs).await; let cancel_token = Arc::new(AtomicBool::new(false)); let child_cancel = Arc::clone(&cancel_token); diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 7b3e33b46..a00695024 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -1,6 +1,7 @@ #![expect( clippy::disallowed_methods, - reason = "FOLLOW-UP: per-run `workflow create` operation; writes .fabro/ scaffolding to disk via sync std::fs" + reason = "sync workflow creation path: reads workflow.toml during workflow load and persists \ + .fabro scaffolding outside the Tokio execution hot path" )] use std::collections::{BTreeMap, HashMap}; @@ -18,6 +19,7 @@ use fabro_types::settings::run::RunMode; use fabro_types::settings::{Settings, SettingsLayer}; use fabro_types::{RunId, RunProvenance}; use fabro_util::json::normalize_json_value; +use tokio::task::spawn_blocking; use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow}; use crate::error::Error; @@ -135,6 +137,8 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result Some(RunDefinition::new( workflow_path.clone(), @@ -143,25 +147,30 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result None, }; - let persisted = create_from_source( - &resolved.raw_source, - PersistCreateOptions { - settings, - run_id: Some(run_id), - run_dir: Some(run_dir.clone()), - workflow_slug: workflow_slug.or(resolved.workflow_slug.clone()), - labels: combined_labels(&resolved_settings), - base_branch, - working_directory, - host_repo_path, - repo_origin_url, - provenance, - configured_providers, - }, - current_dir, - file_resolver, - goal_override.as_deref(), - )?; + let raw_source = resolved.raw_source.clone(); + let persisted = spawn_blocking(move || { + create_from_source( + &raw_source, + PersistCreateOptions { + settings, + run_id: Some(run_id), + run_dir: Some(persisted_run_dir), + workflow_slug: workflow_slug.or(resolved_workflow_slug), + labels: combined_labels(&resolved_settings), + base_branch, + working_directory, + host_repo_path, + repo_origin_url, + provenance, + configured_providers, + }, + current_dir, + file_resolver, + goal_override.as_deref(), + ) + }) + .await + .map_err(|err| Error::engine(format!("workflow create task failed: {err}")))??; let workflow_config = resolved .workflow_toml_path diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index 0a8c9a222..4b38c9e04 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -1,8 +1,3 @@ -#![expect( - clippy::disallowed_methods, - reason = "FOLLOW-UP: rebuild metadata uses sync std::fs::canonicalize during async checkpoint rebuild; per-run, not per-request" -)] - use std::collections::HashMap; use std::fmt::Write; use std::path::PathBuf; @@ -13,6 +8,7 @@ use fabro_checkpoint::git::Store as GitStore; use fabro_store::{Database as DurableStore, RunDatabase as DurableRunStore}; use fabro_types::{RunId, StageId}; use git2::{Repository, Signature}; +use tokio::task::spawn_blocking; use ulid::Ulid; use super::rewind::{self, RunTimeline, build_timeline}; @@ -185,7 +181,12 @@ pub async fn find_run_id_by_prefix_or_store( return Ok(run_id); } - let current_repo_root = canonical_repo_root(repo)?; + let current_repo_root = { + let repo_root = repo_root_path(repo); + spawn_blocking(move || canonical_repo_root(&repo_root)) + .await + .map_err(|err| anyhow::anyhow!("repo root canonicalize task failed: {err}"))?? + }; let mut matches = Vec::new(); for summary in fabro_store .list_runs(&fabro_store::ListRunsQuery::default()) @@ -199,12 +200,15 @@ pub async fn find_run_id_by_prefix_or_store( let Some(host_repo_path) = summary.host_repo_path.as_deref() else { continue; }; - let Ok(host_repo) = Repository::discover(host_repo_path) else { - continue; - }; - let Ok(host_repo_root) = canonical_repo_root(&host_repo) else { - continue; - }; + let host_repo_path = host_repo_path.to_string(); + let host_repo_root = + match spawn_blocking(move || canonical_repo_root_for_path(&host_repo_path)).await { + Ok(Ok(root)) => root, + Ok(Err(_)) => continue, + Err(err) => { + return Err(anyhow::anyhow!("host repo canonicalize task failed: {err}")); + } + }; if host_repo_root == current_repo_root { return Ok(summary.run_id); } @@ -214,12 +218,15 @@ pub async fn find_run_id_by_prefix_or_store( let Some(host_repo_path) = summary.host_repo_path.as_deref() else { continue; }; - let Ok(host_repo) = Repository::discover(host_repo_path) else { - continue; - }; - let Ok(host_repo_root) = canonical_repo_root(&host_repo) else { - continue; - }; + let host_repo_path = host_repo_path.to_string(); + let host_repo_root = + match spawn_blocking(move || canonical_repo_root_for_path(&host_repo_path)).await { + Ok(Ok(root)) => root, + Ok(Err(_)) => continue, + Err(err) => { + return Err(anyhow::anyhow!("host repo canonicalize task failed: {err}")); + } + }; if host_repo_root == current_repo_root && summary.run_id.to_string().starts_with(prefix) { matches.push(summary.run_id); } @@ -312,15 +319,28 @@ fn find_run_id_by_prefix_in_refs(repo: &Repository, prefix: &str) -> Result Result { - let root = repo - .workdir() +fn repo_root_path(repo: &Repository) -> PathBuf { + repo.workdir() .or_else(|| repo.path().parent()) - .unwrap_or(repo.path()); + .unwrap_or(repo.path()) + .to_path_buf() +} + +#[expect( + clippy::disallowed_methods, + reason = "sync repo-root canonicalize helper; async callers wrap it in spawn_blocking" +)] +fn canonical_repo_root(root: &std::path::Path) -> Result { std::fs::canonicalize(root) .with_context(|| format!("failed to canonicalize repo root {}", root.display())) } +fn canonical_repo_root_for_path(path: &str) -> Result { + let repo = Repository::discover(path)?; + let root = repo_root_path(&repo); + canonical_repo_root(&root) +} + fn resolve_prefix_matches(prefix: &str, matches: Vec) -> Result { match matches.len() { 0 => bail!("no run found matching '{prefix}'"), diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index f3fa9e80c..f1997653b 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -1,6 +1,6 @@ #![expect( clippy::disallowed_methods, - reason = "FOLLOW-UP: run dump writes serialized run contents to disk; invoked from sync CLI command handlers" + reason = "sync run dump writer used by CLI export paths; async callers wrap it in spawn_blocking" )] use std::collections::HashMap; diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index 130b285f5..386f61945 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -1,6 +1,7 @@ #![expect( clippy::disallowed_methods, - reason = "FOLLOW-UP: run lookup walks the runs directory; invoked from sync CLI and async server paths" + reason = "directory walk for CLI run listing; async server paths call the heavier filesystem \ + scans from spawn_blocking boundaries" )] use std::collections::HashMap;