From 1884a4b072e67d3aaa9efb4ac43765218181744e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 31 Jul 2026 13:17:21 -0400 Subject: [PATCH] Reduce large stage event payloads --- docs/internal/events.md | 4 +- docs/public/agents/outputs.mdx | 7 ++-- docs/public/execution/observability.mdx | 8 ++-- lib/components/fabro-dump/src/lib.rs | 34 +++++++++++++++ lib/components/fabro-workflow/src/artifact.rs | 17 ++++++++ .../fabro-workflow/src/handler/agent.rs | 42 +++++++++++++++++-- .../fabro-workflow/src/handler/prompt.rs | 3 +- .../fabro-workflow/src/lifecycle/event.rs | 15 ++----- 8 files changed, 106 insertions(+), 24 deletions(-) diff --git a/docs/internal/events.md b/docs/internal/events.md index 63948d376..db621c438 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -424,7 +424,7 @@ Emitted when a workflow node finishes execution. | `failure_signature` | string? | Dedup key for repeated failures | | `context_updates` | object? | Context delta written by this stage | | `jump_to_node` | string? | Non-edge jump target | -| `context_values` | object? | Full context snapshot after the stage | +| `context_values` | object? | Durable context snapshot after the stage; runtime-only keys such as `current.preamble` are omitted | | `node_visits` | object? | Node visit counts after the stage | | `loop_failure_signatures` | object? | Loop failure signature counts | | `restart_failure_signatures` | object? | Restart failure signature counts | @@ -508,7 +508,7 @@ Emitted when a prompt is rendered for an LLM stage. | Property | Type | Description | |----------|------|-------------| -| `text` | string | Rendered prompt text | +| `text` | string | Rendered prompt text, or a `blob://sha256/...` reference when the serialized text exceeds 100KB | --- diff --git a/docs/public/agents/outputs.mdx b/docs/public/agents/outputs.mdx index bbe1624d1..3614f6806 100644 --- a/docs/public/agents/outputs.mdx +++ b/docs/public/agents/outputs.mdx @@ -193,18 +193,19 @@ For the **CLI** and **ACP** backends, Fabro takes a different approach: it runs ## Artifact offloading -When a stage produces a large context value -- an LLM response or any context update -- Fabro automatically offloads it into a global content-addressed blob store instead of leaving the full value inline in durable context. Command output is streamed to a stage log file while the command runs, then finalized into a durable blob ref after completion. +When a stage produces a large context value or assembled model prompt, Fabro automatically offloads it into a global content-addressed blob store instead of leaving the full value inline in durable state. Command output is streamed to a stage log file while the command runs, then finalized into a durable blob ref after completion. ### How offloading works -After each node completes, Fabro checks every context update. If the serialized JSON of a value exceeds **100KB**, it is stored once by SHA-256 hash and replaced with a durable blob ref. Command output is always stored this way after completion, even when it is small or empty: +Fabro checks every context update and each `stage.prompt` event. If the serialized JSON of a value exceeds **100KB**, it is stored once by SHA-256 hash and replaced with a durable blob ref. Command output is always stored this way after completion, even when it is small or empty: ``` response.plan --> blob://sha256/2cf24dba5fb0... +stage.prompt --> blob://sha256/7a1d831f0064... command.output --> blob://sha256/a4f3c1d9c2e1... ``` -Non-command values under 100KB remain inline. +Non-command values under 100KB remain inline. `fabro dump` resolves an offloaded stage prompt when it writes the stage's `prompt.md` file. Checkpoints, checkpoint-completed events, forks, and resumes persist these `blob://` refs, not host-specific file paths. diff --git a/docs/public/execution/observability.mdx b/docs/public/execution/observability.mdx index d2e694742..829dcdc25 100644 --- a/docs/public/execution/observability.mdx +++ b/docs/public/execution/observability.mdx @@ -69,9 +69,11 @@ Only `id`, `ts`, `run_id`, and `event` are always present. Optional fields are o For runtime `for_each` branches, `parallel.branch.started` and `parallel.branch.completed` include the zero-based `index` and an optional `item_label`. They do not include the raw item. The final prompt is recorded by -the existing `stage.prompt` event, including the fenced item data, so event -streams, run dumps, and retained logs are source-bearing data. Apply the same -access controls and retention policy you use for workflow inputs. +the existing `stage.prompt` event, including the fenced item data. When its +serialized text exceeds 100KB, the event contains a managed blob reference +instead of inline text, and `fabro dump` resolves that reference. Event streams, +run dumps, and retained logs are source-bearing data. Apply the same access +controls and retention policy you use for workflow inputs. ## Reading the event stream diff --git a/lib/components/fabro-dump/src/lib.rs b/lib/components/fabro-dump/src/lib.rs index a47cd283f..7f0a3b642 100644 --- a/lib/components/fabro-dump/src/lib.rs +++ b/lib/components/fabro-dump/src/lib.rs @@ -722,6 +722,40 @@ mod tests { assert!(paths.contains(&"dump.log")); } + #[test] + fn hydrate_referenced_blobs_resolves_stage_prompt_text_ref() { + let prompt = "assembled stage prompt"; + let blob = serde_json::to_vec(prompt).unwrap(); + let blob_id = fabro_types::RunBlobId::new(&blob); + let mut projection = RunProjection::new("Demo".to_string(), sample_run_spec(), Utc::now()); + projection + .stage_entry("build", 1, first_event_seq(1)) + .prompt = Some(fabro_types::format_blob_ref(&blob_id)); + let mut dump = RunDump::from_projection(&projection).unwrap(); + + executor::block_on(async { + dump.hydrate_referenced_blobs_with_reader(|read_blob_id| { + let blob = blob.clone(); + Box::pin(async move { + assert_eq!(read_blob_id, blob_id); + Ok(Some(bytes::Bytes::from(blob))) + }) + }) + .await + }) + .unwrap(); + + let prompt_entry = dump + .entries() + .iter() + .find(|entry| entry.path == "stages/001-build@1/prompt.md") + .expect("prompt.md should be emitted"); + let RunDumpContents::Text(text) = &prompt_entry.contents else { + panic!("prompt.md should be text"); + }; + assert_eq!(text, prompt); + } + #[test] fn hydrate_referenced_blobs_ignores_legacy_artifact_file_refs() { let blob = serde_json::to_vec("hydrated legacy text").unwrap(); diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index 517f5a435..2fbe3463b 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -50,6 +50,23 @@ pub async fn offload_large_values( Ok(()) } +/// Offload a text value when its serialized JSON exceeds the blob threshold. +/// +/// Returns the original text when it is small and a `blob://sha256/...` +/// reference when it is offloaded. +/// +/// # Errors +/// +/// Returns an error if serialization or blob persistence fails. +pub(crate) async fn offload_large_text(text: &str, run_store: &RunStoreHandle) -> Result { + let mut value = Value::String(text.to_owned()); + offload_value(&mut value, run_store).await?; + value + .as_str() + .map(ToOwned::to_owned) + .ok_or_else(|| Error::engine("offloaded text was not a string")) +} + /// Offload large context-update values from typed parallel branch results /// before they are emitted through `parallel.completed` and stored in /// projections. diff --git a/lib/components/fabro-workflow/src/handler/agent.rs b/lib/components/fabro-workflow/src/handler/agent.rs index 0c6b75118..2bfb94d38 100644 --- a/lib/components/fabro-workflow/src/handler/agent.rs +++ b/lib/components/fabro-workflow/src/handler/agent.rs @@ -13,6 +13,7 @@ use super::structured_output::{ self, OutputSchemaKind, StructuredOutputError, ValidatedStructuredOutput, }; use super::{EngineServices, Handler, NodeTimeoutPolicy}; +use crate::artifact; use crate::context::{Context, WorkflowContext, keys}; use crate::error::Error; use crate::event::{Emitter, Event, StageScope}; @@ -70,7 +71,7 @@ pub struct OneShotRequest<'a> { /// `provider`/`model` overrides over run-level defaults, and the backend's /// `EffectiveRequestControls` (or `Default::default()` when no backend is /// attached) — live in one place. -pub(crate) fn emit_stage_prompt( +pub(crate) async fn emit_stage_prompt( services: &EngineServices, context: &Context, node: &Node, @@ -91,11 +92,12 @@ pub(crate) fn emit_stage_prompt( .map(|b| b.effective_request_controls(node)) .transpose()? .unwrap_or_default(); + let stored_prompt = artifact::offload_large_text(prompt, &services.run.run_store).await?; services.run.emitter.emit_scoped( &Event::Prompt { stage: node.id.clone(), visit: stage_scope.visit, - text: prompt.to_string(), + text: stored_prompt, mode: Some(mode.to_string()), provider: prompt_provider, model: prompt_model, @@ -270,7 +272,8 @@ impl Handler for AgentHandler { &prompt, StageModelUsage::MODE_AGENT, self.backend.as_deref(), - )?; + ) + .await?; let agent_tool_runtime = fabro_agent::AgentToolRuntime::with_question_runtime(Arc::new( WorkflowAgentQuestionRuntime::new( Arc::clone(&services.interviewer), @@ -1492,4 +1495,37 @@ Some text in between. "prompt.md should contain original prompt" ); } + + #[tokio::test] + async fn codergen_handler_offloads_large_stage_prompt() { + let handler = AgentHandler::new(None); + let prompt = "x".repeat(101 * 1024); + let mut node = Node::new("report"); + node.attrs + .insert("prompt".to_string(), AttrValue::String(prompt.clone())); + let context = test_context(); + let graph = Graph::new("test"); + let tmp = TempDir::new().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; + + handler + .execute(&node, &context, &graph, tmp.path(), &services) + .await + .unwrap(); + logger.flush().await; + + let state = run_store.state().await.unwrap(); + let node_state = state.stage(&StageId::new("report", 1)).unwrap(); + let prompt_ref = node_state.prompt.as_deref().unwrap(); + let blob_id = fabro_types::parse_blob_ref(prompt_ref) + .expect("large stage prompt should be stored as a blob reference"); + let blob = run_store + .read_blob(&blob_id) + .await + .unwrap() + .expect("stage prompt blob should exist"); + let stored_prompt: String = serde_json::from_slice(&blob).unwrap(); + + assert_eq!(stored_prompt, prompt); + } } diff --git a/lib/components/fabro-workflow/src/handler/prompt.rs b/lib/components/fabro-workflow/src/handler/prompt.rs index 57ef7c740..8e355f14c 100644 --- a/lib/components/fabro-workflow/src/handler/prompt.rs +++ b/lib/components/fabro-workflow/src/handler/prompt.rs @@ -111,7 +111,8 @@ impl Handler for PromptHandler { &prompt, StageModelUsage::MODE_PROMPT, self.backend.as_deref(), - )?; + ) + .await?; // 3. Call LLM backend (one_shot) let (response_text, stage_usage, backend_files_touched, timing) = diff --git a/lib/components/fabro-workflow/src/lifecycle/event.rs b/lib/components/fabro-workflow/src/lifecycle/event.rs index 9a6ca4b0e..8922708e0 100644 --- a/lib/components/fabro-workflow/src/lifecycle/event.rs +++ b/lib/components/fabro-workflow/src/lifecycle/event.rs @@ -95,16 +95,10 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } -/// Context values for `StageCompleted` events. Runtime-only keys are stripped, -/// except for `CURRENT_PREAMBLE`, which stage events have historically -/// included. +/// Context values for `StageCompleted` events. Runtime-only keys are stripped. fn stage_context_values(workflow_context: &Context) -> Option> { let mut snapshot = workflow_context.snapshot(); - let preamble = snapshot.get(context::keys::CURRENT_PREAMBLE).cloned(); artifact::strip_transient_keys(&mut snapshot); - if let Some(preamble) = preamble { - snapshot.insert(context::keys::CURRENT_PREAMBLE.to_owned(), preamble); - } (!snapshot.is_empty()).then(|| snapshot.into_iter().collect()) } @@ -512,7 +506,7 @@ mod tests { use super::*; #[test] - fn stage_context_values_drops_runtime_keys_but_keeps_current_preamble() { + fn stage_context_values_drops_runtime_keys_including_current_preamble() { let workflow_context = Context::new(); workflow_context.set( context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, @@ -532,10 +526,7 @@ mod tests { assert!(!values.contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES)); assert!(!values.contains_key(context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL)); - assert_eq!( - values.get(context::keys::CURRENT_PREAMBLE), - Some(&serde_json::json!("active preamble")) - ); + assert!(!values.contains_key(context::keys::CURRENT_PREAMBLE)); assert_eq!( values.get("response.work"), Some(&serde_json::json!("durable"))