scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Add per-asset metadata (mime, md5, sha256) to asset capture

Browse source 
Replace the batch AssetsCaptured event with per-file AssetCaptured events
that include content hashes and MIME type. The asset collection manifest
now stores a captured_assets array with full metadata instead of bare
path strings, enabling downstream integrity verification and content
type awareness.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-29 22:10:09 -04:00
parent be6e37fa26
commit b6947af7af
10 changed files with 206 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -2013,6 +2013,8 @@ dependencies = [
"futures",
"git2",
"hex",
"md5",
"mime_guess",
"predicates",
"rand 0.8.5",
"regex",

1
Cargo.toml
View file

@ -46,6 +46,7 @@ sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "chrono"] }
dirs = "6"
mac_address = "1"
md5 = "0.7"
mime_guess = "2"
indicatif = "0.18"
termimad = "0.34"
toml = "0.8"

13
docs/agents/outputs.mdx
View file

@ -276,10 +276,15 @@ Each collection writes a `manifest.json` summarizing what was captured:
"total_bytes": 245760,
"files_skipped": 0,
"download_errors": 0,
"copied_paths": [
"test-results/screenshot.png",
"test-results/video.webm",
"playwright-report/index.html"
"hash_errors": 0,
"captured_assets": [
{
"path": "test-results/screenshot.png",
"mime": "image/png",
"content_md5": "a1b2c3...",
"content_sha256": "d4e5f6...",
"bytes": 81920
}
]
}
```

2
lib/crates/fabro-cli/tests/it/scenario.rs
View file

@ -788,7 +788,7 @@ fn local_run_lifecycle() {
std::fs::write(asset_dir.join("output.txt"), "asset-content-42").unwrap();
std::fs::write(
asset_dir.join("manifest.json"),
r#"{"files_copied":1,"total_bytes":16,"files_skipped":0,"download_errors":0,"copied_paths":["output.txt"]}"#,
r#"{"files_copied":1,"total_bytes":16,"files_skipped":0,"download_errors":0,"hash_errors":0,"captured_assets":[{"path":"output.txt","mime":"text/plain","content_md5":"f02439728c0a94b7bfc465acb1201a1f","content_sha256":"0af9dea3e1c2dec968531c18c9331659b8268e8c9cf24b01cda7b8ce51d2ff00","bytes":16}]}"#,
)
.unwrap();

2
lib/crates/fabro-workflows/Cargo.toml
View file

@ -54,8 +54,10 @@ chrono = { workspace = true, features = ["serde"] }
dirs = "6"
regex.workspace = true
scopeguard = "1"
md5.workspace = true
hex.workspace = true
sha2 = { workspace = true }
mime_guess.workspace = true
shlex = "1"
git2.workspace = true
tokio-util.workspace = true

109
lib/crates/fabro-workflows/src/asset_snapshot.rs
View file

@ -1,5 +1,6 @@
use fabro_agent::Sandbox;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::path::Path;
use tokio::fs;
use tracing::{debug, warn};
@ -12,6 +13,16 @@ pub struct DiscoveredFile {
pub mtime_epoch_secs: f64,
}
/// Metadata for a single captured asset file.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CapturedAssetInfo {
pub path: String,
pub mime: String,
pub content_md5: String,
pub content_sha256: String,
pub bytes: u64,
}
/// Summary of an asset collection run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AssetCollectionSummary {
@ -19,7 +30,8 @@ pub struct AssetCollectionSummary {
pub total_bytes: u64,
pub files_skipped: usize,
pub download_errors: usize,
pub copied_paths: Vec<String>,
pub hash_errors: usize,
pub captured_assets: Vec<CapturedAssetInfo>,
}
/// Directories to exclude from the find search and checkpoint commits.
@ -242,6 +254,27 @@ fn normalize_paths(discovered: Vec<DiscoveredFile>, root: &str) -> Vec<Discovere
.collect()
}
fn compute_asset_info(
relative_path: &str,
local_path: &Path,
) -> std::result::Result<CapturedAssetInfo, String> {
let mime = mime_guess::from_path(relative_path)
.first_or_octet_stream()
.to_string();
let data = std::fs::read(local_path)
.map_err(|e| format!("failed to read {}: {e}", local_path.display()))?;
let bytes = u64::try_from(data.len()).unwrap_or(u64::MAX);
let content_md5 = format!("{:x}", md5::compute(&data));
let content_sha256 = hex::encode(Sha256::digest(&data));
Ok(CapturedAssetInfo {
path: relative_path.to_string(),
mime,
content_md5,
content_sha256,
bytes,
})
}
/// Collect asset files matching the configured globs that were created during this stage.
pub async fn collect_assets(
sandbox: &dyn Sandbox,
@ -268,7 +301,8 @@ pub async fn collect_assets(
let mut files_copied: usize = 0;
let mut total_bytes: u64 = 0;
let mut download_errors: usize = 0;
let mut copied_paths: Vec<String> = Vec::new();
let mut hash_errors: usize = 0;
let mut captured_assets: Vec<CapturedAssetInfo> = Vec::new();
for file in &to_collect {
let dest = stage_dir.join(&file.relative_path);
@ -276,11 +310,22 @@ pub async fn collect_assets(
.download_file_to_local(&file.relative_path, &dest)
.await
{
Ok(()) => {
files_copied += 1;
total_bytes += file.size;
copied_paths.push(file.relative_path.clone());
}
Ok(()) => match compute_asset_info(&file.relative_path, &dest) {
Ok(info) => {
files_copied += 1;
total_bytes += info.bytes;
captured_assets.push(info);
}
Err(e) => {
warn!(
path = file.relative_path.as_str(),
error = e.as_str(),
"Asset hash failed"
);
let _ = std::fs::remove_file(&dest);
hash_errors += 1;
}
},
Err(e) => {
warn!(
path = file.relative_path.as_str(),
@ -298,7 +343,8 @@ pub async fn collect_assets(
total_bytes,
files_skipped,
download_errors,
copied_paths,
hash_errors,
captured_assets,
};
if files_copied > 0 {
@ -339,8 +385,8 @@ pub fn collect_asset_paths(assets_dir: &Path) -> Vec<String> {
continue;
};
let retry_dir = retry_entry.path();
for relative_path in &summary.copied_paths {
let full_path = retry_dir.join(relative_path);
for asset in &summary.captured_assets {
let full_path = retry_dir.join(&asset.path);
all_paths.push(full_path.to_string_lossy().into_owned());
}
}
@ -639,9 +685,19 @@ mod tests {
.unwrap();
assert_eq!(summary.files_copied, 1);
assert_eq!(summary.total_bytes, 1024);
assert_eq!(summary.total_bytes, 7);
assert_eq!(summary.download_errors, 0);
assert_eq!(summary.copied_paths, vec!["test-results/r.xml"]);
assert_eq!(summary.hash_errors, 0);
assert_eq!(summary.captured_assets.len(), 1);
let asset = &summary.captured_assets[0];
assert_eq!(asset.path, "test-results/r.xml");
assert_eq!(asset.mime, "text/xml");
assert_eq!(asset.bytes, 7);
assert_eq!(asset.content_md5, "f1430934c390c118ed2f148e1d44d36c");
assert_eq!(
asset.content_sha256,
"28e51ddac37391b99c2b9053f1122d0bf84b02365e6fd8c6e8667378bd00f436"
);
// Check that the file was written to the stage dir
let dest = stage_dir.path().join("test-results/r.xml");
@ -690,6 +746,7 @@ mod tests {
assert_eq!(summary.files_copied, 0);
assert_eq!(summary.download_errors, 2);
assert_eq!(summary.hash_errors, 0);
}
#[test]
@ -708,9 +765,22 @@ mod tests {
total_bytes: 2048,
files_skipped: 0,
download_errors: 0,
copied_paths: vec![
"test-results/report.xml".to_string(),
"test-results/screenshot.png".to_string(),
hash_errors: 0,
captured_assets: vec![
CapturedAssetInfo {
path: "test-results/report.xml".to_string(),
mime: "text/xml".to_string(),
content_md5: "md5-report".to_string(),
content_sha256: "sha256-report".to_string(),
bytes: 1024,
},
CapturedAssetInfo {
path: "test-results/screenshot.png".to_string(),
mime: "image/png".to_string(),
content_md5: "md5-screenshot".to_string(),
content_sha256: "sha256-screenshot".to_string(),
bytes: 1024,
},
],
})
.unwrap(),
@ -726,7 +796,14 @@ mod tests {
total_bytes: 512,
files_skipped: 0,
download_errors: 0,
copied_paths: vec!["coverage/lcov.info".to_string()],
hash_errors: 0,
captured_assets: vec![CapturedAssetInfo {
path: "coverage/lcov.info".to_string(),
mime: "application/octet-stream".to_string(),
content_md5: "md5-lcov".to_string(),
content_sha256: "sha256-lcov".to_string(),
bytes: 512,
}],
})
.unwrap(),
)

11
lib/crates/fabro-workflows/src/assets.rs
View file

@ -57,17 +57,14 @@ pub fn scan_assets(assets_dir: &Path, node_filter: Option<&str>) -> Result<Vec<A
continue;
};
for relative_path in &summary.copied_paths {
let absolute_path = retry_dir.join(relative_path);
let size = std::fs::metadata(&absolute_path)
.map(|metadata| metadata.len())
.unwrap_or(0);
for asset in &summary.captured_assets {
let absolute_path = retry_dir.join(&asset.path);
entries.push(AssetEntry {
node_slug: node_slug.clone(),
retry,
relative_path: relative_path.clone(),
relative_path: asset.path.clone(),
absolute_path,
size,
size: asset.bytes,
});
}
}

71
lib/crates/fabro-workflows/src/event.rs
View file

@ -250,11 +250,15 @@ pub enum WorkflowRunEvent {
node: String,
idle_seconds: u64,
},
AssetsCaptured {
AssetCaptured {
node_id: String,
files_copied: usize,
total_bytes: u64,
files_skipped: usize,
attempt: u32,
node_slug: String,
path: String,
mime: String,
content_md5: String,
content_sha256: String,
bytes: u64,
},
SshAccessReady {
ssh_command: String,
@ -619,16 +623,15 @@ impl WorkflowRunEvent {
Self::StallWatchdogTimeout { node, idle_seconds } => {
warn!(node, idle_seconds, "Stall watchdog timeout");
}
Self::AssetsCaptured {
Self::AssetCaptured {
node_id,
files_copied,
total_bytes,
files_skipped,
node_slug,
attempt,
path,
bytes,
..
} => {
debug!(
node_id,
files_copied, total_bytes, files_skipped, "Assets captured"
);
debug!(node_id, node_slug, attempt, path, bytes, "Asset captured");
}
Self::SshAccessReady { ssh_command } => {
info!(ssh_command, "SSH access ready");
@ -1164,6 +1167,8 @@ fn rename_fields(event_name: &str, fields: &mut serde_json::Map<String, serde_js
rename(fields, "stage", "node_id");
default_node_label(fields);
rename(fields, "text", "prompt_text");
} else if event_name == "AssetCaptured" {
default_node_label(fields);
} else if event_name.starts_with("Interview") && event_name != "InterviewCompleted" {
// InterviewStarted, InterviewTimeout have `stage`
rename(fields, "stage", "node_id");
@ -1844,6 +1849,27 @@ mod tests {
);
}
#[test]
fn serde_round_trip_asset_captured() {
let event = WorkflowRunEvent::AssetCaptured {
node_id: "work".to_string(),
attempt: 2,
node_slug: "work-visit_3".to_string(),
path: "coverage/lcov.info".to_string(),
mime: "application/octet-stream".to_string(),
content_md5: "abc123".to_string(),
content_sha256: "def456".to_string(),
bytes: 512,
};
let json = serde_json::to_string(&event).unwrap();
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
WorkflowRunEvent::AssetCaptured { node_slug, bytes, .. }
if node_slug == "work-visit_3" && bytes == 512
));
}
#[test]
fn flatten_event_simple_variant() {
let event = WorkflowRunEvent::StageStarted {
@ -2127,6 +2153,27 @@ mod tests {
assert!(!fields.contains_key("text"));
}
#[test]
fn rename_fields_asset_captured() {
let event = WorkflowRunEvent::AssetCaptured {
node_id: "test_stage".to_string(),
attempt: 1,
node_slug: "test_stage".to_string(),
path: "test-results/report.xml".to_string(),
mime: "text/xml".to_string(),
content_md5: "d41d8cd98f00b204e9800998ecf8427e".to_string(),
content_sha256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
.to_string(),
bytes: 1024,
};
let (name, fields) = flatten_event(&event);
assert_eq!(name, "AssetCaptured");
assert_eq!(fields["node_id"], "test_stage");
assert_eq!(fields["node_label"], "test_stage");
assert_eq!(fields["path"], "test-results/report.xml");
assert_eq!(fields["bytes"], 1024);
}
#[test]
fn rename_fields_interview_started() {
let event = WorkflowRunEvent::InterviewStarted {

20
lib/crates/fabro-workflows/src/lifecycle/artifact.rs
View file

@ -97,18 +97,24 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
};
let stage_dir = self
.assets_dir
.join(node_slug)
.join(&node_slug)
.join(format!("retry_{}", ctx.attempt));
let _ = std::fs::create_dir_all(&stage_dir);
match collect_assets(&*self.sandbox, &stage_dir, &self.asset_globs, epoch).await {
Ok(summary) if summary.files_copied > 0 => {
self.emitter.emit(&WorkflowRunEvent::AssetsCaptured {
node_id: node_id.to_string(),
files_copied: summary.files_copied,
total_bytes: summary.total_bytes,
files_skipped: summary.files_skipped,
});
for asset in &summary.captured_assets {
self.emitter.emit(&WorkflowRunEvent::AssetCaptured {
node_id: node_id.to_string(),
attempt: ctx.attempt,
node_slug: node_slug.clone(),
path: asset.path.clone(),
mime: asset.mime.clone(),
content_md5: asset.content_md5.clone(),
content_sha256: asset.content_sha256.clone(),
bytes: asset.bytes,
});
}
}
Ok(_) => {} // no files collected
Err(e) => {

27
lib/crates/fabro-workflows/tests/it/integration.rs
View file

@ -12892,16 +12892,33 @@ async fn asset_collection_local_sandbox_success() {
serde_json::from_str(&std::fs::read_to_string(&manifest_path).unwrap()).unwrap();
assert!(manifest["files_copied"].as_u64().unwrap() >= 1);
// Check that AssetsCaptured event was emitted
// Check that AssetCaptured events were emitted
let captured_events = events.lock().unwrap();
let assets_events: Vec<&WorkflowRunEvent> = captured_events
let asset_events: Vec<&WorkflowRunEvent> = captured_events
.iter()
.filter(|e| matches!(e, WorkflowRunEvent::AssetsCaptured { .. }))
.filter(|e| matches!(e, WorkflowRunEvent::AssetCaptured { .. }))
.collect();
assert!(
!assets_events.is_empty(),
"should emit at least one AssetsCaptured event"
!asset_events.is_empty(),
"should emit at least one AssetCaptured event"
);
if let WorkflowRunEvent::AssetCaptured {
path,
mime,
content_md5,
content_sha256,
bytes,
attempt,
..
} = asset_events[0]
{
assert!(!path.is_empty());
assert!(!mime.is_empty());
assert_eq!(content_md5.len(), 32);
assert_eq!(content_sha256.len(), 64);
assert!(*bytes > 0);
assert_eq!(*attempt, 1);
}
}
/// Local sandbox: assets are still collected even when the handler fails.