diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index a457e7395..b9f5c9747 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -19,6 +19,19 @@ use crate::runtime_store::RunStoreHandle; /// Threshold above which values are persisted as blobs (100KB). const BLOB_OFFLOAD_THRESHOLD: usize = 100 * 1024; +/// Largest serialized JSON one context or outcome value may contribute to a +/// prompt preamble before it is demoted to a preview plus a file reference. +const PROMPT_INLINE_VALUE_MAX: usize = 8 * 1024; + +/// Largest serialized JSON one `for_each` item may contribute to a branch +/// prompt before it is demoted. The item is the branch's work assignment, so +/// its budget is deliberately more generous than [`PROMPT_INLINE_VALUE_MAX`]. +const PROMPT_INLINE_ITEM_MAX: usize = 64 * 1024; + +/// Rendered head carried inline by a demotion marker so the reader can tell +/// what the value is without opening the file. +const LARGE_VALUE_PREVIEW_CHARS: usize = 600; + /// Prefix used to identify artifact pointer strings in context values. const ARTIFACT_POINTER_PREFIX: &str = "file://"; @@ -111,6 +124,154 @@ async fn offload_value(value: &mut Value, run_store: &RunStoreHandle) -> Result< Ok(()) } +/// Bound every value the prompt preamble may inline. +/// +/// The resolved context and outcomes passed to the preamble builders exist +/// only to render prompt text, so any value whose serialized JSON exceeds +/// [`PROMPT_INLINE_VALUE_MAX`] is replaced with a small marker object holding +/// a preview and the sandbox path of the full value. The agent reads the file +/// when it needs the data; the preamble stays within its budget no matter how +/// much state the run has accumulated. Context keys the preamble never +/// renders are left alone. +/// +/// Demotion is an optimization of prompt size, not a correctness gate: a +/// value that fails to demote is left inline and logged rather than failing +/// the node. +pub async fn demote_large_values_for_prompt( + context: &Context, + node_outcomes: &mut HashMap, + run_store: &RunStoreHandle, + env: &dyn Sandbox, + run_dir: &Path, +) { + let mut locality = SandboxLocality::default(); + for (key, mut value) in context.snapshot() { + if context::keys::is_preamble_hidden_key(&key) { + continue; + } + match demote_value_for_prompt( + &mut value, + PROMPT_INLINE_VALUE_MAX, + run_store, + env, + run_dir, + &mut locality, + ) + .await + { + Ok(true) => context.set(key, value), + Ok(false) => {} + Err(err) => tracing::warn!(key, %err, "prompt value demotion failed; kept inline"), + } + } + for (node_id, outcome) in node_outcomes.iter_mut() { + for (key, value) in &mut outcome.context_updates { + if let Err(err) = demote_value_for_prompt( + value, + PROMPT_INLINE_VALUE_MAX, + run_store, + env, + run_dir, + &mut locality, + ) + .await + { + tracing::warn!( + node_id, + key, + %err, + "prompt value demotion failed; kept inline" + ); + } + } + } +} + +/// Bound every `for_each` item rendered into a branch prompt. +/// +/// Items above [`PROMPT_INLINE_ITEM_MAX`] are demoted the same way as context +/// values; the branch reads the file for its full assignment. An item that +/// fails to demote is left inline and logged. +pub async fn demote_large_items_for_prompt( + items: &mut [Value], + run_store: &RunStoreHandle, + env: &dyn Sandbox, + run_dir: &Path, +) { + let mut locality = SandboxLocality::default(); + for (index, item) in items.iter_mut().enumerate() { + if let Err(err) = demote_value_for_prompt( + item, + PROMPT_INLINE_ITEM_MAX, + run_store, + env, + run_dir, + &mut locality, + ) + .await + { + tracing::warn!(index, %err, "for_each item demotion failed; kept inline"); + } + } +} + +/// Replace `value` with a preview-plus-path marker when its serialized JSON +/// exceeds `max_inline_bytes`. Returns whether the value was demoted. +/// +/// The full value is persisted as a content-addressed blob and materialized +/// as a real file in the sandbox, so the marker's `path` is readable by the +/// agent that receives the prompt. +async fn demote_value_for_prompt( + value: &mut Value, + max_inline_bytes: usize, + run_store: &RunStoreHandle, + env: &dyn Sandbox, + run_dir: &Path, + locality: &mut SandboxLocality, +) -> Result { + let bytes = serde_json::to_vec(&*value) + .map_err(|e| Error::engine_with_source("prompt value serialize failed", e))?; + if bytes.len() <= max_inline_bytes { + return Ok(false); + } + let blob_hash = run_store + .write_blob(&bytes) + .await + .map_err(|e| Error::engine_with_anyhow("prompt value blob write failed", e))?; + let pointer = materialize_blob_ref(&blob_hash, run_store, env, run_dir, locality).await?; + let path = pointer + .strip_prefix(ARTIFACT_POINTER_PREFIX) + .unwrap_or(&pointer); + *value = large_value_marker(path, bytes.len(), &rendered_head(value, &bytes)); + Ok(true) +} + +/// Head of the value as the preamble would have rendered it: the raw text for +/// strings, compact JSON otherwise. +fn rendered_head(value: &Value, serialized: &[u8]) -> String { + if let Some(text) = value.as_str() { + return text.chars().take(LARGE_VALUE_PREVIEW_CHARS).collect(); + } + // Four bytes covers the widest UTF-8 character, so this slice always + // holds at least LARGE_VALUE_PREVIEW_CHARS characters of the rendering. + let head = &serialized[..serialized.len().min(LARGE_VALUE_PREVIEW_CHARS * 4)]; + String::from_utf8_lossy(head) + .chars() + .take(LARGE_VALUE_PREVIEW_CHARS) + .collect() +} + +fn large_value_marker(path: &str, bytes: usize, preview: &str) -> Value { + serde_json::json!({ + "fabroLargeValue": { + "bytes": bytes, + "path": path, + "hint": "too large to inline; read this file for the full value", + "preview": preview, + } + }) +} + /// Extract the file path from an artifact pointer value. /// /// Returns `Some(path)` if the value is a string starting with `"file://"`, @@ -1168,4 +1329,95 @@ mod tests { assert_eq!(updates["count"], serde_json::json!(42)); assert_eq!(updates["nested"], serde_json::json!({"a": 1})); } + + #[tokio::test] + async fn demote_replaces_oversized_prompt_values_with_preview_markers() { + let run_store: RunStoreHandle = make_run_store("prompt-demote").await.into(); + let tmp = tempfile::tempdir().unwrap(); + let run_dir = tmp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + let sandbox = fabro_agent::LocalSandbox::new(tmp.path().to_path_buf()); + + let dataset = serde_json::json!({ + "rows": vec![serde_json::json!({"payload": "x".repeat(64)}); 256] + }); + let context = Context::new(); + context.set("dataset", dataset.clone()); + context.set("small", serde_json::json!("kept inline")); + let mut outcomes = HashMap::from([("work".to_string(), Outcome { + context_updates: HashMap::from([( + context::keys::COMMAND_OUTPUT.to_string(), + serde_json::json!("o".repeat(PROMPT_INLINE_VALUE_MAX + 1)), + )]), + ..Outcome::success() + })]); + + demote_large_values_for_prompt(&context, &mut outcomes, &run_store, &sandbox, &run_dir) + .await; + + let marker = context.get("dataset").unwrap(); + let details = marker + .get("fabroLargeValue") + .expect("oversized context value should demote"); + assert_eq!( + usize::try_from(details["bytes"].as_u64().unwrap()).unwrap(), + serde_json::to_vec(&dataset).unwrap().len() + ); + let stored: Value = + serde_json::from_slice(&std::fs::read(details["path"].as_str().unwrap()).unwrap()) + .unwrap(); + assert_eq!(stored, dataset); + assert!( + details["preview"] + .as_str() + .unwrap() + .starts_with("{\"rows\"") + ); + assert!(serde_json::to_vec(&marker).unwrap().len() <= PROMPT_INLINE_VALUE_MAX); + + assert_eq!( + context.get("small").unwrap(), + serde_json::json!("kept inline") + ); + + let output = &outcomes["work"].context_updates[context::keys::COMMAND_OUTPUT]; + let details = output + .get("fabroLargeValue") + .expect("oversized command output should demote"); + assert!(details["preview"].as_str().unwrap().starts_with("ooo")); + let stored: Value = + serde_json::from_slice(&std::fs::read(details["path"].as_str().unwrap()).unwrap()) + .unwrap(); + assert_eq!(stored.as_str().unwrap().len(), PROMPT_INLINE_VALUE_MAX + 1); + } + + #[tokio::test] + async fn demote_skips_keys_the_preamble_never_renders() { + let run_store: RunStoreHandle = make_run_store("prompt-demote-hidden").await.into(); + let tmp = tempfile::tempdir().unwrap(); + let run_dir = tmp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + let sandbox = fabro_agent::LocalSandbox::new(tmp.path().to_path_buf()); + + let inherited_preamble = "p".repeat(PROMPT_INLINE_VALUE_MAX + 1); + let context = Context::new(); + context.set( + context::keys::CURRENT_PREAMBLE, + serde_json::json!(inherited_preamble.clone()), + ); + + demote_large_values_for_prompt( + &context, + &mut HashMap::new(), + &run_store, + &sandbox, + &run_dir, + ) + .await; + + assert_eq!( + context.get(context::keys::CURRENT_PREAMBLE).unwrap(), + serde_json::json!(inherited_preamble) + ); + } } diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 31566e8c9..b63bbb319 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -68,6 +68,22 @@ pub mod keys { pub const RESPONSE_PREFIX: &str = "response."; pub const INTERNAL_RETRY_COUNT_PREFIX: &str = "internal.retry_count."; + /// Keys the prompt preamble never renders as context values: engine + /// bookkeeping, per-thread cursors, and values the per-stage sections + /// already present. + #[must_use] + pub(crate) fn is_preamble_hidden_key(key: &str) -> bool { + key.starts_with(INTERNAL_PREFIX) + || key.starts_with(CURRENT_PREFIX) + || key.starts_with(GRAPH_PREFIX) + || key.starts_with(THREAD_PREFIX) + || key.starts_with(RESPONSE_PREFIX) + || key == OUTCOME + || key == LAST_STAGE + || key == LAST_RESPONSE + || key == PREFERRED_LABEL + } + // --- Helper functions for dynamic keys --- #[must_use] diff --git a/lib/components/fabro-workflow/src/handler/llm/preamble.rs b/lib/components/fabro-workflow/src/handler/llm/preamble.rs index 482e186ef..d1bad9c8b 100644 --- a/lib/components/fabro-workflow/src/handler/llm/preamble.rs +++ b/lib/components/fabro-workflow/src/handler/llm/preamble.rs @@ -97,15 +97,7 @@ fn is_blank_value(val: Option<&serde_json::Value>) -> bool { } fn is_context_key_excluded(key: &str) -> bool { - key.starts_with(keys::INTERNAL_PREFIX) - || key.starts_with(keys::CURRENT_PREFIX) - || key.starts_with(keys::GRAPH_PREFIX) - || key.starts_with(keys::THREAD_PREFIX) - || key.starts_with(keys::RESPONSE_PREFIX) - || key == keys::OUTCOME - || key == keys::LAST_STAGE - || key == keys::LAST_RESPONSE - || key == keys::PREFERRED_LABEL + keys::is_preamble_hidden_key(key) } fn format_value(val: &serde_json::Value) -> String { diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 1d245659f..274003177 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -160,6 +160,7 @@ async fn build_branch_plan( node: &Node, context: &Context, graph: &Graph, + run_dir: &Path, services: &EngineServices, simulated: bool, ) -> Result { @@ -252,14 +253,34 @@ async fn build_branch_plan( ))); } + // Labels come from the full items; demotion below may replace an + // oversized item with a preview-plus-path marker before it is rendered + // into the branch prompt. + let mut items = items; + let labels: Vec = items + .iter() + .enumerate() + .map(|(index, item)| item_label(item, index)) + .collect(); + if !simulated { + artifact::demote_large_items_for_prompt( + &mut items, + &services.run.run_store, + &*services.run.sandbox, + run_dir, + ) + .await; + } + Ok(BranchPlan { work_items: items .into_iter() + .zip(labels) .enumerate() - .map(|(index, item)| BranchWorkItem { + .map(|(index, (item, label))| BranchWorkItem { index, target_id: target_id.clone(), - item_label: Some(item_label(&item, index)), + item_label: Some(label), item: Some(item), }) .collect(), @@ -332,10 +353,11 @@ async fn run_branches( simulated: bool, ) -> Result { let parallel_start = Instant::now(); - let branch_plan = match build_branch_plan(node, context, graph, services, simulated).await { - Ok(plan) => plan, - Err(outcome) => return Ok(outcome), - }; + let branch_plan = + match build_branch_plan(node, context, graph, run_dir, services, simulated).await { + Ok(plan) => plan, + Err(outcome) => return Ok(outcome), + }; let is_for_each = branch_plan.is_for_each(); let BranchPlan { work_items, @@ -1649,6 +1671,69 @@ mod tests { ); } + #[tokio::test] + async fn for_each_demotes_oversized_items_before_prompt_render() { + let captures = Arc::new(Mutex::new(Vec::new())); + let handler = ItemRecordingHandler { + captures: Arc::clone(&captures), + active: Arc::new(AtomicUsize::new(0)), + max_active: Arc::new(AtomicUsize::new(0)), + delay: Duration::ZERO, + fail_marker: None, + }; + let store = test_store(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); + let run_dir = tempfile::tempdir().unwrap(); + let mut services = make_services(); + services.registry = Arc::new(super::super::HandlerRegistry::new(Box::new(handler))); + services.run = services + .run + .with_run_store(run_store.into()) + .with_sandbox(Arc::new(fabro_agent::LocalSandbox::new( + run_dir.path().to_path_buf(), + ))); + let (node, graph) = for_each_graph("context.items", 2); + let context = test_context(); + let oversized_payload = "x".repeat(65 * 1024); + context.set( + "items", + serde_json::json!([ + {"name": "small", "path": "src/auth.rs"}, + {"name": "huge", "payload": oversized_payload} + ]), + ); + + let outcome = ParallelHandler + .execute(&node, &context, &graph, run_dir.path(), &services) + .await + .unwrap(); + assert_eq!(outcome.status, StageOutcome::Succeeded); + + let captures = captures.lock().unwrap(); + let small = captures + .iter() + .find(|capture| capture.prompt.contains("src/auth.rs")) + .expect("small item renders inline"); + assert!(!small.prompt.contains("fabroLargeValue")); + + let huge = captures + .iter() + .find(|capture| capture.prompt.contains("fabroLargeValue")) + .expect("oversized item demotes to a marker"); + assert!(!huge.prompt.contains(&oversized_payload)); + assert!(huge.prompt.len() < 8 * 1024); + + // The label still comes from the full item, not the marker. + let results: Vec = + serde_json::from_value(outcome.context_updates[keys::PARALLEL_RESULTS].clone()) + .unwrap(); + assert!( + results + .iter() + .any(|result| result.item_label.as_deref() == Some("huge")) + ); + } + #[tokio::test] async fn for_each_refuses_an_array_above_the_item_limit() { // The array is runtime data, so its length is not something a workflow diff --git a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs index 37d2c7e33..76d89ce9a 100644 --- a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs @@ -183,7 +183,7 @@ impl RunLifecycle for FidelityLifecycle { ) .await .map_err(|err| CoreError::Other(err.to_string()))?; - let resolved_outcomes = artifact::resolve_outcomes_for_execution( + let mut resolved_outcomes = artifact::resolve_outcomes_for_execution( &state.node_outcomes, &self.run_store, &*self.sandbox, @@ -192,6 +192,17 @@ impl RunLifecycle for FidelityLifecycle { .await .map_err(|err| CoreError::Other(err.to_string()))?; + // The resolved copies exist only to render prompt preambles, so bound + // what any one value may contribute before the builders see them. + artifact::demote_large_values_for_prompt( + &resolved_context, + &mut resolved_outcomes, + &self.run_store, + &*self.sandbox, + &self.run_dir, + ) + .await; + let preamble = build_preamble( fidelity, &resolved_context, diff --git a/lib/components/fabro-workflow/tests/it/integration.rs b/lib/components/fabro-workflow/tests/it/integration.rs index 8104f7802..8439c6aed 100644 --- a/lib/components/fabro-workflow/tests/it/integration.rs +++ b/lib/components/fabro-workflow/tests/it/integration.rs @@ -23,7 +23,6 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; -use fabro_config::RunScratch; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_graphviz::parser::parse; use fabro_interview::{ @@ -10344,15 +10343,11 @@ async fn downstream_local_execution_resolves_response_blob_refs_as_text() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageOutcome::Succeeded); + // The downstream handler saw the full inline text, so resolution itself + // did not swap the value for a file reference. Prompt-preamble demotion + // may still materialize the oversized response under `runtime/blobs`. let captured_value = captured.lock().unwrap().first().cloned().unwrap(); assert_eq!(captured_value, "x".repeat(150 * 1024)); - assert!( - !RunScratch::new(dir.path()) - .runtime_dir() - .join("blobs") - .exists(), - "textual response values should resolve without file materialization" - ); } #[tokio::test] @@ -10420,11 +10415,20 @@ async fn downstream_remote_execution_resolves_response_blob_refs_as_text() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageOutcome::Succeeded); + // The downstream handler saw the full inline text, so resolution itself + // did not swap the value for a file reference. Prompt-preamble demotion + // may still materialize the oversized response into the sandbox blob + // directory, but nowhere else. let captured_value = captured.lock().unwrap().first().cloned().unwrap(); assert_eq!(captured_value, "x".repeat(150 * 1024)); assert!( - remote_env.written.lock().unwrap().is_empty(), - "textual response values should resolve without sandbox file materialization" + remote_env + .written + .lock() + .unwrap() + .iter() + .all(|(path, _)| path.contains("/.fabro/blobs/")), + "resolution should write nothing outside the sandbox blob directory" ); }