Skip to main content

atman_runtime/
exec.rs

1use std::{collections::HashMap, path::PathBuf};
2
3use atman_dsl::ast::{CmpOp, Expr, FlowDecl, Node, Stmt, WatchAction, WatchDecl, WatchEvent};
4
5use crate::env::Env;
6use crate::error::RuntimeError;
7use crate::eval::{EvalCtx, eval_expr};
8use crate::provider::LlmRequest;
9use crate::streaming::{LlmStream, WarnRule, WatchRules};
10use crate::tool::{BoxFut, ToolCtx, ToolRegistry};
11use crate::value::Value;
12
13fn bind_pattern(
14    pattern: &atman_dsl::ast::Pattern,
15    value: Value,
16    env: &mut Env,
17) -> Result<(), RuntimeError> {
18    use atman_dsl::ast::{Pattern, PatternFieldBinding};
19    match pattern {
20        Pattern::Ident(id) => {
21            env.bind(id.name.clone(), value);
22            Ok(())
23        }
24        Pattern::Struct { fields } => {
25            let pairs = match value {
26                Value::Struct(pairs) => pairs,
27                other => {
28                    return Err(RuntimeError::TypeMismatch {
29                        expected: "struct for destructuring bind".into(),
30                        actual: other.kind_name().into(),
31                    });
32                }
33            };
34            for field in fields {
35                let Some((_, matched)) = pairs.iter().find(|(k, _)| k == &field.source.name) else {
36                    return Err(RuntimeError::MissingArg(format!(
37                        "destructure: struct has no field `{}`",
38                        field.source.name
39                    )));
40                };
41                match &field.binding {
42                    PatternFieldBinding::Same => {
43                        env.bind(field.source.name.clone(), matched.clone());
44                    }
45                    PatternFieldBinding::Rename(target) => {
46                        env.bind(target.name.clone(), matched.clone());
47                    }
48                    PatternFieldBinding::Nested(inner) => {
49                        bind_pattern(inner, matched.clone(), env)?;
50                    }
51                }
52            }
53            Ok(())
54        }
55    }
56}
57
58pub enum StmtOutcome {
59    Continue,
60    Return(Value),
61    Err(RuntimeError),
62}
63
64pub fn exec_stmts<'a>(
65    stmts: &'a [Stmt],
66    env: &'a mut Env,
67    ctx: &'a EvalCtx<'a>,
68) -> BoxFut<'a, StmtOutcome> {
69    exec_stmts_prefixed(stmts, env, ctx, String::new())
70}
71
72pub fn exec_stmts_prefixed<'a>(
73    stmts: &'a [Stmt],
74    env: &'a mut Env,
75    ctx: &'a EvalCtx<'a>,
76    prefix: String,
77) -> BoxFut<'a, StmtOutcome> {
78    Box::pin(async move {
79        let watches = collect_watches(stmts);
80        let parent_node_id = ctx.current_node_id.clone();
81        for (i, stmt) in stmts.iter().enumerate() {
82            let node_id = if prefix.is_empty() {
83                format!("{i}")
84            } else {
85                format!("{prefix}.{i}")
86            };
87            // Check for pending L4 stop / L3 redirect between statements.
88            if let Some(session) = ctx.session_runtime.as_ref()
89                && let Some(turn_id) = ctx.turn_id.as_ref()
90            {
91                if let Some(inj) = session.peek_pending_l2_or_higher(turn_id) {
92                    match inj.level {
93                        crate::injection::InjectionLevel::L4HardStop => {
94                            session.mark_injection_consumed(&inj.id);
95                            emit_flow_node_start(ctx, &node_id, stmt, parent_node_id.as_deref());
96                            emit_flow_node_end(
97                                ctx,
98                                &node_id,
99                                &StmtOutcome::Continue,
100                                parent_node_id.as_deref(),
101                                Some("cancelled: hard stop"),
102                            );
103                            return StmtOutcome::Err(RuntimeError::Cancelled(
104                                "hard stop from user".into(),
105                            ));
106                        }
107                        crate::injection::InjectionLevel::L3Redirect => {
108                            if let Some(target) = inj.redirect_target.clone() {
109                                session.mark_injection_consumed(&inj.id);
110                                return StmtOutcome::Err(RuntimeError::Redirect(target));
111                            }
112                        }
113                        _ => {}
114                    }
115                }
116            }
117            emit_flow_node_start(ctx, &node_id, stmt, parent_node_id.as_deref());
118            let stmt_ctx = ctx.with_node(&node_id);
119            let (outcome, preview) = exec_stmt(stmt, env, &stmt_ctx, &watches).await;
120            emit_flow_node_end(
121                ctx,
122                &node_id,
123                &outcome,
124                parent_node_id.as_deref(),
125                preview.as_deref(),
126            );
127            match outcome {
128                StmtOutcome::Continue => continue,
129                other => return other,
130            }
131        }
132        StmtOutcome::Continue
133    })
134}
135
136fn emit_flow_node_start(
137    ctx: &EvalCtx<'_>,
138    node_id: &str,
139    stmt: &Stmt,
140    parent_node_id: Option<&str>,
141) {
142    let Some(run_id) = ctx.flow_run_id.clone() else {
143        return;
144    };
145    let (kind, label) = stmt_to_node_kind_label(stmt);
146    if let Some(sink) = ctx.events {
147        sink.emit(crate::event::Event::FlowNodeStart {
148            run_id: run_id.clone(),
149            node_id: node_id.to_string(),
150            kind: kind.clone(),
151            label: label.clone(),
152            parent_node_id: parent_node_id.map(String::from),
153        });
154    }
155    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
156        let _ = tx.send(crate::stream::StreamFrame::FlowNodeStart {
157            run_id: run_id.0.to_string(),
158            node_id: node_id.to_string(),
159            kind,
160            label,
161            parent_node_id: parent_node_id.map(String::from),
162        });
163    }
164}
165
166fn value_preview(v: &Value) -> Option<String> {
167    let raw = match v {
168        Value::Str(s) => s.clone(),
169        Value::Message(m) => {
170            let text = m.text_concat();
171            let tool_uses: Vec<String> = m
172                .parts
173                .iter()
174                .filter_map(|p| match p {
175                    crate::message::MessagePart::ToolUse { name, .. } => Some(name.clone()),
176                    _ => None,
177                })
178                .collect();
179            match (text.trim().is_empty(), tool_uses.is_empty()) {
180                (false, true) => text,
181                (false, false) => format!("{}\n\n→ tool_uses: {}", text, tool_uses.join(", ")),
182                (true, false) => format!("→ tool_uses: {}", tool_uses.join(", ")),
183                (true, true) => return None,
184            }
185        }
186        Value::Path(p) => p.display().to_string(),
187        Value::Int(n) => n.to_string(),
188        Value::Float(n) => n.to_string(),
189        Value::Bool(b) => b.to_string(),
190        Value::Unit => return None,
191        Value::Err(e) => format!("err: {e}"),
192        Value::List(items) => format!("list[{}]", items.len()),
193        Value::Struct(fields) => format!(
194            "{{{}}}",
195            fields
196                .iter()
197                .map(|(k, _)| k.as_str())
198                .collect::<Vec<_>>()
199                .join(", ")
200        ),
201        Value::EditProposal(_) => "<edit proposal>".into(),
202    };
203    let trimmed = raw.trim();
204    if trimmed.is_empty() {
205        None
206    } else {
207        Some(trimmed.chars().take(4000).collect())
208    }
209}
210
211fn emit_flow_node_end(
212    ctx: &EvalCtx<'_>,
213    node_id: &str,
214    outcome: &StmtOutcome,
215    parent_node_id: Option<&str>,
216    output_preview: Option<&str>,
217) {
218    let Some(run_id) = ctx.flow_run_id.clone() else {
219        return;
220    };
221    let status = match outcome {
222        StmtOutcome::Err(_) => crate::event::FlowNodeStatus::Err,
223        _ => crate::event::FlowNodeStatus::Ok,
224    };
225    let preview_owned = output_preview.map(String::from);
226    if let Some(sink) = ctx.events {
227        sink.emit(crate::event::Event::FlowNodeEnd {
228            run_id: run_id.clone(),
229            node_id: node_id.to_string(),
230            status: status.clone(),
231            output_preview: preview_owned.clone(),
232        });
233    }
234    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
235        let _ = tx.send(crate::stream::StreamFrame::FlowNodeEnd {
236            run_id: run_id.0.to_string(),
237            node_id: node_id.to_string(),
238            status,
239            output_preview: preview_owned,
240            parent_node_id: parent_node_id.map(String::from),
241        });
242    }
243}
244
245fn stmt_to_node_kind_label(stmt: &Stmt) -> (crate::nodegraph::NodeKind, String) {
246    use crate::nodegraph::NodeKind;
247    match stmt {
248        Stmt::Bind { value, .. } | Stmt::Expr(value) => expr_to_node_kind_label(value),
249        Stmt::Return { .. } => (NodeKind::Return, "return".into()),
250        Stmt::When { .. } => (
251            NodeKind::When {
252                condition_preview: "when".into(),
253            },
254            "when …".into(),
255        ),
256        Stmt::Watch(_) => (NodeKind::Return, "watch".into()),
257    }
258}
259
260fn expr_to_node_kind_label(expr: &Expr) -> (crate::nodegraph::NodeKind, String) {
261    use crate::nodegraph::NodeKind;
262    match expr {
263        Expr::Node(Node::Llm { .. }) => (NodeKind::Llm { model: None }, "llm".into()),
264        Expr::Node(Node::ToolCall { path, .. }) => {
265            let p = path
266                .iter()
267                .map(|s| s.name.clone())
268                .collect::<Vec<_>>()
269                .join(".");
270            (NodeKind::ToolCall { path: p.clone() }, format!("⟶ {p}"))
271        }
272        Expr::Node(Node::Fanout { items, collect }) => (
273            NodeKind::Fanout {
274                collect: (*collect).into(),
275            },
276            format!("fanout ×{}", items.len()),
277        ),
278        Expr::Node(Node::Subflow { name, .. }) => (
279            NodeKind::Subflow {
280                name: name.name.clone(),
281            },
282            format!("subflow({})", name.name),
283        ),
284        _ => (NodeKind::Return, "expr".into()),
285    }
286}
287
288fn collect_watches(stmts: &[Stmt]) -> HashMap<String, Vec<&WatchDecl>> {
289    let mut out: HashMap<String, Vec<&WatchDecl>> = HashMap::new();
290    for stmt in stmts {
291        if let Stmt::Watch(w) = stmt {
292            out.entry(w.target.name.clone()).or_default().push(w);
293        }
294    }
295    out
296}
297
298fn exec_stmt<'a>(
299    stmt: &'a Stmt,
300    env: &'a mut Env,
301    ctx: &'a EvalCtx<'a>,
302    watches: &'a HashMap<String, Vec<&'a WatchDecl>>,
303) -> BoxFut<'a, (StmtOutcome, Option<String>)> {
304    Box::pin(async move {
305        match stmt {
306            Stmt::Bind { name, value } => {
307                let watch_target = name.as_single_ident().map(|id| id.name.clone());
308                let v = if let Some(target) = watch_target.as_ref()
309                    && let Some(ws) = watches.get(target)
310                {
311                    match eval_bind_with_watches(value, env, ctx, ws).await {
312                        Ok(v) => v,
313                        Err(e) => return (StmtOutcome::Err(e), None),
314                    }
315                } else {
316                    eval_expr(value, env, ctx).await
317                };
318                if let Value::Err(e) = v {
319                    return (StmtOutcome::Err(e), None);
320                }
321                let preview = value_preview(&v);
322                if let Err(e) = bind_pattern(name, v, env) {
323                    return (StmtOutcome::Err(e), None);
324                }
325                (StmtOutcome::Continue, preview)
326            }
327            Stmt::When { cond, body } => {
328                let c = eval_expr(cond, env, ctx).await;
329                let truthy = match c {
330                    Value::Bool(b) => b,
331                    Value::Unit => false,
332                    Value::Err(e) => {
333                        return (StmtOutcome::Err(e), None);
334                    }
335                    _ => true,
336                };
337                if truthy {
338                    (exec_stmts(body, env, ctx).await, Some("true".into()))
339                } else {
340                    (StmtOutcome::Continue, Some("false".into()))
341                }
342            }
343            Stmt::Return { value } => {
344                let v = eval_expr(value, env, ctx).await;
345                if let Value::Err(e) = v {
346                    return (StmtOutcome::Err(e), None);
347                }
348                let preview = value_preview(&v);
349                (StmtOutcome::Return(v), preview)
350            }
351            Stmt::Expr(e) => {
352                let v = eval_expr(e, env, ctx).await;
353                if let Value::Err(err) = v {
354                    return (StmtOutcome::Err(err), None);
355                }
356                let preview = value_preview(&v);
357                (StmtOutcome::Continue, preview)
358            }
359            Stmt::Watch(_) => (StmtOutcome::Continue, None),
360        }
361    })
362}
363
364async fn eval_bind_with_watches(
365    expr: &Expr,
366    env: &mut Env,
367    ctx: &EvalCtx<'_>,
368    watches: &[&WatchDecl],
369) -> Result<Value, RuntimeError> {
370    let Expr::Node(Node::Llm { kwargs }) = expr else {
371        return Ok(eval_expr(expr, env, ctx).await);
372    };
373
374    let mut model: Option<String> = None;
375    let mut prompt: Option<String> = None;
376    let mut input = Value::Unit;
377    let mut cache_prompt = false;
378    let mut context_budget: Option<u64> = None;
379    for (k, v) in kwargs {
380        if k.name == "schema" || k.name == "fallback" || k.name == "retry" {
381            continue;
382        }
383        let val = eval_expr(v, env, ctx).await;
384        if val.is_err() {
385            return Ok(val);
386        }
387        match k.name.as_str() {
388            "model" => match val {
389                Value::Str(s) => model = Some(s),
390                other => {
391                    return Ok(Value::Err(RuntimeError::TypeMismatch {
392                        expected: "string".into(),
393                        actual: other.kind_name().into(),
394                    }));
395                }
396            },
397            "prompt" => match val {
398                Value::Str(s) => prompt = Some(s),
399                other => {
400                    return Ok(Value::Err(RuntimeError::TypeMismatch {
401                        expected: "string".into(),
402                        actual: other.kind_name().into(),
403                    }));
404                }
405            },
406            "input" => input = val,
407            "cache" => match val {
408                Value::Bool(b) => cache_prompt = b,
409                other => {
410                    return Ok(Value::Err(RuntimeError::TypeMismatch {
411                        expected: "bool".into(),
412                        actual: other.kind_name().into(),
413                    }));
414                }
415            },
416            "context_budget" => match val {
417                Value::Int(n) if n > 0 => context_budget = Some(n as u64),
418                other => {
419                    return Ok(Value::Err(RuntimeError::TypeMismatch {
420                        expected: "positive int".into(),
421                        actual: other.kind_name().into(),
422                    }));
423                }
424            },
425            _ => {}
426        }
427    }
428    let Some(model) = model else {
429        return Ok(Value::Err(RuntimeError::MissingArg("llm.model".into())));
430    };
431    let Some(mut prompt) = prompt else {
432        return Ok(Value::Err(RuntimeError::MissingArg("llm.prompt".into())));
433    };
434    if let Some(budget) = context_budget {
435        let (truncated, stat) = crate::eval::truncate_prompt_to_budget_tracked(prompt, budget);
436        prompt = truncated;
437        if let (Some(sink), Some(stat)) = (ctx.events, stat) {
438            sink.emit(crate::event::Event::ContextTruncated {
439                turn_id: ctx.turn_id.clone(),
440                flow_run_id: ctx.flow_run_id.clone(),
441                original_chars: stat.original_chars as u64,
442                result_chars: stat.result_chars as u64,
443                dropped_chars: stat.dropped_chars as u64,
444                budget_tokens: stat.budget_tokens,
445            });
446        }
447    }
448    let Some(provider) = ctx.providers.resolve(&model) else {
449        return Ok(Value::Err(RuntimeError::ToolFailed(format!(
450            "no provider registered for model `{model}`"
451        ))));
452    };
453
454    let messages = vec![crate::provider::user_text_message(prompt.clone())];
455    let req = LlmRequest {
456        model: model.clone(),
457        messages,
458        system: None,
459        input: input.clone(),
460        schema: None,
461        cache_prompt,
462        tools: Vec::new(),
463        thinking_enabled: false,
464        stall_timeout_secs: 120,
465    };
466    let frame_tx = ctx
467        .tool_ctx
468        .agent_entry
469        .as_ref()
470        .map(|e| e.frame_tx.clone());
471    let mut stream = LlmStream::new(provider.as_ref(), req)
472        .with_stream_tx(ctx.tool_ctx.stream_tx.clone())
473        .with_frame_tx(frame_tx)
474        .with_watch_rules(collect_watch_rules(watches))
475        .with_event_sink(ctx.events)
476        .with_turn_id(ctx.turn_id.clone())
477        .with_flow_run_id(ctx.flow_run_id.clone());
478    if let Some(session) = ctx.session_runtime.as_ref() {
479        stream = stream.with_session(session);
480    }
481    if let Some(entry) = ctx.tool_ctx.agent_entry.as_ref() {
482        stream = stream.with_entry(entry);
483    }
484    match stream.run().await {
485        Ok(am) => Ok(crate::provider::assistant_message_to_value(&am)),
486        Err(e) => Ok(Value::Err(e)),
487    }
488}
489
490fn render_warn_msg(msg: &Option<Expr>, fallback: &str) -> String {
491    match msg {
492        Some(Expr::Literal(atman_dsl::ast::Literal::Str(s))) => s.clone(),
493        _ => fallback.to_string(),
494    }
495}
496
497fn collect_watch_rules(watches: &[&WatchDecl]) -> WatchRules {
498    let mut rules = WatchRules::default();
499    for w in watches {
500        for on in &w.on_blocks {
501            let has_abort = on
502                .actions
503                .iter()
504                .any(|a| matches!(a, WatchAction::Abort { .. }));
505            let warn_msg_expr = on.actions.iter().find_map(|a| match a {
506                WatchAction::Warn { msg } => Some(msg),
507                _ => None,
508            });
509            if !has_abort && warn_msg_expr.is_none() {
510                continue;
511            }
512            match &on.event {
513                WatchEvent::Token { patterns } => {
514                    for p in patterns {
515                        if has_abort {
516                            rules
517                                .token_matches
518                                .push((p.clone(), format!("token match: {p}")));
519                        }
520                        if let Some(msg_expr) = warn_msg_expr {
521                            rules.warn_token.push(WarnRule {
522                                target: w.target.name.clone(),
523                                message: render_warn_msg(
524                                    msg_expr,
525                                    &format!("watch warn: token `{p}`"),
526                                ),
527                                pattern: p.clone(),
528                            });
529                        }
530                    }
531                }
532                WatchEvent::TokensConsumed { cmp, value }
533                    if matches!(cmp, CmpOp::Gt | CmpOp::Ge) =>
534                {
535                    let threshold = if matches!(cmp, CmpOp::Ge) {
536                        value.saturating_sub(1)
537                    } else {
538                        *value
539                    };
540                    if has_abort {
541                        rules.tokens_gt = Some(match rules.tokens_gt {
542                            Some(existing) => existing.min(threshold),
543                            None => threshold,
544                        });
545                    }
546                    if let Some(msg_expr) = warn_msg_expr {
547                        rules.warn_tokens_gt.push((
548                            threshold,
549                            WarnRule {
550                                target: w.target.name.clone(),
551                                message: render_warn_msg(
552                                    msg_expr,
553                                    &format!("watch warn: tokens_consumed > {threshold}"),
554                                ),
555                                pattern: format!("tokens_consumed>{threshold}"),
556                            },
557                        ));
558                    }
559                }
560                WatchEvent::Elapsed { cmp, duration_ms }
561                    if matches!(cmp, CmpOp::Gt | CmpOp::Ge) =>
562                {
563                    let threshold = if matches!(cmp, CmpOp::Ge) {
564                        duration_ms.saturating_sub(1)
565                    } else {
566                        *duration_ms
567                    };
568                    if has_abort {
569                        rules.elapsed_ms_gt = Some(match rules.elapsed_ms_gt {
570                            Some(existing) => existing.min(threshold),
571                            None => threshold,
572                        });
573                    }
574                    if let Some(msg_expr) = warn_msg_expr {
575                        rules.warn_elapsed_ms_gt.push((
576                            threshold,
577                            WarnRule {
578                                target: w.target.name.clone(),
579                                message: render_warn_msg(
580                                    msg_expr,
581                                    &format!("watch warn: elapsed > {threshold}ms"),
582                                ),
583                                pattern: format!("elapsed>{threshold}ms"),
584                            },
585                        ));
586                    }
587                }
588                _ => {}
589            }
590        }
591    }
592    rules
593}
594
595pub async fn exec_flow(
596    flow: &FlowDecl,
597    args: Vec<(String, Value)>,
598    tools: &ToolRegistry,
599    tool_ctx: &ToolCtx,
600    providers: &crate::provider::ProviderRegistry,
601    source_dir: Option<PathBuf>,
602) -> Result<Value, RuntimeError> {
603    let flows = std::collections::HashMap::new();
604    exec_flow_with_siblings(
605        flow,
606        args,
607        tools,
608        tool_ctx,
609        providers,
610        &flows,
611        None,
612        None,
613        None,
614        None,
615        tokio_util::sync::CancellationToken::new(),
616        None,
617        source_dir,
618    )
619    .await
620}
621
622#[allow(clippy::too_many_arguments)]
623pub async fn exec_flow_with_siblings(
624    flow: &FlowDecl,
625    args: Vec<(String, Value)>,
626    tools: &ToolRegistry,
627    tool_ctx: &ToolCtx,
628    providers: &crate::provider::ProviderRegistry,
629    flows: &std::collections::HashMap<String, FlowDecl>,
630    events: Option<&crate::event::EventSink>,
631    turn_id: Option<crate::event::TurnId>,
632    flow_run_id: Option<crate::event::FlowRunId>,
633    session: Option<std::sync::Arc<crate::session::Session>>,
634    flow_cancel: tokio_util::sync::CancellationToken,
635    safety: Option<&crate::safety::SafetyConfig>,
636    source_dir: Option<PathBuf>,
637) -> Result<Value, RuntimeError> {
638    let ctx = EvalCtx {
639        tools,
640        tool_ctx,
641        providers,
642        flows,
643        contract: flow.contract.as_ref(),
644        events,
645        turn_id,
646        flow_run_id,
647        session_runtime: session,
648        flow_cancel,
649        safety,
650        current_node_id: None,
651        source_dir,
652    };
653    let mut env = Env::new();
654    let provided: std::collections::HashSet<String> = args.iter().map(|(n, _)| n.clone()).collect();
655    for (name, value) in args {
656        env.bind(name, value);
657    }
658    for p in &flow.params {
659        if !provided.contains(&p.name.name) {
660            if let Some(default) = &p.default {
661                let val = crate::eval::eval_expr(default, &env, &ctx).await;
662                env.bind(p.name.name.clone(), val);
663            }
664        }
665    }
666    match exec_stmts(&flow.body, &mut env, &ctx).await {
667        StmtOutcome::Return(v) => Ok(v),
668        StmtOutcome::Err(e) => Err(e),
669        StmtOutcome::Continue => Ok(Value::Unit),
670    }
671}
672
673#[cfg(test)]
674mod tests {
675    use super::*;
676    use atman_dsl::parse::parse_file;
677
678    async fn run(src: &str, args: Vec<(String, Value)>) -> Result<Value, RuntimeError> {
679        let file = parse_file(src).expect("parse test src");
680        let tools = ToolRegistry::new();
681        let tool_ctx = ToolCtx::new();
682        let providers = crate::provider::ProviderRegistry::new();
683        exec_flow(&file.flows[0], args, &tools, &tool_ctx, &providers, None).await
684    }
685
686    #[tokio::test]
687    async fn bind_and_return() {
688        let out = run(
689            r#"flow t() -> Int {
690    x = 1
691    y = x + 2
692    return y
693}
694"#,
695            vec![],
696        )
697        .await
698        .unwrap();
699        assert!(matches!(out, Value::Int(3)));
700    }
701
702    #[tokio::test]
703    async fn when_true_executes_body() {
704        let out = run(
705            r#"flow t() -> Int {
706    x = 5
707    when x > 3 {
708        return 42
709    }
710    return 0
711}
712"#,
713            vec![],
714        )
715        .await
716        .unwrap();
717        assert!(matches!(out, Value::Int(42)));
718    }
719
720    #[tokio::test]
721    async fn when_false_skips_body() {
722        let out = run(
723            r#"flow t() -> Int {
724    x = 1
725    when x > 3 {
726        return 42
727    }
728    return 0
729}
730"#,
731            vec![],
732        )
733        .await
734        .unwrap();
735        assert!(matches!(out, Value::Int(0)));
736    }
737
738    #[tokio::test]
739    async fn err_in_bind_stops_flow() {
740        let err = run(
741            r#"flow t() -> Int {
742    x = missing
743    return 1
744}
745"#,
746            vec![],
747        )
748        .await
749        .unwrap_err();
750        assert!(matches!(err, RuntimeError::UndefinedVar(n) if n == "missing"));
751    }
752
753    #[tokio::test]
754    async fn flow_args_bind_before_body() {
755        let out = run(
756            r#"flow t() -> Int {
757    return n + 1
758}
759"#,
760            vec![("n".into(), Value::Int(4))],
761        )
762        .await
763        .unwrap();
764        assert!(matches!(out, Value::Int(5)));
765    }
766
767    #[tokio::test]
768    async fn when_cond_unit_is_falsy() {
769        let out = run(
770            r#"flow t() -> Int {
771    when 1 {
772        return 1
773    }
774    return 0
775}
776"#,
777            vec![],
778        )
779        .await
780        .unwrap();
781        assert!(matches!(out, Value::Int(1)));
782    }
783
784    #[tokio::test]
785    async fn when_cond_struct_is_truthy() {
786        let out = run(
787            r#"flow t() -> Int {
788    x = { a: 1 }
789    when x {
790        return 42
791    }
792    return 0
793}
794"#,
795            vec![],
796        )
797        .await
798        .unwrap();
799        assert!(matches!(out, Value::Int(42)));
800    }
801
802    #[tokio::test]
803    async fn flow_falls_through_to_unit_without_return() {
804        let out = run(
805            r#"flow t() {
806    x = 1
807}
808"#,
809            vec![],
810        )
811        .await
812        .unwrap();
813        assert!(matches!(out, Value::Unit));
814    }
815}