rbt/core/
run_scope.rs

1//! Run-scoped variables and partition binds (P5a).
2//!
3//! Orchestrators and CLI pass key/value pairs that:
4//! 1. Expand `{name}` / `${name}` templates in frontmatter paths and partition values
5//! 2. Merge into effective `require_partitions` for keys listed in `partition_by`
6//!
7//! Named project `roots:` still use `$root` / `${root}` expansion via [`crate::core::paths`].
8
9use anyhow::{bail, Context, Result};
10use serde::{Deserialize, Serialize};
11use std::collections::BTreeMap;
12use std::fmt;
13
14/// How a bronze scan behaves when the scan root is missing or filters match no files.
15#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum OnMissing {
18    /// Fail with `E_RBT_BRONZE_*` (default — fail-closed).
19    #[default]
20    Error,
21    /// Register an empty typed table using declared column dtypes (+ partition keys).
22    Empty,
23}
24
25impl OnMissing {
26    pub fn parse(s: &str) -> Result<Self> {
27        match s.trim().to_ascii_lowercase().as_str() {
28            "error" | "fail" | "strict" => Ok(Self::Error),
29            "empty" | "empty_frame" | "optional" => Ok(Self::Empty),
30            other => bail!(
31                "E_RBT_ON_MISSING: unknown on_missing '{other}' (expected error|empty)"
32            ),
33        }
34    }
35
36    pub fn as_str(self) -> &'static str {
37        match self {
38            Self::Error => "error",
39            Self::Empty => "empty",
40        }
41    }
42}
43
44impl fmt::Display for OnMissing {
45    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
46        f.write_str(self.as_str())
47    }
48}
49
50/// Variables and policy for one pipeline invocation (CLI / library / orchestrator).
51#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
52pub struct RunScope {
53    /// Free-form and partition binds (`report_date`, `run_id`, `domain`, …).
54    #[serde(default)]
55    pub vars: BTreeMap<String, String>,
56    /// Optional contract version stamped into fingerprints / receipts.
57    /// Overrides `rbt_project.yml` `contract_version` when set.
58    #[serde(default)]
59    pub contract_version: Option<String>,
60    /// When true, skip materialize if bronze fingerprint matches last successful receipt.
61    #[serde(default)]
62    pub skip_if_fingerprint_match: bool,
63    /// Write `.rbt/runs/{run_id}.json` receipt after run (default true for CLI run).
64    #[serde(default = "default_true")]
65    pub write_receipt: bool,
66    /// Explicit run id; generated when empty.
67    #[serde(default)]
68    pub run_id: Option<String>,
69}
70
71fn default_true() -> bool {
72    true
73}
74
75impl RunScope {
76    pub fn new() -> Self {
77        Self {
78            write_receipt: true,
79            ..Default::default()
80        }
81    }
82
83    pub fn with_var(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
84        self.vars.insert(key.into(), value.into());
85        self
86    }
87
88    /// Parse repeated `key=value` tokens (CLI `--var`).
89    pub fn extend_from_kv_pairs<'a, I>(&mut self, pairs: I) -> Result<()>
90    where
91        I: IntoIterator<Item = &'a str>,
92    {
93        for raw in pairs {
94            let raw = raw.trim();
95            if raw.is_empty() {
96                continue;
97            }
98            let (k, v) = raw
99                .split_once('=')
100                .with_context(|| {
101                    format!(
102                        "E_RBT_VAR: expected key=value, got '{raw}'. Example: --var report_date=2026-07-29"
103                    )
104                })?;
105            let k = k.trim();
106            if k.is_empty() {
107                bail!("E_RBT_VAR: empty key in '{raw}'");
108            }
109            self.vars.insert(k.to_string(), v.trim().to_string());
110        }
111        Ok(())
112    }
113
114    /// Load `RBT_VAR_<KEY>` env vars (e.g. `RBT_VAR_REPORT_DATE=2026-07-29` → `report_date`).
115    pub fn extend_from_env(&mut self) {
116        for (key, val) in std::env::vars() {
117            let Some(rest) = key.strip_prefix("RBT_VAR_") else {
118                continue;
119            };
120            if rest.is_empty() {
121                continue;
122            }
123            let var_name = rest.to_ascii_lowercase();
124            self.vars.entry(var_name).or_insert(val);
125        }
126        if let Ok(csv) = std::env::var("RBT_VARS") {
127            let _ = self.extend_from_kv_pairs(csv.split(',').map(str::trim).filter(|s| !s.is_empty()));
128        }
129    }
130
131    /// Expand `{name}` and `${name}` using run vars (not project `$roots`).
132    pub fn expand_template(&self, input: &str) -> String {
133        expand_braced_vars(input, &self.vars)
134    }
135
136    /// Merge frontmatter `require_partitions` with run vars for keys in `partition_by`.
137    ///
138    /// Run vars win over static frontmatter values for the same key.
139    /// Values are template-expanded.
140    pub fn effective_require_partitions(
141        &self,
142        partition_by: &[String],
143        frontmatter: &std::collections::HashMap<String, String>,
144    ) -> std::collections::HashMap<String, String> {
145        let mut out = std::collections::HashMap::new();
146        for (k, v) in frontmatter {
147            out.insert(k.clone(), self.expand_template(v));
148        }
149        for key in partition_by {
150            if let Some(v) = self.vars.get(key) {
151                out.insert(key.clone(), v.clone());
152            }
153        }
154        // Also apply any var that was explicitly put only in require-style names
155        // already handled via partition_by + frontmatter.
156        out
157    }
158
159    /// Stable short key for receipts scoped by vars (not including run_id).
160    pub fn scope_key(&self) -> String {
161        let mut parts: Vec<String> = self
162            .vars
163            .iter()
164            .map(|(k, v)| format!("{k}={v}"))
165            .collect();
166        parts.sort();
167        let body = parts.join("&");
168        format!("s{:016x}", fnv1a64(body.as_bytes()))
169    }
170
171    pub fn resolve_run_id(&self) -> String {
172        if let Some(id) = &self.run_id {
173            if !id.trim().is_empty() {
174                return id.trim().to_string();
175            }
176        }
177        let ts = std::time::SystemTime::now()
178            .duration_since(std::time::UNIX_EPOCH)
179            .map(|d| d.as_millis())
180            .unwrap_or(0);
181        format!("run_{ts}_{:08x}", fnv1a64(self.scope_key().as_bytes()) as u32)
182    }
183}
184
185/// Expand `{key}` and `${key}` placeholders from `vars`. Unknown keys left unchanged.
186pub fn expand_braced_vars(input: &str, vars: &BTreeMap<String, String>) -> String {
187    let mut out = String::with_capacity(input.len());
188    let bytes = input.as_bytes();
189    let mut i = 0;
190    while i < bytes.len() {
191        if bytes[i] == b'$' && i + 1 < bytes.len() && bytes[i + 1] == b'{' {
192            if let Some(end) = input[i + 2..].find('}') {
193                let key = &input[i + 2..i + 2 + end];
194                if let Some(v) = vars.get(key) {
195                    out.push_str(v);
196                } else {
197                    out.push_str(&input[i..i + 3 + end]);
198                }
199                i += 3 + end;
200                continue;
201            }
202        }
203        if bytes[i] == b'{' {
204            if let Some(end) = input[i + 1..].find('}') {
205                let key = &input[i + 1..i + 1 + end];
206                // Avoid matching JSON-like braces with spaces / operators
207                if is_simple_ident(key) {
208                    if let Some(v) = vars.get(key) {
209                        out.push_str(v);
210                        i += 2 + end;
211                        continue;
212                    }
213                }
214            }
215        }
216        out.push(bytes[i] as char);
217        i += 1;
218    }
219    out
220}
221
222fn is_simple_ident(s: &str) -> bool {
223    let mut chars = s.chars();
224    match chars.next() {
225        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
226        _ => return false,
227    }
228    chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
229}
230
231/// Stable 64-bit FNV-1a (not crypto; fine for fingerprint / scope keys).
232pub fn fnv1a64(data: &[u8]) -> u64 {
233    const OFFSET: u64 = 0xcbf29ce484222325;
234    const PRIME: u64 = 0x100000001b3;
235    let mut hash = OFFSET;
236    for b in data {
237        hash ^= u64::from(*b);
238        hash = hash.wrapping_mul(PRIME);
239    }
240    hash
241}
242
243#[cfg(test)]
244mod tests {
245    use super::*;
246
247    #[test]
248    fn expand_braced_and_dollar() {
249        let mut vars = BTreeMap::new();
250        vars.insert("report_date".into(), "2026-07-29".into());
251        vars.insert("domain".into(), "acme.com".into());
252        assert_eq!(
253            expand_braced_vars("d={domain}/dt={report_date}", &vars),
254            "d=acme.com/dt=2026-07-29"
255        );
256        assert_eq!(
257            expand_braced_vars("x=${report_date}/y", &vars),
258            "x=2026-07-29/y"
259        );
260    }
261
262    #[test]
263    fn merge_partitions_run_wins() {
264        let scope = RunScope::new()
265            .with_var("report_date", "2026-07-29")
266            .with_var("domain", "acme.com");
267        let mut fm = std::collections::HashMap::new();
268        fm.insert("report_date".into(), "static".into());
269        let pb = vec!["domain".into(), "report_date".into(), "run_id".into()];
270        let eff = scope.effective_require_partitions(&pb, &fm);
271        assert_eq!(eff.get("report_date").map(String::as_str), Some("2026-07-29"));
272        assert_eq!(eff.get("domain").map(String::as_str), Some("acme.com"));
273        assert!(!eff.contains_key("run_id"));
274    }
275
276    #[test]
277    fn parse_kv_pairs() {
278        let mut s = RunScope::new();
279        s.extend_from_kv_pairs(["report_date=2026-07-29", "run_id=r1"])
280            .unwrap();
281        assert_eq!(s.vars.get("run_id").map(String::as_str), Some("r1"));
282    }
283}
rbt/core/run_scope.rs

rbt/core/
run_scope.rs
