Reduce large stage event payloads

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 |
 
 ---
 

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.
 

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
 

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();

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<String> {
+    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.

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

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) =

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

@@ -95,16 +95,10 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option<String> {
         .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<BTreeMap<String, serde_json::Value>> {
     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"))