fv_compute/
manifest.rs

1//! The `transform.toml` manifest — the single self-describing unit the registry
2//! discovers. This crate defines + validates it (the publish gate); the registry discovers directories
3//! of them. Every IMPL (builtin/expression/wasm/container/llm) uses this one shape.
4
5use crate::capability::CapabilityEnvelope;
6use crate::types::SchemaSpec;
7use serde::{Deserialize, Serialize};
8use std::collections::BTreeMap;
9
10/// The IMPL axis: which backend runs the transform.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
12#[serde(rename_all = "snake_case")]
13pub enum ImplKind {
14    /// A native primitive (fv-value / a structural step-kind); `ref` points at it. NOT re-authored
15    /// as a slow plugin — the manifest + golden vectors are co-located, the code stays native.
16    Builtin,
17    /// A value-dialect payload (`entry` → `expr.fvx`).
18    Expression,
19    /// A WebAssembly component (`entry` → `main.wasm`) — the flagship polyglot backend.
20    Wasm,
21    /// A container image / process speaking the stdio protocol (`entry` → image ref/Dockerfile).
22    Container,
23    /// An ONNX model artifact (`entry` → `model.onnx`), run by the native `fv-infer` backend
24    /// A model is just a transform: inputs = feature columns, output = the
25    /// prediction column(s). Pure/deterministic/no-io by default, so it is a legal derived/stream
26    /// compute — the "model as a pipeline node" shape, pluggable via this same registry.
27    Onnx,
28    /// An LLM-backed compute (nondeterministic/effectful — Action bindings only).
29    Llm,
30}
31
32/// The wire protocol a `container` transform speaks over stdio (ignored by other impls).
33/// `arrow-ipc` is the fast Arrow handoff; `json` is the stdlib-friendly row-JSON contract
34/// (no Arrow dependency in the transform — works under the stripped-env isolation).
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
36#[serde(rename_all = "kebab-case")]
37pub enum ContainerProtocol {
38    #[default]
39    ArrowIpc,
40    Json,
41}
42
43/// How output-dataset labels are derived (label propagation, fail-closed).
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
45#[serde(rename_all = "snake_case")]
46pub enum LabelsRule {
47    /// Output inherits the union of the inputs' labels regardless of what the transform does.
48    #[default]
49    Propagate,
50}
51
52#[derive(Debug, Clone, Serialize, Deserialize, Default)]
53pub struct Labels {
54    #[serde(default)]
55    pub rule: LabelsRule,
56}
57
58/// A parsed, not-yet-validated `transform.toml`. Call [`TransformManifest::validate`] (or
59/// [`parse_and_validate`]) before trusting it — an unvalidated manifest never executes.
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct TransformManifest {
62    /// Stable identifier (camelCase), unique within a registry root.
63    pub id: String,
64    /// Semver — enables `id@version` selection + hot-swap.
65    pub version: String,
66    #[serde(rename = "impl")]
67    pub impl_kind: ImplKind,
68    /// For `builtin`: the native impl this points at (e.g. "haversineKm").
69    #[serde(default, rename = "ref")]
70    pub reference: Option<String>,
71    /// For `wasm`/`container`/`expression`: the built artifact / payload path, relative to the dir.
72    #[serde(default)]
73    pub entry: Option<String>,
74    /// Input signature(s). Empty for a pure generator; >1 for a JOIN.
75    #[serde(default)]
76    pub inputs: Vec<SchemaSpec>,
77    /// The single output signature.
78    pub output: SchemaSpec,
79    #[serde(default)]
80    pub capabilities: CapabilityEnvelope,
81    /// output column -> source column(s). Bare `col` for single-input; qualified `input.col` when
82    /// there is >1 input. Undeclared = opaque (dataset-level edge), like a raw sql step.
83    #[serde(default, rename = "columnLineage")]
84    pub column_lineage: BTreeMap<String, Vec<String>>,
85    #[serde(default)]
86    pub labels: Labels,
87    /// For `impl = "container"`: the stdio wire protocol. Default `arrow-ipc`.
88    #[serde(default)]
89    pub protocol: ContainerProtocol,
90}
91
92#[derive(Debug, thiserror::Error)]
93pub enum ManifestError {
94    // Boxed: toml::de::Error is large, and this is a cold error path — keeps the Ok path small
95    // (clippy::result_large_err).
96    #[error("TOML parse error: {0}")]
97    Toml(Box<toml::de::Error>),
98    #[error("invalid manifest `{id}`: {msg}")]
99    Invalid { id: String, msg: String },
100}
101
102impl TransformManifest {
103    /// Parse from a `transform.toml` string (does NOT validate).
104    pub fn from_toml_str(s: &str) -> Result<Self, ManifestError> {
105        toml::from_str(s).map_err(|e| ManifestError::Toml(Box::new(e)))
106    }
107
108    fn invalid(&self, msg: impl Into<String>) -> ManifestError {
109        ManifestError::Invalid {
110            id: self.id.clone(),
111            msg: msg.into(),
112        }
113    }
114
115    /// The publish gate: reject anything malformed before it can execute.
116    pub fn validate(&self) -> Result<(), ManifestError> {
117        // id
118        if self.id.is_empty() || !is_ident(&self.id) {
119            return Err(self.invalid("id must be a non-empty identifier (alnum, starting with a letter)"));
120        }
121        // version
122        if semver::Version::parse(&self.version).is_err() {
123            return Err(self.invalid(format!("version `{}` is not valid semver", self.version)));
124        }
125        // impl-specific artifact presence
126        match self.impl_kind {
127            ImplKind::Builtin => {
128                if self.entry.is_some() {
129                    return Err(self.invalid("builtin must not set `entry` (it points at native code via `ref`)"));
130                }
131                // `ref` defaults to `id` if omitted.
132            }
133            ImplKind::Expression | ImplKind::Wasm | ImplKind::Container | ImplKind::Onnx => {
134                if self.entry.as_deref().unwrap_or("").is_empty() {
135                    return Err(self.invalid("this impl requires a non-empty `entry` (the artifact/payload path)"));
136                }
137                if self.reference.is_some() {
138                    return Err(self.invalid("`ref` is only for builtin impls"));
139                }
140            }
141            ImplKind::Llm => { /* entry/config are impl-defined; nothing structural to enforce yet */ }
142        }
143        // output must declare at least one column
144        if self.output.columns.is_empty() {
145            return Err(self.invalid("output must declare at least one column"));
146        }
147        // multi-input datasets must be named + distinct (so qualified lineage resolves)
148        if self.inputs.len() > 1 {
149            let mut seen = std::collections::HashSet::new();
150            for inp in &self.inputs {
151                match &inp.name {
152                    None => return Err(self.invalid("every input must be named when there is >1 input")),
153                    Some(n) if !seen.insert(n.clone()) => {
154                        return Err(self.invalid(format!("duplicate input name `{n}`")));
155                    }
156                    _ => {}
157                }
158            }
159        }
160        // column lineage integrity
161        self.validate_lineage()?;
162        Ok(())
163    }
164
165    fn validate_lineage(&self) -> Result<(), ManifestError> {
166        let multi = self.inputs.len() > 1;
167        for (out_col, sources) in &self.column_lineage {
168            if !self.output.has_column(out_col) {
169                return Err(self.invalid(format!("columnLineage references unknown output column `{out_col}`")));
170            }
171            if sources.is_empty() {
172                return Err(self.invalid(format!("columnLineage for `{out_col}` has no sources")));
173            }
174            for src in sources {
175                if multi {
176                    // qualified `input.col` — the input must exist and carry the column
177                    let (inp, col) = src.split_once('.').ok_or_else(|| {
178                        self.invalid(format!(
179                            "multi-input lineage source `{src}` must be qualified as `input.column`"
180                        ))
181                    })?;
182                    let found = self
183                        .inputs
184                        .iter()
185                        .find(|i| i.name.as_deref() == Some(inp))
186                        .ok_or_else(|| self.invalid(format!("lineage source references unknown input `{inp}`")))?;
187                    if !found.has_column(col) {
188                        return Err(
189                            self.invalid(format!("input `{inp}` has no column `{col}` (lineage for `{out_col}`)"))
190                        );
191                    }
192                } else {
193                    // single input: bare column name must exist in the (sole) input, if declared
194                    if let Some(inp) = self.inputs.first() {
195                        if !inp.has_column(src) {
196                            return Err(self.invalid(format!(
197                                "lineage source `{src}` is not a column of the input (for `{out_col}`)"
198                            )));
199                        }
200                    }
201                }
202            }
203        }
204        Ok(())
205    }
206
207    /// The `ref` a builtin resolves to (defaults to `id`).
208    pub fn builtin_ref(&self) -> &str {
209        self.reference.as_deref().unwrap_or(&self.id)
210    }
211}
212
213/// Parse + validate in one step — the normal entry point.
214pub fn parse_and_validate(s: &str) -> Result<TransformManifest, ManifestError> {
215    let m = TransformManifest::from_toml_str(s)?;
216    m.validate()?;
217    Ok(m)
218}
219
220/// An identifier: starts with an ASCII letter, then alnum. (camelCase ids like `haversineKm`.)
221fn is_ident(s: &str) -> bool {
222    let mut chars = s.chars();
223    match chars.next() {
224        Some(c) if c.is_ascii_alphabetic() => {}
225        _ => return false,
226    }
227    s.chars().all(|c| c.is_ascii_alphanumeric())
228}
229
230#[cfg(test)]
231mod tests {
232    use super::*;
233
234    #[test]
235    fn is_ident_rules() {
236        assert!(is_ident("haversineKm"));
237        assert!(is_ident("rename"));
238        assert!(!is_ident("2cool"));
239        assert!(!is_ident("has space"));
240        assert!(!is_ident("snake_case"));
241        assert!(!is_ident(""));
242    }
243
244    #[test]
245    fn builtin_ref_defaults_to_id() {
246        let m = parse_and_validate(
247            r#"
248            id = "haversineKm"
249            version = "0.1.0"
250            impl = "builtin"
251            [output]
252            columns = [{ name = "km", type = "float64" }]
253            "#,
254        )
255        .unwrap();
256        assert_eq!(m.builtin_ref(), "haversineKm");
257    }
258}
fv_compute/manifest.rs

fv_compute/
manifest.rs
