use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::time::{SystemTime, UNIX_EPOCH};
use candle_graph::cli::trace_cli::{
load_evidence, run_query, run_summary, QueryLabelFilter, QueryOptions, TraceQueryKind,
};
use candle_graph::nsight::{
CaptureCorrelation, CaptureHardware, CaptureManifest, CaptureRun, CaptureTool,
GpuEvidenceStatus, ManifestArtifact, ProvenanceBindingState, CAPTURE_MANIFEST_SCHEMA,
};
use candle_graph::trace::{
write_jsonl, DeviceIntervalEvent, GradientEvent, GradientState, MemoryAction, MemoryCategory,
MemoryEvent, OpEvent, RunOutcome, SpanRecord, TensorEvent, TensorStatsEvent, TerminalEvent,
TraceRunMeta, SCHEMA as TRACE_SCHEMA,
};
use candle_graph::{
publish_bundle, verify_bundle, BundleManifest, CaptureContract, CoverageLevel, ExecutionPhase,
MeasurementScope, SpanKind, TimingMode, TraceDocument,
};
use sha2::{Digest, Sha256};
struct TempRoot(PathBuf);
impl TempRoot {
fn new(label: &str) -> Self {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let path = std::env::temp_dir().join(format!(
"candle-graph-cli-{label}-{}-{nonce}",
std::process::id()
));
fs::create_dir(&path).unwrap();
Self(path)
}
}
impl Drop for TempRoot {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.0);
}
}
fn trace_document() -> TraceDocument {
TraceDocument {
schema: TRACE_SCHEMA.into(),
run: TraceRunMeta {
run_id: "cli-bundle-run".into(),
correlation_id: "cli/bundle/run".into(),
entrypoint: "demo::update".into(),
phase: ExecutionPhase::Infer,
timestamp: "2026-08-23T00:00:00Z".into(),
capture_step: 1,
warmup_steps: 0,
device: "cuda:0".into(),
measured_region_device_synchronized: true,
timing_mode: TimingMode::Host,
capture_contract: CaptureContract {
measurement_scope: MeasurementScope::ProductionEquivalent,
device_timing: CoverageLevel::None,
required_semantic_labels: vec!["phase/gpu".into()],
gpu_expected_semantic_labels: vec!["phase/gpu".into()],
..CaptureContract::default()
},
comparison_identity: None,
tags: BTreeMap::new(),
candle_version: None,
},
spans: vec![
SpanRecord {
id: "root".into(),
parent_id: None,
name: "demo::update".into(),
kind: SpanKind::Function,
measured: true,
start_ns: 0,
closed: true,
duration_ns: 200,
step: None,
},
SpanRecord {
id: "gpu".into(),
parent_id: Some("root".into()),
name: "phase/gpu".into(),
kind: SpanKind::Module,
measured: false,
start_ns: 10,
closed: true,
duration_ns: 100,
step: None,
},
],
ops: Vec::new(),
tensors: Vec::new(),
tensor_stats: Vec::new(),
memory: Vec::new(),
device_memory: Vec::new(),
device_intervals: Vec::new(),
gradients: Vec::new(),
edges: Vec::new(),
terminal: TerminalEvent {
outcome: RunOutcome::Complete,
timestamp_ns: 200,
reason: None,
},
}
}
fn paged_tensor_stats_document() -> TraceDocument {
let mut document = trace_document();
document.tensor_stats = (0..105)
.map(|index| TensorStatsEvent {
span_id: "gpu".into(),
label: if index < 90 {
format!("bulk/{index:03}")
} else {
format!("target/{:03}", index - 90)
},
shape: vec![1],
dtype: "f32".into(),
elements: 1,
non_finite: 0,
rms: index as f64,
abs_max: index as f64,
mean: index as f64,
})
.collect();
document
}
fn label_catalog_document() -> TraceDocument {
let mut document = trace_document();
document.tensors = vec![
TensorEvent {
span_id: "gpu".into(),
tensor_id: "tensor-z-1".into(),
label: Some("tensor/z".into()),
shape: vec![1],
dtype: "f32".into(),
device: "cuda:0".into(),
requires_grad: false,
dense_bytes: Some(4),
category: MemoryCategory::Activation,
},
TensorEvent {
span_id: "gpu".into(),
tensor_id: "tensor-a".into(),
label: Some("tensor/a".into()),
shape: vec![1],
dtype: "f32".into(),
device: "cuda:0".into(),
requires_grad: false,
dense_bytes: Some(4),
category: MemoryCategory::Activation,
},
TensorEvent {
span_id: "gpu".into(),
tensor_id: "tensor-z-2".into(),
label: Some("tensor/z".into()),
shape: vec![1],
dtype: "f32".into(),
device: "cuda:0".into(),
requires_grad: false,
dense_bytes: Some(4),
category: MemoryCategory::Activation,
},
];
document.tensor_stats = ["stat/z", "stat/a", "stat/a"]
.into_iter()
.map(|label| TensorStatsEvent {
span_id: "gpu".into(),
label: label.into(),
shape: vec![1],
dtype: "f32".into(),
elements: 1,
non_finite: 0,
rms: 1.0,
abs_max: 1.0,
mean: 1.0,
})
.collect();
document.gradients = [
("gradient-z-1", "z"),
("gradient-a", "a"),
("gradient-z-2", "z"),
]
.into_iter()
.map(|(event_id, key)| GradientEvent {
event_id: event_id.into(),
root: "parameters".into(),
key: key.into(),
state: GradientState::Present,
norm: Some(1.0),
})
.collect();
document
}
fn manifest_artifact(root: &Path, path: &Path) -> ManifestArtifact {
let bytes = fs::read(path).unwrap();
ManifestArtifact {
path: path
.strip_prefix(root)
.unwrap()
.to_string_lossy()
.replace('\\', "/"),
size_bytes: bytes.len() as u64,
sha256: format!("{:x}", Sha256::digest(&bytes)),
}
}
fn publish_augmented_bundle(root: &Path) -> (PathBuf, PathBuf) {
publish_bundle_with_reports(root, true, true, true)
}
fn publish_bundle_with_reports(
root: &Path,
include_kernels: bool,
include_projection: bool,
include_gpu_timeline: bool,
) -> (PathBuf, PathBuf) {
publish_bundle_with_report_options(
root,
include_kernels,
include_projection,
include_gpu_timeline,
false,
)
}
fn publish_bundle_with_report_options(
root: &Path,
include_kernels: bool,
include_projection: bool,
include_gpu_timeline: bool,
include_broken_kernel: bool,
) -> (PathBuf, PathBuf) {
let trace = root.join("raw.jsonl");
write_jsonl(&trace, &trace_document().to_events()).unwrap();
let nsight = root.join("nsight-input");
fs::create_dir(&nsight).unwrap();
let raw = nsight.join("capture.nsys-rep");
let kernels = nsight.join("sample_cuda_gpu_kern_sum.csv");
let broken_kernel = nsight.join("broken_cuda_gpu_kern_sum.csv");
let projection = nsight.join("sample_nvtx_gpu_proj_trace.csv");
let gpu_timeline = nsight.join("sample_cuda_gpu_trace.csv");
fs::write(&raw, b"retained raw report").unwrap();
if include_kernels {
fs::write(
&kernels,
"Total Time (ns),Instances,Avg (ns),Min (ns),Max (ns),Name\n1200,2,600,500,700,gemm\n",
)
.unwrap();
}
if include_broken_kernel {
fs::write(&broken_kernel, "unsupported,columns\n1,2\n").unwrap();
}
if include_projection {
fs::write(
&projection,
"Name,Start (ns),Duration (ns),Projected Start (ns),Projected Duration (ns),Num GPU Ops,CorrId\nphase/gpu,10,100,20,80,2,1\n",
)
.unwrap();
}
if include_gpu_timeline {
fs::write(
&gpu_timeline,
"Name,Start (ns),Duration (ns),CorrId\nkernel/a,20,40,1\nkernel/b,50,20,1\n",
)
.unwrap();
}
let mut artifact_paths = vec![raw];
if include_kernels {
artifact_paths.push(kernels);
}
if include_broken_kernel {
artifact_paths.push(broken_kernel);
}
if include_projection {
artifact_paths.push(projection);
}
if include_gpu_timeline {
artifact_paths.push(gpu_timeline);
}
let artifacts = artifact_paths
.iter()
.map(|path| manifest_artifact(&nsight, path))
.collect();
let manifest = CaptureManifest {
schema: CAPTURE_MANIFEST_SCHEMA.into(),
run: CaptureRun {
id: "cli-bundle-run".into(),
started_at: None,
},
correlation: CaptureCorrelation {
id: "cli/bundle/run".into(),
},
tool: CaptureTool {
name: "nsys".into(),
version: "test".into(),
},
commands: vec!["nsys profile demo".into()],
hardware: CaptureHardware {
host: Some("test-host".into()),
devices: vec!["test-gpu".into()],
},
source_revisions: BTreeMap::new(),
required_semantic_labels: vec!["phase/gpu".into()],
gpu_expected_semantic_labels: vec!["phase/gpu".into()],
cpu_only_semantic_labels: Vec::new(),
artifacts,
};
fs::write(
nsight.join("capture-manifest.json"),
serde_json::to_vec_pretty(&manifest).unwrap(),
)
.unwrap();
let bundle = root.join("profile");
publish_bundle(&bundle, &trace, Some(&nsight)).unwrap();
(trace, bundle)
}
fn read_json(path: &Path) -> serde_json::Value {
serde_json::from_slice(&fs::read(path).unwrap()).unwrap()
}
fn run_cli_query(input: &Path, kind: &str, flags: &[&str], output: &Path) -> Output {
Command::new(env!("CARGO_BIN_EXE_candle-graph"))
.arg("query")
.arg(input)
.arg("--kind")
.arg(kind)
.args(flags)
.arg("--output")
.arg(output)
.output()
.unwrap()
}
fn write_packet_and_rebind_manifest(bundle: &Path, packet: &serde_json::Value) {
let evidence_path = bundle.join("evidence.json");
let packet_bytes = serde_json::to_vec_pretty(packet).unwrap();
fs::write(&evidence_path, &packet_bytes).unwrap();
let manifest_path = bundle.join("bundle.json");
let mut manifest: BundleManifest =
serde_json::from_slice(&fs::read(&manifest_path).unwrap()).unwrap();
let evidence = manifest
.files
.iter_mut()
.find(|file| file.path == "evidence.json")
.unwrap();
evidence.bytes = packet_bytes.len() as u64;
evidence.sha256 = format!("{:x}", Sha256::digest(&packet_bytes));
fs::write(manifest_path, serde_json::to_vec_pretty(&manifest).unwrap()).unwrap();
}
#[test]
fn bundle_root_and_bundled_trace_retain_verified_gpu_evidence() {
let root = TempRoot::new("bundle-input");
let (_, bundle) = publish_augmented_bundle(&root.0);
let bundled_trace = bundle.join("trace.jsonl");
for input in [&bundle, &bundled_trace] {
let evidence = load_evidence(input).unwrap();
assert_eq!(evidence.gpu.status, GpuEvidenceStatus::Available);
assert_eq!(
evidence.gpu.provenance.binding,
ProvenanceBindingState::Bound
);
assert!(evidence.gpu.correlation.complete);
assert_eq!(evidence.gpu.kernels[0].name, "gemm");
assert_eq!(evidence.gpu.phase_attribution[0].semantic_key, "phase/gpu");
}
}
#[test]
fn raw_trace_gpu_status_is_explicitly_unavailable() {
let root = TempRoot::new("raw-trace");
let (trace, _) = publish_augmented_bundle(&root.0);
let output = root.0.join("raw-status.json");
run_query(
&trace,
TraceQueryKind::GpuStatus,
&QueryOptions::default(),
Some(&output),
)
.unwrap();
let query = read_json(&output);
assert_eq!(query["input"]["kind"], "raw_trace");
assert_eq!(query["result"]["status"], "unavailable");
assert_eq!(
query["result"]["reason"],
"Nsight capture was not requested"
);
assert_eq!(
query["result"]["normalized_rows"]["kernels"]["status"],
"unavailable"
);
assert!(query["result"]["normalized_rows"]["kernels"]["total"].is_null());
}
#[test]
fn gpu_summary_and_queries_use_verified_bounded_bundle_evidence() {
let root = TempRoot::new("gpu-queries");
let (_, bundle) = publish_augmented_bundle(&root.0);
let summary_path = root.0.join("summary.json");
run_summary(&bundle, Some(&summary_path), false).unwrap();
let summary = read_json(&summary_path);
assert_eq!(summary["schema"], "candle-graph/summary/6");
assert_eq!(summary["input"]["kind"], "verified_bundle");
assert!(summary["input"]["gpu_identity_bound"].as_bool().unwrap());
assert_eq!(summary["gpu"]["status"], "available");
assert_eq!(summary["gpu"]["normalized_rows"]["kernels"]["total"], 1);
let cases = [
(TraceQueryKind::GpuStatus, "gpu-status"),
(TraceQueryKind::GpuCorrelation, "gpu-correlation"),
(TraceQueryKind::GpuPhases, "gpu-phases"),
(TraceQueryKind::GpuKernels, "gpu-kernels"),
(TraceQueryKind::GpuAttributionGaps, "gpu-attribution-gaps"),
];
for (kind, name) in cases {
let output = root.0.join(format!("{name}.json"));
run_query(&bundle, kind, &QueryOptions::default(), Some(&output)).unwrap();
let query = read_json(&output);
assert_eq!(query["schema"], "candle-graph/trace-query/7");
assert_eq!(query["kind"], name);
assert_eq!(query["input"]["kind"], "verified_bundle");
assert_eq!(query["result"]["status"], "available");
}
let correlation = read_json(&root.0.join("gpu-correlation.json"));
assert!(correlation["result"]["complete"].as_bool().unwrap());
assert_eq!(correlation["result"]["ledger"]["matched"]["total"], 1);
let phases = read_json(&root.0.join("gpu-phases.json"));
assert_eq!(phases["result"]["projected_ranges"]["population_total"], 1);
assert_eq!(phases["result"]["attributed_phases"]["population_total"], 1);
assert_eq!(
phases["result"]["projected_ranges"]["sample_selection"],
"earliest_original_start_rows"
);
assert_eq!(
phases["result"]["projected_ranges"]["global_duration_ranking"],
false
);
let kernels = read_json(&root.0.join("gpu-kernels.json"));
assert_eq!(kernels["result"]["rows"][0]["name"], "gemm");
let gaps = read_json(&root.0.join("gpu-attribution-gaps.json"));
assert_eq!(
gaps["result"]["matched_without_exact_gpu_busy_attribution"]["total"],
0
);
}
#[test]
fn partial_gpu_reports_remain_unknown_in_report_specific_queries() {
let kernel_root = TempRoot::new("partial-kernel");
let (_, kernel_bundle) = publish_bundle_with_reports(&kernel_root.0, true, false, false);
let status_path = kernel_root.0.join("status.json");
run_query(
&kernel_bundle,
TraceQueryKind::GpuStatus,
&QueryOptions::default(),
Some(&status_path),
)
.unwrap();
let status = read_json(&status_path);
assert_eq!(
status["result"]["normalized_rows"]["kernels"]["status"],
"available"
);
assert_eq!(status["result"]["normalized_rows"]["kernels"]["total"], 1);
assert_eq!(
status["result"]["normalized_rows"]["projected_ranges"]["status"],
"unavailable"
);
assert!(status["result"]["normalized_rows"]["projected_ranges"]["total"].is_null());
for (kind, name) in [
(TraceQueryKind::GpuCorrelation, "correlation"),
(TraceQueryKind::GpuPhases, "phases"),
(TraceQueryKind::GpuAttributionGaps, "gaps"),
] {
let output = kernel_root.0.join(format!("{name}.json"));
run_query(
&kernel_bundle,
kind,
&QueryOptions::default(),
Some(&output),
)
.unwrap();
let query = read_json(&output);
assert_eq!(query["result"]["status"], "unavailable");
assert!(query["result"]["reason"]
.as_str()
.unwrap()
.contains("unknown, not zero"));
}
let correlation = read_json(&kernel_root.0.join("correlation.json"));
assert!(correlation["result"]["complete"].is_null());
assert!(correlation["result"]["ledger"].is_null());
let gaps = read_json(&kernel_root.0.join("gaps.json"));
assert!(gaps["result"]["missing_expected"].is_null());
let projection_root = TempRoot::new("partial-projection");
let (_, projection_bundle) =
publish_bundle_with_reports(&projection_root.0, false, true, false);
let correlation_path = projection_root.0.join("correlation.json");
run_query(
&projection_bundle,
TraceQueryKind::GpuCorrelation,
&QueryOptions::default(),
Some(&correlation_path),
)
.unwrap();
let correlation = read_json(&correlation_path);
assert_eq!(correlation["result"]["status"], "available");
assert_eq!(correlation["result"]["complete"], true);
let phases_path = projection_root.0.join("phases.json");
run_query(
&projection_bundle,
TraceQueryKind::GpuPhases,
&QueryOptions::default(),
Some(&phases_path),
)
.unwrap();
let phases = read_json(&phases_path);
assert_eq!(phases["result"]["status"], "unavailable");
assert_eq!(phases["result"]["projected_ranges"]["status"], "available");
assert_eq!(phases["result"]["projected_ranges"]["population_total"], 1);
assert_eq!(
phases["result"]["attributed_phases"]["status"],
"unavailable"
);
assert!(phases["result"]["attributed_phases"]["population_total"].is_null());
let kernels_path = projection_root.0.join("kernels.json");
run_query(
&projection_bundle,
TraceQueryKind::GpuKernels,
&QueryOptions::default(),
Some(&kernels_path),
)
.unwrap();
let kernels = read_json(&kernels_path);
assert_eq!(kernels["result"]["status"], "unavailable");
assert!(kernels["result"]["population_total"].is_null());
assert_eq!(kernels["result"]["rows"].as_array().unwrap().len(), 0);
let broken_root = TempRoot::new("partial-broken-kernel");
let (_, broken_bundle) =
publish_bundle_with_report_options(&broken_root.0, true, true, true, true);
let broken_path = broken_root.0.join("kernels.json");
run_query(
&broken_bundle,
TraceQueryKind::GpuKernels,
&QueryOptions::default(),
Some(&broken_path),
)
.unwrap();
let broken = read_json(&broken_path);
assert_eq!(broken["result"]["status"], "failed");
assert!(broken["result"]["reason"]
.as_str()
.unwrap()
.contains("failed to normalize"));
assert!(broken["result"]["population_total"].is_null());
}
#[test]
fn summary_and_query_outputs_cannot_modify_a_verified_bundle() {
let root = TempRoot::new("protected-output");
let (_, bundle) = publish_augmented_bundle(&root.0);
let initial_receipt = verify_bundle(&bundle).unwrap();
let initial_evidence = fs::read(bundle.join("evidence.json")).unwrap();
let direct_overwrite = bundle.join("evidence.json");
let direct_error = run_summary(&bundle, Some(&direct_overwrite), false).unwrap_err();
assert!(direct_error.to_string().contains("inside verified bundle"));
let injected = bundle.join("injected.json");
let injection_error = run_query(
&bundle,
TraceQueryKind::GpuStatus,
&QueryOptions::default(),
Some(&injected),
)
.unwrap_err();
assert!(injection_error
.to_string()
.contains("inside verified bundle"));
assert!(!injected.exists());
let dotdot = bundle.join("nsight/../dotdot.json");
let dotdot_error = run_summary(&bundle, Some(&dotdot), false).unwrap_err();
assert!(dotdot_error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("dotdot.json").exists());
let traversing = bundle.join("new-directory/../../outside.json");
let traversing_error = run_summary(&bundle, Some(&traversing), false).unwrap_err();
assert!(traversing_error
.to_string()
.contains("inside verified bundle"));
assert!(!bundle.join("new-directory").exists());
let root_error = run_summary(&bundle, Some(&bundle), false).unwrap_err();
assert!(root_error.to_string().contains("inside verified bundle"));
assert_eq!(verify_bundle(&bundle).unwrap(), initial_receipt);
assert_eq!(
fs::read(bundle.join("evidence.json")).unwrap(),
initial_evidence
);
}
#[test]
fn import_view_compare_verify_and_report_cannot_modify_a_verified_bundle() {
use candle_graph::cli::trace_cli::{run_compare, run_import, run_report, run_verify};
let root = TempRoot::new("protected-output-all-commands");
let (trace, bundle) = publish_augmented_bundle(&root.0);
let initial_receipt = verify_bundle(&bundle).unwrap();
let import_error = run_import(&bundle, Some(&bundle.join("import.json"))).unwrap_err();
assert!(import_error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("import.json").exists());
let verify_error = run_verify(&bundle, false, Some(&bundle.join("receipt.json"))).unwrap_err();
assert!(verify_error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("receipt.json").exists());
let compare_error = run_compare(
std::slice::from_ref(&bundle),
std::slice::from_ref(&bundle),
false,
false,
Some(&bundle.join("comparison.json")),
)
.unwrap_err();
assert!(compare_error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("comparison.json").exists());
#[cfg(feature = "visualizer")]
{
use candle_graph::cli::trace_cli::run_view;
let view_error =
run_view(&bundle.join("trace.jsonl"), &bundle.join("view.html"), None).unwrap_err();
assert!(view_error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("view.html").exists());
}
let report_error = run_report(&trace, None, &bundle.join("nested-bundle"), None).unwrap_err();
assert!(report_error.to_string().contains("inside existing bundle"));
assert!(!bundle.join("nested-bundle").exists());
assert_eq!(verify_bundle(&bundle).unwrap(), initial_receipt);
}
#[cfg(unix)]
#[test]
fn output_symlink_alias_into_verified_bundle_is_rejected() {
let root = TempRoot::new("protected-output-symlink");
let (_, bundle) = publish_augmented_bundle(&root.0);
let initial_receipt = verify_bundle(&bundle).unwrap();
let alias = root.0.join("bundle-alias");
std::os::unix::fs::symlink(&bundle, &alias).unwrap();
let output = alias.join("injected.json");
let error = run_query(
&bundle,
TraceQueryKind::GpuStatus,
&QueryOptions::default(),
Some(&output),
)
.unwrap_err();
assert!(error.to_string().contains("inside verified bundle"));
assert!(!bundle.join("injected.json").exists());
assert_eq!(verify_bundle(&bundle).unwrap(), initial_receipt);
}
#[test]
fn verified_bundle_consumer_rejects_old_evidence_and_graph_schemas() {
let evidence_root = TempRoot::new("old-evidence-schema");
let (_, evidence_bundle) = publish_augmented_bundle(&evidence_root.0);
let mut packet = read_json(&evidence_bundle.join("evidence.json"));
packet["schema"] = serde_json::json!("candle-graph/evidence/4");
write_packet_and_rebind_manifest(&evidence_bundle, &packet);
verify_bundle(&evidence_bundle).unwrap();
let error = load_evidence(&evidence_bundle).unwrap_err();
assert!(format!("{error:#}").contains("unsupported evidence schema"));
let graph_root = TempRoot::new("old-graph-schema");
let (_, graph_bundle) = publish_augmented_bundle(&graph_root.0);
let mut packet = read_json(&graph_bundle.join("evidence.json"));
packet["graph"]["schema"] = serde_json::json!("candle-graph/graph/4");
write_packet_and_rebind_manifest(&graph_bundle, &packet);
verify_bundle(&graph_bundle).unwrap();
let error = load_evidence(&graph_bundle).unwrap_err();
assert!(format!("{error:#}").contains("unsupported graph schema"));
}
#[test]
fn slowest_host_query_only_reports_measured_scope_headlines() {
let root = TempRoot::new("slowest-host-scope");
let (trace, _) = publish_augmented_bundle(&root.0);
let output = root.0.join("slowest-host.json");
run_query(
&trace,
TraceQueryKind::SlowestHost,
&QueryOptions::default(),
Some(&output),
)
.unwrap();
let query = read_json(&output);
assert!(query["result"].get("slowest_host_spans").is_some());
assert!(query["result"].get("slowest_host_ops").is_none());
}
#[test]
fn activation_query_ranks_each_evidence_plane_independently() {
let root = TempRoot::new("activation-query");
let trace = root.0.join("trace.jsonl");
let mut document = trace_document();
document.run.capture_contract.operations = CoverageLevel::Complete;
document.run.capture_contract.activations = CoverageLevel::Complete;
document.run.capture_contract.tensors = CoverageLevel::Partial;
document.run.capture_contract.logical_memory = CoverageLevel::Complete;
document.run.capture_contract.device_timing = CoverageLevel::Complete;
document.spans.extend([
SpanRecord {
id: "activation-a".into(),
parent_id: Some("root".into()),
name: "host-heavy".into(),
kind: SpanKind::Op,
measured: false,
start_ns: 10,
closed: true,
duration_ns: 50,
step: None,
},
SpanRecord {
id: "activation-b".into(),
parent_id: Some("root".into()),
name: "device-heavy".into(),
kind: SpanKind::Op,
measured: false,
start_ns: 70,
closed: true,
duration_ns: 50,
step: None,
},
]);
document.ops = vec![
OpEvent {
span_id: "activation-a".into(),
op_name: "host-heavy".into(),
inputs: Vec::new(),
output: Some("ta".into()),
shape: vec![100],
dtype: "f32".into(),
device: "cuda:0".into(),
duration_ns: 40,
timestamp_ns: 10,
output_dense_bytes: None,
input_dense_bytes: 0,
},
OpEvent {
span_id: "activation-b".into(),
op_name: "device-heavy".into(),
inputs: Vec::new(),
output: Some("tb".into()),
shape: vec![200],
dtype: "f32".into(),
device: "cuda:0".into(),
duration_ns: 10,
timestamp_ns: 70,
output_dense_bytes: None,
input_dense_bytes: 0,
},
];
document.tensors = [
("activation-a", "ta", 100usize),
("activation-b", "tb", 200usize),
]
.into_iter()
.map(|(span_id, tensor_id, elements)| TensorEvent {
span_id: span_id.into(),
tensor_id: tensor_id.into(),
label: None,
shape: vec![elements],
dtype: "f32".into(),
device: "cuda:0".into(),
requires_grad: false,
dense_bytes: None,
category: MemoryCategory::Activation,
})
.collect();
let memory =
|timestamp_ns, span_id: &str, op_name: &str, tensor_id: &str, bytes, action| MemoryEvent {
timestamp_ns,
storage_id: format!("storage-{tensor_id}"),
tensor_id: tensor_id.into(),
span_id: span_id.into(),
op_name: Some(op_name.into()),
device: "cuda:0".into(),
bytes,
action,
shape: vec![bytes as usize],
dtype: "u8".into(),
category: MemoryCategory::Activation,
};
document.memory = vec![
memory(
15,
"activation-a",
"host-heavy",
"ta",
1_000,
MemoryAction::Alloc,
),
memory(
55,
"activation-a",
"host-heavy",
"ta",
1_000,
MemoryAction::Free,
),
memory(
75,
"activation-b",
"device-heavy",
"tb",
500,
MemoryAction::Alloc,
),
memory(
115,
"activation-b",
"device-heavy",
"tb",
500,
MemoryAction::Free,
),
];
document.device_intervals = vec![
DeviceIntervalEvent {
span_id: "activation-a".into(),
device: "cuda:0".into(),
stream_id: "0".into(),
clock_id: "cuda-event".into(),
backend: "cuda-event".into(),
start_ns: 0,
duration_ns: 20,
},
DeviceIntervalEvent {
span_id: "activation-b".into(),
device: "cuda:0".into(),
stream_id: "0".into(),
clock_id: "cuda-event".into(),
backend: "cuda-event".into(),
start_ns: 30,
duration_ns: 90,
},
];
write_jsonl(&trace, &document.to_events()).unwrap();
let output = root.0.join("activations.json");
run_query(
&trace,
TraceQueryKind::Activations,
&QueryOptions::default(),
Some(&output),
)
.unwrap();
let query = read_json(&output);
assert_eq!(query["schema"], "candle-graph/trace-query/7");
assert_eq!(query["kind"], "activations");
assert_eq!(
query["result"]["qualifications"]["coverage"]["level"],
"complete"
);
assert_eq!(
query["result"]["rankings"]["observed_host_duration_ns"]["rows"][0]["name"],
"host-heavy"
);
assert_eq!(
query["result"]["rankings"]["device_busy_ns_by_clock"]["rows"][0]["operation"]["name"],
"device-heavy"
);
assert_eq!(
query["result"]["rankings"]["dense_output_bytes"]["rows"][0]["name"],
"device-heavy"
);
assert_eq!(
query["result"]["rankings"]["logical_allocated_bytes"]["rows"][0]["name"],
"host-heavy"
);
assert_eq!(
query["result"]["rankings"]["logical_byte_nanoseconds"]["rows"][0]["name"],
"host-heavy"
);
assert_eq!(
query["result"]["rankings"]["logical_byte_nanoseconds"]["rows"][0]
["logical_byte_nanoseconds"],
"40000"
);
}
#[test]
fn summary_and_query_surface_ordered_tensor_stats() {
let root = TempRoot::new("tensor-stats");
let trace = root.0.join("trace.jsonl");
let mut document = trace_document();
document.tensor_stats = vec![
TensorStatsEvent {
span_id: "gpu".into(),
label: "seam/out_y".into(),
shape: vec![2, 3],
dtype: "f32".into(),
elements: 6,
non_finite: 0,
rms: 1.0,
abs_max: 2.0,
mean: 0.25,
},
TensorStatsEvent {
span_id: "gpu".into(),
label: "seam/gate_logits".into(),
shape: vec![2],
dtype: "f32".into(),
elements: 2,
non_finite: 1,
rms: 0.0,
abs_max: 0.0,
mean: 0.0,
},
];
write_jsonl(&trace, &document.to_events()).unwrap();
let summary_path = root.0.join("summary.json");
run_summary(&trace, Some(&summary_path), false).unwrap();
let summary = read_json(&summary_path);
assert_eq!(summary["tensor_stats"]["events"], 2);
assert_eq!(summary["tensor_stats"]["non_finite_events"], 1);
let query_path = root.0.join("query.json");
run_query(
&trace,
TraceQueryKind::TensorStats,
&QueryOptions::default(),
Some(&query_path),
)
.unwrap();
let query = read_json(&query_path);
assert_eq!(query["result"]["total"], 2);
assert_eq!(query["result"]["matched"], 2);
assert_eq!(query["result"]["displayed"], 2);
assert_eq!(query["result"]["offset"], 0);
assert!(query["result"]["next_offset"].is_null());
assert_eq!(query["result"]["truncated"], false);
assert_eq!(query["result"]["rows"][0]["label"], "seam/out_y");
assert_eq!(query["result"]["rows"][1]["label"], "seam/gate_logits");
}
#[test]
fn augmented_inputs_fail_closed_instead_of_falling_back_to_trace_only() {
let root = TempRoot::new("fail-closed");
let (trace, bundle) = publish_augmented_bundle(&root.0);
let evidence_path = bundle.join("evidence.json");
let mut evidence_bytes = fs::read(&evidence_path).unwrap();
let trailing_newline = evidence_bytes
.iter()
.rposition(|byte| *byte == b'\n')
.unwrap();
evidence_bytes[trailing_newline] = b' ';
fs::write(&evidence_path, evidence_bytes).unwrap();
let error = load_evidence(&bundle.join("trace.jsonl")).unwrap_err();
assert!(format!("{error:#}").contains("mismatch"));
let unverified = root.0.join("unverified-profile");
fs::create_dir(&unverified).unwrap();
fs::copy(trace, unverified.join("trace.jsonl")).unwrap();
fs::write(unverified.join("evidence.json"), b"{}").unwrap();
let error = load_evidence(&unverified.join("trace.jsonl")).unwrap_err();
assert!(error.to_string().contains("refusing to discard"));
}
#[test]
fn overview_is_bounded_and_identifies_the_tool() {
use candle_graph::cli::trace_cli::{run_overview, run_protocol};
let root = TempRoot::new("overview");
let (_, bundle) = publish_augmented_bundle(&root.0);
let output = root.0.join("overview.json");
run_overview(&bundle, Some(&output)).unwrap();
let overview = read_json(&output);
assert_eq!(overview["schema"], "candle-graph/overview/2");
assert_eq!(overview["tool"]["package"], "candle-graph");
assert_eq!(overview["tool"]["version"], env!("CARGO_PKG_VERSION"));
assert_eq!(overview["input"]["kind"], "verified_bundle");
assert!(overview["health"]["structurally_valid"].as_bool().unwrap());
assert!(overview["health"]["capture_complete"].as_bool().unwrap());
assert!(overview["health"]["error_count"].is_u64());
assert!(overview["health"]["warning_count"].is_u64());
for section in ["findings", "gaps"] {
for field in ["total", "displayed", "truncated"] {
assert!(
overview[section].get(field).is_some(),
"overview.{section}.{field} must exist"
);
}
}
assert_eq!(overview["counts"]["graph_spans"], 2);
assert_eq!(overview["timing"]["entrypoint"], "demo::update");
assert!(
overview["timing"]["slowest_host_spans"]["rows"]
.as_array()
.unwrap()
.len()
<= 5
);
assert_eq!(overview["gpu"]["status"], "available");
assert_eq!(
overview["next_queries"][0]["argv"],
serde_json::json!([
"candle-graph",
"query",
bundle.to_string_lossy(),
"--kind",
"labels"
])
);
assert_eq!(overview["next_queries"].as_array().unwrap().len(), 2);
let protocol_path = root.0.join("protocol.json");
run_protocol(Some(&protocol_path)).unwrap();
assert_eq!(
read_json(&protocol_path)["schemas"]["query"],
"candle-graph/trace-query/7"
);
let rendered = serde_json::to_string(&overview).unwrap();
assert!(!rendered.contains("parent_id"));
}
#[test]
fn query_label_prefix_filters_tensor_stats_completely() {
let root = TempRoot::new("label-prefix");
let trace = root.0.join("trace.jsonl");
let mut document = trace_document();
document.tensor_stats = vec![
TensorStatsEvent {
span_id: "gpu".into(),
label: "seam/out_y".into(),
shape: vec![2, 3],
dtype: "f32".into(),
elements: 6,
non_finite: 0,
rms: 1.0,
abs_max: 2.0,
mean: 0.25,
},
TensorStatsEvent {
span_id: "gpu".into(),
label: "seam/gate_logits".into(),
shape: vec![2],
dtype: "f32".into(),
elements: 2,
non_finite: 1,
rms: 0.0,
abs_max: 0.0,
mean: 0.0,
},
];
write_jsonl(&trace, &document.to_events()).unwrap();
let output = root.0.join("query.json");
run_query(
&trace,
TraceQueryKind::TensorStats,
&QueryOptions {
filter: Some(QueryLabelFilter::Prefix("seam/gate".into())),
..QueryOptions::default()
},
Some(&output),
)
.unwrap();
let query = read_json(&output);
assert_eq!(query["schema"], "candle-graph/trace-query/7");
assert_eq!(query["tool"]["package"], "candle-graph");
assert_eq!(query["filter"]["label_prefix"], "seam/gate");
assert!(query["filter"]["label"].is_null());
assert_eq!(query["result"]["total"], 2);
assert_eq!(query["result"]["matched"], 1);
assert_eq!(query["result"]["displayed"], 1);
assert_eq!(query["result"]["offset"], 0);
assert!(query["result"]["next_offset"].is_null());
assert_eq!(query["result"]["truncated"], false);
let rows = query["result"]["rows"].as_array().unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["label"], "seam/gate_logits");
}
#[test]
fn query_label_filter_on_unsupported_kind_errors() {
let root = TempRoot::new("label-unsupported");
let (trace, _) = publish_augmented_bundle(&root.0);
let error = run_query(
&trace,
TraceQueryKind::Memory,
&QueryOptions {
filter: Some(QueryLabelFilter::Exact("seam/out_y".into())),
..QueryOptions::default()
},
Some(&root.0.join("query.json")),
)
.unwrap_err();
assert_eq!(
error.to_string(),
"query kind memory does not support label filtering; --label and --label-prefix apply only to labels, spans, tensors, tensor-stats, and gradients"
);
assert!(!root.0.join("query.json").exists());
}
#[test]
fn cli_collection_pages_are_stable_filtered_before_paging_and_exportable() {
let root = TempRoot::new("collection-pages");
let trace = root.0.join("trace.jsonl");
write_jsonl(&trace, &paged_tensor_stats_document().to_events()).unwrap();
let bundle = root.0.join("verified");
publish_bundle(&bundle, &trace, None).unwrap();
let first_path = root.0.join("first.json");
let first_output = run_cli_query(&bundle, "tensor-stats", &[], &first_path);
assert!(first_output.status.success(), "{:?}", first_output.stderr);
let first = read_json(&first_path);
assert_eq!(first["schema"], "candle-graph/trace-query/7");
assert_eq!(first["input"]["kind"], "verified_bundle");
assert_eq!(first["result"]["total"], 105);
assert_eq!(first["result"]["matched"], 105);
assert_eq!(first["result"]["displayed"], 50);
assert_eq!(first["result"]["offset"], 0);
assert_eq!(first["result"]["next_offset"], 50);
assert_eq!(first["result"]["truncated"], true);
let second_path = root.0.join("second.json");
let second_output = run_cli_query(
&bundle,
"tensor-stats",
&["--limit", "50", "--offset", "50"],
&second_path,
);
assert!(second_output.status.success(), "{:?}", second_output.stderr);
let second = read_json(&second_path);
assert_eq!(second["result"]["offset"], 50);
assert_eq!(second["result"]["next_offset"], 100);
let first_labels = first["result"]["rows"]
.as_array()
.unwrap()
.iter()
.map(|row| row["label"].as_str().unwrap())
.collect::<BTreeSet<_>>();
let second_labels = second["result"]["rows"]
.as_array()
.unwrap()
.iter()
.map(|row| row["label"].as_str().unwrap())
.collect::<BTreeSet<_>>();
assert_eq!(first_labels.len(), 50);
assert_eq!(second_labels.len(), 50);
assert!(first_labels.is_disjoint(&second_labels));
assert_eq!(first["result"]["rows"][49]["label"], "bulk/049");
assert_eq!(second["result"]["rows"][0]["label"], "bulk/050");
let filtered_path = root.0.join("filtered.json");
let filtered_output = run_cli_query(
&bundle,
"tensor-stats",
&["--label-prefix", "target/", "--limit", "4", "--offset", "3"],
&filtered_path,
);
assert!(
filtered_output.status.success(),
"{:?}",
filtered_output.stderr
);
let filtered = read_json(&filtered_path);
assert_eq!(filtered["result"]["total"], 105);
assert_eq!(filtered["result"]["matched"], 15);
assert_eq!(filtered["result"]["displayed"], 4);
assert_eq!(filtered["result"]["offset"], 3);
assert_eq!(filtered["result"]["next_offset"], 7);
assert_eq!(filtered["result"]["rows"][0]["label"], "target/003");
assert_eq!(filtered["result"]["rows"][3]["label"], "target/006");
let empty_path = root.0.join("empty.json");
let empty_output = run_cli_query(
&bundle,
"tensor-stats",
&["--label", "missing", "--offset", "12"],
&empty_path,
);
assert!(empty_output.status.success(), "{:?}", empty_output.stderr);
let empty = read_json(&empty_path);
assert_eq!(empty["result"]["total"], 105);
assert_eq!(empty["result"]["matched"], 0);
assert_eq!(empty["result"]["displayed"], 0);
assert_eq!(empty["result"]["offset"], 12);
assert!(empty["result"]["next_offset"].is_null());
assert_eq!(empty["result"]["truncated"], false);
assert_eq!(empty["result"]["rows"], serde_json::json!([]));
let all_path = root.0.join("all.json");
let all_output = run_cli_query(&bundle, "tensor-stats", &["--all"], &all_path);
assert!(all_output.status.success(), "{:?}", all_output.stderr);
let all = read_json(&all_path);
assert_eq!(all["result"]["displayed"], 105);
assert_eq!(all["result"]["offset"], 0);
assert!(all["result"]["next_offset"].is_null());
assert_eq!(all["result"]["truncated"], false);
}
#[test]
fn cli_labels_query_is_grouped_sorted_filterable_and_uses_gradient_composites() {
let root = TempRoot::new("labels");
let trace = root.0.join("trace.jsonl");
write_jsonl(&trace, &label_catalog_document().to_events()).unwrap();
let bundle = root.0.join("verified");
publish_bundle(&bundle, &trace, None).unwrap();
let labels_path = root.0.join("labels.json");
let labels_output = run_cli_query(&bundle, "labels", &["--all"], &labels_path);
assert!(labels_output.status.success(), "{:?}", labels_output.stderr);
let labels = read_json(&labels_path);
let rows = labels["result"]["rows"].as_array().unwrap();
let actual = rows
.iter()
.map(|row| {
(
row["kind"].as_str().unwrap(),
row["label"].as_str().unwrap(),
row["events"].as_u64().unwrap(),
)
})
.collect::<Vec<_>>();
assert_eq!(
actual,
vec![
("spans", "demo::update", 1),
("spans", "phase/gpu", 1),
("spans", "tensor/a", 1),
("spans", "tensor/z", 2),
("tensors", "tensor/a", 1),
("tensors", "tensor/z", 2),
("tensor-stats", "stat/a", 2),
("tensor-stats", "stat/z", 1),
("gradients", "parameters/a", 1),
("gradients", "parameters/z", 2),
]
);
assert_eq!(labels["result"]["total"], 10);
assert_eq!(labels["result"]["matched"], 10);
for kind in ["labels", "spans", "tensors", "tensor-stats", "gradients"] {
let page_path = root.0.join(format!("{kind}-page.json"));
let page_output = run_cli_query(&bundle, kind, &["--limit", "1"], &page_path);
assert!(page_output.status.success(), "{:?}", page_output.stderr);
let page = read_json(&page_path);
for field in [
"total",
"matched",
"displayed",
"offset",
"next_offset",
"truncated",
"rows",
] {
assert!(
page["result"].get(field).is_some(),
"{kind} result is missing {field}"
);
}
assert_eq!(page["result"]["displayed"], 1);
}
let filtered_path = root.0.join("gradient-labels.json");
let filtered_output = run_cli_query(
&bundle,
"labels",
&[
"--label-prefix",
"parameters/",
"--limit",
"1",
"--offset",
"1",
],
&filtered_path,
);
assert!(
filtered_output.status.success(),
"{:?}",
filtered_output.stderr
);
let filtered = read_json(&filtered_path);
assert_eq!(filtered["result"]["total"], 10);
assert_eq!(filtered["result"]["matched"], 2);
assert_eq!(filtered["result"]["displayed"], 1);
assert_eq!(filtered["result"]["rows"][0]["label"], "parameters/z");
assert_eq!(filtered["result"]["rows"][0]["events"], 2);
}
#[test]
fn cli_collection_queries_remain_bounded_on_failed_captures() {
let root = TempRoot::new("failed-collection");
let trace = root.0.join("failed.jsonl");
let mut document = paged_tensor_stats_document();
document.terminal.outcome = RunOutcome::Failed;
document.terminal.reason = Some("intentional test failure".into());
write_jsonl(&trace, &document.to_events()).unwrap();
let stats_path = root.0.join("stats.json");
let stats_output = run_cli_query(&trace, "tensor-stats", &[], &stats_path);
assert!(stats_output.status.success(), "{:?}", stats_output.stderr);
let stats = read_json(&stats_path);
assert_eq!(
stats["capabilities"]["outer_wall_time"]["level"],
"unavailable"
);
assert_eq!(
stats["capabilities"]["outer_wall_time"]["reason"],
"capture did not complete"
);
assert_eq!(stats["result"]["total"], 105);
assert_eq!(stats["result"]["displayed"], 50);
assert_eq!(stats["result"]["truncated"], true);
let labels_path = root.0.join("labels.json");
let labels_output = run_cli_query(
&trace,
"labels",
&["--label-prefix", "target/"],
&labels_path,
);
assert!(labels_output.status.success(), "{:?}", labels_output.stderr);
let labels = read_json(&labels_path);
assert_eq!(labels["result"]["matched"], 15);
assert!(labels["result"]["rows"]
.as_array()
.unwrap()
.iter()
.all(|row| row["kind"] == "tensor-stats"));
let spans_path = root.0.join("spans.json");
let spans_output = run_cli_query(&trace, "spans", &[], &spans_path);
assert!(!spans_output.status.success());
assert!(String::from_utf8_lossy(&spans_output.stderr)
.contains("complete, structurally valid capture"));
assert!(!spans_path.exists());
}
#[test]
fn cli_span_edges_require_unfiltered_all_and_invalid_flags_are_rejected() {
let root = TempRoot::new("query-rejections");
let trace = root.0.join("trace.jsonl");
write_jsonl(&trace, &trace_document().to_events()).unwrap();
let paged_path = root.0.join("paged-spans.json");
let paged_output = run_cli_query(&trace, "spans", &[], &paged_path);
assert!(paged_output.status.success(), "{:?}", paged_output.stderr);
let paged = read_json(&paged_path);
assert!(paged["result"]["edges"].is_null());
assert!(paged["result"]["edges_reason"]
.as_str()
.unwrap()
.contains("paged or label-filtered"));
let all_path = root.0.join("all-spans.json");
let all_output = run_cli_query(&trace, "spans", &["--all"], &all_path);
assert!(all_output.status.success(), "{:?}", all_output.stderr);
let all = read_json(&all_path);
assert!(!all["result"]["edges"].as_array().unwrap().is_empty());
assert!(all["result"]["edges_reason"].is_null());
let rejected = [
("tensor-stats", vec!["--limit", "0"], "1..=1000"),
("tensors", vec!["--limit", "1001"], "1..=1000"),
(
"gradients",
vec!["--all", "--offset", "0"],
"cannot be used with",
),
(
"memory",
vec!["--offset", "0"],
"does not support pagination",
),
(
"activations",
vec!["--label", "x"],
"does not support label filtering",
),
];
for (index, (kind, flags, expected)) in rejected.into_iter().enumerate() {
let output_path = root.0.join(format!("rejected-{index}.json"));
let output = run_cli_query(&trace, kind, &flags, &output_path);
assert!(!output.status.success(), "{kind} unexpectedly succeeded");
assert!(
String::from_utf8_lossy(&output.stderr).contains(expected),
"stderr for {kind}: {}",
String::from_utf8_lossy(&output.stderr)
);
assert!(!output_path.exists());
}
}
#[test]
fn verify_envelope_reports_receipt_and_semantic_rederivation() {
use candle_graph::cli::trace_cli::run_verify;
let root = TempRoot::new("verify-envelope");
let (_, bundle) = publish_augmented_bundle(&root.0);
let plain = root.0.join("verify.json");
run_verify(&bundle, false, Some(&plain)).unwrap();
let envelope = read_json(&plain);
assert_eq!(envelope["schema"], "candle-graph/verify/1");
assert_eq!(envelope["tool"]["package"], "candle-graph");
assert_eq!(envelope["tool"]["version"], env!("CARGO_PKG_VERSION"));
assert_eq!(envelope["receipt"]["run_id"], "cli-bundle-run");
assert_eq!(
envelope["receipt"]["schema"],
"candle-graph/bundle-verification/1"
);
assert!(envelope["semantic"].is_null());
let semantic = root.0.join("verify-semantic.json");
run_verify(&bundle, true, Some(&semantic)).unwrap();
let envelope = read_json(&semantic);
assert_eq!(envelope["semantic"]["status"], "rederived_match");
}
#[cfg(feature = "visualizer")]
#[test]
fn view_rejects_external_nsight_dir_for_bundle_inputs() {
use candle_graph::cli::trace_cli::run_view;
let root = TempRoot::new("view-bundle-nsight");
let (_, bundle) = publish_augmented_bundle(&root.0);
let output = root.0.join("view.html");
let error = run_view(&bundle, &output, Some(&root.0.join("nsight-input"))).unwrap_err();
assert!(error
.to_string()
.contains("already bind their Nsight evidence"));
assert!(!output.exists());
run_view(&bundle, &output, None).unwrap();
assert!(output.exists());
}
#[test]
fn report_emits_a_publication_receipt() {
use candle_graph::cli::trace_cli::run_report;
let root = TempRoot::new("report-receipt");
let trace = root.0.join("raw.jsonl");
write_jsonl(&trace, &trace_document().to_events()).unwrap();
let bundle = root.0.join("published");
let receipt_path = root.0.join("receipt.json");
run_report(&trace, None, &bundle, Some(&receipt_path)).unwrap();
let receipt = read_json(&receipt_path);
assert_eq!(receipt["schema"], "candle-graph/publication/1");
assert_eq!(receipt["status"], "published");
assert_eq!(receipt["run_id"], "cli-bundle-run");
assert_eq!(
receipt["verification"]["schema"],
"candle-graph/bundle-verification/1"
);
assert!(bundle.join("bundle.json").is_file());
let error = run_report(
&trace,
None,
&root.0.join("second"),
Some(&root.0.join("second/receipt.json")),
)
.unwrap_err();
assert!(error.to_string().contains("inside verified bundle"));
}
#[test]
fn campaign_status_and_series_reconcile_published_bundles() {
use candle_graph::cli::trace_cli::{run_campaign_status, run_series};
let root = TempRoot::new("campaign");
for step in [1_u64, 2] {
let mut document = trace_document();
document.run.capture_step = step;
document.run.run_id = format!("cli-bundle-run-{step}");
let trace = root.0.join(format!("raw-{step}.jsonl"));
write_jsonl(&trace, &document.to_events()).unwrap();
publish_bundle(&root.0.join(format!("bundles/step-{step}")), &trace, None).unwrap();
}
let manifest = serde_json::json!({
"schema": "candle-graph/campaign/1",
"campaign_id": "demo-campaign",
"entrypoint": "demo::update",
"planned": [
{"capture_step": 1, "bundle": "bundles/step-1"},
{"capture_step": 2, "bundle": "bundles/step-2"},
{"capture_step": 3, "bundle": "bundles/step-3"},
],
});
let manifest_path = root.0.join("campaign.json");
fs::write(
&manifest_path,
serde_json::to_vec_pretty(&manifest).unwrap(),
)
.unwrap();
let status_path = root.0.join("status.json");
run_campaign_status(&manifest_path, Some(&status_path)).unwrap();
let status = read_json(&status_path);
assert_eq!(status["schema"], "candle-graph/campaign-status/1");
assert_eq!(status["published"], 2);
assert_eq!(status["missing"], 1);
let error = run_series(Some(&manifest_path), &[], None, None).unwrap_err();
let message = error.to_string();
assert!(message.contains("step 3"));
assert!(message.contains("missing"));
assert!(message.contains("campaign-status"));
let series_path = root.0.join("series.json");
run_series(
None,
&[root.0.join("bundles/step-1"), root.0.join("bundles/step-2")],
Some("seam/"),
Some(&series_path),
)
.unwrap();
let series = read_json(&series_path);
assert_eq!(series["schema"], "candle-graph/series/1");
assert_eq!(series["inputs"].as_array().unwrap().len(), 2);
assert_eq!(series["label_prefix"], "seam/");
}
#[test]
fn arbitrary_trace_filename_cannot_bypass_augmented_parent_detection() {
let root = TempRoot::new("arbitrary-name");
let (trace, _) = publish_augmented_bundle(&root.0);
let unverified = root.0.join("unverified-custom-profile");
fs::create_dir(&unverified).unwrap();
let custom_trace = unverified.join("captured-update.data");
fs::copy(trace, &custom_trace).unwrap();
fs::create_dir(unverified.join("nsight")).unwrap();
let error = load_evidence(&custom_trace).unwrap_err();
assert!(error.to_string().contains("refusing to discard"));
}