scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge remote-tracking branch 'origin/main'

Browse source 
Resolve the artifact lifecycle merge by keeping object-backed upload and sandbox sync behavior alongside durable blob-ref normalization.
This commit is contained in:
Bryan Helmkamp 2026-04-07 17:52:07 -04:00
commit ffe3854c8f
No known key found for this signature in database
23 changed files with 1018 additions and 194 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.config/nextest.toml
View file

@ -1,6 +1,7 @@
[profile.default]
# Unit tests: flag SLOW after 1.5s, hard-kill after 3s for most crates
slow-timeout = { period = "1.5s", terminate-after = 2 }
leak-timeout = "500ms"
[[profile.default.overrides]]
filter = "package(fabro-cli) & kind(test)"
@ -17,3 +18,4 @@ slow-timeout = { period = "1.5s", terminate-after = 2 }
[profile.e2e]
# E2E (ignored) tests: flag SLOW after 10s, hard-kill after 30s
slow-timeout = { period = "10s", terminate-after = 3 }
leak-timeout = "500ms"

68
docs/agents/outputs.mdx
View file

@ -3,7 +3,7 @@ title: "Outputs & Artifacts"
description: "How Fabro captures agent responses, tracks file changes, and collects test assets"
---
When an agent or prompt node finishes, Fabro captures its response text and produces an **outcome** that feeds into context, transition logic, and downstream nodes. Fabro also tracks every file change per stage, offloads large outputs to disk, and automatically collects test artifacts like screenshots and reports.
When an agent or prompt node finishes, Fabro captures its response text and produces an **outcome** that feeds into context, transition logic, and downstream nodes. Fabro also tracks every file change per stage, offloads large outputs into content-addressed blob storage, and automatically collects test artifacts like screenshots and reports.
## Response capture
@ -132,88 +132,52 @@ For the **CLI backend**, Fabro takes a different approach: it runs `git diff --n
## Artifact offloading
When a stage produces a large context value -- an LLM response, command output, or any context update -- Fabro automatically offloads it to disk instead of keeping it in memory. This prevents large outputs from bloating checkpoint files and overwhelming preamble summaries.
When a stage produces a large context value -- an LLM response, command output, 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.
### How offloading works
After each node completes, Fabro checks every context update. If the serialized JSON of a value exceeds **100KB**, it is written to a file-backed artifact on disk and replaced in the context with a `file://` pointer:
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:
```
response.plan --> file:///path/to/logs/cache/artifacts/values/response.plan.json
command.output --> file:///path/to/logs/cache/artifacts/values/command.output.json
response.plan --> blob://sha256/2cf24dba5fb0...
command.output --> blob://sha256/a4f3c1d9c2e1...
```
Values under 100KB remain in the context as-is.
Values under 100KB remain inline.
### File-backed artifact layout
Offloaded artifacts are written to the run's directory:
```
~/.fabro/scratch/{run_id}/
cache/
artifacts/
values/
response.plan.json
response.implement.json
command.output.json
```
Each file contains the full serialized JSON value. The `ArtifactStore` manages reads and writes, and cleans up files when artifacts are removed.
Checkpoints, checkpoint-completed events, forks, and resumes persist these `blob://` refs, not host-specific file paths.
### Preamble rendering
When Fabro builds a [preamble](/execution/context#preamble-construction) for a downstream stage, it resolves `file://` pointers and renders a reference instead of inlining the full content:
When Fabro builds a [preamble](/execution/context#preamble-construction) for a downstream stage, it first materializes any blob refs into execution-local files and then renders a reference instead of inlining the full content:
```markdown
## Completed stages
- **plan**: success
- Model: claude-sonnet-4-5, 12.4k tokens in / 3.2k out
- Files: src/main.rs, tests/api_test.rs
- Response: See: /path/to/logs/cache/artifacts/values/response.plan.json
- Response: See: /path/to/runtime/blobs/<blob_id>.json
- **test**: success
- Script: `cargo test 2>&1 || true`
- Stdout: See: /path/to/logs/cache/artifacts/values/command.output.json
- Stdout: See: /path/to/runtime/blobs/<blob_id>.json
```
This keeps preambles concise while still giving agents a path to read the full output if needed.
## Git storage
Artifact data is persisted on the Git [metadata branch](/execution/checkpoints#metadata-branch) alongside checkpoint data. Each time a checkpoint is written, any file-backed artifacts are included as additional entries:
Large offloaded context values are not stored on the Git [metadata branch](/execution/checkpoints#metadata-branch). The metadata branch keeps checkpoint JSON and stage metadata; blob payloads live in the durable blob store and are referenced by `blob://sha256/...`.
```
fabro/meta/{run_id}
run.json
start.json
checkpoint.json
artifacts/
response.plan.json
command.output.json
```
This means file-backed artifact data survives process restarts and can be recovered when resuming a run from a Git branch.
Captured stage artifacts such as screenshots, videos, reports, and traces still use the artifact store and metadata export paths described below.
## Remote sandbox syncing
For remote sandboxes (Docker, Daytona), artifact files stored on the host are not directly accessible inside the sandbox. Before a stage executes, Fabro syncs any `file://` pointers to the sandbox filesystem.
For remote sandboxes (Docker, Daytona), execution-time file access happens inside the sandbox filesystem.
For each pointer in the context updates:
- Blob refs are materialized into `{working_directory}/.fabro/blobs/{blob_id}.json`
- Explicit non-blob `file://` refs keep the existing copy-on-demand behavior and are copied into `{working_directory}/.fabro/artifacts/{filename}` when needed
1. Fabro checks whether the file is already accessible inside the sandbox
2. If not, it reads the local file and uploads it via the sandbox's `write_file` interface
3. The file is placed at `{working_directory}/.fabro/artifacts/{filename}`
4. The pointer is rewritten to reference the remote path
```
# Before sync (host path)
file:///home/user/.fabro/scratch/01JK.../cache/artifacts/values/response.plan.json
# After sync (sandbox path)
file:///workspace/.fabro/artifacts/response.plan.json
```
This ensures agents running in remote sandboxes can read offloaded artifacts using the same `file://` pointer mechanism.
In both cases, downstream handlers and agents continue to consume ordinary `file://` pointers during execution.
<Note>
For local sandboxes, syncing is a no-op since the agent can already access the host filesystem directly.

15
docs/execution/context.mdx
View file

@ -196,15 +196,22 @@ Internal keys (prefixed with `internal.`, `current`, `graph.`, `thread.`, `respo
## Artifact offloading
When a stage produces a large output (over 100KB of serialized JSON), Fabro automatically offloads it to a file-backed artifact on disk rather than keeping it in the in-memory context. The context value is replaced with a `file://` pointer:
When a stage produces a large output (over 100KB of serialized JSON), Fabro stores the serialized bytes in a global content-addressed blob store and replaces the context value with a durable blob ref:
```
response.plan → file:///tmp/logs/cache/artifacts/values/response.plan.json
response.plan → blob://sha256/2cf24dba5fb0...
```
The preamble renderer resolves these pointers and displays a reference to the file path. For remote sandboxes (Docker, Daytona), Fabro syncs artifact files to the sandbox at `{working_directory}/.fabro/artifacts/` so agents can read them.
Checkpoints and checkpoint-completed events persist these `blob://` refs, not host-specific file paths.
This keeps the context lean — large LLM responses, test output, and file listings don't bloat checkpoint files or overwhelm preamble summaries.
Before Fabro builds a preamble or starts the next stage, it resolves any blob refs into execution-local files so handlers and agents still see normal `file://` references:
- Local execution materializes blobs under `{run_dir}/runtime/blobs/{blob_id}.json`
- Remote sandboxes materialize blobs under `{working_directory}/.fabro/blobs/{blob_id}.json`
These materialized `file://` paths are runtime-only. They are not written back into durable context snapshots.
This keeps the durable context lean and portable while still giving downstream stages a filesystem path they can read.
## Context compaction

71
lib/crates/fabro-cli/src/commands/run/output.rs
View file

@ -1,10 +1,12 @@
use std::path::Path;
use std::time::Duration;
use anyhow::Result;
use anyhow::{Context as _, Result};
use fabro_api::types;
use fabro_config::RunScratch;
use fabro_types::PullRequestRecord;
use fabro_types::{
PullRequestRecord, RunBlobId, RunId, parse_blob_ref, parse_legacy_blob_file_ref,
};
use fabro_util::check_report::{CheckDetail, CheckReport, CheckResult, CheckSection, CheckStatus};
use fabro_util::terminal::Styles;
use fabro_util::text::strip_goal_decoration;
@ -149,7 +151,9 @@ pub(crate) async fn print_run_summary_with_client(
pr_url.as_deref(),
styles,
);
print_final_output(checkpoint.as_ref(), styles);
let final_output =
resolve_final_output_with_client(client, run_id, checkpoint.as_ref()).await?;
print_final_output(final_output.as_deref(), styles);
if let Some(run_dir) = local_run_dir {
print_assets(run_dir, styles);
}
@ -254,22 +258,65 @@ pub(crate) fn print_run_conclusion(
}
}
pub(crate) fn print_final_output(checkpoint: Option<&fabro_types::Checkpoint>, styles: &Styles) {
let Some(checkpoint) = checkpoint else {
pub(crate) fn print_final_output(output: Option<&str>, styles: &Styles) {
let Some(output) = output else {
return;
};
let text = output.trim();
if !text.is_empty() {
eprintln!("\n{}", styles.bold.apply_to("=== Output ==="));
eprintln!("{}", styles.render_markdown(text));
}
}
async fn resolve_final_output_with_client(
client: &server_client::ServerStoreClient,
run_id: &RunId,
checkpoint: Option<&fabro_types::Checkpoint>,
) -> Result<Option<String>> {
let Some(checkpoint) = checkpoint else {
return Ok(None);
};
for node_id in checkpoint.completed_nodes.iter().rev() {
let key = format!("response.{node_id}");
if let Some(serde_json::Value::String(response)) = checkpoint.context_values.get(&key) {
let text = response.trim();
if !text.is_empty() {
eprintln!("\n{}", styles.bold.apply_to("=== Output ==="));
eprintln!("{}", styles.render_markdown(text));
}
return;
let Some(serde_json::Value::String(response)) = checkpoint.context_values.get(&key) else {
continue;
};
let Some(output) = resolve_response_string(client, run_id, response).await? else {
continue;
};
if !output.trim().is_empty() {
return Ok(Some(output));
}
}
Ok(None)
}
async fn resolve_response_string(
client: &server_client::ServerStoreClient,
run_id: &RunId,
response: &str,
) -> Result<Option<String>> {
let Some(blob_id) = blob_id_from_response(response) else {
return Ok(Some(response.to_string()));
};
let Some(bytes) = client.read_run_blob(run_id, &blob_id).await? else {
return Ok(None);
};
let value: serde_json::Value =
serde_json::from_slice(&bytes).context("blob-backed final output should be valid JSON")?;
Ok(Some(match value {
serde_json::Value::String(text) => text,
other => other.to_string(),
}))
}
fn blob_id_from_response(response: &str) -> Option<RunBlobId> {
parse_blob_ref(response).or_else(|| parse_legacy_blob_file_ref(response))
}
pub(crate) fn print_assets(run_dir: &Path, styles: &Styles) {

32
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -305,11 +305,21 @@ mod tests {
let run_record = sample_run_record(run_id, created_at);
let start_record = sample_start_record(run_id, created_at);
let status_record = sample_status();
let first_checkpoint = sample_checkpoint("plan", 1);
let second_checkpoint = sample_checkpoint("code", 2);
let mut first_checkpoint = sample_checkpoint("plan", 1);
let mut second_checkpoint = sample_checkpoint("code", 2);
let conclusion = sample_conclusion();
let retro = sample_retro(run_id);
let sandbox = sample_sandbox();
let summary_blob = run.write_blob(br#"{"done":true}"#).await.unwrap();
let plan_blob = run.write_blob(br#"{"steps":3}"#).await.unwrap();
first_checkpoint.context_values.insert(
"artifact".to_string(),
serde_json::json!(fabro_types::format_blob_ref(&plan_blob)),
);
second_checkpoint.context_values.insert(
"artifact".to_string(),
serde_json::json!(fabro_types::format_blob_ref(&summary_blob)),
);
let node = StageId::new("code", 2);
append_event(
@ -558,8 +568,6 @@ mod tests {
)
.await
.unwrap();
let summary_blob = run.write_blob(br#"{"done":true}"#).await.unwrap();
let plan_blob = run.write_blob(br#"{"steps":3}"#).await.unwrap();
artifact_store
.put(&run_id, &node, "src/lib.rs", b"fn main() {}")
.await
@ -575,7 +583,7 @@ mod tests {
let file_count = export_run(&run, &artifact_store, output.path())
.await
.unwrap();
assert_eq!(file_count, 22);
assert_eq!(file_count, 20);
let exported_run: RunRecord = read_json(&output.path().join("run.json"));
assert_eq!(exported_run.run_id, run_id);
@ -588,6 +596,10 @@ mod tests {
let exported_checkpoint: Checkpoint = read_json(&output.path().join("checkpoint.json"));
assert_eq!(exported_checkpoint.current_node, "code");
assert_eq!(
exported_checkpoint.context_values.get("artifact"),
Some(&serde_json::json!({"done": true}))
);
assert_eq!(
std::fs::read_to_string(output.path().join("graph.fabro")).unwrap(),
"digraph night_sky {}"
@ -635,15 +647,15 @@ mod tests {
let second_checkpoint: Checkpoint = read_json(&output.path().join("checkpoints/0005.json"));
assert_eq!(first_checkpoint.current_node, "plan");
assert_eq!(second_checkpoint.current_node, "code");
assert_eq!(
std::fs::read(output.path().join("blobs").join(plan_blob.to_string())).unwrap(),
br#"{"steps":3}"#
first_checkpoint.context_values.get("artifact"),
Some(&serde_json::json!({"steps": 3}))
);
assert_eq!(
std::fs::read(output.path().join("blobs").join(summary_blob.to_string())).unwrap(),
br#"{"done":true}"#
second_checkpoint.context_values.get("artifact"),
Some(&serde_json::json!({"done": true}))
);
assert!(!output.path().join("blobs").exists());
assert_eq!(
std::fs::read(

1
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -838,7 +838,6 @@ fn attach_json_errors_without_prompting_for_human_input() {
"start"
],
"context_values": {
"current.preamble": "Goal: Wait for approval/n",
"current_node": "start",
"failure_class": "",
"failure_signature": "",

3
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -914,7 +914,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
"start"
],
"context_values": {
"current.preamble": "Goal: Route through the default approval path/n",
"current_node": "start",
"failure_class": "",
"failure_signature": "",
@ -1026,7 +1025,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
"approve"
],
"context_values": {
"current.preamble": "Goal: Route through the default approval path/n",
"current_node": "approve",
"failure_class": "",
"failure_signature": "",
@ -1185,7 +1183,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
"context_values": {
"command.output": "shipped/n",
"command.stderr": "",
"current.preamble": "Goal: Route through the default approval path/n/n## Completed stages/n- **approve**: success/n/n## Context/n- human.gate.label: [A] Approve/n- human.gate.selected: A/n",
"current_node": "ship",
"failure_class": "",
"failure_signature": "",

21
lib/crates/fabro-store/src/keys.rs
View file

@ -3,6 +3,7 @@ use fabro_types::{RunBlobId, RunId};
const RUNS_PREFIX: &str = "runs#";
const RUNS_INDEX_BY_START_PREFIX: &str = "runs#_index#by-start#";
const BLOBS_PREFIX: &str = "blobs#";
const GLOBAL_BLOBS_PREFIX: &str = "blobs#sha256#";
pub(crate) fn runs_index_by_start_prefix() -> &'static str {
RUNS_INDEX_BY_START_PREFIX
@ -28,11 +29,17 @@ pub(crate) fn run_event_key(run_id: &RunId, seq: u32, epoch_ms: i64) -> String {
}
pub(crate) fn blobs_prefix(run_id: &RunId) -> String {
format!("{BLOBS_PREFIX}{run_id}#")
let _ = run_id;
GLOBAL_BLOBS_PREFIX.to_string()
}
pub(crate) fn blob_key(run_id: &RunId, id: &RunBlobId) -> String {
format!("{}{id}", blobs_prefix(run_id))
let _ = run_id;
format!("{GLOBAL_BLOBS_PREFIX}{id}")
}
pub(crate) fn legacy_blob_key(run_id: &RunId, id: &RunBlobId) -> String {
format!("{BLOBS_PREFIX}{run_id}#{id}")
}
pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
@ -45,6 +52,10 @@ pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
}
pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
if let Some(blob_id) = key.strip_prefix(GLOBAL_BLOBS_PREFIX) {
return blob_id.parse().ok();
}
let rest = key.strip_prefix(BLOBS_PREFIX)?;
let (_, blob_id) = rest.split_once('#')?;
blob_id.parse().ok()
@ -92,7 +103,7 @@ mod tests {
let blob_id = RunBlobId::new(b"summary");
assert_eq!(
blob_key(&run_id, &blob_id),
format!("blobs#{run_id}#{blob_id}")
format!("blobs#sha256#{blob_id}")
);
}
@ -103,6 +114,10 @@ mod tests {
Some(7)
);
let blob_id = RunBlobId::new(b"summary");
assert_eq!(
parse_blob_id(&format!("blobs#sha256#{blob_id}")),
Some(blob_id)
);
assert_eq!(
parse_blob_id(&format!("blobs#01JT56VE4Z5NZ814GZN2JZD65A#{blob_id}")),
Some(blob_id)

59
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -305,10 +305,19 @@ impl RunDatabase {
}
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
Ok(self
let global = self
.inner
.db
.get(keys::blob_key(&self.inner.run_id, id))
.await?;
if global.is_some() {
return Ok(global);
}
Ok(self
.inner
.db
.get(keys::legacy_blob_key(&self.inner.run_id, id))
.await?)
}
@ -398,3 +407,51 @@ fn key_to_string(key: &Bytes) -> Result<String> {
String::from_utf8(key.to_vec())
.map_err(|err| StoreError::Other(format!("stored key is not valid UTF-8: {err}")))
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::time::Duration;
use object_store::memory::InMemory;
use crate::Database;
use crate::keys;
#[tokio::test]
async fn read_blob_falls_back_to_legacy_run_scoped_key() {
let object_store = Arc::new(InMemory::new());
let store = Database::new(object_store, "", Duration::from_millis(1));
let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let run = store.create_run(&run_id).await.unwrap();
let blob = br#"{"legacy":true}"#;
let blob_id = fabro_types::RunBlobId::new(blob);
run.inner
.db
.put(keys::legacy_blob_key(&run_id, &blob_id), blob.as_slice())
.await
.unwrap();
let read = run.read_blob(&blob_id).await.unwrap().unwrap();
assert_eq!(read.as_ref(), blob);
}
#[tokio::test]
async fn list_blobs_reads_global_cas_namespace() {
let object_store = Arc::new(InMemory::new());
let store = Database::new(object_store, "", Duration::from_millis(1));
let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let run = store.create_run(&run_id).await.unwrap();
let first_blob = br#"{"a":1}"#;
let second_blob = br#"{"b":2}"#;
let first_id = run.write_blob(first_blob).await.unwrap();
let second_id = run.write_blob(second_blob).await.unwrap();
let mut blob_ids = run.list_blobs().await.unwrap();
blob_ids.sort();
assert_eq!(blob_ids, vec![first_id, second_id]);
}
}

112
lib/crates/fabro-types/src/blob_ref.rs Normal file
View file

@ -0,0 +1,112 @@
use std::path::Path;
use crate::RunBlobId;
const BLOB_REF_PREFIX: &str = "blob://sha256/";
#[must_use]
pub fn format_blob_ref(blob_id: &RunBlobId) -> String {
format!("{BLOB_REF_PREFIX}{blob_id}")
}
#[must_use]
pub fn parse_blob_ref(value: &str) -> Option<RunBlobId> {
value.strip_prefix(BLOB_REF_PREFIX)?.parse().ok()
}
#[must_use]
pub fn parse_legacy_blob_file_ref(value: &str) -> Option<RunBlobId> {
let path = value.strip_prefix("file://")?;
let blob_id = parse_blob_file_name(path)?;
if has_path_suffix(path, &["cache", "artifacts", "values"])
|| has_path_suffix(path, &[".fabro", "artifacts"])
{
Some(blob_id)
} else {
None
}
}
#[must_use]
pub fn parse_managed_blob_file_ref(value: &str) -> Option<RunBlobId> {
let path = value.strip_prefix("file://")?;
let blob_id = parse_blob_file_name(path)?;
if has_path_suffix(path, &["runtime", "blobs"]) || has_path_suffix(path, &[".fabro", "blobs"]) {
Some(blob_id)
} else {
None
}
}
fn parse_blob_file_name(path: &str) -> Option<RunBlobId> {
let file_name = Path::new(path).file_name()?.to_str()?;
let blob_id = file_name.strip_suffix(".json")?;
blob_id.parse().ok()
}
fn has_path_suffix(path: &str, suffix: &[&str]) -> bool {
let components = Path::new(path)
.parent()
.into_iter()
.flat_map(Path::components)
.filter_map(|component| component.as_os_str().to_str())
.collect::<Vec<_>>();
components.ends_with(suffix)
}
#[cfg(test)]
mod tests {
use super::{
format_blob_ref, parse_blob_ref, parse_legacy_blob_file_ref, parse_managed_blob_file_ref,
};
use crate::RunBlobId;
#[test]
fn blob_ref_round_trips() {
let blob_id = RunBlobId::new(br#"{"kind":"summary"}"#);
let formatted = format_blob_ref(&blob_id);
assert_eq!(parse_blob_ref(&formatted), Some(blob_id));
}
#[test]
fn legacy_local_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///tmp/run/cache/artifacts/values/{blob_id}.json");
assert_eq!(parse_legacy_blob_file_ref(&value), Some(blob_id));
}
#[test]
fn legacy_remote_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///sandbox/.fabro/artifacts/{blob_id}.json");
assert_eq!(parse_legacy_blob_file_ref(&value), Some(blob_id));
}
#[test]
fn managed_local_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///tmp/run/runtime/blobs/{blob_id}.json");
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_id));
}
#[test]
fn managed_remote_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///sandbox/.fabro/blobs/{blob_id}.json");
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_id));
}
#[test]
fn ordinary_file_refs_are_not_treated_as_blob_refs() {
assert_eq!(parse_legacy_blob_file_ref("file:///tmp/report.json"), None);
assert_eq!(parse_managed_blob_file_ref("file:///tmp/report.json"), None);
}
}

4
lib/crates/fabro-types/src/lib.rs
View file

@ -1,6 +1,7 @@
extern crate self as fabro_types;
pub mod billing;
pub mod blob_ref;
pub mod checkpoint;
pub mod combine;
pub mod conclusion;
@ -26,6 +27,9 @@ pub use billing::{
ModelBillingFacts, ModelBillingInput, ModelPricing, ModelPricingPolicy, ModelRef, ModelUsage,
OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, Speed, TokenCounts, UsdMicros,
};
pub use blob_ref::{
format_blob_ref, parse_blob_ref, parse_legacy_blob_file_ref, parse_managed_blob_file_ref,
};
pub use checkpoint::Checkpoint;
pub use conclusion::{Conclusion, StageSummary};
pub use fabro_macros::Combine;

351
lib/crates/fabro-workflow/src/artifact.rs
View file

@ -1,24 +1,33 @@
use std::collections::HashMap;
use std::path::Path;
use std::path::{Path, PathBuf};
use futures::future::BoxFuture;
use serde_json::Value;
use fabro_agent::Sandbox;
use fabro_config::RunScratch;
use fabro_types::{
RunBlobId, format_blob_ref, parse_blob_ref, parse_legacy_blob_file_ref,
parse_managed_blob_file_ref,
};
use crate::context::{self, Context};
use crate::error::{FabroError, Result};
use crate::outcome::Outcome;
use crate::records::Checkpoint;
use crate::runtime_store::RunStoreHandle;
/// Threshold above which values are persisted as blobs and materialized to disk (100KB).
/// Threshold above which values are persisted as blobs (100KB).
const BLOB_OFFLOAD_THRESHOLD: usize = 100 * 1024;
/// Prefix used to identify artifact pointer strings in context values.
const ARTIFACT_POINTER_PREFIX: &str = "file://";
/// Offload context values exceeding the blob threshold into SlateDB and materialize cache files.
/// Offload context values exceeding the blob threshold into the blob store.
///
/// For each entry in `updates` whose serialized JSON exceeds `BLOB_OFFLOAD_THRESHOLD`,
/// the value is persisted as a blob in `run_store`, materialized in `cache_dir`, and
/// replaced with a `"file://{path}"` pointer.
/// the value is persisted as a blob in `run_store` and replaced with a
/// `"blob://sha256/{blob_id}"` reference.
/// Small values are left untouched.
///
/// # Errors
@ -29,7 +38,7 @@ pub async fn offload_large_values(
run_store: &RunStoreHandle,
cache_dir: &Path,
) -> Result<()> {
std::fs::create_dir_all(cache_dir)?;
let _ = cache_dir;
for value in updates.values_mut() {
let bytes = serde_json::to_vec(&*value)
@ -40,11 +49,7 @@ pub async fn offload_large_values(
.write_blob(&bytes)
.await
.map_err(|e| FabroError::engine(format!("artifact blob write failed: {e}")))?;
let cache_path = cache_dir.join(format!("{blob_id}.json"));
if !cache_path.exists() {
std::fs::write(&cache_path, &bytes)?;
}
*value = Value::String(format!("{ARTIFACT_POINTER_PREFIX}{}", cache_path.display()));
*value = Value::String(format_blob_ref(&blob_id));
}
}
Ok(())
@ -76,6 +81,71 @@ pub fn format_artifact_reference(path: &str) -> String {
format!("See: {path}")
}
pub fn durable_context_snapshot(context: &Context) -> HashMap<String, Value> {
let mut snapshot = context.snapshot();
snapshot.remove(context::keys::CURRENT_PREAMBLE);
normalize_durable_updates(&mut snapshot);
snapshot
}
pub fn normalize_durable_updates(updates: &mut HashMap<String, Value>) {
for value in updates.values_mut() {
normalize_durable_value(value);
}
}
pub fn normalize_durable_outcomes(node_outcomes: &mut HashMap<String, Outcome>) {
for outcome in node_outcomes.values_mut() {
normalize_durable_updates(&mut outcome.context_updates);
}
}
pub fn normalize_checkpoint_for_resume(checkpoint: &mut Checkpoint) {
checkpoint
.context_values
.remove(context::keys::CURRENT_PREAMBLE);
normalize_durable_updates(&mut checkpoint.context_values);
normalize_durable_outcomes(&mut checkpoint.node_outcomes);
}
pub async fn resolve_context_for_execution(
context: &Context,
run_store: &RunStoreHandle,
env: &dyn Sandbox,
run_dir: &Path,
) -> Result<Context> {
let values = resolved_context_snapshot(context, run_store, env, run_dir).await?;
let resolved = Context::new();
for (key, value) in values {
resolved.set(key, value);
}
Ok(resolved)
}
pub async fn resolve_outcomes_for_execution(
node_outcomes: &HashMap<String, Outcome>,
run_store: &RunStoreHandle,
env: &dyn Sandbox,
run_dir: &Path,
) -> Result<HashMap<String, Outcome>> {
let mut resolved = node_outcomes.clone();
for outcome in resolved.values_mut() {
resolve_execution_values(&mut outcome.context_updates, run_store, env, run_dir).await?;
}
Ok(resolved)
}
pub async fn resolved_context_snapshot(
context: &Context,
run_store: &RunStoreHandle,
env: &dyn Sandbox,
run_dir: &Path,
) -> Result<HashMap<String, Value>> {
let mut values = context.snapshot();
resolve_execution_values(&mut values, run_store, env, run_dir).await?;
Ok(values)
}
/// Sync artifact files to a remote sandbox.
///
/// For each `file://` pointer in `updates`, checks whether the file is accessible
@ -126,6 +196,166 @@ pub async fn sync_artifacts_to_env(
Ok(())
}
fn normalize_durable_value(value: &mut Value) {
match value {
Value::String(current) => {
if let Some(blob_id) =
parse_legacy_blob_file_ref(current).or_else(|| parse_managed_blob_file_ref(current))
{
*current = format_blob_ref(&blob_id);
}
}
Value::Array(items) => {
for item in items {
normalize_durable_value(item);
}
}
Value::Object(map) => {
for item in map.values_mut() {
normalize_durable_value(item);
}
}
Value::Null | Value::Bool(_) | Value::Number(_) => {}
}
}
fn resolve_execution_values<'a>(
values: &'a mut HashMap<String, Value>,
run_store: &'a RunStoreHandle,
env: &'a dyn Sandbox,
run_dir: &'a Path,
) -> BoxFuture<'a, Result<()>> {
Box::pin(async move {
for value in values.values_mut() {
resolve_execution_value(value, run_store, env, run_dir).await?;
}
Ok(())
})
}
fn resolve_execution_value<'a>(
value: &'a mut Value,
run_store: &'a RunStoreHandle,
env: &'a dyn Sandbox,
run_dir: &'a Path,
) -> BoxFuture<'a, Result<()>> {
Box::pin(async move {
match value {
Value::String(current) => {
if let Some(blob_id) =
parse_blob_ref(current).or_else(|| parse_legacy_blob_file_ref(current))
{
*current = materialize_blob_ref(&blob_id, run_store, env, run_dir).await?;
} else if current.starts_with(ARTIFACT_POINTER_PREFIX)
&& parse_managed_blob_file_ref(current).is_none()
{
*current = resolve_explicit_file_ref(current, env).await?;
}
}
Value::Array(items) => {
for item in items {
resolve_execution_value(item, run_store, env, run_dir).await?;
}
}
Value::Object(map) => {
for item in map.values_mut() {
resolve_execution_value(item, run_store, env, run_dir).await?;
}
}
Value::Null | Value::Bool(_) | Value::Number(_) => {}
}
Ok(())
})
}
async fn materialize_blob_ref(
blob_id: &RunBlobId,
run_store: &RunStoreHandle,
env: &dyn Sandbox,
run_dir: &Path,
) -> Result<String> {
let bytes = run_store
.read_blob(blob_id)
.await
.map_err(|e| FabroError::engine(format!("artifact blob read failed: {e}")))?
.ok_or_else(|| FabroError::engine(format!("artifact blob missing: {blob_id}")))?;
if is_local_execution(env, run_dir).await? {
let path = local_materialized_blob_path(run_dir, blob_id);
if !path.exists() {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(&path, &bytes)?;
}
return Ok(format!("{ARTIFACT_POINTER_PREFIX}{}", path.display()));
}
let remote_path = format!("{}/.fabro/blobs/{blob_id}.json", env.working_directory());
if !env
.file_exists(&remote_path)
.await
.map_err(|e| FabroError::engine(format!("failed to check blob existence: {e}")))?
{
let content = String::from_utf8(bytes.to_vec()).map_err(|e| {
FabroError::engine(format!("artifact blob was not valid UTF-8 JSON: {e}"))
})?;
env.write_file(&remote_path, &content).await.map_err(|e| {
FabroError::engine(format!("failed to write artifact blob to sandbox: {e}"))
})?;
}
Ok(format!("{ARTIFACT_POINTER_PREFIX}{remote_path}"))
}
async fn resolve_explicit_file_ref(value: &str, env: &dyn Sandbox) -> Result<String> {
let local_path = value
.strip_prefix(ARTIFACT_POINTER_PREFIX)
.ok_or_else(|| FabroError::engine(format!("invalid artifact pointer: {value}")))?;
if env
.file_exists(local_path)
.await
.map_err(|e| FabroError::engine(format!("failed to check artifact existence: {e}")))?
{
return Ok(value.to_string());
}
let content = std::fs::read_to_string(local_path).map_err(|e| {
FabroError::engine(format!("failed to read local artifact {local_path}: {e}"))
})?;
let filename = Path::new(local_path)
.file_name()
.and_then(|file| file.to_str())
.unwrap_or("artifact.json");
let remote_path = format!("{}/.fabro/artifacts/{filename}", env.working_directory());
if !env
.file_exists(&remote_path)
.await
.map_err(|e| FabroError::engine(format!("failed to check artifact existence: {e}")))?
{
env.write_file(&remote_path, &content).await.map_err(|e| {
FabroError::engine(format!("failed to write artifact to remote env: {e}"))
})?;
}
Ok(format!("{ARTIFACT_POINTER_PREFIX}{remote_path}"))
}
async fn is_local_execution(env: &dyn Sandbox, run_dir: &Path) -> Result<bool> {
env.file_exists(&run_dir.to_string_lossy())
.await
.map_err(|e| FabroError::engine(format!("failed to inspect sandbox locality: {e}")))
}
fn local_materialized_blob_path(run_dir: &Path, blob_id: &RunBlobId) -> PathBuf {
RunScratch::new(run_dir)
.runtime_dir()
.join("blobs")
.join(format!("{blob_id}.json"))
}
#[cfg(test)]
mod tests {
use std::hash::{Hash, Hasher};
@ -166,13 +396,9 @@ mod tests {
.unwrap();
let pointer = updates.get("response.plan").unwrap();
let path = artifact_path(pointer).expect("should be an artifact pointer");
assert_eq!(
path,
dir.path()
.join(format!("{expected_blob_id}.json"))
.to_str()
.unwrap()
pointer,
&serde_json::json!(fabro_types::format_blob_ref(&expected_blob_id))
);
let blob = run_store
@ -183,8 +409,8 @@ mod tests {
let blob_value: serde_json::Value = serde_json::from_slice(&blob).unwrap();
assert_eq!(blob_value, serde_json::json!(large_string));
assert!(
dir.path().join(format!("{expected_blob_id}.json")).exists(),
"materialized cache file should exist"
std::fs::read_dir(dir.path()).unwrap().next().is_none(),
"offload should not materialize host cache files"
);
}
@ -225,6 +451,93 @@ mod tests {
assert_eq!(artifact_path(&value), None);
}
#[test]
fn normalize_durable_updates_rewrites_managed_blob_file_refs_recursively() {
let blob_id = fabro_types::RunBlobId::new(b"hello");
let mut updates = HashMap::from([(
"nested".to_string(),
serde_json::json!({
"items": [
format!("file:///tmp/run/runtime/blobs/{blob_id}.json"),
format!("file:///sandbox/.fabro/blobs/{blob_id}.json"),
"file:///tmp/report.json",
]
}),
)]);
normalize_durable_updates(&mut updates);
assert_eq!(
updates["nested"],
serde_json::json!({
"items": [
fabro_types::format_blob_ref(&blob_id),
fabro_types::format_blob_ref(&blob_id),
"file:///tmp/report.json",
]
})
);
}
#[test]
fn normalize_checkpoint_for_resume_converts_legacy_blob_file_refs_and_drops_preamble() {
let blob_id = fabro_types::RunBlobId::new(b"legacy");
let mut checkpoint = crate::records::Checkpoint {
timestamp: chrono::Utc::now(),
current_node: "work".to_string(),
completed_nodes: vec!["work".to_string()],
node_retries: HashMap::new(),
context_values: HashMap::from([
(
crate::context::keys::CURRENT_PREAMBLE.to_string(),
serde_json::json!("runtime only"),
),
(
"response.work".to_string(),
serde_json::json!(format!(
"file:///tmp/run/cache/artifacts/values/{blob_id}.json"
)),
),
]),
node_outcomes: HashMap::from([(
"work".to_string(),
crate::outcome::Outcome {
context_updates: HashMap::from([(
"response.work".to_string(),
serde_json::json!(format!(
"file:///sandbox/.fabro/artifacts/{blob_id}.json"
)),
)]),
..crate::outcome::Outcome::success()
},
)]),
next_node_id: Some("exit".to_string()),
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::new(),
};
normalize_checkpoint_for_resume(&mut checkpoint);
assert!(
!checkpoint
.context_values
.contains_key(crate::context::keys::CURRENT_PREAMBLE)
);
assert_eq!(
checkpoint.context_values.get("response.work"),
Some(&serde_json::json!(fabro_types::format_blob_ref(&blob_id)))
);
assert_eq!(
checkpoint
.node_outcomes
.get("work")
.and_then(|outcome| outcome.context_updates.get("response.work")),
Some(&serde_json::json!(fabro_types::format_blob_ref(&blob_id)))
);
}
// --- sync_artifacts_to_env tests ---
use std::sync::Mutex;

4
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -12,7 +12,7 @@ use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle};
use fabro_core::outcome::NodeResult;
use fabro_core::state::ExecutionState;
use crate::artifact::{offload_large_values, sync_artifacts_to_env};
use crate::artifact::{normalize_durable_updates, offload_large_values, sync_artifacts_to_env};
use crate::artifact_snapshot::{CapturedArtifactInfo, collect_artifacts};
use crate::artifact_upload::StageArtifactUploader;
use crate::event::{Emitter, Event, RunNoticeLevel};
@ -188,6 +188,8 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
});
}
normalize_durable_updates(&mut result.outcome.context_updates);
// Sync file-backed artifacts to sandbox environment
if let Err(e) =
sync_artifacts_to_env(&mut result.outcome.context_updates, &*self.sandbox).await

21
lib/crates/fabro-workflow/src/lifecycle/event.rs
View file

@ -15,6 +15,7 @@ use fabro_core::state::ExecutionState;
use super::circuit_breaker::CircuitBreakerLifecycle;
use super::git::GitCheckpointResult;
use crate::artifact;
use crate::context;
use crate::error::FabroError;
use crate::event::{Emitter, Event};
@ -320,6 +321,10 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let diff = git_result.as_ref().and_then(|r| r.diff.clone());
let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker);
let context_values = artifact::durable_context_snapshot(&state.context);
let mut node_outcomes = state.node_outcomes.clone();
node_outcomes.insert(node.id().to_string(), result.outcome.clone());
artifact::normalize_durable_outcomes(&mut node_outcomes);
self.emitter.emit(&Event::CheckpointCompleted {
node_id: node.id().to_string(),
@ -331,20 +336,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>(),
context_values: state
.context
.snapshot()
.into_iter()
.collect::<BTreeMap<_, _>>(),
node_outcomes: state
.node_outcomes
.clone()
.into_iter()
.chain(std::iter::once((
node.id().to_string(),
result.outcome.clone(),
)))
.collect::<BTreeMap<_, _>>(),
context_values: context_values.into_iter().collect::<BTreeMap<_, _>>(),
node_outcomes: node_outcomes.into_iter().collect::<BTreeMap<_, _>>(),
next_node_id: next_node_id.map(ToOwned::to_owned),
git_commit_sha: git_sha.clone(),
loop_failure_signatures: loop_failure_signatures.unwrap_or_default(),

39
lib/crates/fabro-workflow/src/lifecycle/fidelity.rs
View file

@ -1,18 +1,23 @@
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_core::error::CoreError;
use fabro_core::error::Result as CoreResult;
use fabro_core::graph::NodeSpec;
use fabro_core::lifecycle::{EdgeContext, EdgeDecision, NodeDecision, RunLifecycle};
use fabro_core::state::ExecutionState;
use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode};
use crate::artifact;
use crate::context::keys;
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::handler::llm::preamble::build_preamble;
use crate::outcome::BilledModelUsage;
use crate::runtime_store::RunStoreHandle;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeDecision = NodeDecision<Option<BilledModelUsage>>;
@ -27,15 +32,26 @@ struct IncomingEdgeData {
/// Sub-lifecycle responsible for fidelity/thread resolution and context key setup.
pub(crate) struct FidelityLifecycle {
pub graph: Arc<GvGraph>,
pub sandbox: Arc<dyn Sandbox>,
pub run_store: RunStoreHandle,
pub run_dir: PathBuf,
incoming_edge_data: Mutex<Option<IncomingEdgeData>>,
/// True on the first node after checkpoint resume when prior fidelity was Full.
degrade_fidelity_on_resume: Mutex<bool>,
}
impl FidelityLifecycle {
pub(crate) fn new(graph: Arc<GvGraph>) -> Self {
pub(crate) fn new(
graph: Arc<GvGraph>,
sandbox: Arc<dyn Sandbox>,
run_store: RunStoreHandle,
run_dir: PathBuf,
) -> Self {
Self {
graph,
sandbox,
run_store,
run_dir,
incoming_edge_data: Mutex::new(None),
degrade_fidelity_on_resume: Mutex::new(false),
}
@ -84,12 +100,29 @@ impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
);
// 4. Preamble building: if Full, empty preamble; otherwise build from context
let resolved_context = artifact::resolve_context_for_execution(
&state.context,
&self.run_store,
&*self.sandbox,
&self.run_dir,
)
.await
.map_err(|err| CoreError::Other(err.to_string()))?;
let resolved_outcomes = artifact::resolve_outcomes_for_execution(
&state.node_outcomes,
&self.run_store,
&*self.sandbox,
&self.run_dir,
)
.await
.map_err(|err| CoreError::Other(err.to_string()))?;
let preamble = build_preamble(
fidelity,
&state.context,
&resolved_context,
&self.graph,
&state.completed_nodes,
&state.node_outcomes,
&resolved_outcomes,
);
state
.context

4
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -13,6 +13,7 @@ use fabro_core::lifecycle::RunLifecycle;
use fabro_core::outcome::NodeResult;
use fabro_core::state::ExecutionState;
use crate::artifact;
use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::git::MetadataStore;
use crate::graph::WorkflowGraph;
@ -37,6 +38,7 @@ fn build_checkpoint(
) -> fabro_types::Checkpoint {
let mut node_outcomes = state.node_outcomes.clone();
node_outcomes.insert(node.id().to_string(), result.outcome.clone());
artifact::normalize_durable_outcomes(&mut node_outcomes);
fabro_types::Checkpoint {
timestamp: chrono::Utc::now(),
@ -44,7 +46,7 @@ fn build_checkpoint(
completed_nodes: state.completed_nodes.clone(),
node_outcomes,
node_retries: state.node_retries.clone(),
context_values: state.context.snapshot(),
context_values: artifact::durable_context_snapshot(&state.context),
next_node_id: next_node_id.map(String::from),
git_commit_sha,
node_visits: state.node_visits.clone(),

9
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -141,7 +141,12 @@ impl WorkflowLifecycle {
graph_name: graph.name.clone(),
};
let fidelity = FidelityLifecycle::new(Arc::clone(&graph));
let fidelity = FidelityLifecycle::new(
Arc::clone(&graph),
Arc::clone(sandbox),
run_store.clone(),
run_dir.clone(),
);
let start_node_id = graph.find_start_node().map(|n| n.id.clone());
@ -310,9 +315,9 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
) -> CoreResult<()> {
self.auto_status.after_node(node, result, state).await?;
self.circuit_breaker.after_node(node, result, state).await?;
self.artifact.after_node(node, result, state).await?;
self.event.after_node(node, result, state).await?;
self.hook.after_node(node, result, state).await?;
self.artifact.after_node(node, result, state).await?;
Ok(())
}

25
lib/crates/fabro-workflow/src/node_handler.rs
View file

@ -10,6 +10,7 @@ use fabro_core::handler::NodeHandler;
use fabro_core::outcome::FailureCategory;
use fabro_core::retry::RetryPolicy as CoreRetryPolicy;
use crate::artifact;
use crate::context::Context;
use crate::graph::WorkflowGraph;
@ -42,9 +43,22 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
let gv_node = node.inner();
let handler = self.services.registry.resolve(gv_node);
// Fork the context so handler writes don't leak back unless we diff+apply.
let snapshot = context.snapshot();
let wf_context = context.fork();
let wf_context = artifact::resolve_context_for_execution(
context,
&self.services.run_store,
&*self.services.sandbox,
&self.run_dir,
)
.await
.map_err(|err| {
CoreError::handler(HandlerErrorDetail {
message: err.to_string(),
retryable: true,
category: Some(FailureCategory::TransientInfra),
signature: None,
})
})?;
let execution_snapshot = wf_context.snapshot();
// Timeout from the node
let node_timeout = gv_node.timeout();
@ -79,9 +93,10 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
// 2. After handler returns, diff the forked context against the snapshot
// and apply changes back to the original context
let new_values = wf_context.snapshot();
let mut new_values = wf_context.snapshot();
artifact::normalize_durable_updates(&mut new_values);
for (k, v) in &new_values {
if snapshot.get(k) != Some(v) {
if execution_snapshot.get(k) != Some(v) {
context.set(k.clone(), v.clone());
}
}

6
lib/crates/fabro-workflow/src/pipeline/execute.rs
View file

@ -7,6 +7,7 @@ use fabro_core::state::ExecutionState;
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use crate::artifact;
use crate::context::{self, Context};
use crate::error::FabroError;
use crate::event::Event;
@ -57,6 +58,11 @@ pub async fn execute(init: Initialized) -> Executed {
provider,
} = init;
let mut checkpoint = checkpoint;
if let Some(cp) = checkpoint.as_mut() {
artifact::normalize_checkpoint_for_resume(cp);
}
let start = Instant::now();
let graph_arc = Arc::new(graph.clone());
let wf_graph = WorkflowGraph(Arc::clone(&graph_arc));

7
lib/crates/fabro-workflow/src/records/checkpoint.rs
View file

@ -1,5 +1,6 @@
use std::collections::HashMap;
use crate::artifact;
use crate::context::Context;
use crate::outcome::Outcome;
pub use fabro_types::checkpoint::Checkpoint;
@ -25,18 +26,20 @@ impl CheckpointExt for Checkpoint {
current_node: &str,
completed_nodes: Vec<String>,
node_retries: HashMap<String, u32>,
node_outcomes: HashMap<String, Outcome>,
mut node_outcomes: HashMap<String, Outcome>,
next_node_id: Option<String>,
loop_failure_signatures: HashMap<FailureSignature, usize>,
restart_failure_signatures: HashMap<FailureSignature, usize>,
node_visits: HashMap<String, usize>,
) -> Self {
artifact::normalize_durable_outcomes(&mut node_outcomes);
Self {
timestamp: chrono::Utc::now(),
current_node: current_node.to_string(),
completed_nodes,
node_retries,
context_values: context.snapshot(),
context_values: artifact::durable_context_snapshot(context),
node_outcomes,
next_node_id,
git_commit_sha: None,

74
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -1,8 +1,11 @@
use std::collections::HashMap;
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_store::{ArtifactStore, RunDatabase, RunProjection};
use fabro_types::{RunBlobId, parse_blob_ref, parse_legacy_blob_file_ref};
use futures::future::BoxFuture;
use crate::git::MetadataStore;
@ -211,18 +214,6 @@ impl RunDump {
);
}
for blob_id in run_store.list_blobs().await? {
let blob_name = validate_single_path_segment("blob id", &blob_id.to_string())?;
let value = run_store
.read_blob(&blob_id)
.await?
.with_context(|| format!("blob {blob_id:?} is missing from the store"))?;
entries.push(RunDumpEntry::bytes_path(
&PathBuf::from("blobs").join(blob_name),
value.to_vec(),
));
}
if let Some(run_record) = run_record {
for asset in artifact_store.list_for_run(&run_record.run_id).await? {
let node_id_segment =
@ -250,6 +241,8 @@ impl RunDump {
}
}
hydrate_referenced_blobs(&mut entries, run_store).await?;
Ok(Self { entries })
}
@ -410,6 +403,63 @@ fn validate_relative_path(kind: &str, value: &str) -> Result<PathBuf> {
Ok(normalized)
}
async fn hydrate_referenced_blobs(
entries: &mut [RunDumpEntry],
run_store: &RunDatabase,
) -> Result<()> {
let mut cache = HashMap::new();
for entry in entries {
if let RunDumpContents::Json(value) = &mut entry.contents {
hydrate_blob_refs_in_value(value, run_store, &mut cache).await?;
}
}
Ok(())
}
fn hydrate_blob_refs_in_value<'a>(
value: &'a mut serde_json::Value,
run_store: &'a RunDatabase,
cache: &'a mut HashMap<RunBlobId, serde_json::Value>,
) -> BoxFuture<'a, Result<()>> {
Box::pin(async move {
match value {
serde_json::Value::String(current) => {
let Some(blob_id) =
parse_blob_ref(current).or_else(|| parse_legacy_blob_file_ref(current))
else {
return Ok(());
};
if let Some(cached) = cache.get(&blob_id).cloned() {
*value = cached;
return Ok(());
}
let blob = run_store
.read_blob(&blob_id)
.await?
.with_context(|| format!("blob {blob_id:?} is missing from the store"))?;
let hydrated: serde_json::Value = serde_json::from_slice(&blob)
.with_context(|| format!("blob {blob_id:?} is not valid JSON"))?;
cache.insert(blob_id, hydrated.clone());
*value = hydrated;
}
serde_json::Value::Array(items) => {
for item in items {
hydrate_blob_refs_in_value(item, run_store, cache).await?;
}
}
serde_json::Value::Object(map) => {
for item in map.values_mut() {
hydrate_blob_refs_in_value(item, run_store, cache).await?;
}
}
serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => {
}
}
Ok(())
})
}
fn ensure_parent_dir(path: &Path) -> Result<()> {
let parent = path
.parent()

26
lib/crates/fabro-workflow/tests/it/daytona_integration.rs
View file

@ -489,7 +489,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
// Checkpoint should have a pointer rewritten for Daytona
// Checkpoint should persist a durable blob ref.
let checkpoint =
load_checkpoint(&dir.path().join("checkpoint.json")).expect("checkpoint should load");
let pointer_value = checkpoint
@ -497,24 +497,14 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
.get("response.big_output")
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
let expected_prefix = format!("file://{}/.fabro/artifacts/", env.working_directory());
assert!(
pointer_str.starts_with(&expected_prefix),
"pointer should reference Daytona path, got: {pointer_str}"
let expected_blob_id = fabro_types::RunBlobId::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
// Verify the artifact file is readable in the sandbox
let remote_path = pointer_str.strip_prefix("file://").unwrap();
assert!(
env.file_exists(remote_path).await.unwrap(),
"artifact should exist in Daytona sandbox at {remote_path}"
);
let remote_content = env.read_file(remote_path, None, None).await.unwrap();
assert!(
remote_content.len() > 100 * 1024,
"remote artifact should be >100KB, got {} bytes",
remote_content.len()
assert_eq!(
pointer_str,
fabro_types::format_blob_ref(&expected_blob_id),
"checkpoint should persist a blob ref"
);
env.cleanup().await.unwrap();

258
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -1525,6 +1525,31 @@ impl Handler for LargeOutputHandler {
}
}
#[derive(Clone)]
struct ContextValueCaptureHandler {
values: Arc<std::sync::Mutex<Vec<String>>>,
key: String,
}
#[async_trait::async_trait]
impl Handler for ContextValueCaptureHandler {
async fn execute(
&self,
_node: &Node,
context: &Context,
_graph: &Graph,
_run_dir: &Path,
_services: &fabro_workflow::handler::EngineServices,
) -> Result<Outcome, FabroError> {
let value = context
.get(&self.key)
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.expect("captured context value should be a string");
self.values.lock().unwrap().push(value);
Ok(Outcome::success())
}
}
/// A handler that sets `context_updates` = {"`my_flag"`: "set"}.
struct ContextSetterHandler;
@ -5126,6 +5151,10 @@ async fn fidelity_stored_in_checkpoint_context() {
Some(&serde_json::json!("summary:low")),
"checkpoint should record the resolved fidelity"
);
assert!(
!cp.context_values.contains_key("current.preamble"),
"checkpoint should exclude runtime-only preamble state"
);
}
#[tokio::test]
@ -8586,7 +8615,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
// The checkpoint context should contain an artifact pointer, not the full value
// The checkpoint context should contain a durable blob ref, not the full value.
let checkpoint =
load_checkpoint(&dir.path().join("checkpoint.json")).expect("checkpoint should load");
let pointer_value = checkpoint
@ -8594,35 +8623,23 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
.get("response.big_output")
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
assert!(
pointer_str.starts_with("file://"),
"value should be an artifact pointer, got: {pointer_str}"
);
let expected_blob_id = fabro_types::RunBlobId::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
// The artifact file should exist on disk
let artifact_file = RunScratch::new(dir.path())
.blob_cache_dir()
.join(format!("{expected_blob_id}.json"));
assert!(
artifact_file.exists(),
"artifact file should exist at {artifact_file:?}"
assert_eq!(
pointer_str,
fabro_types::format_blob_ref(&expected_blob_id),
"value should be a durable blob ref"
);
// The artifact file should contain the original large value
let artifact_content =
std::fs::read_to_string(&artifact_file).expect("should read artifact file");
let artifact_value: serde_json::Value =
serde_json::from_str(&artifact_content).expect("should parse artifact JSON");
let artifact_str = artifact_value.as_str().expect("should be a string");
assert_eq!(
artifact_str.len(),
150 * 1024,
"artifact should contain the original 150KB value"
assert!(
!RunScratch::new(dir.path())
.blob_cache_dir()
.join(format!("{expected_blob_id}.json"))
.exists(),
"legacy host blob cache file should not exist"
);
// WorkflowRunCompleted artifact_count now tracks captured artifacts, not offloaded values.
@ -8649,6 +8666,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
struct RemoteMockEnv {
working_dir: String,
written: std::sync::Mutex<Vec<(String, String)>>,
existing_paths: std::sync::Mutex<std::collections::HashSet<String>>,
}
impl RemoteMockEnv {
@ -8656,6 +8674,7 @@ impl RemoteMockEnv {
Self {
working_dir: working_dir.to_string(),
written: std::sync::Mutex::new(Vec::new()),
existing_paths: std::sync::Mutex::new(std::collections::HashSet::new()),
}
}
}
@ -8676,6 +8695,7 @@ impl fabro_agent::Sandbox for RemoteMockEnv {
.lock()
.unwrap()
.push((path.to_string(), content.to_string()));
self.existing_paths.lock().unwrap().insert(path.to_string());
Ok(())
}
@ -8683,8 +8703,8 @@ impl fabro_agent::Sandbox for RemoteMockEnv {
Err("not implemented".to_string())
}
async fn file_exists(&self, _path: &str) -> std::result::Result<bool, String> {
Ok(false)
async fn file_exists(&self, path: &str) -> std::result::Result<bool, String> {
Ok(self.existing_paths.lock().unwrap().contains(path))
}
async fn list_directory(
@ -8807,7 +8827,7 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() {
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
// The checkpoint context should contain a pointer rewritten for the remote env
// The checkpoint context should contain a durable blob ref.
let checkpoint =
load_checkpoint(&dir.path().join("checkpoint.json")).expect("checkpoint should load");
let pointer_value = checkpoint
@ -8815,14 +8835,190 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() {
.get("response.big_output")
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
assert!(
pointer_str.starts_with("file:///sandbox/.fabro/artifacts/"),
"pointer should reference remote path, got: {pointer_str}"
let expected_blob_id = fabro_types::RunBlobId::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
assert_eq!(
pointer_str,
fabro_types::format_blob_ref(&expected_blob_id),
"checkpoint should persist a blob ref"
);
// The RemoteMockEnv should have received exactly one write with >100KB content
let written = remote_env.written.lock().unwrap();
assert_eq!(written.len(), 1, "should have written 1 artifact");
assert!(
written.is_empty(),
"blob materialization should not happen until a downstream execution needs it"
);
}
#[tokio::test]
async fn downstream_local_execution_materializes_blob_refs_to_runtime_files() {
let mut graph = make_graph_with_start_exit("ArtifactMaterializeLocal");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Test local blob materialization".to_string()),
);
let mut big_output = Node::new("big_output");
big_output.attrs.insert(
"label".to_string(),
AttrValue::String("Big Output".to_string()),
);
graph.nodes.insert("big_output".to_string(), big_output);
let mut inspect = Node::new("inspect");
inspect.attrs.insert(
"label".to_string(),
AttrValue::String("Inspect".to_string()),
);
inspect.attrs.insert(
"type".to_string(),
AttrValue::String("capture_context".to_string()),
);
graph.nodes.insert("inspect".to_string(), inspect);
graph.edges.push(Edge::new("start", "big_output"));
graph.edges.push(Edge::new("big_output", "inspect"));
graph.edges.push(Edge::new("inspect", "exit"));
let dir = tempfile::tempdir().unwrap();
let captured = Arc::new(std::sync::Mutex::new(Vec::new()));
let mut registry = HandlerRegistry::new(Box::new(LargeOutputHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register(
"capture_context",
Box::new(ContextValueCaptureHandler {
values: Arc::clone(&captured),
key: "response.big_output".to_string(),
}),
);
let engine = WorkflowRunner::new(registry, Arc::new(Emitter::default()), local_env());
let run_options = RunOptions {
settings: Settings::default(),
run_dir: dir.path().to_path_buf(),
cancel_token: None,
run_id: test_run_id("test-run"),
labels: std::collections::HashMap::new(),
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
let (outcome, _state) = engine
.run_with_state(&graph, &run_options)
.await
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
let expected_blob_id = fabro_types::RunBlobId::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
let captured_value = captured.lock().unwrap().first().cloned().unwrap();
let expected_path = RunScratch::new(dir.path())
.runtime_dir()
.join("blobs")
.join(format!("{expected_blob_id}.json"));
assert_eq!(
captured_value,
format!("file://{}", expected_path.display()),
"downstream handlers should receive a local file ref"
);
let artifact_content = std::fs::read_to_string(&expected_path).expect("should read artifact");
let artifact_value: serde_json::Value =
serde_json::from_str(&artifact_content).expect("should parse artifact JSON");
let artifact_str = artifact_value
.as_str()
.expect("artifact should be a string");
assert_eq!(artifact_str.len(), 150 * 1024);
}
#[tokio::test]
async fn downstream_remote_execution_materializes_blob_refs_to_sandbox_files() {
let mut graph = make_graph_with_start_exit("ArtifactMaterializeRemote");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Test remote blob materialization".to_string()),
);
let mut big_output = Node::new("big_output");
big_output.attrs.insert(
"label".to_string(),
AttrValue::String("Big Output".to_string()),
);
graph.nodes.insert("big_output".to_string(), big_output);
let mut inspect = Node::new("inspect");
inspect.attrs.insert(
"label".to_string(),
AttrValue::String("Inspect".to_string()),
);
inspect.attrs.insert(
"type".to_string(),
AttrValue::String("capture_context".to_string()),
);
graph.nodes.insert("inspect".to_string(), inspect);
graph.edges.push(Edge::new("start", "big_output"));
graph.edges.push(Edge::new("big_output", "inspect"));
graph.edges.push(Edge::new("inspect", "exit"));
let dir = tempfile::tempdir().unwrap();
let captured = Arc::new(std::sync::Mutex::new(Vec::new()));
let mut registry = HandlerRegistry::new(Box::new(LargeOutputHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register(
"capture_context",
Box::new(ContextValueCaptureHandler {
values: Arc::clone(&captured),
key: "response.big_output".to_string(),
}),
);
let remote_env = Arc::new(RemoteMockEnv::new("/sandbox"));
let engine = WorkflowRunner::new(registry, Arc::new(Emitter::default()), remote_env.clone());
let run_options = RunOptions {
settings: Settings::default(),
run_dir: dir.path().to_path_buf(),
cancel_token: None,
run_id: test_run_id("test-run"),
labels: std::collections::HashMap::new(),
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
let (outcome, _state) = engine
.run_with_state(&graph, &run_options)
.await
.expect("pipeline should succeed");
assert_eq!(outcome.status, StageStatus::Success);
let expected_blob_id = fabro_types::RunBlobId::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
let captured_value = captured.lock().unwrap().first().cloned().unwrap();
assert_eq!(
captured_value,
format!("file:///sandbox/.fabro/blobs/{expected_blob_id}.json"),
"downstream handlers should receive a sandbox-local file ref"
);
let written = remote_env.written.lock().unwrap();
assert_eq!(written.len(), 1, "should materialize the blob once");
assert_eq!(
written[0].0,
format!("/sandbox/.fabro/blobs/{expected_blob_id}.json")
);
assert!(
written[0].1.len() > 100 * 1024,
"written content should be >100KB, got {} bytes",