diff --git a/docs/public/execution/context.mdx b/docs/public/execution/context.mdx index bc7e19d15..d86909af4 100644 --- a/docs/public/execution/context.mdx +++ b/docs/public/execution/context.mdx @@ -234,6 +234,8 @@ Internal keys (prefixed with `internal.`, `current`, `graph.`, `thread.`, `respo In fidelity modes that render context or completed-stage output, one value can contribute at most 8 KiB of serialized JSON inline. Fabro stores larger values as content-addressed blobs, materializes them as readable files, and puts the size, file path, and a 300-character preview in the preamble. The agent can read the file when it needs the full value. +Only values the selected fidelity actually renders are materialized this way. For example, compact stage details do not include a prior stage's raw model response, so an oversized `response.` creates no blob or sandbox file at compact fidelity, while `summary:high` — which includes the response in its stage section — does materialize and reference it. Parallel branch preambles follow the same rule at each branch's own fidelity. + This prompt limit is separate from durable artifact offloading. If Fabro cannot demote a value, it logs a warning and keeps that value inline so the stage can continue. ## Artifact offloading diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index 3fb27588f..ffcab34ad 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use fabro_agent::Sandbox; @@ -155,18 +155,71 @@ fn serialized_if_over(value: &Value, threshold: usize) -> Result> Ok((bytes.len() > threshold).then_some(bytes)) } -/// Bound every value the prompt preamble may inline. +/// The exact set of resolved context and outcome values the preamble builders +/// will render for the fidelities in play, keyed the way the demotion pass +/// looks them up. +/// +/// Built by `preamble::rendered_value_selection` for one fidelity and merged +/// across a parallel node's branch fidelities, so demotion materializes only +/// values some generated preamble actually references. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct PromptValueSelection { + context_keys: HashSet, + outcome_keys: HashMap>, +} + +impl PromptValueSelection { + /// Mark a resolved context snapshot key as rendered. + pub fn select_context_value(&mut self, key: &str) { + self.context_keys.insert(key.to_string()); + } + + /// Mark one key of a completed node's outcome context updates as rendered. + pub fn select_outcome_value(&mut self, node_id: &str, key: &str) { + self.outcome_keys + .entry(node_id.to_string()) + .or_default() + .insert(key.to_string()); + } + + /// Union another fidelity's selection into this one. + pub fn merge(&mut self, other: Self) { + self.context_keys.extend(other.context_keys); + for (node_id, keys) in other.outcome_keys { + self.outcome_keys.entry(node_id).or_default().extend(keys); + } + } + + /// True when no preamble in play renders any value, so demotion has + /// nothing to bound. + #[must_use] + pub fn is_empty(&self) -> bool { + self.context_keys.is_empty() && self.outcome_keys.is_empty() + } + + pub(crate) fn renders_context_value(&self, key: &str) -> bool { + self.context_keys.contains(key) + } + + pub(crate) fn renders_outcome_value(&self, node_id: &str, key: &str) -> bool { + self.outcome_keys + .get(node_id) + .is_some_and(|keys| keys.contains(key)) + } +} + +/// Bound every value the prompt preamble will inline. /// /// The resolved context snapshot and outcomes passed here exist only to -/// render prompt text, so any value whose serialized JSON exceeds +/// render prompt text, so any selected 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 skipped. Outcome updates are -/// demoted wholesale: the set is small, and over-demoting a prompt-only copy -/// is harmless. +/// Only values named by `selection` are demoted: those are exactly the values +/// the selected fidelity's preamble path renders, so no blob or sandbox file +/// is created for a value the generated prompt omits. /// /// 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 @@ -174,13 +227,14 @@ fn serialized_if_over(value: &Value, threshold: usize) -> Result> pub async fn demote_large_values_for_prompt( values: &mut HashMap, node_outcomes: &mut HashMap, + selection: &PromptValueSelection, run_store: &RunStoreHandle, env: &dyn Sandbox, run_dir: &Path, ) { let mut locality = SandboxLocality::default(); for (key, value) in &mut *values { - if context::keys::is_preamble_hidden_key(key) { + if !selection.renders_context_value(key) { continue; } if let Err(err) = demote_value_for_prompt( @@ -198,6 +252,9 @@ pub async fn demote_large_values_for_prompt( } for (node_id, outcome) in &mut *node_outcomes { for (key, value) in &mut outcome.context_updates { + if !selection.renders_outcome_value(node_id, key) { + continue; + } if let Err(err) = demote_value_for_prompt( value, PROMPT_INLINE_VALUE_MAX, @@ -1464,9 +1521,20 @@ mod tests { )]), ..Outcome::success() })]); + let mut selection = PromptValueSelection::default(); + selection.select_context_value("dataset"); + selection.select_context_value("small"); + selection.select_outcome_value("work", context::keys::COMMAND_OUTPUT); - demote_large_values_for_prompt(&mut values, &mut outcomes, &run_store, &sandbox, &run_dir) - .await; + demote_large_values_for_prompt( + &mut values, + &mut outcomes, + &selection, + &run_store, + &sandbox, + &run_dir, + ) + .await; let details = values["dataset"] .get("fabroLargeValue") @@ -1500,22 +1568,31 @@ mod tests { } #[tokio::test] - async fn demote_skips_keys_the_preamble_never_renders() { - let run_store: RunStoreHandle = make_run_store("prompt-demote-hidden").await.into(); + async fn demote_skips_values_the_selection_omits() { + let run_store: RunStoreHandle = make_run_store("prompt-demote-unselected").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 unrendered_context = "p".repeat(PROMPT_INLINE_VALUE_MAX + 1); + let unrendered_response = "r".repeat(PROMPT_INLINE_VALUE_MAX + 1); let mut values = HashMap::from([( context::keys::CURRENT_PREAMBLE.to_string(), - serde_json::json!(inherited_preamble.clone()), + serde_json::json!(unrendered_context.clone()), )]); + let mut outcomes = HashMap::from([("work".to_string(), Outcome { + context_updates: HashMap::from([( + "response.work".to_string(), + serde_json::json!(unrendered_response.clone()), + )]), + ..Outcome::success() + })]); demote_large_values_for_prompt( &mut values, - &mut HashMap::new(), + &mut outcomes, + &PromptValueSelection::default(), &run_store, &sandbox, &run_dir, @@ -1524,7 +1601,33 @@ mod tests { assert_eq!( values[context::keys::CURRENT_PREAMBLE], - serde_json::json!(inherited_preamble) + serde_json::json!(unrendered_context) + ); + assert_eq!( + outcomes["work"].context_updates["response.work"], + serde_json::json!(unrendered_response) + ); + let blobs_dir = RunScratch::new(&run_dir).runtime_dir().join("blobs"); + assert!( + !blobs_dir.exists(), + "no blob file may be materialized for an unselected value" ); } + + #[test] + fn prompt_value_selection_merges_and_reports_emptiness() { + let mut selection = PromptValueSelection::default(); + assert!(selection.is_empty()); + + let mut other = PromptValueSelection::default(); + other.select_context_value("output.plan"); + other.select_outcome_value("run_cmd", context::keys::COMMAND_OUTPUT); + selection.merge(other); + + assert!(!selection.is_empty()); + assert!(selection.renders_context_value("output.plan")); + assert!(!selection.renders_context_value("response.plan")); + assert!(selection.renders_outcome_value("run_cmd", context::keys::COMMAND_OUTPUT)); + assert!(!selection.renders_outcome_value("other", context::keys::COMMAND_OUTPUT)); + } } diff --git a/lib/components/fabro-workflow/src/handler/llm/preamble.rs b/lib/components/fabro-workflow/src/handler/llm/preamble.rs index 87a43dca7..e51962ce9 100644 --- a/lib/components/fabro-workflow/src/handler/llm/preamble.rs +++ b/lib/components/fabro-workflow/src/handler/llm/preamble.rs @@ -3,12 +3,14 @@ use std::fmt::Write; use fabro_graphviz::graph::{Graph, Node, is_llm_handler_type}; -use crate::artifact::{self, PromptLargeValue}; +use crate::artifact::{self, PromptLargeValue, PromptValueSelection}; use crate::context::{Context, WorkflowContext, keys}; use crate::outcome::{Outcome, OutcomeExt}; const COMPACT_OUTPUT_MAX_LINES: usize = 25; const SUMMARY_HIGH_OUTPUT_MAX_LINES: usize = 50; +const SUMMARY_MEDIUM_RECENT_STAGES: usize = 5; +const SUMMARY_LOW_RECENT_STAGES: usize = 2; /// Build a fidelity-appropriate preamble string for non-full context modes. /// @@ -80,10 +82,85 @@ pub fn build_preamble( } } +/// The context and outcome values [`build_preamble`] will render at +/// `fidelity`, keyed the way the prompt demotion pass looks them up. +/// +/// This mirrors the builders in this module and must stay in lockstep with +/// them: demotion materializes exactly the values named here, so a value the +/// builders render but this selection omits would reach the prompt inline at +/// full size, and a value named here but never rendered would create an +/// unreferenced materialization. +#[must_use] +pub(crate) fn rendered_value_selection( + fidelity: keys::Fidelity, + context_values: &HashMap, + graph: &Graph, + completed_nodes: &[String], + node_outcomes: &HashMap, +) -> PromptValueSelection { + use keys::Fidelity; + + let mut selection = PromptValueSelection::default(); + // Full renders no preamble, Truncate renders only goal and run ID, and + // SummaryLow renders stage statuses without any context or output values. + let stage_window: &[String] = match fidelity { + Fidelity::Full | Fidelity::Truncate | Fidelity::SummaryLow => return selection, + Fidelity::Compact | Fidelity::SummaryHigh => completed_nodes, + Fidelity::SummaryMedium => { + recent_stage_window(completed_nodes, SUMMARY_MEDIUM_RECENT_STAGES) + } + }; + + let mut all_rendered_keys = HashSet::new(); + for node_id in stage_window { + if is_meta_handler(graph, node_id) { + continue; + } + let Some(outcome) = node_outcomes.get(node_id) else { + continue; + }; + let node = graph.nodes.get(node_id); + let handler = node.and_then(|n| n.handler_type()); + match handler { + Some("command") if outcome.context_updates.contains_key(keys::COMMAND_OUTPUT) => { + selection.select_outcome_value(node_id, keys::COMMAND_OUTPUT); + } + // Only summary:high stage sections include the model response. + h if is_llm_handler_type(h) && matches!(fidelity, Fidelity::SummaryHigh) => { + let response_key = keys::response_key(node_id); + if outcome.context_updates.contains_key(&response_key) { + selection.select_outcome_value(node_id, &response_key); + } + } + _ => {} + } + all_rendered_keys.extend(stage_rendered_keys(node_id, outcome)); + } + + for (key, value) in context_values { + if keys::is_preamble_hidden_key(key) + || all_rendered_keys.contains(key) + || is_blank_value(Some(value)) + { + continue; + } + selection.select_context_value(key); + } + + selection +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- +/// The trailing slice of completed nodes a recency-limited summary shows. +/// Meta stages are filtered later, so they still consume window slots. +fn recent_stage_window(completed_nodes: &[String], recent_count: usize) -> &[String] { + let skipped = completed_nodes.len().saturating_sub(recent_count); + &completed_nodes[skipped..] +} + fn is_meta_handler(graph: &Graph, node_id: &str) -> bool { graph .nodes @@ -479,18 +556,15 @@ fn build_summary_preamble( let stage_count = completed_nodes.len(); parts.push(format!("Completed {stage_count} stage(s) so far.")); - let recent_count = 5; - let stages_to_show: Vec<&String> = if stage_count > recent_count { - let skipped = stage_count - recent_count; + let stages_to_show = recent_stage_window(completed_nodes, SUMMARY_MEDIUM_RECENT_STAGES); + let skipped = stage_count - stages_to_show.len(); + if skipped > 0 { parts.push(format!("\n({skipped} earlier stage(s) omitted)")); - completed_nodes.iter().skip(skipped).collect() - } else { - completed_nodes.iter().collect() - }; + } { let mut header_emitted = false; - for node_id in &stages_to_show { + for node_id in stages_to_show { if is_meta_handler(graph, node_id) { continue; } @@ -498,7 +572,7 @@ fn build_summary_preamble( parts.push(String::from("\nRecent stages:")); header_emitted = true; } - if let Some(outcome) = node_outcomes.get(*node_id) { + if let Some(outcome) = node_outcomes.get(node_id) { let status = outcome.status.to_string(); let mut line = format!("- {node_id}: {status}"); if let Some(notes) = outcome.notes.as_deref() { @@ -509,7 +583,7 @@ fn build_summary_preamble( } parts.push(line); - let node = graph.nodes.get(*node_id); + let node = graph.nodes.get(node_id); let details = render_compact_stage_details(node_id, node, outcome); parts.extend(details); @@ -526,18 +600,15 @@ fn build_summary_preamble( let stage_count = completed_nodes.len(); parts.push(format!("Completed {stage_count} stage(s) so far.")); - let recent_count = 2; - let stages_to_show: Vec<&String> = if stage_count > recent_count { - let skipped = stage_count - recent_count; + let stages_to_show = recent_stage_window(completed_nodes, SUMMARY_LOW_RECENT_STAGES); + let skipped = stage_count - stages_to_show.len(); + if skipped > 0 { parts.push(format!("\n({skipped} earlier stage(s) omitted)")); - completed_nodes.iter().skip(skipped).collect() - } else { - completed_nodes.iter().collect() - }; + } { let mut header_emitted = false; - for node_id in &stages_to_show { + for node_id in stages_to_show { if is_meta_handler(graph, node_id) { continue; } @@ -545,7 +616,7 @@ fn build_summary_preamble( parts.push(String::from("\nRecent stages:")); header_emitted = true; } - if let Some(outcome) = node_outcomes.get(*node_id) { + if let Some(outcome) = node_outcomes.get(node_id) { let status = outcome.status.to_string(); let mut line = format!("- {node_id}: {status}"); if let Some(notes) = outcome.notes.as_deref() { @@ -556,7 +627,7 @@ fn build_summary_preamble( } parts.push(line); - let node = graph.nodes.get(*node_id); + let node = graph.nodes.get(node_id); let handler = node.and_then(|n| n.handler_type()); if let Some(h) = handler { parts.push(format!(" - Handler: {h}")); @@ -2362,4 +2433,222 @@ mod tests { "should not contain parent section when no parent preamble" ); } + + // --- rendered_value_selection --- + + /// Distinctive fill patterns so a leaked raw value is detectable in a + /// rendered preamble. + const RAW_RESPONSE_FILL: &str = "RRRR"; + const STRUCTURED_OUTPUT_FILL: &str = "OOOO"; + const COMMAND_OUTPUT_FILL: &str = "DDDD"; + const EXTRA_CONTEXT_FILL: &str = "NNNN"; + + /// A completed agent stage (`plan`) with a raw response and a structured + /// output, followed by a completed command stage (`run_cmd`), with all + /// values present both in the outcomes and in the context snapshot the + /// way `ExecutionState::record` applies them. + fn selection_fixture() -> ( + Graph, + Vec, + HashMap, + HashMap, + ) { + let mut graph = Graph::new("selection"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Ship the fix".to_string()), + ); + let mut plan = Node::new("plan"); + plan.attrs + .insert("shape".to_string(), AttrValue::String("box".to_string())); + let mut run_cmd = Node::new("run_cmd"); + run_cmd.attrs.insert( + "shape".to_string(), + AttrValue::String("parallelogram".to_string()), + ); + run_cmd.attrs.insert( + "script".to_string(), + AttrValue::String("make test".to_string()), + ); + graph.nodes.insert("plan".to_string(), plan); + graph.nodes.insert("run_cmd".to_string(), run_cmd); + + let raw_response = RAW_RESPONSE_FILL.repeat(200); + let structured_output = serde_json::json!({"rows": STRUCTURED_OUTPUT_FILL.repeat(200)}); + let command_output = COMMAND_OUTPUT_FILL.repeat(200); + + let mut plan_outcome = Outcome::success(); + plan_outcome.context_updates.insert( + "response.plan".to_string(), + serde_json::json!(raw_response.clone()), + ); + plan_outcome + .context_updates + .insert("output.plan".to_string(), structured_output.clone()); + let mut cmd_outcome = Outcome::success(); + cmd_outcome.context_updates.insert( + keys::COMMAND_OUTPUT.to_string(), + serde_json::json!(command_output.clone()), + ); + + let completed = vec!["plan".to_string(), "run_cmd".to_string()]; + let outcomes = HashMap::from([ + ("plan".to_string(), plan_outcome), + ("run_cmd".to_string(), cmd_outcome), + ]); + let context_values = HashMap::from([ + ("response.plan".to_string(), serde_json::json!(raw_response)), + ("output.plan".to_string(), structured_output), + ( + keys::COMMAND_OUTPUT.to_string(), + serde_json::json!(command_output), + ), + ( + "notes.data".to_string(), + serde_json::json!(EXTRA_CONTEXT_FILL.repeat(200)), + ), + ( + keys::INTERNAL_RUN_ID.to_string(), + serde_json::json!("run-1"), + ), + ]); + (graph, completed, outcomes, context_values) + } + + #[test] + fn rendered_selection_is_empty_for_full_truncate_and_summary_low() { + let (graph, completed, outcomes, context_values) = selection_fixture(); + + for fidelity in [ + keys::Fidelity::Full, + keys::Fidelity::Truncate, + keys::Fidelity::SummaryLow, + ] { + let selection = + rendered_value_selection(fidelity, &context_values, &graph, &completed, &outcomes); + assert!( + selection.is_empty(), + "{fidelity} renders no values, so nothing may be selected" + ); + } + } + + #[test] + fn rendered_selection_compact_omits_llm_response() { + let (graph, completed, outcomes, context_values) = selection_fixture(); + + let selection = rendered_value_selection( + keys::Fidelity::Compact, + &context_values, + &graph, + &completed, + &outcomes, + ); + + let mut expected = PromptValueSelection::default(); + expected.select_outcome_value("run_cmd", keys::COMMAND_OUTPUT); + expected.select_context_value("output.plan"); + expected.select_context_value("notes.data"); + assert_eq!(selection, expected); + } + + #[test] + fn rendered_selection_summary_high_includes_llm_response() { + let (graph, completed, outcomes, context_values) = selection_fixture(); + + let selection = rendered_value_selection( + keys::Fidelity::SummaryHigh, + &context_values, + &graph, + &completed, + &outcomes, + ); + + let mut expected = PromptValueSelection::default(); + expected.select_outcome_value("plan", &keys::response_key("plan")); + expected.select_outcome_value("run_cmd", keys::COMMAND_OUTPUT); + expected.select_context_value("output.plan"); + expected.select_context_value("notes.data"); + assert_eq!(selection, expected); + } + + #[test] + fn rendered_selection_medium_ignores_stages_outside_recent_window() { + let (graph, _, outcomes, context_values) = selection_fixture(); + // run_cmd falls outside the medium window of recent stages, so its + // outcome output is not rendered — but the context copy of + // command.output then is, because no shown stage supersedes it. + let mut completed = vec!["run_cmd".to_string()]; + completed.extend((0..SUMMARY_MEDIUM_RECENT_STAGES).map(|i| format!("later_{i}"))); + + let selection = rendered_value_selection( + keys::Fidelity::SummaryMedium, + &context_values, + &graph, + &completed, + &outcomes, + ); + + assert!(!selection.renders_outcome_value("run_cmd", keys::COMMAND_OUTPUT)); + assert!(selection.renders_context_value(keys::COMMAND_OUTPUT)); + assert!(selection.renders_context_value("output.plan")); + } + + /// Drift guard between [`rendered_value_selection`] and the builders: at + /// every fidelity, demoting exactly the selected values yields a preamble + /// that references every demoted path and inlines no unselected fill. + #[test] + fn rendered_selection_matches_builders_at_every_fidelity() { + let (graph, completed, outcomes, context_values) = selection_fixture(); + + for &fidelity in keys::Fidelity::variants() { + let selection = + rendered_value_selection(fidelity, &context_values, &graph, &completed, &outcomes); + + let mut demoted_values = context_values.clone(); + let mut demoted_outcomes = outcomes.clone(); + let mut expected_paths = Vec::new(); + for (key, value) in &mut demoted_values { + if selection.renders_context_value(key) { + let path = format!("/blobs/context-{key}.json"); + *value = large_prompt_value(9_000, &path, "preview"); + expected_paths.push(path); + } + } + for (node_id, outcome) in &mut demoted_outcomes { + for (key, value) in &mut outcome.context_updates { + if selection.renders_outcome_value(node_id, key) { + let path = format!("/blobs/{node_id}-{key}.json"); + *value = large_prompt_value(9_000, &path, "preview"); + expected_paths.push(path); + } + } + } + + let context = Context::new(); + for (key, value) in demoted_values { + context.set(key, value); + } + let preamble = + build_preamble(fidelity, &context, &graph, &completed, &demoted_outcomes); + + for path in &expected_paths { + assert!( + preamble.contains(path), + "{fidelity}: selected value was not referenced: {path}\n{preamble}" + ); + } + for fill in [ + RAW_RESPONSE_FILL, + STRUCTURED_OUTPUT_FILL, + COMMAND_OUTPUT_FILL, + EXTRA_CONTEXT_FILL, + ] { + assert!( + !preamble.contains(fill), + "{fidelity}: unselected large value leaked inline ({fill})\n{preamble}" + ); + } + } + } } diff --git a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs index 8d624bb76..05501d4f6 100644 --- a/lib/components/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/components/fabro-workflow/src/lifecycle/fidelity.rs @@ -13,7 +13,7 @@ use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvN use crate::artifact; use crate::context::{Context, ParallelBranchPreamble, keys}; use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::handler::llm::preamble::build_preamble; +use crate::handler::llm::preamble; use crate::outcome::{BilledModelUsage, Outcome}; use crate::runtime_store::RunStoreHandle; @@ -104,7 +104,7 @@ impl FidelityLifecycle { let entry = ParallelBranchPreamble { fidelity: branch_fidelity, - preamble: build_preamble( + preamble: preamble::build_preamble( branch_fidelity, resolved_context, &self.graph, @@ -121,6 +121,29 @@ impl FidelityLifecycle { preambles } + + /// Distinct effective fidelities of a parallel node's branch preamble + /// entries, in outgoing-edge order. Branches that inherit the fork's + /// preamble contribute nothing beyond the fork's own fidelity. + fn parallel_branch_effective_fidelities( + &self, + node_id: &str, + fork_fidelity: keys::Fidelity, + ) -> Vec { + let mut fidelities = Vec::new(); + for edge in self.graph.outgoing_edges(node_id) { + let Some(target_node) = self.graph.nodes.get(&edge.to) else { + continue; + }; + let resolution = resolve_parallel_branch_fidelity(edge, target_node, fork_fidelity); + if let Some(fidelity) = resolution.effective { + if !fidelities.contains(&fidelity) { + fidelities.push(fidelity); + } + } + } + fidelities + } } #[async_trait] @@ -194,16 +217,33 @@ impl RunLifecycle for FidelityLifecycle { // The resolved copies exist only to render prompt preambles, so bound // what any one value may contribute before the builders see them. - // Full renders no preamble and Truncate renders no context values, so - // there is nothing to bound — except for a parallel node, whose branch - // stash may render at a richer fidelity. - let preamble_renders_values = - !matches!(fidelity, keys::Fidelity::Full | keys::Fidelity::Truncate) - || gv_node.handler_type() == Some("parallel"); - if preamble_renders_values { + // Demote exactly the values the selected fidelity renders — plus, for + // a parallel node, whatever its branch stash renders at other + // fidelities — so no blob or sandbox file is materialized for a value + // absent from every generated preamble. + let mut selection = preamble::rendered_value_selection( + fidelity, + &resolved_values, + &self.graph, + &state.completed_nodes, + &resolved_outcomes, + ); + if gv_node.handler_type() == Some("parallel") { + for branch_fidelity in self.parallel_branch_effective_fidelities(node.id(), fidelity) { + selection.merge(preamble::rendered_value_selection( + branch_fidelity, + &resolved_values, + &self.graph, + &state.completed_nodes, + &resolved_outcomes, + )); + } + } + if !selection.is_empty() { artifact::demote_large_values_for_prompt( &mut resolved_values, &mut resolved_outcomes, + &selection, &self.run_store, &*self.sandbox, &self.run_dir, @@ -212,7 +252,7 @@ impl RunLifecycle for FidelityLifecycle { } let resolved_context = Context::from_values(resolved_values); - let preamble = build_preamble( + let preamble = preamble::build_preamble( fidelity, &resolved_context, &self.graph, @@ -405,6 +445,10 @@ fn resolve_thread_id( } #[cfg(test)] +#[expect( + clippy::disallowed_methods, + reason = "tests inspect materialized blob files on disk" +)] mod tests { use std::path::Path; use std::time::Duration; @@ -478,6 +522,274 @@ mod tests { ) } + /// Over the 8 KiB prompt inline budget, under the 100 KiB durable + /// offload threshold, so values stay inline in context until demotion. + const OVERSIZED_LEN: usize = 20_000; + + fn oversized_response() -> String { + "R".repeat(OVERSIZED_LEN) + } + + fn oversized_output() -> serde_json::Value { + serde_json::json!({"rows": "O".repeat(OVERSIZED_LEN)}) + } + + fn linear_workflow_graph(work_fidelity: Option<&str>) -> WorkflowGraph { + let mut graph = Graph::new("linear-fidelity"); + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), str_attr("Mdiamond")); + let mut consolidate = Node::new("consolidate"); + consolidate + .attrs + .insert("shape".to_string(), str_attr("box")); + let mut work = Node::new("work"); + work.attrs.insert("shape".to_string(), str_attr("box")); + if let Some(fidelity) = work_fidelity { + work.attrs + .insert("fidelity".to_string(), str_attr(fidelity)); + } + + graph.nodes.insert(start.id.clone(), start); + graph.nodes.insert(consolidate.id.clone(), consolidate); + graph.nodes.insert(work.id.clone(), work); + graph.edges.push(Edge::new("start", "consolidate")); + graph.edges.push(Edge::new("consolidate", "work")); + + WorkflowGraph(Arc::new(graph)) + } + + /// A state where the agent stage `node_id` completed with an oversized + /// raw response and an oversized structured output, applied to context + /// the way `ExecutionState::record` does. + fn state_with_completed_llm_stage(graph: &WorkflowGraph, node_id: &str) -> WfRunState { + let mut state: WfRunState = ExecutionState::new(graph).unwrap(); + let mut outcome = Outcome::success(); + outcome.context_updates.insert( + keys::response_key(node_id), + serde_json::json!(oversized_response()), + ); + outcome + .context_updates + .insert(format!("output.{node_id}"), oversized_output()); + state.context.apply_updates(&outcome.context_updates); + state.completed_nodes.push(node_id.to_string()); + state.node_outcomes.insert(node_id.to_string(), outcome); + state + } + + fn materialized_blob_files(run_dir: &Path) -> Vec { + let blobs_dir = fabro_config::RunScratch::new(run_dir) + .runtime_dir() + .join("blobs"); + if !blobs_dir.exists() { + return Vec::new(); + } + let mut files: Vec = std::fs::read_dir(&blobs_dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect(); + files.sort(); + files + } + + fn assert_durable_state_unchanged(state: &WfRunState, node_id: &str) { + assert_eq!( + state.context.get(&keys::response_key(node_id)), + Some(serde_json::json!(oversized_response())), + "durable context must keep the raw response" + ); + assert_eq!( + state.context.get(&format!("output.{node_id}")), + Some(oversized_output()), + "durable context must keep the structured output" + ); + let outcome = &state.node_outcomes[node_id]; + assert_eq!( + outcome.context_updates[&keys::response_key(node_id)], + serde_json::json!(oversized_response()), + "node outcome must keep the raw response" + ); + assert_eq!( + outcome.context_updates[&format!("output.{node_id}")], + oversized_output(), + "node outcome must keep the structured output" + ); + } + + #[tokio::test] + async fn compact_materializes_only_values_its_preamble_renders() { + let graph = linear_workflow_graph(None); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state = state_with_completed_llm_stage(&graph, "consolidate"); + let work = graph.get_node("work").unwrap(); + + lifecycle.before_node(&work, &state).await.unwrap(); + + let files = materialized_blob_files(run_dir.path()); + assert_eq!( + files.len(), + 1, + "compact must not materialize the unrendered LLM response" + ); + let stored: serde_json::Value = + serde_json::from_slice(&std::fs::read(&files[0]).unwrap()).unwrap(); + assert_eq!(stored, oversized_output()); + + let preamble = state.context.preamble(); + assert!( + preamble.contains(files[0].to_str().unwrap()), + "the materialized output must be referenced by the preamble" + ); + assert!( + !preamble.contains("RRRR"), + "the raw response must not appear in a compact preamble" + ); + assert!( + !preamble.contains(&"O".repeat(1000)), + "no oversized value may be inlined" + ); + assert_durable_state_unchanged(&state, "consolidate"); + } + + #[tokio::test] + async fn summary_high_materializes_and_references_the_llm_response() { + let graph = linear_workflow_graph(Some("summary:high")); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state = state_with_completed_llm_stage(&graph, "consolidate"); + let work = graph.get_node("work").unwrap(); + + lifecycle.before_node(&work, &state).await.unwrap(); + + let files = materialized_blob_files(run_dir.path()); + assert_eq!( + files.len(), + 2, + "summary:high renders both the response and the output" + ); + let preamble = state.context.preamble(); + for file in &files { + assert!( + preamble.contains(file.to_str().unwrap()), + "every materialized blob must be referenced by the preamble: {}", + file.display() + ); + } + assert!( + !preamble.contains(&"R".repeat(1000)), + "no oversized value may be inlined" + ); + assert!( + !preamble.contains(&"O".repeat(1000)), + "no oversized value may be inlined" + ); + assert_durable_state_unchanged(&state, "consolidate"); + } + + #[tokio::test] + async fn value_free_fidelities_materialize_nothing() { + for fidelity in ["full", "truncate", "summary:low"] { + let graph = linear_workflow_graph(Some(fidelity)); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state = state_with_completed_llm_stage(&graph, "consolidate"); + let work = graph.get_node("work").unwrap(); + + lifecycle.before_node(&work, &state).await.unwrap(); + + assert!( + materialized_blob_files(run_dir.path()).is_empty(), + "{fidelity} renders no values, so nothing may be materialized" + ); + assert_durable_state_unchanged(&state, "consolidate"); + } + } + + #[tokio::test] + async fn parallel_branch_materializes_values_only_its_fidelity_renders() { + let graph = parallel_workflow_graph(None, Some("summary:high")); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state = state_with_completed_llm_stage(&graph, "work"); + let fork = graph.get_node("fork").unwrap(); + + lifecycle.before_node(&fork, &state).await.unwrap(); + + // The compact fork renders the output; the summary:high branch also + // renders the response. + let files = materialized_blob_files(run_dir.path()); + assert_eq!(files.len(), 2); + + let fork_preamble = state.context.preamble(); + assert!( + !fork_preamble.contains("RRRR"), + "the compact fork preamble must not include the response" + ); + + let stash = state + .context + .get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES) + .expect("parallel stash should be set"); + let branch_preamble = stash[0]["preamble"] + .as_str() + .expect("branch_a should render its own preamble"); + + for file in &files { + let path = file.to_str().unwrap(); + assert!( + fork_preamble.contains(path) || branch_preamble.contains(path), + "every materialized blob must be referenced by some preamble: {path}" + ); + } + assert!( + branch_preamble.contains( + materialized_blob_files(run_dir.path()) + .iter() + .find_map(|file| { + let stored: serde_json::Value = + serde_json::from_slice(&std::fs::read(file).unwrap()).unwrap(); + (stored == serde_json::json!(oversized_response())) + .then(|| file.to_str().unwrap().to_string()) + }) + .expect("the raw response must be materialized") + .as_str() + ), + "the summary:high branch must reference the materialized response" + ); + } + + #[tokio::test] + async fn truncate_fork_materializes_only_branch_rendered_values() { + let graph = parallel_workflow_graph(Some("truncate"), Some("compact")); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state = state_with_completed_llm_stage(&graph, "work"); + let fork = graph.get_node("fork").unwrap(); + + lifecycle.before_node(&fork, &state).await.unwrap(); + + // Only the compact branch renders values, and compact omits the + // response — so exactly the structured output is materialized. + let files = materialized_blob_files(run_dir.path()); + assert_eq!(files.len(), 1); + let stored: serde_json::Value = + serde_json::from_slice(&std::fs::read(&files[0]).unwrap()).unwrap(); + assert_eq!(stored, oversized_output()); + + let stash = state + .context + .get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES) + .expect("parallel stash should be set"); + let branch_preamble = stash[0]["preamble"].as_str().unwrap(); + assert!( + branch_preamble.contains(files[0].to_str().unwrap()), + "the compact branch must reference the materialized output" + ); + } + #[test] fn parallel_branch_fidelity_edge_overrides_node() { let mut node = Node::new("branch"); diff --git a/lib/components/fabro-workflow/tests/it/integration.rs b/lib/components/fabro-workflow/tests/it/integration.rs index b72b3cdf8..06694ae01 100644 --- a/lib/components/fabro-workflow/tests/it/integration.rs +++ b/lib/components/fabro-workflow/tests/it/integration.rs @@ -10348,17 +10348,17 @@ async fn downstream_local_execution_resolves_response_blob_refs_as_text() { 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 - // materializes the oversized response for preamble use, confined to the - // run's blob directory. + // did not swap the value for a file reference. No preamble at the default + // compact fidelity renders the response, so prompt demotion materializes + // nothing. let captured_value = captured.lock().unwrap().first().cloned().unwrap(); assert_eq!(captured_value, "x".repeat(150 * 1024)); assert!( - RunScratch::new(dir.path()) + !RunScratch::new(dir.path()) .runtime_dir() .join("blobs") .exists(), - "prompt demotion materializes the oversized response under runtime/blobs" + "no blob file may be materialized for a response no preamble renders" ); } @@ -10428,21 +10428,15 @@ async fn downstream_remote_execution_resolves_response_blob_refs_as_text() { 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. + // did not swap the value for a file reference. No preamble at the default + // compact fidelity renders the response, so prompt demotion writes + // nothing into the sandbox. let captured_value = captured.lock().unwrap().first().cloned().unwrap(); assert_eq!(captured_value, "x".repeat(150 * 1024)); let written = remote_env.written.lock().unwrap(); assert!( - !written.is_empty(), - "prompt demotion materializes the oversized response into the sandbox" - ); - assert!( - written - .iter() - .all(|(path, _)| path.contains("/.fabro/blobs/")), - "nothing is written outside the sandbox blob directory" + written.is_empty(), + "no sandbox file may be written for a response no preamble renders" ); }