diff --git a/crates/arc-api/src/server.rs b/crates/arc-api/src/server.rs
index f6fd41b84..636e99295 100644
--- a/crates/arc-api/src/server.rs
+++ b/crates/arc-api/src/server.rs
@@ -248,6 +248,7 @@ async fn start_pipeline(
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: std::collections::HashMap::new(),
         };
 
         let result = tokio::select! {
diff --git a/crates/arc-cli/src/main.rs b/crates/arc-cli/src/main.rs
index 3ca9a2861..8107db9f1 100644
--- a/crates/arc-cli/src/main.rs
+++ b/crates/arc-cli/src/main.rs
@@ -28,6 +28,8 @@ enum Command {
     Run(arc_workflows::cli::RunArgs),
     /// Validate a pipeline
     Validate(arc_workflows::cli::ValidateArgs),
+    /// List and manage pipeline runs
+    Runs(arc_workflows::cli::runs::RunsArgs),
     /// Start the HTTP API server
     Serve(arc_api::serve::ServeArgs),
 }
@@ -59,6 +61,7 @@ async fn main() -> Result<()> {
         Command::Agent(_) => "agent",
         Command::Run(_) => "run",
         Command::Validate(_) => "validate",
+        Command::Runs(_) => "runs",
         Command::Serve(_) => "serve",
     };
     debug!(command = %command_name, "CLI command started");
@@ -78,6 +81,9 @@ async fn main() -> Result<()> {
             let styles = arc_util::terminal::Styles::detect_stderr();
             arc_workflows::cli::validate::validate_command(&args, &styles)?;
         }
+        Command::Runs(args) => {
+            arc_workflows::cli::runs::runs_command(args)?;
+        }
         Command::Serve(args) => {
             let styles: &'static arc_util::terminal::Styles =
                 Box::leak(Box::new(arc_util::terminal::Styles::detect_stderr()));
diff --git a/crates/arc-workflows/src/cli/mod.rs b/crates/arc-workflows/src/cli/mod.rs
index ee3ba80a3..599bb0934 100644
--- a/crates/arc-workflows/src/cli/mod.rs
+++ b/crates/arc-workflows/src/cli/mod.rs
@@ -1,6 +1,7 @@
 pub mod backend;
 pub mod cli_backend;
 pub mod run;
+pub mod runs;
 pub mod task_config;
 pub mod validate;
 
@@ -112,6 +113,10 @@ pub struct RunArgs {
     /// Execution environment for agent tools
     #[arg(long, value_enum)]
     pub execution_env: Option<ExecutionEnvKind>,
+
+    /// Attach a label to this run (repeatable, format: KEY=VALUE)
+    #[arg(long = "label", value_name = "KEY=VALUE")]
+    pub label: Vec<String>,
 }
 
 #[derive(Args)]
diff --git a/crates/arc-workflows/src/cli/run.rs b/crates/arc-workflows/src/cli/run.rs
index 8721833ec..783a94cc7 100644
--- a/crates/arc-workflows/src/cli/run.rs
+++ b/crates/arc-workflows/src/cli/run.rs
@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
@@ -561,6 +562,12 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu
         base_sha: worktree_base_sha.or(daytona_base_sha),
         run_branch: worktree_branch.or(daytona_branch),
         meta_branch,
+        labels: args
+            .label
+            .iter()
+            .filter_map(|s| s.split_once('='))
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect(),
     };
 
     let run_start = Instant::now();
@@ -907,6 +914,7 @@ async fn run_from_branch(
         base_sha,
         run_branch: Some(run_branch.to_string()),
         meta_branch,
+        labels: HashMap::new(),
     };
 
     let run_start = Instant::now();
diff --git a/crates/arc-workflows/src/cli/runs.rs b/crates/arc-workflows/src/cli/runs.rs
new file mode 100644
index 000000000..74054ddd5
--- /dev/null
+++ b/crates/arc-workflows/src/cli/runs.rs
@@ -0,0 +1,652 @@
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+
+use anyhow::Result;
+use clap::{Args, Subcommand};
+use serde::Serialize;
+use tracing::{debug, info};
+
+/// Arguments for the `arc runs` command.
+#[derive(Args)]
+pub struct RunsArgs {
+    #[command(subcommand)]
+    pub command: Option<RunsCommand>,
+}
+
+#[derive(Subcommand)]
+pub enum RunsCommand {
+    /// List pipeline runs
+    List(RunsListArgs),
+    /// Delete old pipeline runs
+    Prune(RunsPruneArgs),
+}
+
+#[derive(Args)]
+pub struct RunsListArgs {
+    /// Only show runs started before this date (YYYY-MM-DD prefix match)
+    #[arg(long)]
+    pub before: Option<String>,
+
+    /// Filter by pipeline name (substring match)
+    #[arg(long)]
+    pub pipeline: Option<String>,
+
+    /// Filter by label (KEY=VALUE, repeatable, AND semantics)
+    #[arg(long = "label", value_name = "KEY=VALUE")]
+    pub label: Vec<String>,
+
+    /// Include orphan directories (no manifest.json)
+    #[arg(long)]
+    pub orphans: bool,
+
+    /// Output as JSON
+    #[arg(long)]
+    pub json: bool,
+}
+
+#[derive(Args)]
+pub struct RunsPruneArgs {
+    /// Only prune runs started before this date (YYYY-MM-DD prefix match)
+    #[arg(long)]
+    pub before: Option<String>,
+
+    /// Filter by pipeline name (substring match)
+    #[arg(long)]
+    pub pipeline: Option<String>,
+
+    /// Filter by label (KEY=VALUE, repeatable, AND semantics)
+    #[arg(long = "label", value_name = "KEY=VALUE")]
+    pub label: Vec<String>,
+
+    /// Include orphan directories (no manifest.json)
+    #[arg(long)]
+    pub orphans: bool,
+
+    /// Actually delete (default is dry-run)
+    #[arg(long)]
+    pub yes: bool,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct RunInfo {
+    pub run_id: String,
+    pub dir_name: String,
+    pub pipeline_name: String,
+    pub status: String,
+    pub start_time: String,
+    pub labels: HashMap<String, String>,
+    #[serde(skip)]
+    pub path: PathBuf,
+    #[serde(skip)]
+    pub is_orphan: bool,
+}
+
+/// Scan a logs base directory and return info about each run.
+pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
+    let entries = match std::fs::read_dir(base) {
+        Ok(e) => e,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(e) => return Err(e.into()),
+    };
+
+    let mut runs = Vec::new();
+    for entry in entries {
+        let entry = entry?;
+        let path = entry.path();
+        if !path.is_dir() {
+            continue;
+        }
+
+        let dir_name = entry
+            .file_name()
+            .to_string_lossy()
+            .to_string();
+
+        debug!(dir = %dir_name, "scanning run directory");
+
+        let manifest_path = path.join("manifest.json");
+        if manifest_path.exists() {
+            let manifest_text = std::fs::read_to_string(&manifest_path)?;
+            debug!(dir = %dir_name, "reading manifest");
+            let manifest: serde_json::Value = serde_json::from_str(&manifest_text)?;
+
+            let run_id = manifest["run_id"]
+                .as_str()
+                .unwrap_or(&dir_name)
+                .to_string();
+            let pipeline_name = manifest["pipeline_name"]
+                .as_str()
+                .unwrap_or("unknown")
+                .to_string();
+            let start_time = manifest["start_time"]
+                .as_str()
+                .unwrap_or("")
+                .to_string();
+            let labels: HashMap<String, String> = manifest
+                .get("labels")
+                .and_then(|v| serde_json::from_value(v.clone()).ok())
+                .unwrap_or_default();
+
+            let status = read_status(&path);
+
+            runs.push(RunInfo {
+                run_id,
+                dir_name,
+                pipeline_name,
+                status,
+                start_time,
+                labels,
+                path,
+                is_orphan: false,
+            });
+        } else {
+            // Orphan directory — no manifest.json
+            let mtime = entry
+                .metadata()
+                .ok()
+                .and_then(|m| m.modified().ok())
+                .map(|t| {
+                    let dt: chrono::DateTime<chrono::Utc> = t.into();
+                    dt.to_rfc3339()
+                })
+                .unwrap_or_default();
+
+            runs.push(RunInfo {
+                run_id: dir_name.clone(),
+                dir_name,
+                pipeline_name: "[no manifest]".to_string(),
+                status: "unknown".to_string(),
+                start_time: mtime,
+                labels: HashMap::new(),
+                path,
+                is_orphan: true,
+            });
+        }
+    }
+
+    // Sort by start_time descending (newest first)
+    runs.sort_by(|a, b| b.start_time.cmp(&a.start_time));
+    Ok(runs)
+}
+
+fn read_status(run_dir: &Path) -> String {
+    let final_path = run_dir.join("final.json");
+    if final_path.exists() {
+        if let Ok(text) = std::fs::read_to_string(&final_path) {
+            if let Ok(val) = serde_json::from_str::<serde_json::Value>(&text) {
+                if let Some(status) = val["status"].as_str() {
+                    return status.to_string();
+                }
+            }
+        }
+        "unknown".to_string()
+    } else if run_dir.join("run.pid").exists() {
+        "running".to_string()
+    } else {
+        "unknown".to_string()
+    }
+}
+
+/// Filter runs by criteria. Orphans are excluded unless `include_orphans` is true.
+pub fn filter_runs(
+    runs: &[RunInfo],
+    before: Option<&str>,
+    pipeline: Option<&str>,
+    labels: &[(String, String)],
+    include_orphans: bool,
+) -> Vec<RunInfo> {
+    runs.iter()
+        .filter(|r| {
+            if r.is_orphan && !include_orphans {
+                return false;
+            }
+            if let Some(before) = before {
+                if !r.start_time.is_empty() && r.start_time.as_str() >= before {
+                    return false;
+                }
+            }
+            if let Some(pat) = pipeline {
+                if !r.pipeline_name.contains(pat) {
+                    return false;
+                }
+            }
+            for (k, v) in labels {
+                match r.labels.get(k) {
+                    Some(val) if val == v => {}
+                    _ => return false,
+                }
+            }
+            true
+        })
+        .cloned()
+        .collect()
+}
+
+fn parse_label_filters(label_args: &[String]) -> Vec<(String, String)> {
+    label_args
+        .iter()
+        .filter_map(|s| s.split_once('='))
+        .map(|(k, v)| (k.to_string(), v.to_string()))
+        .collect()
+}
+
+fn default_logs_base() -> PathBuf {
+    dirs::home_dir()
+        .expect("could not determine home directory")
+        .join(".arc")
+        .join("logs")
+}
+
+pub fn list_command(args: &RunsListArgs) -> Result<()> {
+    let base = default_logs_base();
+    let runs = scan_runs(&base)?;
+    let label_filters = parse_label_filters(&args.label);
+    let filtered = filter_runs(
+        &runs,
+        args.before.as_deref(),
+        args.pipeline.as_deref(),
+        &label_filters,
+        args.orphans,
+    );
+
+    if args.json {
+        println!("{}", serde_json::to_string_pretty(&filtered)?);
+        return Ok(());
+    }
+
+    if filtered.is_empty() {
+        eprintln!("No runs found.");
+        return Ok(());
+    }
+
+    // Print table header
+    let header = format!(
+        "{:<30} {:<25} {:<10} {:<25} LABELS",
+        "RUN ID", "PIPELINE", "STATUS", "STARTED"
+    );
+    println!("{header}");
+    println!("{}", "-".repeat(100));
+
+    for run in &filtered {
+        let labels_str = run
+            .labels
+            .iter()
+            .map(|(k, v)| format!("{k}={v}"))
+            .collect::<Vec<_>>()
+            .join(", ");
+        let run_id_display = if run.run_id.len() > 28 {
+            format!("{}...", &run.run_id[..25])
+        } else {
+            run.run_id.clone()
+        };
+        let start_display = if run.start_time.len() > 23 {
+            run.start_time[..23].to_string()
+        } else {
+            run.start_time.clone()
+        };
+        println!(
+            "{:<30} {:<25} {:<10} {:<25} {}",
+            run_id_display, run.pipeline_name, run.status, start_display, labels_str
+        );
+    }
+    eprintln!("\n{} run(s) listed.", filtered.len());
+    Ok(())
+}
+
+pub fn prune_command(args: &RunsPruneArgs) -> Result<()> {
+    let base = default_logs_base();
+    prune_from(args, &base)
+}
+
+pub fn prune_from(args: &RunsPruneArgs, base: &Path) -> Result<()> {
+    let runs = scan_runs(base)?;
+    let label_filters = parse_label_filters(&args.label);
+    let filtered = filter_runs(
+        &runs,
+        args.before.as_deref(),
+        args.pipeline.as_deref(),
+        &label_filters,
+        args.orphans,
+    );
+
+    if filtered.is_empty() {
+        eprintln!("No matching runs to prune.");
+        return Ok(());
+    }
+
+    if args.yes {
+        for run in &filtered {
+            info!(run_id = %run.run_id, path = %run.path.display(), "deleting run");
+            std::fs::remove_dir_all(&run.path)?;
+        }
+        eprintln!("{} run(s) deleted.", filtered.len());
+    } else {
+        for run in &filtered {
+            debug!(run_id = %run.run_id, "would delete run (dry-run)");
+            println!(
+                "would delete: {} ({})",
+                run.dir_name, run.pipeline_name
+            );
+        }
+        eprintln!(
+            "\n{} run(s) would be deleted. Pass --yes to confirm.",
+            filtered.len()
+        );
+    }
+    Ok(())
+}
+
+pub fn runs_command(args: RunsArgs) -> Result<()> {
+    match args.command {
+        None => list_command(&RunsListArgs {
+            before: None,
+            pipeline: None,
+            label: Vec::new(),
+            orphans: false,
+            json: false,
+        }),
+        Some(RunsCommand::List(args)) => list_command(&args),
+        Some(RunsCommand::Prune(args)) => prune_command(&args),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs;
+
+    fn make_run_dir(
+        base: &Path,
+        dir_name: &str,
+        manifest: Option<serde_json::Value>,
+        final_json: Option<serde_json::Value>,
+        pid_file: bool,
+    ) -> PathBuf {
+        let dir = base.join(dir_name);
+        fs::create_dir_all(&dir).unwrap();
+        if let Some(m) = manifest {
+            fs::write(dir.join("manifest.json"), serde_json::to_string_pretty(&m).unwrap())
+                .unwrap();
+        }
+        if let Some(f) = final_json {
+            fs::write(dir.join("final.json"), serde_json::to_string_pretty(&f).unwrap())
+                .unwrap();
+        }
+        if pid_file {
+            fs::write(dir.join("run.pid"), "12345").unwrap();
+        }
+        dir
+    }
+
+    #[test]
+    fn scan_runs_reads_manifest_and_final() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base = tmp.path();
+
+        make_run_dir(
+            base,
+            "arc-run-20260101-120000",
+            Some(serde_json::json!({
+                "run_id": "abc123",
+                "pipeline_name": "my-pipeline",
+                "start_time": "2026-01-01T12:00:00Z",
+                "labels": { "env": "prod" }
+            })),
+            Some(serde_json::json!({ "status": "success" })),
+            false,
+        );
+
+        make_run_dir(base, "arc-run-orphan", None, None, false);
+
+        let runs = scan_runs(base).unwrap();
+        assert_eq!(runs.len(), 2);
+
+        let completed = runs.iter().find(|r| r.run_id == "abc123").unwrap();
+        assert_eq!(completed.pipeline_name, "my-pipeline");
+        assert_eq!(completed.status, "success");
+        assert_eq!(completed.labels.get("env").unwrap(), "prod");
+        assert!(!completed.is_orphan);
+
+        let orphan = runs.iter().find(|r| r.is_orphan).unwrap();
+        assert_eq!(orphan.pipeline_name, "[no manifest]");
+        assert_eq!(orphan.status, "unknown");
+    }
+
+    #[test]
+    fn scan_runs_detects_running_status() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base = tmp.path();
+
+        make_run_dir(
+            base,
+            "arc-run-running",
+            Some(serde_json::json!({
+                "run_id": "running-1",
+                "pipeline_name": "pipeline-a",
+                "start_time": "2026-01-15T10:00:00Z"
+            })),
+            None,
+            true,
+        );
+
+        let runs = scan_runs(base).unwrap();
+        assert_eq!(runs.len(), 1);
+        assert_eq!(runs[0].status, "running");
+    }
+
+    #[test]
+    fn scan_runs_empty_dir() {
+        let tmp = tempfile::tempdir().unwrap();
+        let runs = scan_runs(tmp.path()).unwrap();
+        assert!(runs.is_empty());
+    }
+
+    #[test]
+    fn scan_runs_missing_dir() {
+        let runs = scan_runs(Path::new("/tmp/nonexistent-arc-test-dir")).unwrap();
+        assert!(runs.is_empty());
+    }
+
+    #[test]
+    fn filter_runs_before() {
+        let runs = vec![
+            RunInfo {
+                run_id: "old".into(),
+                dir_name: "d1".into(),
+                pipeline_name: "p".into(),
+                status: "success".into(),
+                start_time: "2025-06-01T00:00:00Z".into(),
+                labels: HashMap::new(),
+                path: PathBuf::from("/tmp/d1"),
+                is_orphan: false,
+            },
+            RunInfo {
+                run_id: "new".into(),
+                dir_name: "d2".into(),
+                pipeline_name: "p".into(),
+                status: "success".into(),
+                start_time: "2026-03-01T00:00:00Z".into(),
+                labels: HashMap::new(),
+                path: PathBuf::from("/tmp/d2"),
+                is_orphan: false,
+            },
+        ];
+        let filtered = filter_runs(&runs, Some("2026-01-01"), None, &[], false);
+        assert_eq!(filtered.len(), 1);
+        assert_eq!(filtered[0].run_id, "old");
+    }
+
+    #[test]
+    fn filter_runs_pipeline() {
+        let runs = vec![
+            RunInfo {
+                run_id: "a".into(),
+                dir_name: "d1".into(),
+                pipeline_name: "deploy-prod".into(),
+                status: "success".into(),
+                start_time: "2026-01-01T00:00:00Z".into(),
+                labels: HashMap::new(),
+                path: PathBuf::from("/tmp/d1"),
+                is_orphan: false,
+            },
+            RunInfo {
+                run_id: "b".into(),
+                dir_name: "d2".into(),
+                pipeline_name: "test-suite".into(),
+                status: "success".into(),
+                start_time: "2026-01-01T00:00:00Z".into(),
+                labels: HashMap::new(),
+                path: PathBuf::from("/tmp/d2"),
+                is_orphan: false,
+            },
+        ];
+        let filtered = filter_runs(&runs, None, Some("deploy"), &[], false);
+        assert_eq!(filtered.len(), 1);
+        assert_eq!(filtered[0].run_id, "a");
+    }
+
+    #[test]
+    fn filter_runs_labels() {
+        let runs = vec![
+            RunInfo {
+                run_id: "a".into(),
+                dir_name: "d1".into(),
+                pipeline_name: "p".into(),
+                status: "success".into(),
+                start_time: "2026-01-01T00:00:00Z".into(),
+                labels: HashMap::from([("env".into(), "prod".into())]),
+                path: PathBuf::from("/tmp/d1"),
+                is_orphan: false,
+            },
+            RunInfo {
+                run_id: "b".into(),
+                dir_name: "d2".into(),
+                pipeline_name: "p".into(),
+                status: "success".into(),
+                start_time: "2026-01-01T00:00:00Z".into(),
+                labels: HashMap::from([("env".into(), "staging".into())]),
+                path: PathBuf::from("/tmp/d2"),
+                is_orphan: false,
+            },
+        ];
+        let filtered = filter_runs(
+            &runs,
+            None,
+            None,
+            &[("env".to_string(), "prod".to_string())],
+            false,
+        );
+        assert_eq!(filtered.len(), 1);
+        assert_eq!(filtered[0].run_id, "a");
+    }
+
+    #[test]
+    fn filter_runs_orphans_excluded_by_default() {
+        let runs = vec![RunInfo {
+            run_id: "orphan".into(),
+            dir_name: "d1".into(),
+            pipeline_name: "[no manifest]".into(),
+            status: "unknown".into(),
+            start_time: "".into(),
+            labels: HashMap::new(),
+            path: PathBuf::from("/tmp/d1"),
+            is_orphan: true,
+        }];
+        let filtered = filter_runs(&runs, None, None, &[], false);
+        assert!(filtered.is_empty());
+
+        let filtered = filter_runs(&runs, None, None, &[], true);
+        assert_eq!(filtered.len(), 1);
+    }
+
+    #[test]
+    fn prune_dry_run_preserves_dirs() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base = tmp.path();
+
+        let dir = make_run_dir(
+            base,
+            "arc-run-20250101-120000",
+            Some(serde_json::json!({
+                "run_id": "to-prune",
+                "pipeline_name": "old-pipeline",
+                "start_time": "2025-01-01T12:00:00Z"
+            })),
+            Some(serde_json::json!({ "status": "success" })),
+            false,
+        );
+
+        let args = RunsPruneArgs {
+            before: Some("2026-01-01".into()),
+            pipeline: None,
+            label: Vec::new(),
+            orphans: false,
+            yes: false,
+        };
+
+        prune_from(&args, base).unwrap();
+        assert!(dir.exists(), "dry-run should preserve directory");
+    }
+
+    #[test]
+    fn prune_with_yes_deletes_dirs() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base = tmp.path();
+
+        let dir = make_run_dir(
+            base,
+            "arc-run-20250101-120000",
+            Some(serde_json::json!({
+                "run_id": "to-prune",
+                "pipeline_name": "old-pipeline",
+                "start_time": "2025-01-01T12:00:00Z"
+            })),
+            Some(serde_json::json!({ "status": "success" })),
+            false,
+        );
+
+        // Also add a run that should NOT be pruned (too new)
+        let keep_dir = make_run_dir(
+            base,
+            "arc-run-20260301-120000",
+            Some(serde_json::json!({
+                "run_id": "keep-this",
+                "pipeline_name": "new-pipeline",
+                "start_time": "2026-03-01T12:00:00Z"
+            })),
+            Some(serde_json::json!({ "status": "success" })),
+            false,
+        );
+
+        let args = RunsPruneArgs {
+            before: Some("2026-01-01".into()),
+            pipeline: None,
+            label: Vec::new(),
+            orphans: false,
+            yes: true,
+        };
+
+        prune_from(&args, base).unwrap();
+        assert!(!dir.exists(), "--yes should delete matching directory");
+        assert!(keep_dir.exists(), "non-matching directory should be preserved");
+    }
+
+    #[test]
+    fn prune_orphans_with_yes() {
+        let tmp = tempfile::tempdir().unwrap();
+        let base = tmp.path();
+
+        let orphan_dir = make_run_dir(base, "orphan-dir", None, None, false);
+
+        let args = RunsPruneArgs {
+            before: None,
+            pipeline: None,
+            label: Vec::new(),
+            orphans: true,
+            yes: true,
+        };
+
+        prune_from(&args, base).unwrap();
+        assert!(!orphan_dir.exists(), "orphan directory should be deleted");
+    }
+}
diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs
index 2e4104870..e29abaa48 100644
--- a/crates/arc-workflows/src/engine.rs
+++ b/crates/arc-workflows/src/engine.rs
@@ -304,6 +304,9 @@ fn write_manifest(logs_root: &Path, graph: &Graph, config: &RunConfig) -> serde_
     if let Some(ref base) = config.base_sha {
         manifest["base_sha"] = serde_json::Value::String(base.clone());
     }
+    if !config.labels.is_empty() {
+        manifest["labels"] = serde_json::to_value(&config.labels).unwrap_or_default();
+    }
     if let Ok(json) = serde_json::to_string_pretty(&manifest) {
         let _ = std::fs::create_dir_all(logs_root);
         let _ = std::fs::write(logs_root.join("manifest.json"), json);
@@ -736,6 +739,8 @@ pub struct RunConfig {
     pub run_branch: Option<String>,
     /// Metadata branch name for git-native checkpoint storage (e.g. `refs/arc/{run_id}`).
     pub meta_branch: Option<String>,
+    /// User-defined key-value labels for this run.
+    pub labels: HashMap<String, String>,
 }
 
 /// The pipeline execution engine.
@@ -2306,6 +2311,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -2326,6 +2332,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
         let checkpoint_path = dir.path().join("checkpoint.json");
@@ -2354,6 +2361,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2378,6 +2386,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -2398,6 +2407,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2431,6 +2441,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -2488,6 +2499,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2565,6 +2577,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2579,6 +2592,56 @@ mod tests {
         assert!(manifest["edge_count"].is_number());
     }
 
+    #[tokio::test]
+    async fn manifest_includes_labels_when_present() {
+        let dir = tempfile::tempdir().unwrap();
+        let g = simple_graph();
+        let engine =
+            PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
+        let config = RunConfig {
+            logs_root: dir.path().to_path_buf(),
+            cancel_token: None,
+            dry_run: false,
+            run_id: "labels-run".into(),
+            git_checkpoint: None,
+            base_sha: None,
+            run_branch: None,
+            meta_branch: None,
+            labels: HashMap::from([("env".into(), "test".into())]),
+        };
+        engine.run(&g, &config).await.unwrap();
+
+        let manifest: serde_json::Value =
+            serde_json::from_str(&std::fs::read_to_string(dir.path().join("manifest.json")).unwrap())
+                .unwrap();
+        assert_eq!(manifest["labels"]["env"], "test");
+    }
+
+    #[tokio::test]
+    async fn manifest_omits_labels_when_empty() {
+        let dir = tempfile::tempdir().unwrap();
+        let g = simple_graph();
+        let engine =
+            PipelineEngine::new(make_registry(), Arc::new(EventEmitter::new()), local_env());
+        let config = RunConfig {
+            logs_root: dir.path().to_path_buf(),
+            cancel_token: None,
+            dry_run: false,
+            run_id: "no-labels-run".into(),
+            git_checkpoint: None,
+            base_sha: None,
+            run_branch: None,
+            meta_branch: None,
+            labels: HashMap::new(),
+        };
+        engine.run(&g, &config).await.unwrap();
+
+        let manifest: serde_json::Value =
+            serde_json::from_str(&std::fs::read_to_string(dir.path().join("manifest.json")).unwrap())
+                .unwrap();
+        assert!(manifest.get("labels").is_none());
+    }
+
     #[tokio::test]
     async fn engine_writes_node_status_json() {
         let dir = tempfile::tempdir().unwrap();
@@ -2594,6 +2657,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2620,6 +2684,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2774,6 +2839,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2812,6 +2878,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
 
@@ -2868,6 +2935,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
 
@@ -2926,6 +2994,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
 
@@ -2988,6 +3057,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_ok());
@@ -3039,6 +3109,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -3091,6 +3162,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
 
@@ -3150,6 +3222,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
         // Give spawned inform tasks time to complete
@@ -3184,6 +3257,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         engine.run(&g, &config).await.unwrap();
         // Give spawned inform tasks time to complete
@@ -3219,6 +3293,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -3242,6 +3317,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3264,6 +3340,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -3299,6 +3376,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
 
         // Set cancel after a short delay (while the slow handler is running)
@@ -3371,6 +3449,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3396,6 +3475,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3423,6 +3503,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3511,6 +3592,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
 
         // The engine returns Err because the Fail outcome has no outgoing fail edge,
@@ -3735,6 +3817,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3765,6 +3848,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3802,6 +3886,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3879,6 +3964,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -3969,6 +4055,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let result = engine.run(&g, &config).await;
         assert!(result.is_err());
@@ -4034,6 +4121,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -4088,6 +4176,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let outcome = engine.run(&g, &config).await.unwrap();
         assert_eq!(outcome.status, StageStatus::Success);
@@ -4141,6 +4230,7 @@ mod tests {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: HashMap::new(),
         };
         let _outcome = engine.run(&g, &config).await.unwrap();
 
diff --git a/crates/arc-workflows/tests/daytona_integration.rs b/crates/arc-workflows/tests/daytona_integration.rs
index 366815293..d4e7d6b33 100644
--- a/crates/arc-workflows/tests/daytona_integration.rs
+++ b/crates/arc-workflows/tests/daytona_integration.rs
@@ -284,6 +284,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -474,6 +475,7 @@ async fn daytona_git_checkpoint_remote_emits_events() {
         base_sha: Some(base_sha),
         run_branch: Some(branch_name),
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -653,6 +655,7 @@ async fn daytona_parallel_git_branching_e2e() {
         base_sha: Some(base_sha),
         run_branch: Some(branch_name),
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -977,6 +980,7 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
         base_sha: Some(base_sha),
         run_branch: Some(branch_name),
         meta_branch: Some(meta_branch),
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs
index d4d57845b..d88305cf0 100644
--- a/crates/arc-workflows/tests/integration.rs
+++ b/crates/arc-workflows/tests/integration.rs
@@ -198,6 +198,7 @@ async fn end_to_end_linear_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -334,6 +335,7 @@ async fn end_to_end_branching_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -453,6 +455,7 @@ async fn end_to_end_human_gate_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -558,6 +561,7 @@ async fn goal_gate_routes_to_retry_target_on_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -676,6 +680,7 @@ async fn goal_gate_routes_to_retry_target_when_present() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -981,6 +986,7 @@ async fn retry_on_failure_then_succeed() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -1053,6 +1059,7 @@ async fn pipeline_with_many_nodes() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -1379,6 +1386,7 @@ async fn smoke_test_with_mock_codergen_backend() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -1477,6 +1485,7 @@ async fn end_to_end_parallel_fan_out_fan_in() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -1586,6 +1595,7 @@ async fn resume_from_checkpoint_completes_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -1681,6 +1691,7 @@ async fn resume_from_checkpoint_preserves_goal_gate_outcomes() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     // This should succeed because goal gate for gated_work is satisfied
@@ -1721,6 +1732,7 @@ async fn graph_goal_in_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -1753,6 +1765,7 @@ async fn event_streaming_lifecycle() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -1829,6 +1842,7 @@ async fn context_flow_between_stages() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -1878,6 +1892,7 @@ async fn tool_handler_e2e() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -1946,6 +1961,7 @@ async fn auto_approve_interviewer_e2e() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -1979,6 +1995,7 @@ async fn codergen_without_backend_simulated() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -2080,6 +2097,7 @@ async fn branching_loop_back_on_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2161,6 +2179,7 @@ async fn human_gate_loops_back() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2215,6 +2234,7 @@ async fn scenario_ship_a_feature() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2297,6 +2317,7 @@ async fn scenario_parallel_expert_review() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2373,6 +2394,7 @@ async fn scenario_node_retries_on_retry_status() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2431,6 +2453,7 @@ async fn scenario_loop_restart_resets_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2495,6 +2518,7 @@ async fn scenario_bug_triage_router() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2549,6 +2573,7 @@ async fn scenario_crash_recovery() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine
         .run_from_checkpoint(&graph, &config, &checkpoint)
@@ -2633,6 +2658,7 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
 
@@ -2685,6 +2711,7 @@ async fn manager_loop_max_cycles_exceeded_e2e() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
 
@@ -2818,6 +2845,7 @@ async fn conditional_branching_success_fail_paths() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -2867,6 +2895,7 @@ async fn edge_selection_condition_match_wins_over_weight() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -2910,6 +2939,7 @@ async fn edge_selection_weight_breaks_ties() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -2945,6 +2975,7 @@ async fn edge_selection_lexical_tiebreak() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -2999,6 +3030,7 @@ async fn context_updates_visible_across_nodes() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3039,6 +3071,7 @@ async fn stylesheet_applies_model_override() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -3091,6 +3124,7 @@ async fn custom_handler_registration_and_execution() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3158,6 +3192,7 @@ async fn integration_smoke_plan_implement_review_done() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     let outcome = engine.run(&graph, &config).await.expect("run");
     assert_eq!(outcome.status, StageStatus::Success);
@@ -3230,6 +3265,7 @@ async fn sub_pipeline_e2e_through_engine() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -3369,6 +3405,7 @@ async fn manager_loop_with_child_observer_e2e() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -3502,6 +3539,7 @@ async fn graph_merge_e2e_through_engine() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -3649,6 +3687,7 @@ async fn fidelity_default_is_compact() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3702,6 +3741,7 @@ async fn fidelity_graph_default_applied() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3751,6 +3791,7 @@ async fn fidelity_node_overrides_graph_default() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3806,6 +3847,7 @@ async fn fidelity_edge_overrides_node_and_graph() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3851,6 +3893,7 @@ async fn fidelity_full_produces_empty_preamble() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3906,6 +3949,7 @@ async fn fidelity_truncate_preamble_minimal() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -3974,6 +4018,7 @@ async fn fidelity_summary_low_mode() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4037,6 +4082,7 @@ async fn fidelity_summary_medium_mode() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4100,6 +4146,7 @@ async fn fidelity_summary_high_mode() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4156,6 +4203,7 @@ async fn fidelity_full_sets_thread_id_in_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4223,6 +4271,7 @@ async fn fidelity_full_nodes_share_thread_id() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4299,6 +4348,7 @@ async fn fidelity_resume_degrades_full_to_summary_high() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine
         .run_from_checkpoint(&graph, &config, &checkpoint)
@@ -4391,6 +4441,7 @@ async fn fidelity_resume_degrade_only_affects_first_hop() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine
         .run_from_checkpoint(&graph, &config, &checkpoint)
@@ -4470,6 +4521,7 @@ async fn fidelity_resume_no_degrade_when_not_full() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine
         .run_from_checkpoint(&graph, &config, &checkpoint)
@@ -4508,6 +4560,7 @@ async fn fidelity_stored_in_checkpoint_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4590,6 +4643,7 @@ async fn fidelity_precedence_multi_node_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4654,6 +4708,7 @@ async fn fidelity_compact_preamble_includes_completed_stages_and_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4725,6 +4780,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine_low
         .run(&graph_low, &config_low)
@@ -4788,6 +4844,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine_med
         .run(&graph_med, &config_med)
@@ -4855,6 +4912,7 @@ async fn fidelity_thread_id_fallback_to_previous_node_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4905,6 +4963,7 @@ async fn fidelity_thread_id_from_node_class_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -4958,6 +5017,7 @@ async fn fidelity_edge_thread_id_override_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -5012,6 +5072,7 @@ async fn fidelity_full_without_explicit_thread_id_uses_previous_node() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -5076,6 +5137,7 @@ async fn fidelity_from_parsed_dot_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -5120,6 +5182,7 @@ async fn fidelity_checkpoint_roundtrip_preserves_fidelity() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -5186,6 +5249,7 @@ async fn fidelity_node_thread_id_overrides_edge_thread_id_in_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine.run(&graph, &config).await.expect("run");
 
@@ -5268,6 +5332,7 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
     engine
         .run_from_checkpoint(&graph, &config, &checkpoint)
@@ -5458,6 +5523,7 @@ mod real_llm {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: std::collections::HashMap::new(),
         };
 
         let outcome = tokio::time::timeout(
@@ -5569,6 +5635,7 @@ mod real_llm {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: std::collections::HashMap::new(),
         };
 
         let outcome = tokio::time::timeout(
@@ -5707,6 +5774,7 @@ mod real_llm {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: std::collections::HashMap::new(),
         };
 
         let outcome = tokio::time::timeout(
@@ -5813,6 +5881,7 @@ mod real_llm {
             base_sha: None,
             run_branch: None,
             meta_branch: None,
+            labels: std::collections::HashMap::new(),
         };
 
         let outcome = tokio::time::timeout(
@@ -5908,6 +5977,7 @@ async fn human_gate_freeform_only_routes_text() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6037,6 +6107,7 @@ async fn human_gate_freeform_with_fixed_choice_match() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6151,6 +6222,7 @@ async fn human_gate_freeform_fallback_on_unmatched_text() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6278,6 +6350,7 @@ async fn human_gate_freeform_sets_allow_freeform_on_question() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6385,6 +6458,7 @@ async fn human_gate_without_freeform_sets_allow_freeform_false() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6626,6 +6700,7 @@ async fn tool_hooks_pre_success_allows_pipeline_to_proceed() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6674,6 +6749,7 @@ async fn tool_hooks_pre_failure_skips_tool_call() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     engine
@@ -6725,6 +6801,7 @@ async fn tool_hooks_post_success_does_not_affect_outcome() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6768,6 +6845,7 @@ async fn tool_hooks_post_failure_does_not_block_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6813,6 +6891,7 @@ async fn tool_hooks_graph_level_applies_to_all_nodes() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -6869,6 +6948,7 @@ async fn tool_hooks_node_level_overrides_graph_level() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let _outcome = engine
@@ -6932,6 +7012,7 @@ async fn tool_hooks_pre_receives_node_id_env_var() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     engine
@@ -7033,6 +7114,7 @@ async fn arc_e2e_with_real_llm() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -7158,6 +7240,7 @@ async fn run_fidelity_prompt_pipeline(fidelity: &str) -> String {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     engine
@@ -7354,6 +7437,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -7553,6 +7637,7 @@ async fn artifact_pointers_rewritten_for_remote_execution_env() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -7680,6 +7765,7 @@ async fn node_dir_uses_visit_count_on_revisit() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -8498,6 +8584,7 @@ async fn full_pipeline_with_cli_backend_node() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -8624,6 +8711,7 @@ async fn stylesheet_backend_property_routes_to_cli() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine
@@ -8891,6 +8979,7 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
         base_sha: Some(base_sha.clone()),
         run_branch: Some("arc/run/test-docker".to_string()),
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     // 5. Run pipeline
@@ -9071,6 +9160,7 @@ async fn git_checkpoint_host_writes_shadow_branch() {
         base_sha: Some(base_sha),
         run_branch: Some(format!("arc/run/{run_id}")),
         meta_branch: Some(meta_branch),
+        labels: std::collections::HashMap::new(),
     };
 
     // 5. Run pipeline
@@ -9263,6 +9353,7 @@ async fn parallel_git_branching_host_e2e() {
         base_sha: Some(base_sha.clone()),
         run_branch: Some(run_branch.clone()),
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     // 5. Run pipeline
@@ -9782,6 +9873,7 @@ async fn e2e_circuit_breaker_deterministic_self_loop() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -9826,6 +9918,7 @@ async fn e2e_circuit_breaker_custom_limit() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -9863,6 +9956,7 @@ async fn e2e_circuit_breaker_ignores_transient_failures() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -9907,6 +10001,7 @@ async fn e2e_circuit_breaker_different_reasons_separate_counters() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -9944,6 +10039,7 @@ async fn e2e_circuit_breaker_loop_restart() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10003,6 +10099,7 @@ async fn e2e_failure_signature_persisted_in_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine.run(&graph, &config).await.unwrap();
@@ -10064,6 +10161,7 @@ async fn e2e_failure_signature_hint_overrides_reason_in_context() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let _outcome = engine.run(&graph, &config).await.unwrap();
@@ -10117,6 +10215,7 @@ async fn e2e_signature_maps_persist_in_checkpoint() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine.run(&graph, &config).await.unwrap();
@@ -10240,6 +10339,7 @@ async fn e2e_circuit_breaker_emits_events_before_abort() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10304,6 +10404,7 @@ async fn e2e_circuit_breaker_does_not_fire_below_limit() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine.run(&graph, &config).await.unwrap();
@@ -10397,6 +10498,7 @@ async fn e2e_circuit_breaker_multi_stage_impl_verify_cycle() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10490,6 +10592,7 @@ async fn e2e_loop_restart_blocked_for_deterministic_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10524,6 +10627,7 @@ async fn e2e_loop_restart_blocked_for_structural_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10558,6 +10662,7 @@ async fn e2e_loop_restart_blocked_for_budget_exhausted_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10592,6 +10697,7 @@ async fn e2e_loop_restart_blocked_for_canceled_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10626,6 +10732,7 @@ async fn e2e_loop_restart_blocked_for_compilation_loop_failure() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10661,6 +10768,7 @@ async fn e2e_loop_restart_allowed_for_transient_infra() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10765,6 +10873,7 @@ async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let result = engine.run(&graph, &config).await;
@@ -10818,6 +10927,7 @@ async fn e2e_stall_watchdog_kept_alive_by_handler_events() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine.run(&graph, &config).await.expect("pipeline should succeed");
@@ -10854,6 +10964,7 @@ async fn e2e_stall_watchdog_disabled_with_zero_timeout() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let outcome = engine.run(&graph, &config).await.expect("pipeline should succeed");
@@ -10913,6 +11024,7 @@ async fn e2e_stall_watchdog_with_explicit_timeout_override() {
         base_sha: None,
         run_branch: None,
         meta_branch: None,
+        labels: std::collections::HashMap::new(),
     };
 
     let start = std::time::Instant::now();