Bound what one value may contribute to a prompt preamble

Compact and summary preambles render workflow context values and stage
outputs with no per-value size limit. A late-run node inherits everything
the run has accumulated, and one oversized value (a join result, a jobs
list, a single-line command emit) can push the composed prompt past the
model's context window. A security-review run failed exactly this way:
its dedupe stage assembled a ~1.8M-token prompt against a 1M-token model
limit, made almost entirely of accumulated context the agent never
needed inline.

Reuse the existing blob machinery at the last mile. Before the preamble
builders run, any resolved context or outcome value whose serialized
JSON exceeds 8KB is persisted as a content-addressed blob, materialized
as a real file in the sandbox, and replaced with a small marker holding
a preview, the byte count, and the file path. The agent reads the file
if it needs the data. for_each items get the same treatment at fan-out
with a more generous 64KB budget, since the item is the branch's work
assignment; branch labels still come from the full item. Keys the
preamble never renders are left alone, and a value that fails to demote
stays inline and is logged: demotion bounds prompt size, it does not
gate execution.

The two downstream-resolution integration tests asserted that resolving
text values writes no files; demotion now legitimately materializes the
oversized response for preamble use, so they instead pin that resolution
returned the full inline text and that nothing is written outside the
sandbox blob directory.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FH8Jj9Y4E4Tu5g1jwDtHAb

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<String, Outcome>,
+    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<bool> {
+    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)
+        );
+    }
 }

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]

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 {

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<BranchPlan, Outcome> {
@@ -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<String> = 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<Outcome, Error> {
     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<ParallelBranchResult> =
+            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

lib/components/fabro-workflow/src/lifecycle/fidelity.rs

@@ -183,7 +183,7 @@ impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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,

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"
     );
 }