rbt/core/
receipt.rs

1//! Run receipts and bronze content fingerprints (P5b).
2//!
3//! A receipt is the **consumer authority** for “did this scope succeed?” — not “file exists”.
4//! Fingerprints cover the filtered bronze file set + contract version so identical re-drives
5//! can skip materialize when `skip_if_fingerprint_match` is set.
6
7use crate::core::dag::ModelDag;
8use crate::core::frontmatter::StagingFrontmatter;
9use crate::core::project::RbtProjectConfig;
10use crate::core::run_scope::{fnv1a64, RunScope};
11use crate::scan::{LakeScanner, ScanRequest};
12use anyhow::{Context, Result};
13use serde::{Deserialize, Serialize};
14use std::collections::BTreeMap;
15use std::fs;
16use std::path::{Path, PathBuf};
17use std::time::{SystemTime, UNIX_EPOCH};
18
19/// Machine-readable result of one `rbt run` (or library execute with receipts enabled).
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
21pub struct RunReceipt {
22    pub schema_version: u32,
23    pub run_id: String,
24    pub project: String,
25    pub package_version: String,
26    /// Project or CLI contract version used for skip decisions.
27    pub contract_version: String,
28    pub scope_key: String,
29    pub vars: BTreeMap<String, String>,
30    pub status: RunStatus,
31    pub skipped: bool,
32    pub skip_reason: Option<String>,
33    pub bronze_fingerprint: String,
34    pub models_executed: usize,
35    pub total_rows: usize,
36    pub bronze_sources: usize,
37    pub model_results: Vec<ModelRunResult>,
38    pub started_unix_ms: u128,
39    pub finished_unix_ms: u128,
40    pub wall_ms: u128,
41    pub error: Option<String>,
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
45#[serde(rename_all = "snake_case")]
46pub enum RunStatus {
47    Ok,
48    Skipped,
49    Error,
50}
51
52#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
53pub struct ModelRunResult {
54    pub name: String,
55    pub rows: usize,
56    pub output_path: Option<String>,
57}
58
59impl RunReceipt {
60    pub const SCHEMA_VERSION: u32 = 1;
61
62    pub fn receipt_dir(project_dir: &Path) -> PathBuf {
63        project_dir.join(".rbt").join("runs")
64    }
65
66    pub fn path_for(project_dir: &Path, run_id: &str) -> PathBuf {
67        Self::receipt_dir(project_dir).join(format!("{run_id}.json"))
68    }
69
70    /// Latest successful receipt for this scope (for skip-if-match).
71    pub fn latest_path_for_scope(project_dir: &Path, scope_key: &str) -> PathBuf {
72        Self::receipt_dir(project_dir).join(format!("latest_{scope_key}.json"))
73    }
74
75    pub fn write(&self, project_dir: &Path) -> Result<PathBuf> {
76        let dir = Self::receipt_dir(project_dir);
77        fs::create_dir_all(&dir)?;
78        let path = Self::path_for(project_dir, &self.run_id);
79        let body = serde_json::to_string_pretty(self)
80            .context("E_RBT_RECEIPT: serialize RunReceipt")?;
81        fs::write(&path, body).with_context(|| {
82            format!("E_RBT_RECEIPT: write {}", path.display())
83        })?;
84        if matches!(self.status, RunStatus::Ok | RunStatus::Skipped) {
85            let latest = Self::latest_path_for_scope(project_dir, &self.scope_key);
86            fs::write(&latest, serde_json::to_string_pretty(self)?)?;
87        }
88        Ok(path)
89    }
90
91    pub fn load(path: &Path) -> Result<Self> {
92        let raw = fs::read_to_string(path)
93            .with_context(|| format!("E_RBT_RECEIPT: read {}", path.display()))?;
94        serde_json::from_str(&raw)
95            .with_context(|| format!("E_RBT_RECEIPT: parse {}", path.display()))
96    }
97
98    pub fn load_latest_for_scope(project_dir: &Path, scope_key: &str) -> Option<Self> {
99        let p = Self::latest_path_for_scope(project_dir, scope_key);
100        Self::load(&p).ok()
101    }
102}
103
104/// Effective contract version: run scope override → project yml → `"0"`.
105pub fn effective_contract_version(config: &RbtProjectConfig, scope: &RunScope) -> String {
106    scope
107        .contract_version
108        .clone()
109        .or_else(|| config.contract_version.clone())
110        .filter(|s| !s.trim().is_empty())
111        .unwrap_or_else(|| "0".into())
112}
113
114/// Build a stable bronze fingerprint for the DAG under this scope.
115///
116/// Manifest lines: `model\\tschema.table\\trel_or_abs\\tsize\\tmtime_secs` sorted, plus
117/// `contract_version=…`. Fingerprint is `fnv1a64:` + hex of FNV over the manifest.
118pub fn bronze_fingerprint(
119    dag: &ModelDag,
120    project_dir: &Path,
121    config: &RbtProjectConfig,
122    scope: &RunScope,
123) -> Result<String> {
124    let contract = effective_contract_version(config, scope);
125    let mut lines: Vec<String> = Vec::new();
126    lines.push(format!("contract_version={contract}"));
127
128    for idx in dag.graph.node_indices() {
129        let node = &dag.graph[idx];
130        let Some(fm) = node.frontmatter.as_ref() else {
131            continue;
132        };
133        if !fm.has_scan_contract() {
134            continue;
135        }
136        let fm_eff = apply_scope_to_frontmatter(fm, scope);
137        let mut req = ScanRequest::from_frontmatter_with_config(
138            project_dir,
139            &fm_eff,
140            config.roots.clone(),
141            &config.scan,
142        )?;
143        // Soft list: missing/empty contributes a marker line, not an error.
144        req.allow_empty = true;
145        let scanner = LakeScanner::from_request(&req);
146        match scanner.list_files(&req) {
147            Ok((root, files)) => {
148                if files.is_empty() {
149                    lines.push(format!(
150                        "{}\tEMPTY\t{}",
151                        node.name,
152                        root.display()
153                    ));
154                } else {
155                    for f in files {
156                        let meta = fs::metadata(&f).ok();
157                        let size = meta.as_ref().map(|m| m.len()).unwrap_or(0);
158                        let mtime = meta
159                            .and_then(|m| m.modified().ok())
160                            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
161                            .map(|d| d.as_secs())
162                            .unwrap_or(0);
163                        let rel = f
164                            .strip_prefix(&root)
165                            .map(|p| p.display().to_string())
166                            .unwrap_or_else(|_| f.display().to_string());
167                        lines.push(format!(
168                            "{}\t{}\t{}\t{}",
169                            node.name, rel, size, mtime
170                        ));
171                    }
172                }
173            }
174            Err(_) => {
175                lines.push(format!("{}\tMISSING\t{}", node.name, req.scan_path));
176            }
177        }
178    }
179
180    lines.sort();
181    let manifest = lines.join("\n");
182    let digest = fnv1a64(manifest.as_bytes());
183    Ok(format!("fnv1a64:{digest:016x}"))
184}
185
186/// Clone frontmatter with run-scope templates and effective require_partitions applied.
187pub fn apply_scope_to_frontmatter(fm: &StagingFrontmatter, scope: &RunScope) -> StagingFrontmatter {
188    let mut out = fm.clone();
189    if let Some(sp) = out.scan_path.as_ref() {
190        out.scan_path = Some(scope.expand_template(sp));
191    }
192    if let Some(globs) = out.path_glob.as_ref() {
193        out.path_glob = Some(
194            globs
195                .iter()
196                .map(|g| scope.expand_template(g))
197                .collect(),
198        );
199    }
200    let partition_by = out.partition_by.clone().unwrap_or_default();
201    let base = out.require_partitions.clone().unwrap_or_default();
202    let eff = scope.effective_require_partitions(&partition_by, &base);
203    if !eff.is_empty() {
204        out.require_partitions = Some(eff);
205    }
206    out
207}
208
209pub fn now_unix_ms() -> u128 {
210    SystemTime::now()
211        .duration_since(UNIX_EPOCH)
212        .map(|d| d.as_millis())
213        .unwrap_or(0)
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219    use crate::core::run_scope::RunScope;
220
221    #[test]
222    fn receipt_roundtrip() {
223        let dir = tempfile::tempdir().unwrap();
224        let r = RunReceipt {
225            schema_version: 1,
226            run_id: "run_test".into(),
227            project: "p".into(),
228            package_version: "0.6.0".into(),
229            contract_version: "1".into(),
230            scope_key: "s0".into(),
231            vars: BTreeMap::new(),
232            status: RunStatus::Ok,
233            skipped: false,
234            skip_reason: None,
235            bronze_fingerprint: "fnv1a64:00".into(),
236            models_executed: 1,
237            total_rows: 2,
238            bronze_sources: 1,
239            model_results: vec![],
240            started_unix_ms: 1,
241            finished_unix_ms: 2,
242            wall_ms: 1,
243            error: None,
244        };
245        let p = r.write(dir.path()).unwrap();
246        let loaded = RunReceipt::load(&p).unwrap();
247        assert_eq!(loaded.run_id, "run_test");
248        assert!(RunReceipt::latest_path_for_scope(dir.path(), "s0").exists());
249    }
250
251    #[test]
252    fn apply_scope_expands_scan_path() {
253        let fm = StagingFrontmatter {
254            scan_path: Some("lake/{domain}/raw".into()),
255            partition_by: Some(vec!["report_date".into()]),
256            require_partitions: Some({
257                let mut m = std::collections::HashMap::new();
258                m.insert("report_date".into(), "{report_date}".into());
259                m
260            }),
261            ..Default::default()
262        };
263        let scope = RunScope::new()
264            .with_var("domain", "acme")
265            .with_var("report_date", "2026-07-29");
266        let e = apply_scope_to_frontmatter(&fm, &scope);
267        assert_eq!(e.scan_path.as_deref(), Some("lake/acme/raw"));
268        assert_eq!(
269            e.require_partitions
270                .as_ref()
271                .and_then(|m| m.get("report_date"))
272                .map(String::as_str),
273            Some("2026-07-29")
274        );
275    }
276}
rbt/core/receipt.rs

rbt/core/
receipt.rs
