feat(sdk): parallel pipeline fan-out + typed JSON report with metrics

Base::run_parallel runs steps as concurrent copy-on-write MicroVM forks
(bounded, collect-all, input-ordered) and returns a dependency-free JSON
Report. StepResult gains separated stdout/stderr, duration_ms, and metrics
parsed from `::metric <key>=<value>` guest-stdout lines (a scoring channel
for matrix/selection workloads).

Steps now take &self (atomic fork counter), so fan-out no longer needs
hand-rolled threads. RAII removes each fork on every path and the base
snapshot on Drop (--force). Box/snapshot names carry per-process+instance
entropy to prevent cross-pipeline name collisions; output is capped to bound
report memory under large fan-out; infra failures use a distinct sentinel.

15 unit tests + doctest, clippy clean; validated end-to-end on a real
/dev/kvm host (real boot, CoW fork-per-step, metrics, JSON report, 0 leaks).

Author: Roy Lin

CHANGELOG.md

@@ -4,6 +4,26 @@ All notable changes to A3S Box will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- **Programmable-CI pipeline: parallel fan-out + typed JSON report (`a3s-box-sdk`).**
+  `Base::run_parallel(steps, max_concurrency)` runs steps concurrently as isolated
+  copy-on-write MicroVM forks (bounded, collect-all, results in input order) and returns a
+  `Report` with a dependency-free `to_json()`. `StepResult` now carries separated
+  `stdout`/`stderr`, `duration_ms`, and `metrics` parsed from `::metric <key>=<value>`
+  guest-stdout lines (a machine-readable scoring channel for matrix/selection workloads).
+  Steps run via `&self` (atomic fork counter), so fan-out no longer needs hand-rolled
+  threads. The base auto-removes its snapshot on `Drop` (`--force`), and each fork is
+  removed on every path (including panic). Box/snapshot names now carry per-process +
+  per-instance entropy, so concurrent pipelines from the same image+setup can no longer
+  collide and tear down each other's boxes. Validated end-to-end on a real `/dev/kvm` host.
+
+### Changed
+
+- **`StepResult.logs` is replaced by separated `stdout` / `stderr` fields**
+  (use `StepResult::combined()` for the old concatenated view). Breaking for
+  direct `.logs` field access on the `a3s-box-sdk` pipeline API.
+
 ## [2.6.0] — 2026-06-26
 
 ### Added

src/sdk/README.md

@@ -11,28 +11,48 @@ state). Set `A3S_BOX` if `a3s-box` is not on `PATH`.
 
 ## Pipelines
 
-Warm a base box **once** (clone + install deps), snapshot it, fork per step:
+Warm a base box **once** (clone + install deps), snapshot it, fork per step.
+Run steps **sequentially** (fail-fast) or **in parallel** (collect-all → a typed
+report):
 
 ```rust
 use a3s_box_sdk::pipeline::{warm_base, WarmBase, FileCache, Step};
 
 fn main() -> Result<(), a3s_box_sdk::pipeline::PipelineError> {
     let cache = FileCache::new(".ci-cache")?;          // skip a step when its inputs are unchanged
-    let mut base = warm_base(
+    let base = warm_base(
         WarmBase::new("node:20", "git clone $REPO /w && cd /w && npm ci")   // runs ONCE
             .env("REPO", "https://github.com/me/app")
             .cache(&cache),
     )?;
+
+    // Sequential, fail-fast: a non-zero exit returns Err.
     base.step(Step::new("lint", "cd /w && npm run lint"))?;
-    base.step(Step::new("test", "cd /w && npm test"))?;   // nonzero exit -> Err (fail-fast)
-    base.step(Step::new("build", "cd /w && npm run build"))?;
-    base.dispose();                                       // drops the snapshot
-    Ok(())
+
+    // Parallel, collect-all: each step is an isolated CoW fork; <=4 at a time.
+    let report = base.run_parallel(vec![
+        Step::new("test",  "cd /w && npm test"),
+        Step::new("build", "cd /w && npm run build"),
+    ], 4);
+
+    println!("{}", report.to_json());   // {"passed":..,"total_ms":..,"steps":[..]}
+    if !report.passed { /* inspect report.failures() */ }
+    Ok(())  // `base` drops here -> snapshot auto-removed (or call base.dispose())
 }
 ```
 
-`Step::allow_failure()` keeps the pipeline going on a non-zero exit; `Step::input(..)`
-adds extra cache-key parts. Parallel steps = spawn threads (each `step` is blocking).
+- `run_parallel` is the way to use a3s-box's cheap (~ms) CoW fork at scale — a
+  matrix / evolution-style batch — without hand-rolling threads (every method
+  takes `&self`).
+- A step reports a metric by printing `::metric <key>=<number>` to stdout; it
+  surfaces as `StepResult::metrics` (the scoring channel for a selection loop).
+- `StepResult` carries separated `stdout`/`stderr`, `duration_ms`, and `cached`;
+  `Report::to_json()` is the machine-readable handoff to an agent/scorer.
+- `Step::allow_failure()` keeps a non-zero step from failing the run; `Step::input(..)`
+  adds extra cache-key parts.
+
+The base **auto-disposes** its snapshot on drop, and each per-step box is removed
+on every path (including a panic), so a long-running batch doesn't leak.
 
 ## Why forking is cheap
 

src/sdk/examples/pipeline.rs

@@ -12,25 +12,36 @@ fn main() -> Result<(), a3s_box_sdk::pipeline::PipelineError> {
     let cache = FileCache::new("/tmp/.a3s-ci-demo")?;
 
     // Warm the base once (here: write a marker = "deps installed"), snapshot it.
-    let mut base = warm_base(WarmBase::new(image, "echo DEPS-INSTALLED > /warmed").cache(&cache))?;
+    let base = warm_base(WarmBase::new(image, "echo DEPS-INSTALLED > /warmed").cache(&cache))?;
 
+    // Sequential, fail-fast: a non-zero exit returns Err.
     let r = base.step(Step::new("read", "cat /warmed"))?;
-    println!(
-        "read   -> code={} out={:?} cached={}",
-        r.exit_code,
-        r.logs.trim(),
-        r.cached
-    );
-
+    println!("read   -> code={} out={:?}", r.exit_code, r.stdout.trim());
     let r2 = base.step(Step::new("read", "cat /warmed"))?; // identical -> cache hit
     println!("read#2 -> cached={}", r2.cached);
 
-    match base.step(Step::new("fail", "exit 7")) {
-        Err(e) => println!("fail-fast ok: {e}"),
-        Ok(_) => println!("ERROR: fail step did not error"),
-    }
+    // Parallel matrix, collect-all: each step is an isolated CoW fork of the base.
+    // A step reports a metric by printing `::metric key=value`.
+    let report = base.run_parallel(
+        vec![
+            Step::new("ok", "echo fine"),
+            Step::new("perf", "echo '::metric duration_ms=12.5'"),
+            Step::new("fail", "exit 7"), // collected, not fatal
+        ],
+        4,
+    );
+    println!("report -> {}", report.to_json());
+    println!(
+        "passed={} failures={:?}",
+        report.passed,
+        report
+            .failures()
+            .iter()
+            .map(|s| &s.name)
+            .collect::<Vec<_>>()
+    );
 
-    base.dispose();
+    // base.dispose() is optional — the snapshot is removed when `base` drops.
     println!("demo ok");
     Ok(())
 }