Skip to main content

dataflow_rs/engine/
task_context.rs

1//! # Task context
2//!
3//! Wraps the per-call state passed to every `AsyncFunctionHandler::execute`
4//! call: the message under processing, a handle to the shared datalogic
5//! engine, and an audit-trail accumulator. Exposes typed helpers so handlers
6//! don't have to reach into `crate::engine::utils::{get,set}_nested_value`
7//! or hand-build `Change` entries.
8//!
9//! Custom handlers should treat `TaskContext` as their *only* mutation surface
10//! for `message.context`: the `set` family records a `Change` automatically
11//! when `message.capture_changes` is true, keeping the audit trail in sync
12//! with the data without per-handler boilerplate.
13
14use crate::engine::error::{ErrorInfo, Result};
15use crate::engine::message::{Change, Message};
16use crate::engine::secrets::{self, Secrets};
17use crate::engine::utils::{get_nested_value, set_nested_value};
18use bumpalo::Bump;
19use datalogic_rs::{Engine as DatalogicEngine, Logic};
20use datavalue::{DataValue, OwnedDataValue};
21use serde_json::Value as JsonValue;
22use std::sync::Arc;
23
24/// Per-call execution context handed to `AsyncFunctionHandler::execute`.
25///
26/// Borrows the message and datalogic engine for the duration of the handler
27/// call; collects `Change` entries that the workflow executor folds into the
28/// audit trail when the handler returns. Drop semantics are trivial — there
29/// is nothing to flush; the executor extracts the buffered changes via
30/// `into_changes()`.
31pub struct TaskContext<'a> {
32    message: &'a mut Message,
33    datalogic: &'a Arc<DatalogicEngine>,
34    /// Changes accumulated through the `set*` family. Only populated when
35    /// `message.capture_changes` is true; otherwise pushes are no-ops to
36    /// keep the bulk-pipeline fast path allocation-free.
37    changes: Vec<Change>,
38    /// Who is executing, when the engine built this context.
39    ///
40    /// Borrowed rather than `Arc`-cloned: the ids live on the `Workflow` and
41    /// `Task`, both of which outlive the dispatch call, and the accessors hand
42    /// back `&str` either way — so a refcount bump per task would buy nothing.
43    identity: Option<TaskIdentity<'a>>,
44    /// Sweep index of the enclosing looping workflow, if any.
45    loop_counter: Option<i64>,
46    /// Fan-out element this call runs for, if the task carries `for_each`.
47    element_index: Option<usize>,
48    /// The engine's secret store — empty for a context built with
49    /// [`Self::new`], for the same reason `identity` is `None` there.
50    secrets: &'a Secrets,
51}
52
53/// Where a handler call sits in a run: the loop sweep and the fan-out
54/// element, either of which may be absent.
55///
56/// One parameter rather than two, so the dispatch path stays under clippy's
57/// argument-count limit.
58#[derive(Debug, Clone, Copy, Default)]
59pub(crate) struct CallStamp {
60    pub loop_counter: Option<i64>,
61    pub element_index: Option<usize>,
62}
63
64/// Which task, in which workflow, the engine is currently running.
65///
66/// All-or-nothing by construction: the engine executing a task inside a
67/// workflow knows both ids, and every other path knows neither. Two separate
68/// `Option<&str>` fields would additionally allow "workflow known, task
69/// unknown" — a state that never occurs — and would be silently swappable at
70/// the call site, being adjacent and identically typed.
71#[derive(Debug, Clone, Copy)]
72pub(crate) struct TaskIdentity<'a> {
73    pub workflow_id: &'a str,
74    pub task_id: &'a str,
75}
76
77impl<'a> TaskContext<'a> {
78    /// Construct a new context. Mostly engine-internal — handlers receive a
79    /// pre-built `&mut TaskContext` from the executor — but exposed `pub` so
80    /// tests and benchmarks can drive `AsyncFunctionHandler::execute`
81    /// directly without going through `Engine::process_message`.
82    /// A context built this way reports `None` from [`Self::workflow_id`],
83    /// [`Self::task_id`] and [`Self::loop_counter`] — there is no workflow run
84    /// to describe, and inventing ids would be worse than admitting their
85    /// absence.
86    pub fn new(message: &'a mut Message, datalogic: &'a Arc<DatalogicEngine>) -> Self {
87        Self {
88            message,
89            datalogic,
90            changes: Vec::new(),
91            identity: None,
92            loop_counter: None,
93            element_index: None,
94            secrets: &secrets::EMPTY,
95        }
96    }
97
98    /// As [`Self::new`], with the identity of the executing task.
99    ///
100    /// Used by the task executor on the dispatch path. A separate constructor
101    /// rather than setters, so there is no window in which a context exists
102    /// with half its identity filled in.
103    pub(crate) fn with_identity(
104        message: &'a mut Message,
105        datalogic: &'a Arc<DatalogicEngine>,
106        identity: Option<TaskIdentity<'a>>,
107        stamp: CallStamp,
108        secrets: &'a Secrets,
109    ) -> Self {
110        Self {
111            message,
112            datalogic,
113            changes: Vec::new(),
114            identity,
115            loop_counter: stamp.loop_counter,
116            element_index: stamp.element_index,
117            secrets,
118        }
119    }
120
121    /// Id of the workflow being executed, when the engine built this context.
122    ///
123    /// `None` for a context built with [`Self::new`] — a test or benchmark
124    /// driving a handler directly is not inside a workflow run.
125    ///
126    /// ```
127    /// # use dataflow_rs::{TaskContext, engine::message::Message};
128    /// # use serde_json::json;
129    /// # let datalogic = std::sync::Arc::new(datalogic_rs::Engine::new());
130    /// # let mut message = Message::from_value(&json!({}));
131    /// let ctx = TaskContext::new(&mut message, &datalogic);
132    /// assert_eq!(ctx.workflow_id(), None);
133    /// ```
134    #[inline]
135    pub fn workflow_id(&self) -> Option<&str> {
136        self.identity.map(|i| i.workflow_id)
137    }
138
139    /// Id of the task being executed, when the engine built this context.
140    ///
141    /// Always a **leaf** task's id. Handlers run only on leaf tasks — a task
142    /// group is span bookkeeping recorded on the task that opens it, never a
143    /// dispatch target — so a group id can never appear here.
144    ///
145    /// `None` for a context built with [`Self::new`].
146    #[inline]
147    pub fn task_id(&self) -> Option<&str> {
148        self.identity.map(|i| i.task_id)
149    }
150
151    /// Sweep index of the enclosing looping workflow, or `None` when the
152    /// workflow does not carry a `loop`.
153    ///
154    /// This is a different fact from identity being unknown: a handler in a
155    /// non-looping workflow has both ids and no counter.
156    ///
157    /// Worth preferring over reading the counter out of `temp_data`, which
158    /// only works when the host gave `LoopConfig` a `counter` name and the
159    /// handler hardcodes that path. A loop with no named counter writes to no
160    /// path at all, and its sweep index is reachable no other way.
161    #[inline]
162    pub fn loop_counter(&self) -> Option<i64> {
163        self.loop_counter
164    }
165
166    /// Index of the element this call runs for, when the task carries a
167    /// [`for_each`](crate::engine::for_each::ForEach); `None` otherwise.
168    ///
169    /// The same value the engine binds at `temp_data.<as>_index` in this
170    /// call's view of the message, without the handler having to know the
171    /// `as` name.
172    ///
173    /// ```
174    /// # use dataflow_rs::{TaskContext, engine::message::Message};
175    /// # use serde_json::json;
176    /// # let datalogic = std::sync::Arc::new(datalogic_rs::Engine::new());
177    /// # let mut message = Message::from_value(&json!({}));
178    /// let ctx = TaskContext::new(&mut message, &datalogic);
179    /// assert_eq!(ctx.element_index(), None);
180    /// ```
181    #[inline]
182    pub fn element_index(&self) -> Option<usize> {
183        self.element_index
184    }
185
186    /// A secret by dotted name, from the store the host configured through
187    /// [`crate::EngineBuilder::with_secrets`].
188    ///
189    /// For handlers whose config names a key (`"key_name": "partner_hmac"`)
190    /// rather than embedding a [`crate::Template`] — a `Template` field can
191    /// simply read `{"secret": "partner_hmac"}` and needs nothing here.
192    ///
193    /// `None` when the key is not declared, and always `None` for a context
194    /// built with [`Self::new`]. The contract for what you do with the value is
195    /// one line: **a handler must not write a secret-derived value into the
196    /// message.** Nothing in the engine records what this returns; whether it
197    /// stays unrecorded is the handler's business from here on.
198    #[inline]
199    pub fn secret(&self, name: &str) -> Option<&OwnedDataValue> {
200        self.secrets.get(name)
201    }
202
203    /// Borrow the message under processing. Use this when you need to inspect
204    /// the message id, payload, or audit trail; for reading and mutating the
205    /// `data` / `metadata` / `temp_data` context, prefer the typed helpers on
206    /// `TaskContext` itself.
207    ///
208    /// Inside a [`for_each`](crate::ForEach) call this is the call's own copy,
209    /// whose audit trail and errors start empty: it carries the context, id
210    /// and payload, not the message's history.
211    #[inline]
212    pub fn message(&self) -> &Message {
213        self.message
214    }
215
216    /// Mutable access to the message. Prefer the typed helpers (`set`,
217    /// `add_error`) over poking at `message.context` directly — direct
218    /// mutations bypass the audit trail.
219    #[inline]
220    pub fn message_mut(&mut self) -> &mut Message {
221        self.message
222    }
223
224    /// Shared datalogic engine, in case the handler needs to evaluate ad-hoc
225    /// JSONLogic. Most handlers can ignore this argument.
226    #[inline]
227    pub fn datalogic(&self) -> &Arc<DatalogicEngine> {
228        self.datalogic
229    }
230
231    /// Read-only view of `data`. Returns `&OwnedDataValue::Null` if missing
232    /// (mirrors the `Index` fallback semantics of `serde_json::Value`).
233    #[inline]
234    pub fn data(&self) -> &OwnedDataValue {
235        self.message.data()
236    }
237
238    /// Read-only view of `metadata`.
239    #[inline]
240    pub fn metadata(&self) -> &OwnedDataValue {
241        self.message.metadata()
242    }
243
244    /// Read-only view of `temp_data`.
245    #[inline]
246    pub fn temp_data(&self) -> &OwnedDataValue {
247        self.message.temp_data()
248    }
249
250    /// The full `{data, metadata, temp_data}` tree — the root every workflow
251    /// JSONLogic expression is written against.
252    ///
253    /// [`Self::data`] / [`Self::metadata`] / [`Self::temp_data`] expose the three
254    /// slots individually; this is the whole-context accessor, so handlers do not
255    /// have to reach through `ctx.message().context`.
256    ///
257    /// Note `payload` is **not** part of this tree, and therefore not part of the
258    /// JSONLogic evaluation context — `{"var": "payload.foo"}` resolves to
259    /// nothing. Parse the payload into `data` first.
260    #[inline]
261    pub fn context(&self) -> &OwnedDataValue {
262        &self.message.context
263    }
264
265    /// Evaluate a pre-compiled expression against the message context, on the
266    /// worker thread's pooled arena.
267    ///
268    /// The same path [`crate::engine::executor::evaluate_condition`] takes, but
269    /// returning the value instead of collapsing it to a bool, and surfacing
270    /// evaluation failures as `Err` instead of `false`. That difference is
271    /// deliberate: a condition that fails to evaluate should not run its task,
272    /// whereas a handler reading a config value needs to know the read failed.
273    ///
274    /// # Errors
275    ///
276    /// [`crate::DataflowError::LogicEvaluation`] if the expression fails to
277    /// evaluate, or [`crate::DataflowError::BudgetExceeded`] if it was aborted
278    /// for crossing the ceiling set by
279    /// [`crate::EngineBuilder::with_ops_budget`].
280    pub fn eval(&self, logic: &Logic) -> Result<OwnedDataValue> {
281        crate::engine::executor::eval_to_owned(self.datalogic, logic, &self.message.context)
282            .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
283    }
284
285    /// As [`Self::eval`], projected straight from the arena to
286    /// `serde_json::Value` in one walk — no `OwnedDataValue` intermediate and no
287    /// `serde_json::from_value` rebuild.
288    pub fn eval_json(&self, logic: &Logic) -> Result<JsonValue> {
289        crate::engine::executor::eval_to_json(self.datalogic, logic, &self.message.context)
290            .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
291    }
292
293    /// As [`Self::eval`], coerced to a *plain* string: a JSON string result
294    /// yields its contents, anything else its compact JSON form.
295    ///
296    /// # This disagrees with datalogic-rs on purpose
297    ///
298    /// datalogic-rs's `String: FromDataValue` — and therefore
299    /// `Session::eval_str` — keeps the JSON quoting, so a string result comes
300    /// back from it as `"\"abc\""`. This method returns `abc`.
301    ///
302    /// The name says `plain_string` rather than `to_string` precisely so the
303    /// difference is visible at the call site: two string semantics in one
304    /// ecosystem is a footgun, and these values end up in URL paths and message
305    /// keys. A test pins both sides, so it fails if either changes.
306    pub fn eval_to_plain_string(&self, logic: &Logic) -> Result<String> {
307        crate::engine::executor::eval_to_plain_string(self.datalogic, logic, &self.message.context)
308            .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
309    }
310
311    /// Run `f` against one arena view of the message context, so several
312    /// expressions cost one view of the context instead of one each.
313    ///
314    /// Every [`Self::eval`] (and every [`Template::resolve`] on a field that
315    /// reads the message) builds an arena view of the *whole* context before it
316    /// evaluates, so a step costs time in proportion to the number of arrays
317    /// and objects in the message, not to what it reads. A handler resolving k
318    /// templated fields pays for k views. Inside `with_view` it pays for one:
319    /// resolve the fields with [`Template::resolve_in`] and friends, or
320    /// evaluate directly with [`ContextView::eval`].
321    ///
322    /// Strings, keys and tensor buffers are borrowed rather than copied
323    /// ([`datavalue::OwnedDataValue::view_in`]), on this path and on the
324    /// per-call one, so what a view saves is the spine walk, not the bytes.
325    ///
326    /// The view borrows `self`, so nothing can write to the message while it
327    /// is alive, and the closure is synchronous, so the view cannot be held
328    /// across an `.await`. Resolve what the call needs up front, then do the
329    /// I/O:
330    ///
331    /// ```rust
332    /// # use dataflow_rs::{Result, TaskContext, Template};
333    /// # fn run(ctx: &mut TaskContext<'_>, url: &Template, timeout: &Template) -> Result<()> {
334    /// let (url, timeout_ms) = ctx.with_view(|view| {
335    ///     Ok::<_, dataflow_rs::DataflowError>((
336    ///         url.resolve_string_in(view)?,
337    ///         timeout.resolve_u64_in(view, "timeout_ms")?,
338    ///     ))
339    /// })?;
340    /// // ... await the call with `url` and `timeout_ms` ...
341    /// # let _ = (url, timeout_ms);
342    /// # Ok(())
343    /// # }
344    /// ```
345    ///
346    /// [`Template::resolve`]: crate::Template::resolve
347    /// [`Template::resolve_in`]: crate::Template::resolve_in
348    pub fn with_view<R>(&self, f: impl FnOnce(&ContextView<'_>) -> R) -> R {
349        crate::engine::executor::with_eval_arena(|arena| {
350            let data = arena.alloc(self.message.context.view_in(arena));
351            f(&ContextView {
352                datalogic: self.datalogic,
353                arena,
354                data,
355            })
356        })
357    }
358
359    /// Look up a value by dot-path against the full context tree (rooted at
360    /// the unified `{data, metadata, temp_data}` object). Returns `None` if
361    /// the path doesn't resolve.
362    ///
363    /// Use the same path syntax as JSONLogic: `"data.user.name"`,
364    /// `"temp_data.items.0"`, `"metadata.progress.status_code"`.
365    #[inline]
366    pub fn get(&self, path: &str) -> Option<&OwnedDataValue> {
367        get_nested_value(&self.message.context, path)
368    }
369
370    /// Set a value at a dot-path on the context. Records a `Change` on the
371    /// audit trail when `message.capture_changes` is true; otherwise the
372    /// write happens but no audit entry is buffered.
373    ///
374    /// Intermediate objects/arrays are created on demand; see
375    /// [`crate::engine::utils::set_nested_value`] for the exact semantics
376    /// (numeric segments → arrays, `#` prefix → escaped object key, etc.).
377    pub fn set(&mut self, path: &str, value: OwnedDataValue) {
378        if self.message.capture_changes {
379            let old_value = get_nested_value(&self.message.context, path)
380                .cloned()
381                .unwrap_or(OwnedDataValue::Null);
382            let new_value = value.clone();
383            self.changes.push(Change {
384                path: Arc::from(path),
385                old_value,
386                new_value,
387                removed: false,
388            });
389        }
390        set_nested_value(&mut self.message.context, path, value);
391    }
392
393    /// Same as [`Self::set`] but accepts a `serde_json::Value` (bridges
394    /// through `OwnedDataValue::from`). Convenience for handlers that
395    /// already speak `serde_json::Value`.
396    #[inline]
397    pub fn set_json(&mut self, path: &str, value: &JsonValue) {
398        self.set(path, OwnedDataValue::from(value));
399    }
400
401    /// Append an error to `message.errors`. Convenience for
402    /// `ctx.message_mut().add_error(...)`.
403    #[inline]
404    pub fn add_error(&mut self, error: ErrorInfo) {
405        self.message.add_error(error);
406    }
407
408    /// Drain the accumulated changes. The workflow executor calls this after
409    /// the handler returns to fold them into the audit trail; tests and
410    /// benchmarks driving the trait directly can use it to inspect what the
411    /// handler buffered.
412    #[inline]
413    pub fn into_changes(self) -> Vec<Change> {
414        self.changes
415    }
416}
417
418/// The message context converted into the arena once, for evaluating several
419/// expressions against it. Built by [`TaskContext::with_view`].
420///
421/// Each method evaluates exactly as its [`TaskContext`] namesake does, with
422/// the same errors, but against this one conversion instead of a fresh one.
423pub struct ContextView<'v> {
424    datalogic: &'v DatalogicEngine,
425    arena: &'v Bump,
426    data: &'v DataValue<'v>,
427}
428
429impl ContextView<'_> {
430    /// As [`TaskContext::eval`], against this view.
431    ///
432    /// # Errors
433    ///
434    /// As [`TaskContext::eval`].
435    pub fn eval(&self, logic: &Logic) -> Result<OwnedDataValue> {
436        self.evaluate(logic).map(|v| v.to_owned())
437    }
438
439    /// As [`TaskContext::eval_json`], against this view.
440    ///
441    /// # Errors
442    ///
443    /// As [`TaskContext::eval`].
444    pub fn eval_json(&self, logic: &Logic) -> Result<JsonValue> {
445        self.evaluate(logic).map(|v| v.to_serde_value())
446    }
447
448    /// As [`TaskContext::eval_to_plain_string`], against this view: a string
449    /// result yields its contents, anything else its compact JSON form.
450    ///
451    /// # Errors
452    ///
453    /// As [`TaskContext::eval`].
454    pub fn eval_to_plain_string(&self, logic: &Logic) -> Result<String> {
455        self.evaluate(logic).map(|v| match v {
456            DataValue::String(s) => s.to_string(),
457            other => other.to_string(),
458        })
459    }
460
461    fn evaluate<'r>(&'r self, logic: &'r Logic) -> Result<&'r DataValue<'r>> {
462        self.datalogic
463            .evaluate(logic, self.data, self.arena)
464            .map_err(|e| crate::engine::error::from_datalogic_eval(&e))
465    }
466}
467
468#[cfg(test)]
469mod tests {
470    use super::*;
471    use crate::engine::error::DataflowError;
472    use crate::engine::executor::with_arena;
473    use crate::engine::utils::set_nested_value;
474    use serde_json::json;
475
476    fn dv(v: serde_json::Value) -> OwnedDataValue {
477        OwnedDataValue::from(&v)
478    }
479
480    fn engine() -> Arc<DatalogicEngine> {
481        Arc::new(crate::engine::compiler::datalogic_engine_builder().build())
482    }
483
484    /// A message with one key in each of the three context slots.
485    fn populated() -> Message {
486        let mut m = Message::from_value(&json!({"payload_key": "payload_value"}));
487        set_nested_value(&mut m.context, "data.x", dv(json!("dx")));
488        set_nested_value(&mut m.context, "metadata.x", dv(json!("mx")));
489        set_nested_value(&mut m.context, "temp_data.x", dv(json!("tx")));
490        m
491    }
492
493    #[test]
494    fn context_matches_the_three_slot_accessors() {
495        let mut m = populated();
496        let dl = engine();
497        let ctx = TaskContext::new(&mut m, &dl);
498
499        let whole = ctx.context();
500        assert_eq!(&whole["data"], ctx.data());
501        assert_eq!(&whole["metadata"], ctx.metadata());
502        assert_eq!(&whole["temp_data"], ctx.temp_data());
503        assert_eq!(whole, &ctx.message().context);
504    }
505
506    #[test]
507    fn eval_roots_at_the_unified_context_not_data_alone() {
508        let mut m = populated();
509        let dl = engine();
510        let ctx = TaskContext::new(&mut m, &dl);
511
512        for (path, expected) in [
513            ("data.x", "dx"),
514            ("metadata.x", "mx"),
515            ("temp_data.x", "tx"),
516        ] {
517            let logic = dl.compile_arc(&json!({"var": path})).unwrap();
518            assert_eq!(ctx.eval(&logic).unwrap(), dv(json!(expected)));
519            assert_eq!(ctx.eval_json(&logic).unwrap(), json!(expected));
520            assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), expected);
521        }
522    }
523
524    #[test]
525    fn payload_is_not_in_the_eval_context() {
526        // Stays true through the new surface: `payload` is a separate field on
527        // Message and never part of the JSONLogic root.
528        let mut m = populated();
529        let dl = engine();
530        let ctx = TaskContext::new(&mut m, &dl);
531
532        let logic = dl
533            .compile_arc(&json!({"var": "payload.payload_key"}))
534            .unwrap();
535        assert_eq!(ctx.eval(&logic).unwrap(), OwnedDataValue::Null);
536        assert_eq!(ctx.eval_json(&logic).unwrap(), serde_json::Value::Null);
537        assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), "null");
538    }
539
540    #[test]
541    fn eval_json_covers_every_result_kind() {
542        let mut m = Message::from_value(&json!({}));
543        let dl = engine();
544        let ctx = TaskContext::new(&mut m, &dl);
545
546        for expected in [
547            json!(null),
548            json!(true),
549            json!(42),
550            json!(1.5),
551            json!("abc"),
552            json!([1, 2]),
553            json!({"a": 1}),
554            json!({"a": [1, {"b": "c"}], "d": {"e": [true, null]}}),
555        ] {
556            let logic = dl.compile_arc(&expected).unwrap();
557            assert_eq!(
558                ctx.eval_json(&logic).unwrap(),
559                expected,
560                "round-trip for {expected}"
561            );
562            // The owned and JSON projections agree on the same result.
563            assert_eq!(
564                serde_json::Value::from(&ctx.eval(&logic).unwrap()),
565                expected
566            );
567        }
568    }
569
570    #[test]
571    fn eval_to_plain_string_unquotes_strings_and_compacts_the_rest() {
572        let mut m = Message::from_value(&json!({}));
573        let dl = engine();
574        let ctx = TaskContext::new(&mut m, &dl);
575
576        let cases = [
577            (json!("abc"), "abc"),
578            (json!(""), ""),
579            (json!(null), "null"),
580            (json!(true), "true"),
581            (json!(42), "42"),
582            (json!({"a": 1}), "{\"a\":1}"),
583            (json!([1, 2]), "[1,2]"),
584        ];
585        for (input, expected) in cases {
586            let logic = dl.compile_arc(&input).unwrap();
587            assert_eq!(
588                ctx.eval_to_plain_string(&logic).unwrap(),
589                expected,
590                "for {input}"
591            );
592        }
593    }
594
595    #[test]
596    fn eval_to_plain_string_diverges_from_datalogics_own_string_projection() {
597        // This test IS the documentation of the divergence — it must fail if
598        // either side changes. datalogic-rs's `String: FromDataValue` keeps the
599        // JSON quoting; ours does not.
600        let mut m = Message::from_value(&json!({}));
601        let dl = engine();
602        let ctx = TaskContext::new(&mut m, &dl);
603
604        // Non-ASCII plus an embedded quote, to cover escaping too.
605        let raw = "héllo \"world\" 世界";
606        let logic = dl.compile_arc(&json!(raw)).unwrap();
607
608        // Ours: contents, byte-identical.
609        assert_eq!(ctx.eval_to_plain_string(&logic).unwrap(), raw);
610
611        // datalogic's: JSON-quoted and escaped.
612        let via_session = dl.session().eval_str(&logic, &m.context).unwrap();
613        assert_ne!(
614            via_session, raw,
615            "if these agree, the divergence this method exists for is gone"
616        );
617        assert!(
618            via_session.starts_with('"') && via_session.contains("\\\""),
619            "datalogic keeps the quoting and escaping, got: {via_session}"
620        );
621    }
622
623    #[test]
624    fn eval_surfaces_an_error_where_evaluate_condition_returns_false() {
625        // Asserted together so the difference is intentional and visible: a
626        // condition that fails should not run its task; a handler reading a
627        // config value needs to know the read failed.
628        let mut m = Message::from_value(&json!({}));
629        let dl = engine();
630
631        // `+` over a non-numeric operand fails to evaluate.
632        let bad = dl.compile_arc(&json!({"+": ["abc", 1]})).unwrap();
633
634        let condition_result =
635            crate::engine::executor::evaluate_condition(&dl, Some(&bad), &m.context);
636
637        let ctx = TaskContext::new(&mut m, &dl);
638        let eval_result = ctx.eval(&bad);
639
640        match (&condition_result, &eval_result) {
641            (Ok(false), Err(DataflowError::LogicEvaluation(msg))) => {
642                assert!(!msg.is_empty(), "the error message must not be empty");
643            }
644            other => panic!(
645                "expected evaluate_condition Ok(false) alongside eval Err(LogicEvaluation), got {other:?}"
646            ),
647        }
648    }
649
650    #[test]
651    fn consecutive_evals_and_interleaved_sets_both_work() {
652        // The arena is rewound between calls, not corrupted; and an eval between
653        // two `set`s leaves the buffered Changes intact.
654        let mut m = populated();
655        let dl = engine();
656        let first = dl.compile_arc(&json!({"var": "data.x"})).unwrap();
657        let second = dl.compile_arc(&json!({"var": "metadata.x"})).unwrap();
658
659        let mut ctx = TaskContext::new(&mut m, &dl);
660
661        assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
662        assert_eq!(ctx.eval_json(&second).unwrap(), json!("mx"));
663        assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
664
665        ctx.set("data.written", dv(json!(1)));
666        assert_eq!(ctx.eval_json(&first).unwrap(), json!("dx"));
667        ctx.set("data.written2", dv(json!(2)));
668
669        let changes = ctx.into_changes();
670        let paths: Vec<&str> = changes.iter().map(|c| &*c.path).collect();
671        assert_eq!(paths, vec!["data.written", "data.written2"]);
672    }
673
674    #[test]
675    fn eval_inside_a_with_arena_scope_falls_back_instead_of_panicking() {
676        // `TaskContext::new` is pub so a test or bench can construct one inside a
677        // `with_arena` closure. The `try_borrow_mut` fallback makes that return
678        // `Ok` on a fresh Bump rather than panicking out of the arena scope.
679        let mut m = populated();
680        let dl = engine();
681        let logic = dl.compile_arc(&json!({"var": "data.x"})).unwrap();
682
683        let got = with_arena(|_| {
684            let ctx = TaskContext::new(&mut m, &dl);
685            ctx.eval_json(&logic)
686        });
687
688        assert_eq!(got.unwrap(), json!("dx"));
689    }
690}