atman_runtime/

exec.rs

use std::{collections::HashMap, path::PathBuf};

use atman_dsl::ast::{Arg, CmpOp, Expr, FlowDecl, Node, Stmt, WatchAction, WatchDecl, WatchEvent};

use crate::env::Env;
use crate::error::RuntimeError;
use crate::eval::{EvalCtx, eval_expr};
use crate::streaming::{WarnRule, WatchRules};
use crate::tool::{BoxFut, Tool, ToolArgs, ToolCtx, ToolRegistry};
use crate::value::Value;

fn bind_pattern(
    pattern: &atman_dsl::ast::Pattern,
    value: Value,
    env: &mut Env,
) -> Result<(), RuntimeError> {
    use atman_dsl::ast::{Pattern, PatternFieldBinding};
    match pattern {
        Pattern::Ident(id) => {
            env.bind(id.name.clone(), value);
            Ok(())
        }
        Pattern::Struct { fields } => {
            let pairs = match value {
                Value::Struct(pairs) => pairs,
                other => {
                    return Err(RuntimeError::TypeMismatch {
                        expected: "struct for destructuring bind".into(),
                        actual: other.kind_name().into(),
                    });
                }
            };
            for field in fields {
                let Some((_, matched)) = pairs.iter().find(|(k, _)| k == &field.source.name) else {
                    return Err(RuntimeError::MissingArg(format!(
                        "destructure: struct has no field `{}`",
                        field.source.name
                    )));
                };
                match &field.binding {
                    PatternFieldBinding::Same => {
                        env.bind(field.source.name.clone(), matched.clone());
                    }
                    PatternFieldBinding::Rename(target) => {
                        env.bind(target.name.clone(), matched.clone());
                    }
                    PatternFieldBinding::Nested(inner) => {
                        bind_pattern(inner, matched.clone(), env)?;
                    }
                }
            }
            Ok(())
        }
    }
}

pub enum StmtOutcome {
    Continue,
    Return(Value),
    Err(RuntimeError),
    LoopBreak,
    LoopContinue,
}

pub fn exec_stmts<'a>(
    stmts: &'a [Stmt],
    env: &'a mut Env,
    ctx: &'a EvalCtx<'a>,
) -> BoxFut<'a, StmtOutcome> {
    exec_stmts_prefixed(stmts, env, ctx, String::new())
}

pub fn exec_stmts_prefixed<'a>(
    stmts: &'a [Stmt],
    env: &'a mut Env,
    ctx: &'a EvalCtx<'a>,
    prefix: String,
) -> BoxFut<'a, StmtOutcome> {
    Box::pin(async move {
        let watches = collect_watches(stmts);
        let parent_node_id = ctx.current_node_id.clone();
        for (i, stmt) in stmts.iter().enumerate() {
            let node_id = if prefix.is_empty() {
                format!("{i}")
            } else {
                format!("{prefix}.{i}")
            };
            // Check for pending L4 stop / L3 redirect between statements.
            if let Some(session) = ctx.session_runtime.as_ref()
                && let Some(turn_id) = ctx.turn_id.as_ref()
            {
                if let Some(inj) = session.peek_pending_l2_or_higher(turn_id) {
                    match inj.level {
                        crate::injection::InjectionLevel::L4HardStop => {
                            session.mark_injection_consumed(&inj.id);
                            emit_flow_node_start(ctx, &node_id, stmt, parent_node_id.as_deref());
                            emit_flow_node_end(
                                ctx,
                                &node_id,
                                &StmtOutcome::Continue,
                                parent_node_id.as_deref(),
                                Some("cancelled: hard stop"),
                            );
                            return StmtOutcome::Err(RuntimeError::Cancelled(
                                "hard stop from user".into(),
                            ));
                        }
                        crate::injection::InjectionLevel::L3Redirect => {
                            if let Some(target) = inj.redirect_target.clone() {
                                session.mark_injection_consumed(&inj.id);
                                return StmtOutcome::Err(RuntimeError::Redirect(target));
                            }
                        }
                        _ => {}
                    }
                }
            }
            emit_flow_node_start(ctx, &node_id, stmt, parent_node_id.as_deref());
            let stmt_ctx = ctx.with_node(&node_id);
            let (outcome, preview) = exec_stmt(stmt, env, &stmt_ctx, &watches).await;
            emit_flow_node_end(
                ctx,
                &node_id,
                &outcome,
                parent_node_id.as_deref(),
                preview.as_deref(),
            );
            match outcome {
                StmtOutcome::Continue => continue,
                other => return other,
            }
        }
        StmtOutcome::Continue
    })
}

fn emit_flow_node_start(
    ctx: &EvalCtx<'_>,
    node_id: &str,
    stmt: &Stmt,
    parent_node_id: Option<&str>,
) {
    let (kind, label) = stmt_to_node_kind_label(stmt);
    emit_flow_node_start_raw(ctx, node_id, kind, &label, parent_node_id);
}

fn emit_flow_node_start_raw(
    ctx: &EvalCtx<'_>,
    node_id: &str,
    kind: crate::nodegraph::NodeKind,
    label: &str,
    parent_node_id: Option<&str>,
) {
    let Some(run_id) = ctx.flow_run_id.clone() else {
        return;
    };
    if let Some(sink) = ctx.events {
        sink.emit(crate::event::Event::FlowNodeStart {
            run_id: run_id.clone(),
            node_id: node_id.to_string(),
            kind: kind.clone(),
            label: label.to_string(),
            parent_node_id: parent_node_id.map(String::from),
        });
    }
    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
        let _ = tx.send(crate::stream::StreamFrame::FlowNodeStart {
            run_id: run_id.0.to_string(),
            node_id: node_id.to_string(),
            kind,
            label: label.to_string(),
            parent_node_id: parent_node_id.map(String::from),
        });
    }
}

fn value_preview(v: &Value) -> Option<String> {
    let raw = match v {
        Value::Str(s) => s.clone(),
        Value::Message(m) => {
            let text = m.text_concat();
            let tool_uses: Vec<String> = m
                .parts
                .iter()
                .filter_map(|p| match p {
                    crate::message::MessagePart::ToolUse { name, .. } => Some(name.clone()),
                    _ => None,
                })
                .collect();
            match (text.trim().is_empty(), tool_uses.is_empty()) {
                (false, true) => text,
                (false, false) => format!("{}\n\n→ tool_uses: {}", text, tool_uses.join(", ")),
                (true, false) => format!("→ tool_uses: {}", tool_uses.join(", ")),
                (true, true) => return None,
            }
        }
        Value::Path(p) => p.display().to_string(),
        Value::Int(n) => n.to_string(),
        Value::Float(n) => n.to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Unit => return None,
        Value::Err(e) => format!("err: {e}"),
        Value::List(items) => format!("list[{}]", items.len()),
        Value::Struct(fields) => format!(
            "{{{}}}",
            fields
                .iter()
                .map(|(k, _)| k.as_str())
                .collect::<Vec<_>>()
                .join(", ")
        ),
        Value::EditProposal(_) => "<edit proposal>".into(),
        Value::Lambda { .. } => "<lambda>".into(),
    };
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        None
    } else {
        Some(trimmed.chars().take(4000).collect())
    }
}

fn emit_flow_node_end(
    ctx: &EvalCtx<'_>,
    node_id: &str,
    outcome: &StmtOutcome,
    parent_node_id: Option<&str>,
    output_preview: Option<&str>,
) {
    let Some(run_id) = ctx.flow_run_id.clone() else {
        return;
    };
    let status = match outcome {
        StmtOutcome::Err(_) => crate::event::FlowNodeStatus::Err,
        StmtOutcome::LoopBreak => crate::event::FlowNodeStatus::Ok,
        StmtOutcome::LoopContinue => crate::event::FlowNodeStatus::Ok,
        _ => crate::event::FlowNodeStatus::Ok,
    };
    let preview_owned = output_preview.map(String::from);
    if let Some(sink) = ctx.events {
        sink.emit(crate::event::Event::FlowNodeEnd {
            run_id: run_id.clone(),
            node_id: node_id.to_string(),
            status: status.clone(),
            output_preview: preview_owned.clone(),
        });
    }
    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
        let _ = tx.send(crate::stream::StreamFrame::FlowNodeEnd {
            run_id: run_id.0.to_string(),
            node_id: node_id.to_string(),
            status,
            output_preview: preview_owned,
            parent_node_id: parent_node_id.map(String::from),
        });
    }
}

fn stmt_to_node_kind_label(stmt: &Stmt) -> (crate::nodegraph::NodeKind, String) {
    use crate::nodegraph::NodeKind;
    match stmt {
        Stmt::Bind { value, .. } | Stmt::Expr(value) => expr_to_node_kind_label(value),
        Stmt::Return { .. } => (NodeKind::Return, "return".into()),
        Stmt::When { cond, .. } => {
            let preview = crate::nodegraph::format_expr_short(cond);
            (
                NodeKind::When {
                    condition_preview: preview.clone(),
                },
                format!("when {preview}"),
            )
        }
        Stmt::Watch(_) => (NodeKind::Return, "watch".into()),
        Stmt::Loop { .. } => (NodeKind::Loop, "loop".into()),
        Stmt::Break => (NodeKind::Return, "break".into()),
        Stmt::Continue => (NodeKind::Return, "continue".into()),
    }
}

fn expr_to_node_kind_label(expr: &Expr) -> (crate::nodegraph::NodeKind, String) {
    use crate::nodegraph::NodeKind;
    match expr {
        Expr::Node(Node::ToolCall { path, .. })
            if path.len() == 2 && path[0].name == "llm" && path[1].name == "call" =>
        {
            (NodeKind::Llm { model: None }, "llm.call".into())
        }
        Expr::Node(Node::ToolCall { path, .. }) => {
            let p = path
                .iter()
                .map(|s| s.name.clone())
                .collect::<Vec<_>>()
                .join(".");
            (NodeKind::ToolCall { path: p.clone() }, format!("⟶ {p}"))
        }
        Expr::Node(Node::Fanout { items, collect }) => (
            NodeKind::Fanout {
                collect: (*collect).into(),
            },
            format!("fanout ×{}", items.len()),
        ),
        Expr::Node(Node::Subflow { name, .. }) => (
            NodeKind::Subflow {
                name: name.name.clone(),
            },
            format!("subflow({})", name.name),
        ),
        _ => (NodeKind::Return, "expr".into()),
    }
}

fn collect_watches(stmts: &[Stmt]) -> HashMap<String, Vec<&WatchDecl>> {
    let mut out: HashMap<String, Vec<&WatchDecl>> = HashMap::new();
    for stmt in stmts {
        if let Stmt::Watch(w) = stmt {
            out.entry(w.target.name.clone()).or_default().push(w);
        }
    }
    out
}

fn exec_stmt<'a>(
    stmt: &'a Stmt,
    env: &'a mut Env,
    ctx: &'a EvalCtx<'a>,
    watches: &'a HashMap<String, Vec<&'a WatchDecl>>,
) -> BoxFut<'a, (StmtOutcome, Option<String>)> {
    Box::pin(async move {
        match stmt {
            Stmt::Bind { name, value } => {
                let watch_target = name.as_single_ident().map(|id| id.name.clone());
                let v = if let Some(target) = watch_target.as_ref()
                    && let Some(ws) = watches.get(target)
                {
                    match eval_bind_with_watches(value, env, ctx, ws).await {
                        Ok(v) => v,
                        Err(e) => return (StmtOutcome::Err(e), None),
                    }
                } else {
                    eval_expr(value, env, ctx).await
                };
                if let Value::Err(e) = v {
                    return (StmtOutcome::Err(e), None);
                }
                let preview = value_preview(&v);
                if let Err(e) = bind_pattern(name, v, env) {
                    return (StmtOutcome::Err(e), None);
                }
                (StmtOutcome::Continue, preview)
            }
            Stmt::When { cond, body } => {
                let c = eval_expr(cond, env, ctx).await;
                let truthy = match c {
                    Value::Bool(b) => b,
                    Value::Unit => false,
                    Value::Err(e) => {
                        return (StmtOutcome::Err(e), None);
                    }
                    _ => true,
                };
                if truthy {
                    let prefix = ctx.current_node_id.clone().unwrap_or_default();
                    (
                        exec_stmts_prefixed(body, env, ctx, prefix).await,
                        Some("true".into()),
                    )
                } else {
                    (StmtOutcome::Continue, Some("false".into()))
                }
            }
            Stmt::Return { value } => {
                let v = eval_expr(value, env, ctx).await;
                if let Value::Err(e) = v {
                    return (StmtOutcome::Err(e), None);
                }
                let preview = value_preview(&v);
                (StmtOutcome::Return(v), preview)
            }
            Stmt::Expr(e) => {
                let v = eval_expr(e, env, ctx).await;
                if let Value::Err(err) = v {
                    return (StmtOutcome::Err(err), None);
                }
                let preview = value_preview(&v);
                (StmtOutcome::Continue, preview)
            }
            Stmt::Watch(_) => (StmtOutcome::Continue, None),
            Stmt::Loop { body } => {
                let loop_node_id = ctx.current_node_id.clone();
                let mut iter = 0u64;
                loop {
                    let iter_id = match &loop_node_id {
                        Some(p) => format!("{p}.iter[{iter}]"),
                        None => format!("iter[{iter}]"),
                    };
                    emit_flow_node_start_raw(
                        ctx,
                        &iter_id,
                        crate::nodegraph::NodeKind::Return,
                        &format!("iteration {iter}"),
                        loop_node_id.as_deref(),
                    );
                    let iter_ctx = ctx.with_node(&iter_id);
                    let outcome = exec_stmts_prefixed(body, env, &iter_ctx, iter_id.clone()).await;
                    let preview = match &outcome {
                        StmtOutcome::LoopBreak => Some("break"),
                        StmtOutcome::LoopContinue => Some("continue"),
                        _ => None,
                    };
                    emit_flow_node_end(ctx, &iter_id, &outcome, loop_node_id.as_deref(), preview);
                    match outcome {
                        StmtOutcome::Continue => {}
                        StmtOutcome::LoopContinue => {}
                        StmtOutcome::LoopBreak => break,
                        other => return (other, Some("loop interrupted".into())),
                    }
                    iter += 1;
                }
                (StmtOutcome::Continue, Some("loop end".into()))
            }
            Stmt::Break => (StmtOutcome::LoopBreak, Some("break".into())),
            Stmt::Continue => (StmtOutcome::LoopContinue, Some("continue".into())),
        }
    })
}

async fn eval_bind_with_watches(
    expr: &Expr,
    env: &mut Env,
    ctx: &EvalCtx<'_>,
    watches: &[&WatchDecl],
) -> Result<Value, RuntimeError> {
    let Expr::Node(Node::ToolCall { path, args }) = expr else {
        return Ok(eval_expr(expr, env, ctx).await);
    };
    if !(path.len() == 2 && path[0].name == "llm" && path[1].name == "call") {
        return Ok(eval_expr(expr, env, ctx).await);
    };

    let mut positional = Vec::new();
    let mut named = Vec::new();
    for arg in args {
        match arg {
            Arg::Positional(expr) => {
                let value = eval_expr(expr, env, ctx).await;
                if value.is_err() {
                    return Ok(value);
                }
                positional.push(value);
            }
            Arg::Named { name, value } => {
                let value = eval_expr(value, env, ctx).await;
                if value.is_err() {
                    return Ok(value);
                }
                named.push((name.name.clone(), value));
            }
        }
    }

    let registry = std::sync::Arc::new(ctx.tools.clone());
    let call_args = ToolArgs { positional, named };
    let mut tool_ctx = ctx
        .tool_ctx
        .clone()
        .with_anchors(
            ctx.turn_id.clone(),
            ctx.flow_run_id.clone(),
            ctx.events.map(|s| s.next_seq_peek()),
        )
        .with_registry(registry)
        .with_current_node(ctx.current_node_id.clone())
        .with_providers(std::sync::Arc::new(ctx.providers.clone()))
        .with_watch_rules(collect_watch_rules(watches));
    if let Some(sink) = ctx.events {
        tool_ctx = tool_ctx.with_events(sink.clone());
    }
    if let Some(session) = ctx.session_runtime.as_ref() {
        tool_ctx = tool_ctx
            .with_session_messages(session.messages_full())
            .with_session_messages_handle(session.messages_handle())
            .with_session_runtime(session.clone())
            .with_watch_hub(std::sync::Arc::clone(&session.watch_hub))
            .with_flow_registry(std::sync::Arc::clone(&session.flow_registry))
            .with_compact_lock_handle(session.compact_lock_handle());
    }
    if let Some(safety) = ctx.safety.cloned() {
        tool_ctx = tool_ctx.with_safety(safety);
    }
    if let Some(model) = &ctx.tool_ctx.current_model {
        tool_ctx = tool_ctx.with_current_model(model.clone());
    }
    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
        tool_ctx = tool_ctx.with_stream_tx(tx);
    }

    let result = crate::tools::llm_call::LlmCallTool
        .call(call_args, &tool_ctx)
        .await;
    Ok(match result {
        Ok(v) => v,
        Err(e) => Value::Err(e),
    })
}

fn render_warn_msg(msg: &Option<Expr>, fallback: &str) -> String {
    match msg {
        Some(Expr::Literal(atman_dsl::ast::Literal::Str(s))) => s.clone(),
        _ => fallback.to_string(),
    }
}

fn collect_watch_rules(watches: &[&WatchDecl]) -> WatchRules {
    let mut rules = WatchRules::default();
    for w in watches {
        for on in &w.on_blocks {
            let has_abort = on
                .actions
                .iter()
                .any(|a| matches!(a, WatchAction::Abort { .. }));
            let warn_msg_expr = on.actions.iter().find_map(|a| match a {
                WatchAction::Warn { msg } => Some(msg),
                _ => None,
            });
            if !has_abort && warn_msg_expr.is_none() {
                continue;
            }
            match &on.event {
                WatchEvent::Token { patterns } => {
                    for p in patterns {
                        if has_abort {
                            rules
                                .token_matches
                                .push((p.clone(), format!("token match: {p}")));
                        }
                        if let Some(msg_expr) = warn_msg_expr {
                            rules.warn_token.push(WarnRule {
                                target: w.target.name.clone(),
                                message: render_warn_msg(
                                    msg_expr,
                                    &format!("watch warn: token `{p}`"),
                                ),
                                pattern: p.clone(),
                            });
                        }
                    }
                }
                WatchEvent::TokensConsumed { cmp, value }
                    if matches!(cmp, CmpOp::Gt | CmpOp::Ge) =>
                {
                    let threshold = if matches!(cmp, CmpOp::Ge) {
                        value.saturating_sub(1)
                    } else {
                        *value
                    };
                    if has_abort {
                        rules.tokens_gt = Some(match rules.tokens_gt {
                            Some(existing) => existing.min(threshold),
                            None => threshold,
                        });
                    }
                    if let Some(msg_expr) = warn_msg_expr {
                        rules.warn_tokens_gt.push((
                            threshold,
                            WarnRule {
                                target: w.target.name.clone(),
                                message: render_warn_msg(
                                    msg_expr,
                                    &format!("watch warn: tokens_consumed > {threshold}"),
                                ),
                                pattern: format!("tokens_consumed>{threshold}"),
                            },
                        ));
                    }
                }
                WatchEvent::Elapsed { cmp, duration_ms }
                    if matches!(cmp, CmpOp::Gt | CmpOp::Ge) =>
                {
                    let threshold = if matches!(cmp, CmpOp::Ge) {
                        duration_ms.saturating_sub(1)
                    } else {
                        *duration_ms
                    };
                    if has_abort {
                        rules.elapsed_ms_gt = Some(match rules.elapsed_ms_gt {
                            Some(existing) => existing.min(threshold),
                            None => threshold,
                        });
                    }
                    if let Some(msg_expr) = warn_msg_expr {
                        rules.warn_elapsed_ms_gt.push((
                            threshold,
                            WarnRule {
                                target: w.target.name.clone(),
                                message: render_warn_msg(
                                    msg_expr,
                                    &format!("watch warn: elapsed > {threshold}ms"),
                                ),
                                pattern: format!("elapsed>{threshold}ms"),
                            },
                        ));
                    }
                }
                _ => {}
            }
        }
    }
    rules
}

pub async fn exec_flow(
    flow: &FlowDecl,
    args: Vec<(String, Value)>,
    tools: &ToolRegistry,
    tool_ctx: &ToolCtx,
    providers: &crate::provider::ProviderRegistry,
    source_dir: Option<PathBuf>,
) -> Result<Value, RuntimeError> {
    let flows = std::collections::HashMap::new();
    exec_flow_with_siblings(
        flow,
        args,
        tools,
        tool_ctx,
        providers,
        &flows,
        None,
        None,
        None,
        None,
        tokio_util::sync::CancellationToken::new(),
        None,
        source_dir,
    )
    .await
}

#[allow(clippy::too_many_arguments)]
pub async fn exec_flow_with_siblings(
    flow: &FlowDecl,
    args: Vec<(String, Value)>,
    tools: &ToolRegistry,
    tool_ctx: &ToolCtx,
    providers: &crate::provider::ProviderRegistry,
    flows: &std::collections::HashMap<String, FlowDecl>,
    events: Option<&crate::event::EventSink>,
    turn_id: Option<crate::event::TurnId>,
    flow_run_id: Option<crate::event::FlowRunId>,
    session: Option<std::sync::Arc<crate::session::Session>>,
    flow_cancel: tokio_util::sync::CancellationToken,
    safety: Option<&crate::safety::SafetyConfig>,
    source_dir: Option<PathBuf>,
) -> Result<Value, RuntimeError> {
    let ctx = EvalCtx {
        tools,
        tool_ctx,
        providers,
        flows,
        contract: flow.contract.as_ref(),
        events,
        turn_id,
        flow_run_id,
        session_runtime: session,
        flow_cancel,
        safety,
        current_node_id: None,
        source_dir,
    };
    let mut env = Env::new();
    let provided: std::collections::HashSet<String> = args.iter().map(|(n, _)| n.clone()).collect();
    for (name, value) in args {
        env.bind(name, value);
    }
    for p in &flow.params {
        if !provided.contains(&p.name.name) {
            if let Some(default) = &p.default {
                let val = crate::eval::eval_expr(default, &env, &ctx).await;
                env.bind(p.name.name.clone(), val);
            }
        }
    }
    let raw_outcome = exec_stmts(&flow.body, &mut env, &ctx).await;
    match raw_outcome {
        StmtOutcome::Return(v) => Ok(v),
        StmtOutcome::Err(e) => Err(e),
        StmtOutcome::Continue => Ok(Value::Unit),
        StmtOutcome::LoopBreak => Err(RuntimeError::ToolFailed("break outside loop".into())),
        StmtOutcome::LoopContinue => Err(RuntimeError::ToolFailed("continue outside loop".into())),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use atman_dsl::parse::parse_file;

    async fn run(src: &str, args: Vec<(String, Value)>) -> Result<Value, RuntimeError> {
        let file = parse_file(src).expect("parse test src");
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        exec_flow(&file.flows[0], args, &tools, &tool_ctx, &providers, None).await
    }

    #[tokio::test]
    async fn bind_and_return() {
        let out = run(
            r#"flow t() -> Int {
    x = 1
    y = x + 2
    return y
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(3)));
    }

    #[tokio::test]
    async fn when_true_executes_body() {
        let out = run(
            r#"flow t() -> Int {
    x = 5
    when x > 3 {
        return 42
    }
    return 0
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(42)));
    }

    #[tokio::test]
    async fn when_false_skips_body() {
        let out = run(
            r#"flow t() -> Int {
    x = 1
    when x > 3 {
        return 42
    }
    return 0
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(0)));
    }

    #[tokio::test]
    async fn err_in_bind_stops_flow() {
        let err = run(
            r#"flow t() -> Int {
    x = missing
    return 1
}
"#,
            vec![],
        )
        .await
        .unwrap_err();
        assert!(matches!(err, RuntimeError::UndefinedVar(n) if n == "missing"));
    }

    #[tokio::test]
    async fn flow_args_bind_before_body() {
        let out = run(
            r#"flow t() -> Int {
    return n + 1
}
"#,
            vec![("n".into(), Value::Int(4))],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(5)));
    }

    #[tokio::test]
    async fn when_cond_unit_is_falsy() {
        let out = run(
            r#"flow t() -> Int {
    when 1 {
        return 1
    }
    return 0
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(1)));
    }

    #[tokio::test]
    async fn when_cond_struct_is_truthy() {
        let out = run(
            r#"flow t() -> Int {
    x = { a: 1 }
    when x {
        return 42
    }
    return 0
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(42)));
    }

    #[tokio::test]
    async fn flow_falls_through_to_unit_without_return() {
        let out = run(
            r#"flow t() {
    x = 1
}
"#,
            vec![],
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Unit));
    }
}