candle_graph/
campaign.rs

1//! Application-neutral campaign/series layer over published evidence bundles.
2//!
3//! A *campaign* is a producer-declared plan of capture steps for one training
4//! entrypoint. This module reconciles that plan against published,
5//! content-verified evidence bundles ([`campaign_status`]) and assembles
6//! long-form metric trajectories across verified bundles ([`build_series`]).
7//! Everything here is stateless and read-only: commands read artifacts;
8//! nothing supervises training.
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::fs;
12use std::path::{Component, Path, PathBuf};
13
14use anyhow::{bail, Context, Result};
15use serde::{Deserialize, Serialize};
16
17use crate::artifact::verify_bundle;
18use crate::trace::schema::GradientState;
19use crate::trace::{parse_trace, TraceDocument};
20
21pub const CAMPAIGN_SCHEMA: &str = "candle-graph/campaign/1";
22pub const CAMPAIGN_STATUS_SCHEMA: &str = "candle-graph/campaign-status/1";
23pub const SERIES_SCHEMA: &str = "candle-graph/series/1";
24
25/// One planned capture: a step coordinate and the bundle path expected to hold
26/// its published evidence, relative to the campaign manifest's directory.
27#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
28pub struct PlannedCapture {
29    pub capture_step: u64,
30    pub bundle: String,
31}
32
33/// Producer-declared plan for one training campaign.
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35pub struct CampaignManifest {
36    pub schema: String,
37    pub campaign_id: String,
38    pub entrypoint: String,
39    pub planned: Vec<PlannedCapture>,
40}
41
42impl CampaignManifest {
43    /// Parse a campaign manifest from JSON on disk and validate it.
44    pub fn load(path: &Path) -> Result<Self> {
45        let bytes =
46            fs::read(path).with_context(|| format!("read campaign manifest {}", path.display()))?;
47        let manifest: Self = serde_json::from_slice(&bytes)
48            .with_context(|| format!("parse campaign manifest {}", path.display()))?;
49        manifest
50            .validate()
51            .with_context(|| format!("validate campaign manifest {}", path.display()))?;
52        Ok(manifest)
53    }
54
55    /// Reject manifests that could not be reconciled deterministically.
56    pub fn validate(&self) -> Result<()> {
57        if self.schema != CAMPAIGN_SCHEMA {
58            bail!(
59                "unsupported campaign schema {:?}; expected {CAMPAIGN_SCHEMA:?}",
60                self.schema
61            );
62        }
63        if self.campaign_id.trim().is_empty() {
64            bail!("campaign manifest requires a non-empty campaign_id");
65        }
66        if self.entrypoint.trim().is_empty() {
67            bail!("campaign manifest requires a non-empty entrypoint");
68        }
69        if self.planned.is_empty() {
70            bail!("campaign manifest requires at least one planned capture");
71        }
72        let mut steps = BTreeSet::new();
73        let mut bundles = BTreeSet::new();
74        for capture in &self.planned {
75            if !steps.insert(capture.capture_step) {
76                bail!(
77                    "campaign manifest plans duplicate capture_step {}",
78                    capture.capture_step
79                );
80            }
81            if !is_safe_relative_path(&capture.bundle) {
82                bail!(
83                    "campaign manifest has an unsafe bundle path {:?}; \
84                     bundle paths must be non-empty relative paths without `..` or backslashes",
85                    capture.bundle
86                );
87            }
88            if !bundles.insert(capture.bundle.as_str()) {
89                bail!(
90                    "campaign manifest plans duplicate bundle path {:?}",
91                    capture.bundle
92                );
93            }
94        }
95        Ok(())
96    }
97}
98
99/// Reconciled state of one planned capture.
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
101#[serde(tag = "state", rename_all = "snake_case")]
102pub enum CaptureState {
103    Missing,
104    Published {
105        run_id: String,
106        manifest_sha256: String,
107    },
108    FailedRun {
109        run_id: String,
110        reason: Option<String>,
111    },
112    VerificationFailed {
113        message: String,
114    },
115    IdentityMismatch {
116        message: String,
117    },
118}
119
120/// One planned capture joined with the state observed on the filesystem.
121#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
122pub struct CaptureStatus {
123    pub capture_step: u64,
124    pub bundle: String,
125    pub state: CaptureState,
126}
127
128/// Deterministic reconciliation of a campaign plan against published bundles.
129#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
130pub struct CampaignStatus {
131    pub schema: String,
132    pub campaign_id: String,
133    pub entrypoint: String,
134    pub planned: usize,
135    pub published: usize,
136    pub missing: usize,
137    pub failed: usize,
138    pub captures: Vec<CaptureStatus>,
139}
140
141/// Reconcile every planned capture in the manifest at `manifest_path` against
142/// the filesystem. Bundle paths resolve relative to the manifest's parent
143/// directory; every planned capture appears exactly once in the result.
144pub fn campaign_status(manifest_path: &Path) -> Result<CampaignStatus> {
145    let manifest = CampaignManifest::load(manifest_path)?;
146    let base = manifest_path.parent().unwrap_or_else(|| Path::new("."));
147
148    let mut captures = Vec::with_capacity(manifest.planned.len());
149    for planned in &manifest.planned {
150        let root = base.join(&planned.bundle);
151        let state = reconcile_capture(&root, &manifest.entrypoint, planned.capture_step);
152        captures.push(CaptureStatus {
153            capture_step: planned.capture_step,
154            bundle: planned.bundle.clone(),
155            state,
156        });
157    }
158    captures.sort_by_key(|capture| capture.capture_step);
159
160    let published = captures
161        .iter()
162        .filter(|capture| matches!(capture.state, CaptureState::Published { .. }))
163        .count();
164    let missing = captures
165        .iter()
166        .filter(|capture| matches!(capture.state, CaptureState::Missing))
167        .count();
168    let failed = captures.len() - published - missing;
169
170    Ok(CampaignStatus {
171        schema: CAMPAIGN_STATUS_SCHEMA.into(),
172        campaign_id: manifest.campaign_id,
173        entrypoint: manifest.entrypoint,
174        planned: captures.len(),
175        published,
176        missing,
177        failed,
178        captures,
179    })
180}
181
182fn reconcile_capture(root: &Path, entrypoint: &str, capture_step: u64) -> CaptureState {
183    if !root.exists() {
184        return CaptureState::Missing;
185    }
186    let receipt = match verify_bundle(root) {
187        Ok(receipt) => receipt,
188        Err(error) => {
189            return CaptureState::VerificationFailed {
190                message: format!("{error:#}"),
191            }
192        }
193    };
194    let document = match parse_trace(root.join("trace.jsonl")) {
195        Ok(document) => document,
196        Err(error) => {
197            return CaptureState::VerificationFailed {
198                message: format!("{error:#}"),
199            }
200        }
201    };
202    if document.run.entrypoint != entrypoint {
203        return CaptureState::IdentityMismatch {
204            message: format!(
205                "entrypoint mismatch: manifest declares {:?}, bundle trace observes {:?}",
206                entrypoint, document.run.entrypoint
207            ),
208        };
209    }
210    if document.run.capture_step != capture_step {
211        return CaptureState::IdentityMismatch {
212            message: format!(
213                "capture_step mismatch: manifest plans {}, bundle trace observes {}",
214                capture_step, document.run.capture_step
215            ),
216        };
217    }
218    match document.terminal.outcome {
219        crate::trace::RunOutcome::Failed => CaptureState::FailedRun {
220            run_id: document.run.run_id,
221            reason: document.terminal.reason,
222        },
223        crate::trace::RunOutcome::Complete => CaptureState::Published {
224            run_id: document.run.run_id,
225            manifest_sha256: receipt.manifest_sha256,
226        },
227    }
228}
229
230/// One verified bundle admitted into a series, keyed by its capture step.
231#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
232pub struct SeriesInput {
233    pub capture_step: u64,
234    pub run_id: String,
235    pub bundle: String,
236    pub manifest_sha256: String,
237}
238
239/// Outer wall time of one run. `Some` only when the trace holds exactly one
240/// measured, closed span; otherwise the coordinate is ambiguous and stays `None`.
241#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
242pub struct TimingSeriesPoint {
243    pub capture_step: u64,
244    pub run_id: String,
245    pub outer_wall_time_ns: Option<u64>,
246}
247
248/// One tensor-statistics observation on the series coordinate.
249#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
250pub struct ScalarSeriesPoint {
251    pub capture_step: u64,
252    pub run_id: String,
253    pub rms: f64,
254    pub abs_max: f64,
255    pub mean: f64,
256    pub non_finite: u64,
257    pub elements: u64,
258}
259
260/// One gradient observation on the series coordinate.
261#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
262pub struct GradientSeriesPoint {
263    pub capture_step: u64,
264    pub run_id: String,
265    pub state: GradientState,
266    pub norm: Option<f64>,
267}
268
269/// Long-form metric trajectories across verified bundles of one entrypoint.
270#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
271pub struct SeriesReport {
272    pub schema: String,
273    pub entrypoint: String,
274    pub coordinate: String,
275    pub label_prefix: Option<String>,
276    pub inputs: Vec<SeriesInput>,
277    pub outer_wall_time: Vec<TimingSeriesPoint>,
278    pub tensor_stats: BTreeMap<String, Vec<ScalarSeriesPoint>>,
279    pub gradients: BTreeMap<String, Vec<GradientSeriesPoint>>,
280}
281
282/// Build metric trajectories across `bundle_roots`, sorted by capture step.
283///
284/// Every bundle must pass deep content verification and all traces must share
285/// one entrypoint and phase; duplicate capture steps are rejected because the
286/// series coordinate would be ambiguous. `label_prefix` filters tensor-stat
287/// labels and gradient `{root}/{key}` composites; `None` keeps all.
288pub fn build_series(bundle_roots: &[PathBuf], label_prefix: Option<&str>) -> Result<SeriesReport> {
289    if bundle_roots.is_empty() {
290        bail!("series requires at least one bundle");
291    }
292
293    struct SeriesEntry {
294        bundle: String,
295        manifest_sha256: String,
296        document: TraceDocument,
297    }
298
299    let mut entries = Vec::with_capacity(bundle_roots.len());
300    for root in bundle_roots {
301        let receipt = verify_bundle(root)
302            .with_context(|| format!("verify series bundle {}", root.display()))?;
303        let document = parse_trace(root.join("trace.jsonl"))
304            .with_context(|| format!("parse trace of series bundle {}", root.display()))?;
305        entries.push(SeriesEntry {
306            bundle: root.display().to_string(),
307            manifest_sha256: receipt.manifest_sha256,
308            document,
309        });
310    }
311
312    let entrypoint = entries[0].document.run.entrypoint.clone();
313    let phase = entries[0].document.run.phase;
314    let mut seen_steps: BTreeMap<u64, &str> = BTreeMap::new();
315    for entry in &entries {
316        let run = &entry.document.run;
317        if run.entrypoint != entrypoint {
318            bail!(
319                "series bundles mix entrypoints: {} observes {:?}, expected {:?} from {}",
320                entry.bundle,
321                run.entrypoint,
322                entrypoint,
323                entries[0].bundle
324            );
325        }
326        if run.phase != phase {
327            bail!(
328                "series bundles mix phases: {} observes {:?}, expected {:?} from {}",
329                entry.bundle,
330                run.phase.as_str(),
331                phase.as_str(),
332                entries[0].bundle
333            );
334        }
335        if let Some(existing) = seen_steps.insert(run.capture_step, entry.bundle.as_str()) {
336            bail!(
337                "ambiguous series coordinate: capture_step {} is provided by both {} and {}",
338                run.capture_step,
339                existing,
340                entry.bundle
341            );
342        }
343    }
344    entries.sort_by_key(|entry| entry.document.run.capture_step);
345
346    let matches_prefix = |value: &str| label_prefix.is_none_or(|prefix| value.starts_with(prefix));
347
348    let mut inputs = Vec::with_capacity(entries.len());
349    let mut outer_wall_time = Vec::with_capacity(entries.len());
350    let mut tensor_stats: BTreeMap<String, Vec<ScalarSeriesPoint>> = BTreeMap::new();
351    let mut gradients: BTreeMap<String, Vec<GradientSeriesPoint>> = BTreeMap::new();
352    for entry in &entries {
353        let run = &entry.document.run;
354        inputs.push(SeriesInput {
355            capture_step: run.capture_step,
356            run_id: run.run_id.clone(),
357            bundle: entry.bundle.clone(),
358            manifest_sha256: entry.manifest_sha256.clone(),
359        });
360
361        let mut measured = entry
362            .document
363            .spans
364            .iter()
365            .filter(|span| span.measured && span.closed);
366        let outer_wall_time_ns = match (measured.next(), measured.next()) {
367            (Some(span), None) => Some(span.duration_ns),
368            _ => None,
369        };
370        outer_wall_time.push(TimingSeriesPoint {
371            capture_step: run.capture_step,
372            run_id: run.run_id.clone(),
373            outer_wall_time_ns,
374        });
375
376        for stats in &entry.document.tensor_stats {
377            if !matches_prefix(&stats.label) {
378                continue;
379            }
380            tensor_stats
381                .entry(stats.label.clone())
382                .or_default()
383                .push(ScalarSeriesPoint {
384                    capture_step: run.capture_step,
385                    run_id: run.run_id.clone(),
386                    rms: stats.rms,
387                    abs_max: stats.abs_max,
388                    mean: stats.mean,
389                    non_finite: stats.non_finite,
390                    elements: stats.elements,
391                });
392        }
393
394        for gradient in &entry.document.gradients {
395            let key = format!("{}/{}", gradient.root, gradient.key);
396            if !matches_prefix(&key) {
397                continue;
398            }
399            gradients.entry(key).or_default().push(GradientSeriesPoint {
400                capture_step: run.capture_step,
401                run_id: run.run_id.clone(),
402                state: gradient.state,
403                norm: gradient.norm,
404            });
405        }
406    }
407
408    Ok(SeriesReport {
409        schema: SERIES_SCHEMA.into(),
410        entrypoint,
411        coordinate: "capture_step".into(),
412        label_prefix: label_prefix.map(str::to_owned),
413        inputs,
414        outer_wall_time,
415        tensor_stats,
416        gradients,
417    })
418}
419
420fn is_safe_relative_path(value: &str) -> bool {
421    let path = Path::new(value);
422    !path.as_os_str().is_empty()
423        && !value.contains('\\')
424        && !path.is_absolute()
425        && path
426            .components()
427            .all(|component| matches!(component, Component::Normal(_)))
428}
429
430#[cfg(test)]
431mod tests {
432    use super::*;
433    use crate::artifact::publish_bundle;
434    use crate::capability::CaptureContract;
435    use crate::trace::{
436        write_jsonl, GradientEvent, RunOutcome, SpanKind, SpanRecord, TensorStatsEvent,
437        TerminalEvent, TimingMode, TraceDocument, TraceRunMeta, SCHEMA as TRACE_SCHEMA,
438    };
439    use std::time::{SystemTime, UNIX_EPOCH};
440
441    fn temp_root(label: &str) -> PathBuf {
442        let nonce = SystemTime::now()
443            .duration_since(UNIX_EPOCH)
444            .unwrap()
445            .as_nanos();
446        let root = std::env::temp_dir().join(format!(
447            "candle-graph-campaign-{label}-{}-{nonce}",
448            std::process::id()
449        ));
450        fs::create_dir_all(&root).unwrap();
451        root
452    }
453
454    fn fixture_document(run_id: &str, entrypoint: &str, capture_step: u64) -> TraceDocument {
455        TraceDocument {
456            schema: TRACE_SCHEMA.into(),
457            run: TraceRunMeta {
458                run_id: run_id.into(),
459                correlation_id: format!("campaign/{run_id}"),
460                entrypoint: entrypoint.into(),
461                phase: crate::ExecutionPhase::Train,
462                timestamp: "2026-08-28T00:00:00Z".into(),
463                capture_step,
464                warmup_steps: 0,
465                device: "cpu".into(),
466                measured_region_device_synchronized: false,
467                timing_mode: TimingMode::Host,
468                capture_contract: CaptureContract::default(),
469                comparison_identity: None,
470                tags: Default::default(),
471                candle_version: None,
472            },
473            spans: vec![SpanRecord {
474                id: "root".into(),
475                parent_id: None,
476                name: entrypoint.into(),
477                kind: SpanKind::Function,
478                measured: true,
479                start_ns: 0,
480                closed: true,
481                duration_ns: 10 + capture_step,
482                step: None,
483            }],
484            ops: vec![],
485            tensors: vec![],
486            tensor_stats: vec![
487                TensorStatsEvent {
488                    span_id: "root".into(),
489                    label: "loss/total".into(),
490                    shape: vec![1],
491                    dtype: "f32".into(),
492                    elements: 1,
493                    non_finite: 0,
494                    rms: 2.0 + capture_step as f64,
495                    abs_max: 2.0 + capture_step as f64,
496                    mean: -(capture_step as f64),
497                },
498                TensorStatsEvent {
499                    span_id: "root".into(),
500                    label: "act/mlp".into(),
501                    shape: vec![2, 3],
502                    dtype: "f32".into(),
503                    elements: 6,
504                    non_finite: 0,
505                    rms: 1.0,
506                    abs_max: 3.0,
507                    mean: 0.5,
508                },
509            ],
510            memory: vec![],
511            device_memory: vec![],
512            device_intervals: vec![],
513            gradients: vec![
514                GradientEvent {
515                    event_id: format!("{run_id}-g1"),
516                    root: "vb".into(),
517                    key: "encoder.weight".into(),
518                    state: GradientState::Present,
519                    norm: Some(0.5),
520                },
521                GradientEvent {
522                    event_id: format!("{run_id}-g2"),
523                    root: "vb".into(),
524                    key: "decoder.bias".into(),
525                    state: GradientState::Zero,
526                    norm: Some(0.0),
527                },
528            ],
529            edges: vec![],
530            terminal: TerminalEvent {
531                outcome: RunOutcome::Complete,
532                timestamp_ns: 10 + capture_step,
533                reason: None,
534            },
535        }
536    }
537
538    fn publish_fixture(root: &Path, name: &str, document: &TraceDocument) -> PathBuf {
539        let trace = root.join(format!("{name}.input.jsonl"));
540        write_jsonl(&trace, &document.to_events()).unwrap();
541        let destination = root.join(name);
542        publish_bundle(&destination, &trace, None).unwrap();
543        fs::remove_file(trace).unwrap();
544        destination
545    }
546
547    fn write_manifest(root: &Path, manifest: &CampaignManifest) -> PathBuf {
548        let path = root.join("campaign.json");
549        fs::write(&path, serde_json::to_string_pretty(manifest).unwrap()).unwrap();
550        path
551    }
552
553    #[test]
554    fn campaign_status_reconciles_every_planned_capture() {
555        let root = temp_root("status");
556        publish_fixture(
557            &root,
558            "b100",
559            &fixture_document("run-100", "demo::train", 100),
560        );
561        publish_fixture(
562            &root,
563            "b200",
564            &fixture_document("run-200", "demo::train", 200),
565        );
566        // Published under the planned path but captured at the wrong step.
567        publish_fixture(
568            &root,
569            "b300",
570            &fixture_document("run-999", "demo::train", 999),
571        );
572        let manifest = CampaignManifest {
573            schema: CAMPAIGN_SCHEMA.into(),
574            campaign_id: "demo-campaign".into(),
575            entrypoint: "demo::train".into(),
576            planned: vec![
577                PlannedCapture {
578                    capture_step: 300,
579                    bundle: "b300".into(),
580                },
581                PlannedCapture {
582                    capture_step: 100,
583                    bundle: "b100".into(),
584                },
585                PlannedCapture {
586                    capture_step: 400,
587                    bundle: "b400".into(),
588                },
589                PlannedCapture {
590                    capture_step: 200,
591                    bundle: "b200".into(),
592                },
593            ],
594        };
595        let manifest_path = write_manifest(&root, &manifest);
596
597        let status = campaign_status(&manifest_path).unwrap();
598        assert_eq!(status.schema, CAMPAIGN_STATUS_SCHEMA);
599        assert_eq!(status.campaign_id, "demo-campaign");
600        assert_eq!(status.entrypoint, "demo::train");
601        assert_eq!(status.planned, 4);
602        assert_eq!(status.published, 2);
603        assert_eq!(status.missing, 1);
604        assert_eq!(status.failed, 1);
605        let steps: Vec<u64> = status
606            .captures
607            .iter()
608            .map(|capture| capture.capture_step)
609            .collect();
610        assert_eq!(steps, vec![100, 200, 300, 400]);
611        match &status.captures[0].state {
612            CaptureState::Published {
613                run_id,
614                manifest_sha256,
615            } => {
616                assert_eq!(run_id, "run-100");
617                assert_eq!(manifest_sha256.len(), 64);
618            }
619            other => panic!("expected published capture, got {other:?}"),
620        }
621        assert!(matches!(
622            status.captures[1].state,
623            CaptureState::Published { .. }
624        ));
625        match &status.captures[2].state {
626            CaptureState::IdentityMismatch { message } => {
627                assert!(message.contains("capture_step mismatch"));
628                assert!(message.contains("300"));
629                assert!(message.contains("999"));
630            }
631            other => panic!("expected identity mismatch, got {other:?}"),
632        }
633        assert_eq!(status.captures[3].state, CaptureState::Missing);
634        fs::remove_dir_all(root).unwrap();
635    }
636
637    #[test]
638    fn manifest_validation_rejects_bad_plans() {
639        let good = CampaignManifest {
640            schema: CAMPAIGN_SCHEMA.into(),
641            campaign_id: "c".into(),
642            entrypoint: "demo::train".into(),
643            planned: vec![PlannedCapture {
644                capture_step: 1,
645                bundle: "bundles/b1".into(),
646            }],
647        };
648        good.validate().unwrap();
649
650        let mut wrong_schema = good.clone();
651        wrong_schema.schema = "candle-graph/campaign/0".into();
652        assert!(wrong_schema
653            .validate()
654            .unwrap_err()
655            .to_string()
656            .contains("schema"));
657
658        let mut duplicate_steps = good.clone();
659        duplicate_steps.planned = vec![
660            PlannedCapture {
661                capture_step: 1,
662                bundle: "a".into(),
663            },
664            PlannedCapture {
665                capture_step: 1,
666                bundle: "b".into(),
667            },
668        ];
669        assert!(duplicate_steps
670            .validate()
671            .unwrap_err()
672            .to_string()
673            .contains("duplicate capture_step"));
674
675        let mut duplicate_bundles = good.clone();
676        duplicate_bundles.planned = vec![
677            PlannedCapture {
678                capture_step: 1,
679                bundle: "same".into(),
680            },
681            PlannedCapture {
682                capture_step: 2,
683                bundle: "same".into(),
684            },
685        ];
686        assert!(duplicate_bundles
687            .validate()
688            .unwrap_err()
689            .to_string()
690            .contains("duplicate bundle path"));
691
692        for unsafe_path in ["", "/abs/b1", "../escape", "a\\b", "a/../b"] {
693            let mut manifest = good.clone();
694            manifest.planned[0].bundle = unsafe_path.into();
695            assert!(
696                manifest
697                    .validate()
698                    .unwrap_err()
699                    .to_string()
700                    .contains("unsafe bundle path"),
701                "path {unsafe_path:?} must be rejected"
702            );
703        }
704
705        let mut empty_plan = good.clone();
706        empty_plan.planned.clear();
707        assert!(empty_plan
708            .validate()
709            .unwrap_err()
710            .to_string()
711            .contains("at least one planned capture"));
712
713        let mut empty_id = good;
714        empty_id.campaign_id = "  ".into();
715        assert!(empty_id
716            .validate()
717            .unwrap_err()
718            .to_string()
719            .contains("campaign_id"));
720    }
721
722    #[test]
723    fn series_sorts_filters_and_rejects_ambiguity() {
724        let root = temp_root("series");
725        let b100 = publish_fixture(
726            &root,
727            "b100",
728            &fixture_document("run-100", "demo::train", 100),
729        );
730        let b200 = publish_fixture(
731            &root,
732            "b200",
733            &fixture_document("run-200", "demo::train", 200),
734        );
735        // Second measured+closed span makes the outer wall coordinate ambiguous.
736        let mut two_measured = fixture_document("run-300", "demo::train", 300);
737        two_measured.spans.push(SpanRecord {
738            id: "root-b".into(),
739            parent_id: None,
740            name: "demo::train#b".into(),
741            kind: SpanKind::Function,
742            measured: true,
743            start_ns: 0,
744            closed: true,
745            duration_ns: 5,
746            step: None,
747        });
748        let b300 = publish_fixture(&root, "b300", &two_measured);
749
750        // Bundles passed out of order; the report must sort by capture_step.
751        let report = build_series(&[b300.clone(), b100.clone(), b200.clone()], None).unwrap();
752        assert_eq!(report.schema, SERIES_SCHEMA);
753        assert_eq!(report.entrypoint, "demo::train");
754        assert_eq!(report.coordinate, "capture_step");
755        assert_eq!(report.label_prefix, None);
756        let input_steps: Vec<u64> = report
757            .inputs
758            .iter()
759            .map(|input| input.capture_step)
760            .collect();
761        assert_eq!(input_steps, vec![100, 200, 300]);
762        assert!(report
763            .inputs
764            .iter()
765            .all(|input| input.manifest_sha256.len() == 64));
766        assert_eq!(report.inputs[0].run_id, "run-100");
767        assert_eq!(
768            report
769                .outer_wall_time
770                .iter()
771                .map(|point| point.outer_wall_time_ns)
772                .collect::<Vec<_>>(),
773            vec![Some(110), Some(210), None]
774        );
775        assert_eq!(
776            report.tensor_stats.keys().collect::<Vec<_>>(),
777            vec!["act/mlp", "loss/total"]
778        );
779        let loss = &report.tensor_stats["loss/total"];
780        assert_eq!(
781            loss.iter()
782                .map(|point| point.capture_step)
783                .collect::<Vec<_>>(),
784            vec![100, 200, 300]
785        );
786        assert_eq!(loss[0].rms, 102.0);
787        assert_eq!(
788            report.gradients.keys().collect::<Vec<_>>(),
789            vec!["vb/decoder.bias", "vb/encoder.weight"]
790        );
791        let encoder = &report.gradients["vb/encoder.weight"];
792        assert_eq!(encoder.len(), 3);
793        assert_eq!(encoder[0].state, GradientState::Present);
794        assert_eq!(encoder[0].norm, Some(0.5));
795
796        let filtered = build_series(&[b100.clone(), b200.clone()], Some("loss")).unwrap();
797        assert_eq!(filtered.label_prefix.as_deref(), Some("loss"));
798        assert_eq!(
799            filtered.tensor_stats.keys().collect::<Vec<_>>(),
800            vec!["loss/total"]
801        );
802        assert!(filtered.gradients.is_empty());
803
804        let gradient_only = build_series(std::slice::from_ref(&b100), Some("vb/enc")).unwrap();
805        assert!(gradient_only.tensor_stats.is_empty());
806        assert_eq!(
807            gradient_only.gradients.keys().collect::<Vec<_>>(),
808            vec!["vb/encoder.weight"]
809        );
810
811        let duplicate = publish_fixture(
812            &root,
813            "b100-dup",
814            &fixture_document("run-100-dup", "demo::train", 100),
815        );
816        let error = build_series(&[b100.clone(), duplicate], None).unwrap_err();
817        assert!(error.to_string().contains("ambiguous series coordinate"));
818
819        let other = publish_fixture(
820            &root,
821            "other",
822            &fixture_document("run-other", "other::train", 400),
823        );
824        let error = build_series(&[b100.clone(), other], None).unwrap_err();
825        assert!(error.to_string().contains("mix entrypoints"));
826
827        assert!(build_series(&[], None)
828            .unwrap_err()
829            .to_string()
830            .contains("at least one bundle"));
831
832        let absent = root.join("never-published");
833        let error = build_series(&[absent], None).unwrap_err();
834        assert!(format!("{error:#}").contains("verify series bundle"));
835
836        fs::remove_dir_all(root).unwrap();
837    }
838}
candle_graph/campaign.rs

candle_graph/
campaign.rs
