scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-06 02:48:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(async-fs): move remaining filesystem work off Tokio paths

Browse source 
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.
This commit is contained in:
Bryan Helmkamp 2026-04-19 19:31:03 -04:00
parent 19939c5f07
commit 32b88d7833
No known key found for this signature in database
14 changed files with 449 additions and 392 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
lib/crates/fabro-agent/src/cli.rs
View file

@ -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;
}
}
}

90
lib/crates/fabro-devcontainer/src/compose.rs
View file

@ -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<ComposeServiceSpec, String> {
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<String, String> {
/// 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<ComposeServiceSpec, String> {
@ -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"));
}
}

77
lib/crates/fabro-devcontainer/src/lib.rs
View file

@ -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<DevcontainerSpec> {
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<PathBuf> = 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,

92
lib/crates/fabro-llm/src/providers/anthropic.rs
View file

@ -232,7 +232,7 @@ fn parse_content_block(block: &serde_json::Value) -> Option<ContentPart> {
}
/// Translate a unified `ContentPart` to an Anthropic content block JSON value.
fn content_part_to_api(part: &ContentPart) -> Option<serde_json::Value> {
async fn content_part_to_api(part: &ContentPart) -> Option<serde_json::Value> {
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<serde_json::Value> {
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<serde_json::Value> {
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<serde_json::Value> {
///
/// 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<ApiMessage> {
async fn translate_messages(messages: &[&Message]) -> Vec<ApiMessage> {
let mut api_messages: Vec<ApiMessage> = Vec::new();
for msg in messages {
@ -330,8 +330,12 @@ fn translate_messages(messages: &[&Message]) -> Vec<ApiMessage> {
Role::System | Role::Developer => continue,
};
let content: Vec<serde_json::Value> =
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"

13
lib/crates/fabro-llm/src/providers/common.rs
View file

@ -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();

222
lib/crates/fabro-llm/src/providers/gemini.rs
View file

@ -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<Content> {
async fn translate_messages(messages: &[&Message]) -> Vec<Content> {
let id_to_name = build_tool_call_id_to_name(messages);
let mut contents: Vec<Content> = Vec::new();
@ -224,10 +224,9 @@ fn translate_messages(messages: &[&Message]) -> Vec<Content> {
Role::System | Role::Developer => continue,
};
let parts: Vec<serde_json::Value> = 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<Content> {
}
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<Content> {
}))
}
_ => 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];

196
lib/crates/fabro-llm/src/providers/openai.rs
View file

@ -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<String>, Vec<serde_json::Value>) {
async fn translate_input(messages: &[Message]) -> (Option<String>, Vec<serde_json::Value>) {
let mut instructions_parts: Vec<String> = Vec::new();
let mut input: Vec<serde_json::Value> = Vec::new();
@ -203,37 +203,40 @@ fn translate_input(messages: &[Message]) -> (Option<String>, Vec<serde_json::Val
instructions_parts.push(msg.text());
}
Role::User => {
let content: Vec<serde_json::Value> = 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<String>, Vec<serde_json::Val
Some(serde_json::json!({"type": "input_text", "text": desc}))
}
_ => 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<serde_json::Valu
/// When `codex_mode` is true, unsupported fields (`temperature`,
/// `max_output_tokens`, `top_p`) are omitted and empty instructions are sent as
/// `""` (required by the Codex endpoint).
fn build_api_request(request: &Request, stream: bool, codex_mode: bool) -> 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());
}

13
lib/crates/fabro-workflow/src/artifact.rs
View file

@ -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;

2
lib/crates/fabro-workflow/src/artifact_snapshot.rs
View file

@ -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;
}
},

3
lib/crates/fabro-workflow/src/handler/manager_loop.rs
View file

@ -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);

49
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -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<Created
let goal_override = resolved.goal_override.clone();
let current_dir = resolved.current_dir.clone();
let file_resolver = resolved.file_resolver.clone();
let resolved_workflow_slug = resolved.workflow_slug.clone();
let persisted_run_dir = run_dir.clone();
let accepted_definition = match (&workflow_path, &workflow_bundle) {
(Some(workflow_path), Some(workflow_bundle)) => Some(RunDefinition::new(
workflow_path.clone(),
@ -143,25 +147,30 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
_ => 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

64
lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
View file

@ -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<Opti
resolve_prefix_matches(prefix, matches).map(Some)
}
fn canonical_repo_root(repo: &Repository) -> Result<PathBuf> {
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<PathBuf> {
std::fs::canonicalize(root)
.with_context(|| format!("failed to canonicalize repo root {}", root.display()))
}
fn canonical_repo_root_for_path(path: &str) -> Result<PathBuf> {
let repo = Repository::discover(path)?;
let root = repo_root_path(&repo);
canonical_repo_root(&root)
}
fn resolve_prefix_matches(prefix: &str, matches: Vec<RunId>) -> Result<RunId> {
match matches.len() {
0 => bail!("no run found matching '{prefix}'"),

2
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -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;

3
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -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;