diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index 6d2eb1e79..743f5c23a 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -6,7 +6,7 @@ use fabro_config::RunScratch; use fabro_types::{ BlobHash, ParallelBranchResult, format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref, }; -use futures::future::BoxFuture; +use futures::future::{BoxFuture, try_join_all}; use serde_json::Value; use tokio::fs; use tokio::io::AsyncWriteExt; @@ -521,23 +521,51 @@ pub async fn resolve_text_or_blob_ref(value: &Value, run_store: &RunStoreHandle) } } -/// Resolve a structured JSON value from inline context or a Fabro-managed -/// blob reference. +/// Resolve a structured JSON value from inline context or Fabro-managed blob +/// references at any depth. /// /// Managed `file://` references are normalized through their content-addressed /// blob hash instead of reading an execution-local path. Ordinary strings and /// ordinary file references remain unchanged for the caller to validate. -pub(crate) async fn resolve_json_value(value: Value, run_store: &RunStoreHandle) -> Result { - let blob_hash = value.as_str().and_then(|reference| { - parse_blob_ref(reference).or_else(|| parse_managed_blob_file_ref(reference)) - }); - let Some(blob_hash) = blob_hash else { - return Ok(value); - }; - - let bytes = read_required_blob(&blob_hash, run_store).await?; - serde_json::from_slice(&bytes) - .map_err(|err| Error::engine_with_source("artifact blob was not valid JSON", err)) +pub(crate) fn resolve_json_value( + value: Value, + run_store: &RunStoreHandle, +) -> BoxFuture<'_, Result> { + Box::pin(async move { + match value { + Value::String(reference) => { + let blob_hash = + parse_blob_ref(&reference).or_else(|| parse_managed_blob_file_ref(&reference)); + let Some(blob_hash) = blob_hash else { + return Ok(Value::String(reference)); + }; + let bytes = read_required_blob(&blob_hash, run_store).await?; + let resolved = serde_json::from_slice(&bytes).map_err(|err| { + Error::engine_with_source("artifact blob was not valid JSON", err) + })?; + resolve_json_value(resolved, run_store).await + } + Value::Array(items) => { + let resolved = try_join_all( + items + .into_iter() + .map(|item| resolve_json_value(item, run_store)), + ) + .await?; + Ok(Value::Array(resolved)) + } + Value::Object(items) => { + let resolved = try_join_all(items.into_iter().map(|(key, item)| async move { + resolve_json_value(item, run_store) + .await + .map(|value| (key, value)) + })) + .await?; + Ok(Value::Object(resolved.into_iter().collect())) + } + primitive => Ok(primitive), + } + }) } /// Resolve a flat workflow context key (`context.NAME` or `NAME`) to a @@ -966,6 +994,40 @@ mod tests { ); } + #[tokio::test] + async fn resolve_json_value_hydrates_nested_parallel_branch_values() { + let run_store = make_run_store("nested-structured-json-resolution").await; + let finder_output = serde_json::json!({ + "findings": [{"file": "src/lib.rs", "line": 7}] + }); + let finder_blob = run_store + .write_blob(&serde_json::to_vec(&finder_output).unwrap()) + .await + .unwrap(); + let parallel_results = serde_json::json!([{ + "id": "finder", + "index": 0, + "status": "succeeded", + "context_updates": { + "output.finder": format_blob_ref(&finder_blob), + "small": "kept inline" + } + }]); + + let resolved = resolve_json_value(parallel_results, &run_store.into()) + .await + .unwrap(); + + assert_eq!( + resolved[0]["context_updates"]["output.finder"], + finder_output + ); + assert_eq!( + resolved[0]["context_updates"]["small"], + serde_json::json!("kept inline") + ); + } + #[tokio::test] async fn offload_preserves_parallel_results_and_replaces_large_context_updates() { let run_store = make_run_store("parallel-result-artifact-offload").await;