improv_engine/
external.rs

1//! Host-side evaluation of external-function measures (Phase 6).
2//!
3//! An external-function measure (`Model.external_calls[m] = ExternalCall {
4//! func, arg_measures }`) is computed *outside* the differential-dataflow graph:
5//! for each coordinate present in all argument measures, we gather the argument
6//! values and invoke the registered external function via `improv_extfn`,
7//! writing the result as an ordinary input cell.
8//!
9//! Doing this host-side (rather than inside a DD operator) keeps the engine's
10//! dataflow pure and `Send`/deterministic — external calls, which shell out to
11//! a runtime, never enter the hot recompute path. It mirrors how SQL live-query
12//! measures refresh: the function's output becomes input cells, and the engine
13//! then recomputes ordinary dependents normally. External calls are gated on an
14//! explicit refresh (they are nondeterministic sources by nature).
15
16use improv_core_model::{Coordinate, MeasureId, Model, Value};
17use std::collections::BTreeMap;
18use std::time::Duration;
19
20/// Default wall-clock deadline for a single external evaluation.
21pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
22
23#[derive(Debug, thiserror::Error)]
24pub enum ExternalError {
25    #[error("{0:?} is not an external-function measure")]
26    NotExternal(MeasureId),
27    #[error("external function {0:?} is not registered on the model")]
28    UnknownFunction(String),
29    #[error("external function {func:?} failed at {coord}: {msg}")]
30    Eval {
31        func: String,
32        coord: String,
33        msg: String,
34    },
35    #[error("external function {func:?} takes {expected} args but the measure supplies {got}")]
36    ArgCountMismatch {
37        func: String,
38        expected: usize,
39        got: usize,
40    },
41}
42
43/// Recompute an external-function measure's cells by invoking its function over
44/// its argument measures, coordinate by coordinate. Returns the number of cells
45/// written. If the function fails at any coordinate (an exception, a timeout,
46/// no usable sandbox), this returns [`ExternalError::Eval`] and leaves the
47/// measure's previous cells untouched: the results are input cells, and an
48/// error value cannot be stored as one.
49///
50/// The measure is populated over the set of coordinates for which *every*
51/// argument measure has a value (the natural join of the arguments' cells).
52pub fn refresh_external_measure(
53    model: &mut Model,
54    timeout: Duration,
55    measure: MeasureId,
56) -> Result<usize, ExternalError> {
57    let call = model
58        .external_calls
59        .get(&measure)
60        .cloned()
61        .ok_or(ExternalError::NotExternal(measure))?;
62
63    let func = model
64        .external_fns
65        .get(&call.func)
66        .cloned()
67        .ok_or_else(|| ExternalError::UnknownFunction(call.func.clone()))?;
68
69    if func.arity() != call.arg_measures.len() {
70        return Err(ExternalError::ArgCountMismatch {
71            func: call.func.clone(),
72            expected: func.arity(),
73            got: call.arg_measures.len(),
74        });
75    }
76
77    // Collect each argument measure's cells: coord -> value.
78    let arg_cells: Vec<BTreeMap<Coordinate, Value>> = call
79        .arg_measures
80        .iter()
81        .map(|mid| {
82            model
83                .inputs
84                .iter()
85                .filter(|((m, _), _)| m == mid)
86                .map(|((_, coord), v)| (coord.clone(), v.clone()))
87                .collect()
88        })
89        .collect();
90
91    // The coordinates to evaluate: those present in every argument. Use the
92    // first argument's coordinates as the candidate set (all args must agree on
93    // the value at each coordinate for a meaningful call). For 0-arg functions
94    // there is a single empty-coordinate cell.
95    let coords: Vec<Coordinate> = match arg_cells.first() {
96        Some(first) => first
97            .keys()
98            .filter(|c| arg_cells.iter().all(|m| m.contains_key(*c)))
99            .cloned()
100            .collect(),
101        None => vec![Coordinate::new()],
102    };
103
104    // Evaluate everything first: a failure leaves the old cells in place.
105    let mut results = Vec::with_capacity(coords.len());
106    for coord in coords {
107        let args: Vec<Value> = arg_cells
108            .iter()
109            .map(|m| m.get(&coord).cloned().unwrap_or(Value::Number(f64::NAN)))
110            .collect();
111        let v = improv_extfn::eval(&func, &args, timeout).map_err(|e| ExternalError::Eval {
112            func: call.func.clone(),
113            coord: coord_label(model, &coord),
114            msg: e.to_string(),
115        })?;
116        results.push((coord, v));
117    }
118    model.inputs.retain(|(m, _), _| *m != measure);
119    let count = results.len();
120    for (coord, v) in results {
121        model.set_input(measure, coord, v);
122    }
123    Ok(count)
124}
125
126/// `Product=WidgetA, Time=Q1` (item names, falling back to ids).
127fn coord_label(model: &Model, coord: &Coordinate) -> String {
128    if coord.dims.is_empty() {
129        return "the scalar cell".into();
130    }
131    coord
132        .dims
133        .iter()
134        .map(|(c, i)| {
135            let cat = model
136                .categories
137                .get(c)
138                .map_or(c.0.to_string(), |c| c.name.0.clone());
139            let item = model
140                .items
141                .get(i)
142                .map_or(i.0.to_string(), |i| i.name.0.clone());
143            format!("{cat}={item}")
144        })
145        .collect::<Vec<_>>()
146        .join(", ")
147}
148
149#[cfg(test)]
150#[cfg_attr(coverage_nightly, coverage(off))]
151mod tests {
152    use super::*;
153    use improv_core_model::{
154        CategoryId, ExternalCall, ExternalFn, ItemId, Language, Measure, MeasureKind, Name,
155        ValueType,
156    };
157
158    fn python_available() -> bool {
159        // python3 must run through the host sandbox: Restricted fails closed
160        // without a working bwrap, and the test then skips.
161        let probe = ExternalFn {
162            name: "probe".into(),
163            language: Language::Python,
164            body: "result = 1".into(),
165            arg_types: vec![],
166            return_type: ValueType::Number,
167            pure: true,
168        };
169        improv_extfn::eval(&probe, &[], std::time::Duration::from_secs(5)).is_ok()
170    }
171
172    fn base_model() -> Model {
173        let mut m = Model::new();
174        let product = CategoryId(2);
175        m.add_category(product, "Product");
176        m.add_item(ItemId(20), product, "A");
177        m.add_item(ItemId(21), product, "B");
178        // Two input measures over Product.
179        for (id, name) in [(MeasureId(100), "X"), (MeasureId(101), "Y")] {
180            m.add_measure(Measure {
181                id,
182                name: Name(name.into()),
183                value_type: ValueType::Number,
184                categories: vec![product],
185                kind: MeasureKind::Input,
186                description: None,
187            });
188        }
189        let c = |i| Coordinate::from_pairs([(product, ItemId(i))]);
190        m.set_input(MeasureId(100), c(20), Value::Number(3.0));
191        m.set_input(MeasureId(100), c(21), Value::Number(10.0));
192        m.set_input(MeasureId(101), c(20), Value::Number(4.0));
193        m.set_input(MeasureId(101), c(21), Value::Number(20.0));
194        m
195    }
196
197    #[test]
198    fn unknown_function_errors() {
199        let mut m = base_model();
200        m.external_calls.insert(
201            MeasureId(200),
202            ExternalCall {
203                func: "nope".into(),
204                arg_measures: vec![MeasureId(100)],
205                refresh_policy: Default::default(),
206            },
207        );
208
209        assert!(matches!(
210            refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(200)),
211            Err(ExternalError::UnknownFunction(_))
212        ));
213    }
214
215    #[test]
216    fn arity_mismatch_errors() {
217        let mut m = base_model();
218        m.external_fns.insert(
219            "hypot".into(),
220            ExternalFn {
221                name: "hypot".into(),
222                language: Language::Python,
223                body: "result = (args[0]**2 + args[1]**2) ** 0.5".into(),
224                arg_types: vec![ValueType::Number, ValueType::Number],
225                return_type: ValueType::Number,
226                pure: true,
227            },
228        );
229        // Supplies one arg but the function takes two.
230        m.external_calls.insert(
231            MeasureId(200),
232            ExternalCall {
233                func: "hypot".into(),
234                arg_measures: vec![MeasureId(100)],
235                refresh_policy: Default::default(),
236            },
237        );
238
239        assert!(matches!(
240            refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(200)),
241            Err(ExternalError::ArgCountMismatch { .. })
242        ));
243    }
244
245    #[test]
246    fn evaluates_hypot_over_coordinates() {
247        if !python_available() {
248            println!("skipped: python3 not found");
249            return;
250        }
251        let mut m = base_model();
252        let hypot = ExternalFn {
253            name: "hypot".into(),
254            language: Language::Python,
255            body: "result = (args[0]**2 + args[1]**2) ** 0.5".into(),
256            arg_types: vec![ValueType::Number, ValueType::Number],
257            return_type: ValueType::Number,
258            pure: true,
259        };
260        m.external_fns.insert("hypot".into(), hypot);
261        m.add_measure(Measure {
262            id: MeasureId(200),
263            name: Name("H".into()),
264            value_type: ValueType::Number,
265            categories: vec![CategoryId(2)],
266            kind: MeasureKind::Input, // populated by refresh
267            description: None,
268        });
269        m.external_calls.insert(
270            MeasureId(200),
271            ExternalCall {
272                func: "hypot".into(),
273                arg_measures: vec![MeasureId(100), MeasureId(101)],
274                refresh_policy: Default::default(),
275            },
276        );
277
278        let n = refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(200)).unwrap();
279        assert_eq!(n, 2);
280        // H[A] = hypot(3,4) = 5 ; H[B] = hypot(10,20) = ~22.36
281        let a = Coordinate::from_pairs([(CategoryId(2), ItemId(20))]);
282        assert_eq!(m.input(MeasureId(200), &a), Some(&Value::Number(5.0)));
283        let b = Coordinate::from_pairs([(CategoryId(2), ItemId(21))]);
284        match m.input(MeasureId(200), &b) {
285            Some(Value::Number(v)) => assert!((v - 500.0_f64.sqrt()).abs() < 1e-9),
286            other => panic!("expected number, got {other:?}"),
287        }
288    }
289
290    #[test]
291    fn coord_label_names_items_and_falls_back_to_ids() {
292        let m = base_model();
293        let product = CategoryId(2);
294        let c = Coordinate::from_pairs([(product, ItemId(20))]);
295        assert_eq!(coord_label(&m, &c), "Product=A");
296        let unknown = Coordinate::from_pairs([(CategoryId(9), ItemId(99)), (product, ItemId(21))]);
297        assert_eq!(coord_label(&m, &unknown), "Product=B, 9=99");
298        assert_eq!(coord_label(&m, &Coordinate::new()), "the scalar cell");
299    }
300
301    #[test]
302    fn zero_arg_function_fills_one_scalar_cell_and_a_failure_keeps_old_cells() {
303        let mut m = base_model();
304        let wasm = |name: &str, body: &str| ExternalFn {
305            name: name.into(),
306            language: Language::Wasm,
307            body: body.into(),
308            arg_types: vec![],
309            return_type: ValueType::Number,
310            pure: true,
311        };
312        m.external_fns.insert(
313            "seven".into(),
314            wasm(
315                "seven",
316                r#"wat:(module (func (export "improv_call") (result f64) f64.const 7))"#,
317            ),
318        );
319        m.external_fns
320            .insert("bad".into(), wasm("bad", "wat:(module)"));
321        for (mid, func) in [(MeasureId(300), "seven"), (MeasureId(301), "bad")] {
322            m.external_calls.insert(
323                mid,
324                ExternalCall {
325                    func: func.into(),
326                    arg_measures: vec![],
327                    refresh_policy: Default::default(),
328                },
329            );
330        }
331        assert_eq!(
332            refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(300)).unwrap(),
333            1
334        );
335        let here = Coordinate::new();
336        assert_eq!(m.input(MeasureId(300), &here), Some(&Value::Number(7.0)));
337        // A failing call is an error naming the function and cell, and the
338        // measure keeps the cells it had (here: one left from earlier).
339        m.set_input(MeasureId(301), here.clone(), Value::Number(1.0));
340        let err = refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(301)).unwrap_err();
341        assert!(
342            matches!(&err, ExternalError::Eval { func, coord, .. } if func == "bad" && coord == "the scalar cell"),
343            "{err}"
344        );
345        assert_eq!(m.input(MeasureId(301), &here), Some(&Value::Number(1.0)));
346        assert!(matches!(
347            refresh_external_measure(&mut m, DEFAULT_TIMEOUT, MeasureId(100)),
348            Err(ExternalError::NotExternal(_))
349        ));
350    }
351}
improv_engine/external.rs

improv_engine/
external.rs
