From b6947af7afa314f86cf703ab7faf5345a4d5dfe1 Mon Sep 17 00:00:00 2001
From: Bryan Helmkamp <bryan@brynary.com>
Date: Sun, 29 Mar 2026 22:10:09 -0400
Subject: [PATCH] Add per-asset metadata (mime, md5, sha256) to asset capture

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>
---
 Cargo.lock                                    |   2 +
 Cargo.toml                                    |   1 +
 docs/agents/outputs.mdx                       |  13 ++-
 lib/crates/fabro-cli/tests/it/scenario.rs     |   2 +-
 lib/crates/fabro-workflows/Cargo.toml         |   2 +
 .../fabro-workflows/src/asset_snapshot.rs     | 109 +++++++++++++++---
 lib/crates/fabro-workflows/src/assets.rs      |  11 +-
 lib/crates/fabro-workflows/src/event.rs       |  71 ++++++++++--
 .../fabro-workflows/src/lifecycle/artifact.rs |  20 ++--
 .../fabro-workflows/tests/it/integration.rs   |  27 ++++-
 10 files changed, 206 insertions(+), 52 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index bfe841776..abc0643d0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2013,6 +2013,8 @@ dependencies = [
  "futures",
  "git2",
  "hex",
+ "md5",
+ "mime_guess",
  "predicates",
  "rand 0.8.5",
  "regex",
diff --git a/Cargo.toml b/Cargo.toml
index 54b15a95b..582bb8e29 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -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"
diff --git a/docs/agents/outputs.mdx b/docs/agents/outputs.mdx
index a4dd0ace1..2cbb143f1 100644
--- a/docs/agents/outputs.mdx
+++ b/docs/agents/outputs.mdx
@@ -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
+    }
   ]
 }
 ```
diff --git a/lib/crates/fabro-cli/tests/it/scenario.rs b/lib/crates/fabro-cli/tests/it/scenario.rs
index 60f144182..3dfc89c5f 100644
--- a/lib/crates/fabro-cli/tests/it/scenario.rs
+++ b/lib/crates/fabro-cli/tests/it/scenario.rs
@@ -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();
 
diff --git a/lib/crates/fabro-workflows/Cargo.toml b/lib/crates/fabro-workflows/Cargo.toml
index e357505da..300f36280 100644
--- a/lib/crates/fabro-workflows/Cargo.toml
+++ b/lib/crates/fabro-workflows/Cargo.toml
@@ -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
diff --git a/lib/crates/fabro-workflows/src/asset_snapshot.rs b/lib/crates/fabro-workflows/src/asset_snapshot.rs
index a4ba2c45b..e462659e1 100644
--- a/lib/crates/fabro-workflows/src/asset_snapshot.rs
+++ b/lib/crates/fabro-workflows/src/asset_snapshot.rs
@@ -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(),
         )
diff --git a/lib/crates/fabro-workflows/src/assets.rs b/lib/crates/fabro-workflows/src/assets.rs
index 4d8e1a722..7e4c9608b 100644
--- a/lib/crates/fabro-workflows/src/assets.rs
+++ b/lib/crates/fabro-workflows/src/assets.rs
@@ -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,
                 });
             }
         }
diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs
index 1e190dcb8..bc8430ef7 100644
--- a/lib/crates/fabro-workflows/src/event.rs
+++ b/lib/crates/fabro-workflows/src/event.rs
@@ -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 {
diff --git a/lib/crates/fabro-workflows/src/lifecycle/artifact.rs b/lib/crates/fabro-workflows/src/lifecycle/artifact.rs
index f816786c4..5c1f18dcc 100644
--- a/lib/crates/fabro-workflows/src/lifecycle/artifact.rs
+++ b/lib/crates/fabro-workflows/src/lifecycle/artifact.rs
@@ -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) => {
diff --git a/lib/crates/fabro-workflows/tests/it/integration.rs b/lib/crates/fabro-workflows/tests/it/integration.rs
index a6f65dded..326d1e67d 100644
--- a/lib/crates/fabro-workflows/tests/it/integration.rs
+++ b/lib/crates/fabro-workflows/tests/it/integration.rs
@@ -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.