atman_runtime/eval/

mod.rs

mod llm_args;
mod llm_context;

use atman_dsl::ast::{Arg, BinOp, Expr, Literal, Node, UnOp};

use crate::env::Env;
use crate::error::RuntimeError;
use crate::streaming::LlmStream;
use crate::tool::{BoxFut, ToolArgs, ToolCtx, ToolRegistry};
use crate::value::Value;

#[derive(Clone)]
pub struct EvalCtx<'a> {
    pub tools: &'a ToolRegistry,
    pub tool_ctx: &'a ToolCtx,
    pub providers: &'a crate::provider::ProviderRegistry,
    pub flows: &'a std::collections::HashMap<String, atman_dsl::ast::FlowDecl>,
    pub contract: Option<&'a atman_dsl::ast::Contract>,
    pub events: Option<&'a crate::event::EventSink>,
    pub turn_id: Option<crate::event::TurnId>,
    pub flow_run_id: Option<crate::event::FlowRunId>,
    pub session_runtime: Option<std::sync::Arc<crate::session::Session>>,
    pub flow_cancel: tokio_util::sync::CancellationToken,
    pub safety: Option<&'a crate::safety::SafetyConfig>,
    pub current_node_id: Option<String>,
    /// Directory of the .at source file. When set, relative `@` paths
    /// are resolved against this directory instead of the process CWD.
    pub source_dir: Option<std::path::PathBuf>,
}

impl<'a> EvalCtx<'a> {
    pub fn with_node(&self, node_id: impl Into<String>) -> Self {
        let mut c = self.clone();
        c.current_node_id = Some(node_id.into());
        c
    }
}

pub fn eval_expr<'a>(expr: &'a Expr, env: &'a Env, ctx: &'a EvalCtx<'a>) -> BoxFut<'a, Value> {
    Box::pin(async move { eval_expr_inner(expr, env, ctx).await })
}

#[derive(Clone, Copy, Debug)]
pub(crate) enum ContextMode {
    None,
    Session,
    SessionRecent(usize),
}

fn parse_context_mode(s: &str) -> ContextMode {
    match s.trim() {
        "session" => ContextMode::Session,
        "none" | "" => ContextMode::None,
        other if other.starts_with("session_recent") => {
            let rest = &other["session_recent".len()..];
            let rest = rest
                .trim()
                .trim_start_matches('(')
                .trim_end_matches(')')
                .trim();
            let n: usize = rest.parse().unwrap_or(10);
            ContextMode::SessionRecent(n.max(1))
        }
        _ => ContextMode::None,
    }
}

fn is_context_overflow_error(err: &RuntimeError) -> bool {
    let RuntimeError::ToolFailed(msg) = err else {
        return false;
    };
    let msg = msg.to_ascii_lowercase();
    msg.contains("maximum context length")
        || msg.contains("context overflow")
        || msg.contains("context window")
        || msg.contains("context length")
        || msg.contains("prompt is too long")
        || msg.contains("input is too long")
        || msg.contains("too many tokens")
}

fn rebuild_session_llm_messages(
    session: &crate::session::Session,
    context_mode: ContextMode,
    turn_id: &crate::event::TurnId,
    prompt: Option<&str>,
    extra_messages: &[crate::message::Message],
) -> Vec<crate::message::Message> {
    let mut messages = match context_mode {
        ContextMode::Session => session.messages().to_vec(),
        ContextMode::SessionRecent(n) => {
            let all = session.messages();
            let start = all.len().saturating_sub(n);
            all[start..].to_vec()
        }
        ContextMode::None => Vec::new(),
    };
    if let Some(prompt) = prompt
        && !prompt.is_empty()
    {
        messages.push(crate::message::Message::user_text(
            turn_id.clone(),
            prompt.to_string(),
        ));
    }
    messages.extend_from_slice(extra_messages);
    messages
}

async fn session_system_context(session: &crate::session::Session) -> Vec<String> {
    let mut parts = Vec::new();
    if let Some(goal) = session.goal() {
        parts.push(format!("[session goal]\n{goal}\n[/session goal]"));
    }
    if let Some(cwd_note) = working_directory_system_prompt(session) {
        parts.push(cwd_note);
    }
    if let Some(plan) = session.plan_system_prompt().await {
        parts.push(format!(
            "[active plan]\n{plan}\n[/active plan]\n\nCall plan.tick to mark a step done. Call plan.write to revise."
        ));
    }
    if let Some(model_info) = available_models_system_prompt() {
        parts.push(model_info);
    }
    parts
}

fn append_system_context(system: &mut Option<String>, parts: Vec<String>) {
    if parts.is_empty() {
        return;
    }
    match system {
        Some(existing) if !existing.is_empty() => {
            existing.push_str("\n\n");
            existing.push_str(&parts.join("\n\n"));
        }
        Some(existing) => {
            *existing = parts.join("\n\n");
        }
        None => {
            *system = Some(parts.join("\n\n"));
        }
    }
}

async fn eval_expr_inner<'a>(expr: &'a Expr, env: &'a Env, ctx: &'a EvalCtx<'a>) -> Value {
    match expr {
        Expr::Literal(lit) => eval_literal(lit),
        Expr::Ident(id) => match env.lookup(&id.name) {
            Some(v) => v.clone(),
            None => Value::Err(RuntimeError::UndefinedVar(id.name.clone())),
        },
        Expr::FileRef(f) => {
            let path = if std::path::Path::new(&f.path).is_relative() {
                if let Some(dir) = &ctx.source_dir {
                    dir.join(&f.path)
                } else {
                    std::path::PathBuf::from(&f.path)
                }
            } else {
                std::path::PathBuf::from(&f.path)
            };
            match tokio::fs::read_to_string(&path).await {
                Ok(s) => Value::Str(s),
                Err(e) => Value::Err(RuntimeError::ToolFailed(format!(
                    "@\"{}\": {e}",
                    path.display()
                ))),
            }
        }
        Expr::Member { base, field } => {
            let base_v = eval_expr(base, env, ctx).await;
            if base_v.is_err() {
                return base_v;
            }
            match base_v.field(&field.name) {
                Some(v) => v.clone(),
                None => Value::Err(RuntimeError::UndefinedVar(format!(".{}", field.name))),
            }
        }
        Expr::Binary { op, left, right } => {
            let l = eval_expr(left, env, ctx).await;
            if l.is_err() {
                return l;
            }
            let r = eval_expr(right, env, ctx).await;
            if r.is_err() {
                return r;
            }
            eval_binop(*op, &l, &r)
        }
        Expr::Unary { op, operand } => {
            let v = eval_expr(operand, env, ctx).await;
            if v.is_err() {
                return v;
            }
            eval_unop(*op, &v)
        }
        Expr::List(items) => {
            let mut acc = Vec::with_capacity(items.len());
            for item in items {
                let v = eval_expr(item, env, ctx).await;
                if v.is_err() {
                    return v;
                }
                acc.push(v);
            }
            Value::List(acc)
        }
        Expr::Struct(fields) => {
            let mut acc = Vec::with_capacity(fields.len());
            for (k, v) in fields {
                let val = eval_expr(v, env, ctx).await;
                if val.is_err() {
                    return val;
                }
                acc.push((k.name.clone(), val));
            }
            Value::Struct(acc)
        }
        Expr::Node(node) => eval_node(node, env, ctx).await,
        Expr::Call { .. } => Value::Err(RuntimeError::ToolFailed(
            "bare function call not supported; use namespaced tool call".into(),
        )),
        Expr::Pipe { lhs, rhs } => eval_pipe(lhs, rhs, env, ctx).await,
    }
}

async fn eval_pipe<'a>(lhs: &'a Expr, rhs: &'a Expr, env: &'a Env, ctx: &'a EvalCtx<'a>) -> Value {
    let piped = eval_expr(lhs, env, ctx).await;
    if piped.is_err() {
        return piped;
    }
    match rhs {
        Expr::Node(Node::ToolCall { path, args }) => {
            dispatch_tool_call(path, args, vec![piped], env, ctx).await
        }
        other => Value::Err(RuntimeError::ToolFailed(format!(
            "pipe rhs must be a tool call like `ns.tool(...)`, got {}",
            expr_shape(other)
        ))),
    }
}

pub(crate) fn expr_shape(e: &Expr) -> &'static str {
    match e {
        Expr::Literal(_) => "literal",
        Expr::Ident(_) => "identifier",
        Expr::FileRef(_) => "file ref",
        Expr::Member { .. } => "member access",
        Expr::Binary { .. } => "binary expr",
        Expr::Unary { .. } => "unary expr",
        Expr::Call { .. } => "bare call",
        Expr::Pipe { .. } => "pipe expr",
        Expr::Struct(_) => "struct literal",
        Expr::List(_) => "list literal",
        Expr::Node(_) => "flow node",
    }
}

async fn dispatch_tool_call<'a>(
    path: &'a [atman_dsl::ast::Ident],
    args: &'a [Arg],
    prefix_positional: Vec<Value>,
    env: &'a Env,
    ctx: &'a EvalCtx<'a>,
) -> Value {
    if ctx.flow_cancel.is_cancelled() {
        return Value::Err(RuntimeError::Cancelled("flow cancelled by user".into()));
    }
    let name = tool_name(path);
    let tool = match ctx.tools.get(&name) {
        Some(t) => t,
        None => {
            if is_type_annotation(path) {
                return Value::Unit;
            }
            return Value::Err(RuntimeError::UndefinedTool(name));
        }
    };
    if matches!(tool.tier(), crate::tool::Tier::Four) && !contract_allows_shell(ctx.contract) {
        return Value::Err(RuntimeError::ToolFailed(format!(
            "tool `{name}` is Tier 4 (shell); flow contract must declare `capabilities {{ shell: true }}`"
        )));
    }
    let mut positional = prefix_positional;
    let mut named = Vec::new();
    for arg in args {
        match arg {
            Arg::Positional(e) => {
                let v = eval_expr(e, env, ctx).await;
                if v.is_err() {
                    return v;
                }
                positional.push(v);
            }
            Arg::Named { name, value } => {
                let v = eval_expr(value, env, ctx).await;
                if v.is_err() {
                    return v;
                }
                named.push((name.name.clone(), v));
            }
        }
    }
    let ctx_with_anchors = 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(std::sync::Arc::new(ctx.tools.clone()));
    let ctx_with_anchors = if let Some(sink) = ctx.events {
        ctx_with_anchors.with_events(sink.clone())
    } else {
        ctx_with_anchors
    };
    let ctx_with_anchors = if matches!(tool.tier(), crate::tool::Tier::Four) {
        ctx_with_anchors
    } else {
        let mut c = ctx_with_anchors;
        c.sandbox = None;
        c
    };
    let ctx_with_anchors = if let Some(session) = ctx.session_runtime.as_ref() {
        ctx_with_anchors
            .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())
    } else {
        ctx_with_anchors
    };
    let ctx_with_anchors = ctx_with_anchors.with_current_node(ctx.current_node_id.clone());
    let ctx_with_anchors =
        ctx_with_anchors.with_providers(std::sync::Arc::new(ctx.providers.clone()));
    let mut ctx_with_anchors = if let Some(model) = &ctx.tool_ctx.current_model {
        ctx_with_anchors.with_current_model(model.clone())
    } else {
        ctx_with_anchors
    };
    if let Some(tx) = ctx.tool_ctx.stream_tx.clone() {
        ctx_with_anchors = ctx_with_anchors.with_stream_tx(tx);
    }
    let ctx_with_anchors = if let Some(session) = ctx.session_runtime.as_ref() {
        let mut c = ctx_with_anchors
            .with_read_files(session.read_files())
            .with_approval(session.approval())
            .with_session_dir(session.dir().to_path_buf())
            .with_session_id(session.id().to_string());
        if let Some(idx) = session.project_index() {
            c = c.with_project_index(idx);
        }
        c = c.with_fs_access(session_fs_access_policy(session));
        c = c.with_forms(session.forms());
        {
            let s = session.clone();
            c.on_memory_recent = Some(std::sync::Arc::new(move |count| {
                s.set_memory_recent_count(count);
            }));
        }
        {
            let store = std::sync::Arc::new(crate::history_store::HistoryStoreImpl::new(
                session.project_index(),
                Some(session.clone()),
                Some(session.id().to_string()),
                Some(
                    session
                        .dir()
                        .parent()
                        .map(|p| p.to_path_buf())
                        .unwrap_or_default(),
                ),
            ));
            c = c.with_history_store(store);
        }
        c
    } else {
        ctx_with_anchors
    };
    let stream_tx = ctx.session_runtime.as_ref().map(|s| s.stream_tx());
    let tool_call_id = uuid::Uuid::now_v7().to_string();
    let args_preview = preview_tool_args(&positional, &named);
    if let (Some(sink), Some(run_id), Some(parent_node)) =
        (ctx.events, ctx.flow_run_id.clone(), &ctx.current_node_id)
    {
        sink.emit(crate::event::Event::ToolNode {
            run_id: run_id.clone(),
            parent_node_id: parent_node.clone(),
            tool_use_id: tool_call_id.clone(),
            tool_name: name.clone(),
            args_preview: args_preview.clone(),
        });
        if let Some(tx) = &stream_tx {
            let _ = tx.send(crate::stream::StreamFrame::ToolNode {
                run_id: run_id.0.to_string(),
                parent_node_id: parent_node.clone(),
                tool_use_id: tool_call_id.clone(),
                tool: name.clone(),
                args_preview: args_preview.clone(),
            });
        }
    }
    if let Some(tx) = &stream_tx {
        let _ = tx.send(crate::stream::StreamFrame::ToolUseStart {
            tool: name.clone(),
            args_preview: args_preview.clone(),
            id: tool_call_id.clone(),
        });
    }
    let call_args = ToolArgs { positional, named };
    let diff_preview = prepare_diff_preview(&name, &call_args);
    let level = tool.approval_level(&call_args, &ctx_with_anchors);
    let gate = crate::approval::request_approval(
        &ctx_with_anchors,
        &tool_call_id,
        &name,
        &call_args,
        level,
        Some(tool.as_ref()),
    )
    .await;
    let outcome = match gate {
        crate::approval::ApprovalOutcome::Deny { reason } => Err(RuntimeError::ToolFailed(
            format!("tool `{name}` denied by user: {reason}"),
        )),
        crate::approval::ApprovalOutcome::Approve => tool.call(call_args, &ctx_with_anchors).await,
    };
    if let Some(tx) = &stream_tx {
        let (ok, preview) = match &outcome {
            Ok(v) => (true, preview_tool_value(v)),
            Err(e) => (false, format!("{e}")),
        };
        let _ = tx.send(crate::stream::StreamFrame::ToolUseDone {
            tool: name.clone(),
            ok,
            preview,
            id: tool_call_id,
        });
    }
    if let Some(session) = ctx.session_runtime.as_ref()
        && (name == "memory.todo.set" || name == "memory.todo.done")
    {
        session.refresh_todos_from_store_async().await;
    }
    if let Some(session) = ctx.session_runtime.as_ref()
        && (name == "plan.write" || name == "plan.tick")
    {
        session.refresh_plans_from_store_async().await;
    }
    if let (Some(sink), Ok(value)) = (ctx.events, &outcome) {
        if let Some((title, old_content, new_content, unified_diff)) =
            complete_diff_preview(diff_preview, &name, value)
        {
            sink.emit(crate::event::Event::DiffPreview {
                turn_id: ctx.turn_id.clone(),
                flow_run_id: ctx.flow_run_id.clone(),
                title,
                old_content,
                new_content,
                unified_diff,
            });
        }
    }
    match outcome {
        Ok(v) => v,
        Err(e) => Value::Err(e),
    }
}

type DiffPreviewData = (String, Option<String>, Option<String>, Option<String>);

fn prepare_diff_preview(name: &str, args: &ToolArgs) -> Option<DiffPreviewData> {
    match name {
        "fs.write" => {
            let path = tool_arg_path(args, "path", 0)?;
            let content = tool_arg_string(args, "content", 1)?;
            let path_str = path.display().to_string();
            let diff = match std::fs::read_to_string(&path).ok() {
                Some(old) => crate::tools::fs::unified_diff_preview(&path_str, &old, &content),
                None => format!("+++ {path_str}\n{content}"),
            };
            Some((path_str, None, None, Some(diff)))
        }
        "fs.edit" => {
            let path = tool_arg_path(args, "path", 0)?;
            let old_string = tool_arg_string(args, "old_string", 1)?;
            let new_string = tool_arg_string(args, "new_string", 2)?;
            let replace_all = matches!(args.named("replace_all"), Some(Value::Bool(true)));
            let old = std::fs::read_to_string(&path).ok()?;
            let new = if replace_all {
                old.replace(&old_string, &new_string)
            } else {
                old.replacen(&old_string, &new_string, 1)
            };
            let path_str = path.display().to_string();
            let diff = crate::tools::fs::unified_diff_preview(&path_str, &old, &new);
            Some((path_str, None, None, Some(diff)))
        }
        _ => None,
    }
}

fn complete_diff_preview(
    prepared: Option<DiffPreviewData>,
    name: &str,
    value: &Value,
) -> Option<DiffPreviewData> {
    if prepared.is_some() {
        return prepared;
    }
    match name {
        "git.diff" => Some((
            "git diff".into(),
            None,
            None,
            value_struct_string(value, "diff"),
        )),
        "git.show" => Some((
            value_struct_string(value, "sha")
                .map(|sha| format!("git show {sha}"))
                .unwrap_or_else(|| "git show".into()),
            None,
            None,
            value_struct_string(value, "diff"),
        )),
        "git.log" => Some((
            "git log HEAD".into(),
            None,
            None,
            value_struct_string(value, "diff"),
        )),
        _ => None,
    }
}

fn tool_arg_string(args: &ToolArgs, name: &str, pos: usize) -> Option<String> {
    let value = args.named(name).or_else(|| args.positional.get(pos))?;
    match value {
        Value::Str(s) => Some(s.clone()),
        _ => None,
    }
}

fn tool_arg_path(args: &ToolArgs, name: &str, pos: usize) -> Option<std::path::PathBuf> {
    let value = args.named(name).or_else(|| args.positional.get(pos))?;
    match value {
        Value::Path(p) => Some(p.clone()),
        Value::Str(s) => Some(std::path::PathBuf::from(s)),
        _ => None,
    }
}

fn value_struct_string(value: &Value, name: &str) -> Option<String> {
    let Value::Struct(fields) = value else {
        return None;
    };
    fields.iter().find_map(|(k, v)| match (k.as_str(), v) {
        (key, Value::Str(s)) if key == name => Some(s.clone()),
        _ => None,
    })
}

#[derive(Default)]
struct StreamCallCtx<'a> {
    session: Option<&'a crate::session::Session>,
    stream_tx: Option<tokio::sync::broadcast::Sender<crate::stream::StreamFrame>>,
    flow_run_id: Option<&'a crate::event::FlowRunId>,
    agent_entry: Option<&'a std::sync::Arc<crate::tools::agent_ctrl::FlowEntry>>,
    event_sink: Option<&'a crate::event::EventSink>,
    turn_id: Option<crate::event::TurnId>,
}

async fn call_and_maybe_stream(
    provider: &dyn crate::provider::Provider,
    req: crate::provider::LlmRequest,
    stream_ctx: StreamCallCtx<'_>,
) -> Result<crate::provider::AssistantMessage, RuntimeError> {
    let stream_tx = stream_ctx
        .stream_tx
        .or_else(|| stream_ctx.session.map(|s| s.stream_tx()));
    let mut stream = LlmStream::new(provider, req)
        .with_stream_tx(stream_tx)
        .with_event_sink(stream_ctx.event_sink)
        .with_turn_id(stream_ctx.turn_id)
        .with_flow_run_id(stream_ctx.flow_run_id.cloned());
    if let Some(session) = stream_ctx.session {
        stream = stream.with_session(session);
    }
    if let Some(entry) = stream_ctx.agent_entry {
        stream = stream.with_entry(entry);
    }
    let result = stream.run().await;
    if let (Some(sess), Err(RuntimeError::AttachmentError { reason })) =
        (stream_ctx.session, &result)
    {
        let count = sess.record_attachment_degrade(reason);
        if count > 0 {
            let _ = sess.stream_tx().send(crate::stream::StreamFrame::Note(format!(
                "attachment degraded ({reason}); {count} image part(s) replaced. re-issue your last message to retry without them."
            )));
        }
    }
    result
}

fn preview_tool_args(positional: &[Value], named: &[(String, Value)]) -> String {
    let mut parts: Vec<String> = positional.iter().map(preview_tool_value).collect();
    for (k, v) in named {
        parts.push(format!("{k}={}", preview_tool_value(v)));
    }
    truncate(&parts.join(", "), 4000)
}

fn available_models_system_prompt() -> Option<String> {
    let mut aliases = crate::model_registry::all_aliases();
    let mut models = crate::model_registry::all_model_entries();
    if aliases.is_empty() && models.is_empty() {
        return None;
    }
    aliases.sort_by(|a, b| a.0.cmp(&b.0));
    models.sort_by(|a, b| a.0.cmp(&b.0));
    let mut lines = vec!["[available models]".to_string()];
    if !aliases.is_empty() {
        lines.push(format!(
            "Aliases: {}",
            aliases
                .into_iter()
                .map(|(alias, model)| format!("{alias} -> {model}"))
                .collect::<Vec<_>>()
                .join(", ")
        ));
    }
    if !models.is_empty() {
        lines.push(format!(
            "Models: {}",
            models
                .into_iter()
                .map(|(name, _)| {
                    let info = crate::model_registry::model_info(&name);
                    let thinking = if info.thinking_enabled() {
                        ", thinking"
                    } else {
                        ""
                    };
                    format!(
                        "{} ({} context{})",
                        info.name,
                        crate::humanize::format_count(info.context_budget),
                        thinking
                    )
                })
                .collect::<Vec<_>>()
                .join(", ")
        ));
    }
    lines.push("Use these names or aliases with flow.spawn's model parameter.".into());
    lines.push("[/available models]".into());
    Some(lines.join("\n"))
}

fn working_directory_system_prompt(session: &crate::session::Session) -> Option<String> {
    let meta = session.meta()?;
    let cwd = meta
        .start_path
        .as_deref()
        .or(meta.project_root.as_deref())?;
    let mut lines = vec!["[working directory]".to_string()];
    lines.push(cwd.display().to_string());
    // Live re-detect in case .git/.atman was created after session start.
    let live_root = crate::session_meta::find_project_root(cwd);
    match (&live_root, &meta.project_root) {
        (Some(live), Some(stored)) if live == stored => {
            if Some(live.as_path()) != meta.start_path.as_deref() {
                lines.push(format!("project root: {}", live.display()));
            }
        }
        (Some(live), _) => {
            lines.push(format!("project root: {}", live.display()));
        }
        (None, Some(stored)) => {
            lines.push(format!(
                "project root (cached): {} (no longer detected)",
                stored.display()
            ));
        }
        (None, None) => {
            lines.push("(no project root — no .git or .atman found)".into());
        }
    }
    lines.push("[/working directory]".into());
    Some(lines.join("\n"))
}

fn preview_tool_value(v: &Value) -> String {
    let raw = match v {
        Value::Str(s) => format!("{s:?}"),
        Value::Int(n) => n.to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Float(f) => f.to_string(),
        Value::Unit => "()".into(),
        Value::List(items) => format!("list[{}]", items.len()),
        Value::Struct(f) => {
            let stdout = f
                .iter()
                .find(|(k, _)| k == "stdout")
                .and_then(|(_, v)| match v {
                    Value::Str(s) => Some(s.as_str()),
                    _ => None,
                });
            let stderr = f
                .iter()
                .find(|(k, _)| k == "stderr")
                .and_then(|(_, v)| match v {
                    Value::Str(s) => Some(s.as_str()),
                    _ => None,
                });
            let exit = f
                .iter()
                .find(|(k, _)| k == "exit")
                .and_then(|(_, v)| match v {
                    Value::Int(n) => Some(*n),
                    _ => None,
                });
            if let (Some(stdout), Some(exit)) = (stdout, exit) {
                let combined = if stdout.is_empty() {
                    stderr.unwrap_or("").to_string()
                } else {
                    stdout.to_string()
                };
                let lines: Vec<&str> = combined.lines().collect();
                if lines.len() > 10 {
                    format!(
                        "exit={exit}\n{}\n… ({} more lines, see `atman logs` for full output)",
                        lines[..10].join("\n"),
                        lines.len() - 10
                    )
                } else {
                    format!("exit={exit}\n{combined}")
                }
            } else {
                format!("struct[{}]", f.len())
            }
        }
        Value::Message(_) => "<message>".into(),
        Value::Err(e) => format!("err({e})"),
        Value::Path(p) => format!("{p:?}"),
        Value::EditProposal(_) => "<edit_proposal>".into(),
    };
    truncate(&raw, 2000)
}

fn truncate(s: &str, max: usize) -> String {
    if s.chars().count() <= max {
        return s.to_string();
    }
    let mut out: String = s.chars().take(max).collect();
    out.push('…');
    out
}

async fn eval_node<'a>(node: &'a Node, env: &'a Env, ctx: &'a EvalCtx<'a>) -> Value {
    if ctx.flow_cancel.is_cancelled() {
        return Value::Err(RuntimeError::Cancelled("flow cancelled by user".into()));
    }
    match node {
        Node::ToolCall { path, args } => dispatch_tool_call(path, args, Vec::new(), env, ctx).await,
        Node::Fanout { items, collect } => match collect {
            atman_dsl::ast::FanoutCollect::All => {
                let parent_id = ctx.current_node_id.clone();
                let branch_ctxs: Vec<EvalCtx<'a>> = (0..items.len())
                    .map(|i| {
                        let branch_id = match &parent_id {
                            Some(p) => format!("{p}.branch[{i}]"),
                            None => format!("branch[{i}]"),
                        };
                        if let (Some(sink), Some(run_id)) = (ctx.events, ctx.flow_run_id.clone()) {
                            sink.emit(crate::event::Event::FlowNodeStart {
                                run_id: run_id.clone(),
                                node_id: branch_id.clone(),
                                kind: crate::nodegraph::NodeKind::UserConfirm,
                                label: format!("branch[{i}]"),
                                parent_node_id: parent_id.clone(),
                            });
                            if let Some(tx) = ctx.tool_ctx.stream_tx.as_ref() {
                                let _ = tx.send(crate::stream::StreamFrame::FlowNodeStart {
                                    run_id: run_id.0.to_string(),
                                    node_id: branch_id.clone(),
                                    kind: crate::nodegraph::NodeKind::UserConfirm,
                                    label: format!("branch[{i}]"),
                                    parent_node_id: parent_id.clone(),
                                });
                            }
                        }
                        ctx.with_node(branch_id)
                    })
                    .collect();
                let futs = items
                    .iter()
                    .zip(branch_ctxs.iter())
                    .map(|(item, bctx)| eval_expr(item, env, bctx));
                let results: Vec<Value> = futures::future::join_all(futs).await;
                for (bctx, v) in branch_ctxs.iter().zip(results.iter()) {
                    if let (Some(sink), Some(run_id), Some(bid)) =
                        (ctx.events, ctx.flow_run_id.clone(), &bctx.current_node_id)
                    {
                        let status = if v.is_err() {
                            crate::event::FlowNodeStatus::Err
                        } else {
                            crate::event::FlowNodeStatus::Ok
                        };
                        sink.emit(crate::event::Event::FlowNodeEnd {
                            run_id: run_id.clone(),
                            node_id: bid.clone(),
                            status: status.clone(),
                            output_preview: None,
                        });
                        if let Some(tx) = ctx.tool_ctx.stream_tx.as_ref() {
                            let _ = tx.send(crate::stream::StreamFrame::FlowNodeEnd {
                                run_id: run_id.0.to_string(),
                                node_id: bid.clone(),
                                status,
                                output_preview: None,
                                parent_node_id: parent_id.clone(),
                            });
                        }
                    }
                }
                for v in &results {
                    if let Value::Err(e) = v {
                        return Value::Err(e.clone());
                    }
                }
                Value::List(results)
            }
            atman_dsl::ast::FanoutCollect::First => Value::Err(RuntimeError::ToolFailed(
                "fanout collect: first not yet implemented".into(),
            )),
        },
        Node::Llm { kwargs } => {
            let args = match llm_args::parse_llm_args(kwargs, env, ctx).await {
                Ok(args) => args,
                Err(v) => return v,
            };
            let Some(model) = args.model.clone() else {
                return Value::Err(RuntimeError::MissingArg("llm.model".into()));
            };
            let mut system = args.system.clone();
            let input = args.input.clone();
            let retry_count = args.retry_count;
            let retry_kinds = args.retry_kinds.clone();
            let cache_prompt = args.cache_prompt;
            let context_mode = parse_context_mode(&args.context_mode);
            let tool_specs = args.tool_specs.clone();
            let stall_timeout_secs = args.stall_timeout_secs;
            if args.messages_override.is_some() && args.prompt.is_some() {
                return Value::Err(RuntimeError::ToolFailed(
                    "llm node: cannot specify both `messages:` and `prompt:` (pick one)".into(),
                ));
            }
            if !matches!(context_mode, ContextMode::None) && args.messages_override.is_some() {
                return Value::Err(RuntimeError::ToolFailed(
                    "llm node: cannot specify both `messages:` and `context:` (pick one)".into(),
                ));
            }
            let Some(provider) = ctx.providers.resolve(&model) else {
                return Value::Err(RuntimeError::ToolFailed(format!(
                    "no provider registered for model `{model}`"
                )));
            };
            let has_messages_override = args.messages_override.is_some();
            if !matches!(context_mode, ContextMode::None)
                && !has_messages_override
                && let Some(session) = ctx.session_runtime.as_ref()
            {
                crate::compaction::start_auto_compact(
                    session.clone(),
                    model.clone(),
                    ctx.providers.clone(),
                )
                .await;
            }
            let mut compact_guard = if !matches!(context_mode, ContextMode::None)
                && !has_messages_override
                && let Some(session) = ctx.session_runtime.as_ref()
            {
                Some(session.acquire_compact_lock().await)
            } else {
                None
            };
            let turn_id = ctx
                .turn_id
                .clone()
                .unwrap_or_else(crate::event::TurnId::now);
            let llm_context = match llm_context::build_llm_context(
                &args,
                context_mode,
                ctx.session_runtime.as_ref(),
                ctx.tool_ctx.session_messages_handle.as_ref(),
                &turn_id,
                ctx.events,
                ctx.flow_run_id.as_ref(),
            ) {
                Ok(context) => context,
                Err(v) => return v,
            };
            let mut final_messages = llm_context.messages;
            let prompt_for_budget = llm_context.budget_text;
            let session_messages_len = llm_context.session_messages_len;
            if let Some(session) = ctx.session_runtime.as_ref()
                && let Some(l3_or_l2) = session.peek_pending_l2_or_higher(&turn_id)
                && matches!(l3_or_l2.level, crate::injection::InjectionLevel::L3Redirect)
                && let Some(target) = &l3_or_l2.redirect_target
            {
                session.mark_injection_consumed(&l3_or_l2.id);
                return Value::Err(RuntimeError::Redirect(target.clone()));
            }
            if let Some(session) = ctx.session_runtime.as_ref() {
                let injections = session.drain_injections(&turn_id);
                let renderable: Vec<crate::injection::Injection> = injections
                    .into_iter()
                    .filter(|i| {
                        matches!(
                            i.level,
                            crate::injection::InjectionLevel::L1Nudge
                                | crate::injection::InjectionLevel::L2CourseCorrect
                        )
                    })
                    .collect();
                if !renderable.is_empty() {
                    let rendered = render_injections(&renderable);
                    final_messages.push(crate::message::Message::user_text(
                        turn_id.clone(),
                        rendered,
                    ));
                }
            }
            let prompt = prompt_for_budget;
            let mut rewrite_used = false;
            if let Some(session) = ctx.session_runtime.as_ref() {
                append_system_context(&mut system, session_system_context(session).await);
            }
            if let Some(safety) = ctx.safety
                && safety.enabled
            {
                let scan_text = final_messages
                    .last()
                    .map(|m| m.text_concat())
                    .unwrap_or_else(|| prompt.clone());
                let verdict = match safety.classifier.scan(&scan_text).await {
                    Ok(v) => v,
                    Err(e) => {
                        crate::notify!(warn, "safety scan skipped: {e}");
                        crate::safety::ScanVerdict::Pass
                    }
                };
                if !verdict.is_pass()
                    && let Some(sink) = ctx.events
                {
                    let action = match (&verdict, safety.mode) {
                        (crate::safety::ScanVerdict::Deny(_), crate::safety::SafetyMode::Deny) => {
                            "blocked"
                        }
                        _ => "warned",
                    };
                    for category in verdict.categories() {
                        sink.emit(crate::event::Event::ContentFilterHit {
                            turn_id: Some(turn_id.clone()),
                            flow_run_id: ctx.flow_run_id.clone(),
                            provider: safety.classifier.kind().to_string(),
                            model: model.clone(),
                            category: category.clone(),
                            action: action.to_string(),
                        });
                    }
                }
                if verdict.is_deny() && safety.mode == crate::safety::SafetyMode::Deny {
                    let cats = verdict.categories().join(", ");
                    return Value::Err(RuntimeError::ToolFailed(format!(
                        "safety: content_filter blocked prompt (categories: {cats})"
                    )));
                }
            }
            let retry_base_messages = final_messages.clone();
            let can_rebuild_from_session = !matches!(context_mode, ContextMode::None)
                && !has_messages_override
                && ctx.session_runtime.is_some();
            let mut compact_after_overflow_used = false;
            let mut saw_context_overflow = false;
            let mut last_err: Option<RuntimeError> = None;
            let retry_kinds_ref = retry_kinds.as_ref();
            let model_info = crate::model_registry::model_info(&model);
            if model_info.context_budget == 0 {
                return Value::Err(RuntimeError::ToolFailed(format!(
                    "model `{model}` is not registered in config.toml — add a [models.{model}] section with context_budget before using it"
                )));
            }
            let mut thinking_enabled = model_info.thinking_enabled();
            let mut signature_retries: u32 = 0;
            'llm_attempts: loop {
                for attempt in 0..=retry_count {
                    let sanitized_messages = sanitize_tool_pairs(final_messages.clone());
                    let req = crate::provider::LlmRequest {
                        model: model.clone(),
                        messages: sanitized_messages,
                        system: system.clone(),
                        input: input.clone(),
                        schema: None,
                        cache_prompt,
                        tools: tool_specs.clone(),
                        thinking_enabled,
                        stall_timeout_secs,
                    };
                    let start = std::time::Instant::now();
                    let outcome = call_and_maybe_stream(
                        provider.as_ref(),
                        req,
                        StreamCallCtx {
                            session: ctx.session_runtime.as_deref(),
                            stream_tx: ctx.tool_ctx.stream_tx.clone(),
                            flow_run_id: ctx.flow_run_id.as_ref(),
                            agent_entry: ctx.tool_ctx.agent_entry.as_ref(),
                            event_sink: ctx.events,
                            turn_id: ctx.turn_id.clone(),
                        },
                    )
                    .await;
                    let elapsed_ms = start.elapsed().as_millis() as u64;
                    let usage = match &outcome {
                        Ok(am) => crate::provider::TokenUsage {
                            input: am
                                .token_usage
                                .input
                                .max(crate::provider::estimate_tokens(&prompt)),
                            cached_input: am.token_usage.cached_input,
                            output: am
                                .token_usage
                                .output
                                .max(crate::provider::estimate_tokens(&am.text_concat())),
                            cache_write: am.token_usage.cache_write,
                            ..Default::default()
                        },
                        Err(_) => crate::provider::TokenUsage {
                            input: crate::provider::estimate_tokens(&prompt),
                            ..Default::default()
                        },
                    };
                    let status = match &outcome {
                        Ok(_) => crate::event::LlmCallStatus::Ok,
                        Err(e) => crate::event::LlmCallStatus::Errored {
                            message: e.to_string(),
                        },
                    };
                    let (ttft_ms, tps) = match &outcome {
                        Ok(am) => (
                            am.timing.ttft_ms,
                            am.timing.tokens_per_second(am.token_usage.output),
                        ),
                        Err(_) => (None, None),
                    };
                    if let Some(sink) = ctx.events {
                        sink.emit(crate::event::Event::LlmCall {
                            model: model.clone(),
                            provider: provider.name().to_string(),
                            usage: usage.clone(),
                            wallclock_ms: elapsed_ms,
                            ttft_ms,
                            tokens_per_second: tps,
                            status,
                            run_id: ctx.flow_run_id.clone(),
                            node_id: ctx.current_node_id.clone(),
                        });
                    }
                    let input_with_cache = input_with_cache_for_window(&usage);
                    if let Some(tx) = ctx.tool_ctx.stream_tx.as_ref() {
                        let _ = tx.send(crate::stream::StreamFrame::LlmCallStats {
                            model: model.clone(),
                            input_tokens: usage.input,
                            output_tokens: usage.output,
                            cache_read: usage.cached_input,
                            cache_write: usage.cache_write,
                            ttft_ms: ttft_ms.unwrap_or(0),
                            tokens_per_second: tps.unwrap_or(0.0),
                            wallclock_ms: elapsed_ms,
                            run_id: ctx.flow_run_id.as_ref().map(|r| r.0.to_string()),
                            node_id: ctx.current_node_id.clone(),
                        });
                    }
                    if let Some(session) = ctx.session_runtime.as_ref() {
                        session.record_llm_call(
                            &model,
                            input_with_cache,
                            usage.output,
                            usage.cached_input,
                            usage.cache_write,
                            ttft_ms,
                            tps,
                        );
                    }
                    match outcome {
                        Ok(am) => {
                            if let Some(session) = ctx.session_runtime.as_ref() {
                                if !matches!(context_mode, ContextMode::None)
                                    && !has_messages_override
                                {
                                    drop(compact_guard.take());
                                }
                                let _append_compact_guard = session.acquire_compact_lock().await;
                                session.append_message(am.message.clone(), None);
                                drop(_append_compact_guard);
                                crate::compaction::start_auto_compact(
                                    session.clone(),
                                    model.clone(),
                                    ctx.providers.clone(),
                                )
                                .await;
                            }
                            if !matches!(context_mode, ContextMode::None) {
                                return Value::Message(am.message.clone());
                            }
                            return crate::provider::assistant_message_to_value(&am);
                        }
                        Err(e) => {
                            if matches!(
                                e,
                                RuntimeError::Cancelled(_) | RuntimeError::L2Restart { .. }
                            ) {
                                return Value::Err(e);
                            }
                            if is_context_overflow_error(&e)
                                && can_rebuild_from_session
                                && !compact_after_overflow_used
                            {
                                compact_after_overflow_used = true;
                                saw_context_overflow = true;
                                let session = ctx
                                    .session_runtime
                                    .as_ref()
                                    .expect("checked by can_rebuild_from_session");
                                session.request_manual_compact();
                                drop(compact_guard.take());
                                crate::compaction::maybe_auto_compact(
                                    session,
                                    &model,
                                    ctx.providers,
                                )
                                .await;
                                final_messages = rebuild_session_llm_messages(
                                    session,
                                    context_mode,
                                    &turn_id,
                                    Some(prompt.as_str()),
                                    &retry_base_messages[session_messages_len..],
                                );
                                last_err = Some(e);
                                continue 'llm_attempts;
                            }
                            if is_context_overflow_error(&e) {
                                last_err = Some(e);
                                break;
                            }
                            if !rewrite_used
                                && let Some(safety) = ctx.safety
                                && safety.enabled
                                && safety.auto_rewrite
                                && matches!(e.kind(), crate::error::ErrorKind::ContentFilter)
                            {
                                rewrite_used = true;
                                if let Some(last) = final_messages.last_mut()
                                    && let Some(part) =
                                        last.parts.iter_mut().find_map(|p| match p {
                                            crate::message::MessagePart::Text { text } => {
                                                Some(text)
                                            }
                                            _ => None,
                                        })
                                {
                                    *part = format!(
                                        "Please rewrite the following in a neutral, safety-compliant way and answer it:\n{part}"
                                    );
                                }
                                if let Some(sink) = ctx.events {
                                    sink.emit(crate::event::Event::ContentFilterHit {
                                        turn_id: Some(turn_id.clone()),
                                        flow_run_id: ctx.flow_run_id.clone(),
                                        provider: provider.name().to_string(),
                                        model: model.clone(),
                                        category: "auto_rewrite".to_string(),
                                        action: "rewritten".to_string(),
                                    });
                                }
                                last_err = Some(e);
                                continue;
                            }
                            if matches!(e, RuntimeError::ThinkingSignatureMissing) {
                                signature_retries += 1;
                                if signature_retries < 3 {
                                    if let Some(tx) =
                                        ctx.session_runtime.as_ref().map(|s| s.stream_tx())
                                    {
                                        let _ = tx.send(crate::stream::StreamFrame::LlmRetry);
                                    }
                                    crate::notify!(
                                        info,
                                        location = Inline,
                                        "thinking signature missing — retry {signature_retries}/3"
                                    );
                                    last_err = Some(e);
                                    continue 'llm_attempts;
                                }
                                // Exhausted — disable thinking
                                thinking_enabled = false;
                                if let Some(tx) =
                                    ctx.session_runtime.as_ref().map(|s| s.stream_tx())
                                {
                                    let _ = tx.send(crate::stream::StreamFrame::LlmRetry);
                                }
                                crate::notify!(
                                    warn,
                                    location = Inline,
                                    "thinking signature missing after 3 retries; disabling thinking…"
                                );
                                last_err = Some(e);
                                continue 'llm_attempts;
                            }
                            if thinking_enabled
                                && matches!(e.kind(), crate::error::ErrorKind::InvalidRequest)
                            {
                                thinking_enabled = false;
                                if let Some(tx) =
                                    ctx.session_runtime.as_ref().map(|s| s.stream_tx())
                                {
                                    let _ = tx.send(crate::stream::StreamFrame::LlmRetry);
                                }
                                crate::notify!(
                                    warn,
                                    location = Inline,
                                    "thinking mode disabled due to API error; retrying…"
                                );
                                last_err = Some(e);
                                continue 'llm_attempts;
                            }
                            if attempt < retry_count {
                                let kind = e.kind();
                                let should_retry = match &retry_kinds_ref {
                                    Some(allowed) => allowed.contains(&kind),
                                    None => true,
                                };
                                if !should_retry {
                                    last_err = Some(e);
                                    break;
                                }
                                if matches!(
                                    kind,
                                    crate::error::ErrorKind::RateLimit
                                        | crate::error::ErrorKind::Timeout
                                        | crate::error::ErrorKind::ProviderDown
                                        | crate::error::ErrorKind::Transient
                                ) {
                                    let delay_ms = 1000u64 << attempt;
                                    if let Some(tx) =
                                        ctx.session_runtime.as_ref().map(|s| s.stream_tx())
                                    {
                                        let _ = tx.send(crate::stream::StreamFrame::Note(format!(
                                            "retrying in {}s…",
                                            delay_ms / 1000
                                        )));
                                    }
                                    tokio::time::sleep(std::time::Duration::from_millis(delay_ms))
                                        .await;
                                }
                                last_err = Some(e);
                            } else {
                                last_err = Some(e);
                            }
                        }
                    }
                }
                break;
            }
            if let Some(fb) = args.fallback_expr.as_ref() {
                return eval_expr(fb, env, ctx).await;
            }
            if let Some(session) = ctx.session_runtime.as_ref()
                && !saw_context_overflow
            {
                crate::compaction::start_auto_compact(
                    session.clone(),
                    model.clone(),
                    ctx.providers.clone(),
                )
                .await;
            }
            Value::Err(last_err.unwrap_or(RuntimeError::ToolFailed("llm failed".into())))
        }
        Node::UserConfirm { msg } => {
            let v = eval_expr(msg, env, ctx).await;
            if v.is_err() {
                return v;
            }
            let prompt = match &v {
                Value::Str(s) => s.clone(),
                other => other.kind_name().to_string(),
            };
            let confirm_kind = crate::form::FormKind::Confirm {
                prompt: prompt.clone(),
            };
            // Daemon clients drive the confirm through the prompt resolver
            // over RPC; in-process TUI subscribes to FormRegistry. Boot /
            // headless / unit tests without either wired keep the historical
            // auto-approve so they don't deadlock.
            if let Some(resolver) = ctx.tool_ctx.prompt_resolver.clone() {
                let id = crate::rendezvous::PromptId::now();
                let payload =
                    serde_json::to_value(&confirm_kind).unwrap_or(serde_json::Value::Null);
                let timeout = std::time::Duration::from_secs(300);
                let result = crate::rendezvous::await_prompt_with_payload(
                    &resolver, id, "form_ask", payload, timeout,
                )
                .await;
                let answer: crate::form::FormAnswer = match result {
                    Ok(v) => {
                        serde_json::from_value(v).unwrap_or(crate::form::FormAnswer::Cancelled)
                    }
                    Err(_) => crate::form::FormAnswer::Cancelled,
                };
                return Value::Bool(matches!(
                    answer,
                    crate::form::FormAnswer::Confirmed { value: true }
                ));
            }
            let Some(session) = ctx.session_runtime.as_ref() else {
                return Value::Bool(true);
            };
            let forms = session.forms();
            if forms.subscriber_count() == 0 {
                return Value::Bool(true);
            }
            let Some(run_id) = ctx.flow_run_id.clone() else {
                return Value::Bool(true);
            };
            let pending = crate::form::PendingForm {
                form_id: uuid::Uuid::now_v7().to_string(),
                run_id,
                tool_use_id: ctx.current_node_id.clone().unwrap_or_default(),
                kind: confirm_kind,
                emitted_at: chrono::Utc::now(),
            };
            let rx = forms.request(pending);
            let answer = rx.await.unwrap_or(crate::form::FormAnswer::Cancelled);
            Value::Bool(matches!(
                answer,
                crate::form::FormAnswer::Confirmed { value: true }
            ))
        }
        Node::FixUntilTestPasses { kwargs } => eval_fix_until_test_passes(kwargs, env, ctx).await,
        Node::Message { role, args } => eval_message_node(*role, args, env, ctx).await,
        Node::Subflow { name, args } => {
            let Some(target) = ctx.flows.get(&name.name) else {
                return Value::Err(RuntimeError::UndefinedTool(format!(
                    "subflow({})",
                    name.name
                )));
            };
            let mut bindings = Vec::with_capacity(args.len());
            for (i, arg) in args.iter().enumerate() {
                let (param_name, value) = match arg {
                    Arg::Positional(e) => {
                        let Some(p) = target.params.get(i) else {
                            return Value::Err(RuntimeError::MissingArg(format!(
                                "subflow({}): too many positional args",
                                name.name
                            )));
                        };
                        let v = eval_expr(e, env, ctx).await;
                        (p.name.name.clone(), v)
                    }
                    Arg::Named { name: n, value } => {
                        let v = eval_expr(value, env, ctx).await;
                        (n.name.clone(), v)
                    }
                };
                if value.is_err() {
                    return value;
                }
                bindings.push((param_name, value));
            }
            let mut sub_env = Env::new();
            for (n, v) in bindings {
                sub_env.bind(n, v);
            }
            let sub_run_id = crate::event::FlowRunId::now();
            if let Some(sink) = ctx.events {
                sink.emit(crate::event::Event::FlowStart {
                    run_id: sub_run_id.clone(),
                    flow_name: name.name.clone(),
                    parent_run_id: ctx.flow_run_id.clone(),
                    parent_node_id: ctx.current_node_id.clone(),
                    spawned: false,
                });
            }
            if let Some(session) = ctx.session_runtime.as_ref() {
                let _ = session
                    .stream_tx()
                    .send(crate::stream::StreamFrame::FlowStart {
                        run_id: sub_run_id.0.to_string(),
                        flow_name: name.name.clone(),
                        parent_run_id: ctx.flow_run_id.as_ref().map(|r| r.0.to_string()),
                        parent_node_id: ctx.current_node_id.clone(),
                    });
            } else if let Some(tx) = ctx.tool_ctx.stream_tx.as_ref() {
                let _ = tx.send(crate::stream::StreamFrame::FlowStart {
                    run_id: sub_run_id.0.to_string(),
                    flow_name: name.name.clone(),
                    parent_run_id: ctx.flow_run_id.as_ref().map(|r| r.0.to_string()),
                    parent_node_id: ctx.current_node_id.clone(),
                });
            }
            let sub_ctx = EvalCtx {
                flow_run_id: Some(sub_run_id.clone()),
                current_node_id: None,
                ..ctx.clone()
            };
            let outcome = crate::exec::exec_stmts(&target.body, &mut sub_env, &sub_ctx).await;
            let (result, status, ok) = match outcome {
                crate::exec::StmtOutcome::Return(v) => (v, crate::event::FlowStatus::Ok, true),
                crate::exec::StmtOutcome::Err(e) => {
                    let status = if matches!(&e, crate::error::RuntimeError::Cancelled(_)) {
                        crate::event::FlowStatus::Cancelled
                    } else {
                        crate::event::FlowStatus::Errored {
                            message: format!("{e}"),
                        }
                    };
                    (Value::Err(e.clone()), status, false)
                }
                crate::exec::StmtOutcome::Continue => {
                    (Value::Unit, crate::event::FlowStatus::Ok, true)
                }
            };
            let cancelled = matches!(status, crate::event::FlowStatus::Cancelled);
            if let Some(sink) = ctx.events {
                sink.emit(crate::event::Event::FlowEnd {
                    run_id: sub_run_id.clone(),
                    flow_name: name.name.clone(),
                    status,
                });
            }
            if let Some(tx) = ctx.tool_ctx.stream_tx.as_ref() {
                let _ = tx.send(crate::stream::StreamFrame::FlowDone {
                    run_id: sub_run_id.0.to_string(),
                    flow_name: name.name.clone(),
                    ok,
                    cancelled,
                });
            }
            result
        }
    }
}

fn session_fs_access_policy(session: &crate::session::Session) -> crate::fs_access::FsAccessPolicy {
    let workspace = session
        .meta()
        .and_then(|m| m.project_root)
        .or_else(|| std::env::current_dir().ok());
    let mode = session
        .fs_access_mode()
        .unwrap_or(crate::fs_access::FsAccessMode::WorkspaceWrite);
    crate::fs_access::FsAccessPolicy { mode, workspace }
}

fn tool_name(path: &[atman_dsl::ast::Ident]) -> String {
    let parts: Vec<&str> = path.iter().map(|i| i.name.as_str()).collect();
    parts.join(".")
}

async fn eval_fix_until_test_passes<'a>(
    kwargs: &'a atman_dsl::ast::Kwargs,
    env: &'a Env,
    ctx: &'a EvalCtx<'a>,
) -> Value {
    let mut edit_flow_expr: Option<&Expr> = None;
    let mut test_expr: Option<&Expr> = None;
    let mut on_giveup_expr: Option<&Expr> = None;
    let mut max_iters: u32 = 5;
    let mut target_path: Option<std::path::PathBuf> = None;

    for (k, v) in kwargs {
        match k.name.as_str() {
            "edit_flow" => edit_flow_expr = Some(v),
            "test" => test_expr = Some(v),
            "on_giveup" => on_giveup_expr = Some(v),
            "max_iters" => match eval_expr(v, env, ctx).await {
                Value::Int(n) if n > 0 => max_iters = n as u32,
                other => {
                    return Value::Err(RuntimeError::TypeMismatch {
                        expected: "positive int (max_iters)".into(),
                        actual: other.kind_name().into(),
                    });
                }
            },
            "target" => match eval_expr(v, env, ctx).await {
                Value::Path(p) => target_path = Some(p),
                Value::Str(s) => target_path = Some(std::path::PathBuf::from(s)),
                Value::Unit => {}
                other => {
                    return Value::Err(RuntimeError::TypeMismatch {
                        expected: "path (target)".into(),
                        actual: other.kind_name().into(),
                    });
                }
            },
            _ => {}
        }
    }

    let Some(edit_flow_expr) = edit_flow_expr else {
        return Value::Err(RuntimeError::MissingArg(
            "fix_until_test_passes.edit_flow".into(),
        ));
    };
    let Some(test_expr) = test_expr else {
        return Value::Err(RuntimeError::MissingArg(
            "fix_until_test_passes.test".into(),
        ));
    };

    let pristine: Option<String> = match &target_path {
        Some(p) => match tokio::fs::read_to_string(p).await {
            Ok(s) => Some(s),
            Err(e) => {
                return Value::Err(RuntimeError::ToolFailed(format!(
                    "fix_until_test_passes: cannot read target {}: {e}",
                    p.display()
                )));
            }
        },
        None => None,
    };

    let mut prev_fail = String::new();
    let mut last_test_result: Option<Value> = None;

    for iter in 0..max_iters {
        let mut loop_env = env.clone();
        loop_env.bind("iter", Value::Int(iter as i64));
        loop_env.bind("prev_fail", Value::Str(prev_fail.clone()));

        let edit_v = eval_expr(edit_flow_expr, &loop_env, ctx).await;
        if edit_v.is_err() {
            return edit_v;
        }
        loop_env.bind("last_edit", edit_v);

        let test_v = eval_expr(test_expr, &loop_env, ctx).await;
        if test_v.is_err() {
            return test_v;
        }
        let exit = test_v
            .field("exit_code")
            .or_else(|| test_v.field("exit"))
            .and_then(|v| match v {
                Value::Int(n) => Some(*n),
                _ => None,
            });
        last_test_result = Some(test_v.clone());
        if let Some(0) = exit {
            return Value::Struct(vec![
                ("status".into(), Value::Str("passed".into())),
                ("iters".into(), Value::Int((iter + 1) as i64)),
                ("test".into(), test_v),
            ]);
        }
        let stderr_tail = test_v
            .field("stderr_tail")
            .or_else(|| test_v.field("output"))
            .and_then(|v| match v {
                Value::Str(s) => Some(s.clone()),
                _ => None,
            })
            .unwrap_or_default();
        let stdout_tail = test_v
            .field("stdout_tail")
            .and_then(|v| match v {
                Value::Str(s) => Some(s.clone()),
                _ => None,
            })
            .unwrap_or_default();
        prev_fail = format!(
            "iter {iter} exit={:?}\n--- stderr ---\n{stderr_tail}\n--- stdout ---\n{stdout_tail}",
            exit
        );

        if let (Some(target), Some(pristine)) = (&target_path, &pristine)
            && let Err(e) = tokio::fs::write(target, pristine.as_bytes()).await
        {
            return Value::Err(RuntimeError::ToolFailed(format!(
                "fix_until_test_passes: revert failed on {}: {e}",
                target.display()
            )));
        }
    }

    if let Some(giveup) = on_giveup_expr {
        let mut giveup_env = env.clone();
        giveup_env.bind("iters", Value::Int(max_iters as i64));
        giveup_env.bind("prev_fail", Value::Str(prev_fail));
        return eval_expr(giveup, &giveup_env, ctx).await;
    }

    Value::Struct(vec![
        ("status".into(), Value::Str("gave_up".into())),
        ("iters".into(), Value::Int(max_iters as i64)),
        ("last_test".into(), last_test_result.unwrap_or(Value::Unit)),
    ])
}

async fn eval_message_node<'a>(
    ast_role: atman_dsl::ast::MessageRole,
    args: &'a [Arg],
    env: &'a Env,
    ctx: &'a EvalCtx<'a>,
) -> Value {
    use crate::message::{ImageData, ImageSource, Message, MessagePart, MessageRole};

    let role = match ast_role {
        atman_dsl::ast::MessageRole::User => MessageRole::User,
        atman_dsl::ast::MessageRole::Assistant => MessageRole::Assistant,
        atman_dsl::ast::MessageRole::System => MessageRole::System,
        atman_dsl::ast::MessageRole::Tool => MessageRole::Tool,
    };
    let turn_id = ctx
        .turn_id
        .clone()
        .unwrap_or_else(crate::event::TurnId::now);

    let mut positional = Vec::new();
    let mut named: Vec<(String, Value)> = Vec::new();
    let mut attachment_paths_raw: Option<Vec<std::path::PathBuf>> = None;
    for arg in args {
        match arg {
            Arg::Positional(e) => {
                let v = eval_expr(e, env, ctx).await;
                if v.is_err() {
                    return v;
                }
                positional.push(v);
            }
            Arg::Named { name, value } => {
                if name.name == "attachments" {
                    if let Expr::List(items) = value {
                        let mut collected = Vec::with_capacity(items.len());
                        let mut all_fileref = true;
                        for it in items {
                            if let Expr::FileRef(f) = it {
                                collected.push(std::path::PathBuf::from(&f.path));
                            } else {
                                all_fileref = false;
                                break;
                            }
                        }
                        if all_fileref {
                            attachment_paths_raw = Some(collected);
                            continue;
                        }
                    }
                }
                let v = eval_expr(value, env, ctx).await;
                if v.is_err() {
                    return v;
                }
                named.push((name.name.clone(), v));
            }
        }
    }
    let take_named = |k: &str, named: &mut Vec<(String, Value)>| -> Option<Value> {
        let pos = named.iter().position(|(n, _)| n == k)?;
        Some(named.remove(pos).1)
    };

    if role == MessageRole::Tool {
        let tool_use_id = match positional.first().or(take_named("id", &mut named).as_ref()) {
            Some(Value::Str(s)) => s.clone(),
            Some(other) => {
                return Value::Err(RuntimeError::TypeMismatch {
                    expected: "string (tool_use_id)".into(),
                    actual: other.kind_name().into(),
                });
            }
            None => {
                return Value::Err(RuntimeError::MissingArg("tool_result: id".into()));
            }
        };
        let content = match positional
            .get(1)
            .or(take_named("content", &mut named).as_ref())
        {
            Some(Value::Str(s)) => s.clone(),
            Some(other) => {
                return Value::Err(RuntimeError::TypeMismatch {
                    expected: "string (content)".into(),
                    actual: other.kind_name().into(),
                });
            }
            None => {
                return Value::Err(RuntimeError::MissingArg("tool_result: content".into()));
            }
        };
        let is_error = match take_named("is_error", &mut named) {
            Some(Value::Bool(b)) => b,
            Some(other) => {
                return Value::Err(RuntimeError::TypeMismatch {
                    expected: "bool (is_error)".into(),
                    actual: other.kind_name().into(),
                });
            }
            None => false,
        };
        return Value::Message(Message {
            role,
            parts: vec![MessagePart::ToolResult {
                tool_use_id,
                content,
                is_error,
            }],
            turn_id,
        });
    }

    let text = match positional.first() {
        Some(Value::Str(s)) => Some(s.clone()),
        Some(other) => {
            return Value::Err(RuntimeError::TypeMismatch {
                expected: "string (message text)".into(),
                actual: other.kind_name().into(),
            });
        }
        None => None,
    };
    let attachment_paths: Vec<std::path::PathBuf> = if let Some(raw) = attachment_paths_raw {
        raw
    } else {
        match take_named("attachments", &mut named) {
            Some(Value::List(items)) => {
                let mut ps = Vec::with_capacity(items.len());
                for it in items {
                    match it {
                        Value::Path(p) => ps.push(p),
                        Value::Str(s) => ps.push(std::path::PathBuf::from(s)),
                        other => {
                            return Value::Err(RuntimeError::TypeMismatch {
                                expected: "path (attachment)".into(),
                                actual: other.kind_name().into(),
                            });
                        }
                    }
                }
                ps
            }
            Some(other) => {
                return Value::Err(RuntimeError::TypeMismatch {
                    expected: "list of path".into(),
                    actual: other.kind_name().into(),
                });
            }
            None => Vec::new(),
        }
    };

    let mut parts: Vec<MessagePart> = attachment_paths
        .into_iter()
        .map(|path| {
            let media_type = guess_image_mime(&path).unwrap_or_else(|| "image/png".to_string());
            MessagePart::Image {
                source: ImageSource {
                    media_type,
                    data: ImageData::Path { path },
                },
            }
        })
        .collect();
    if let Some(t) = text {
        parts.push(MessagePart::Text { text: t });
    }

    Value::Message(Message {
        role,
        parts,
        turn_id,
    })
}

fn render_injections(injections: &[crate::injection::Injection]) -> String {
    use crate::injection::{InjectionLevel, InjectionSource};
    let has_user = injections
        .iter()
        .any(|i| matches!(i.source, InjectionSource::User));
    let has_watcher = injections
        .iter()
        .any(|i| matches!(i.source, InjectionSource::Watcher { .. }));
    let mut out = String::new();
    if has_user {
        out.push_str(
            "The user sent the following steering message(s) while you were working. \
             Apply them to your next step if still relevant.\n\n",
        );
    }
    if has_watcher {
        out.push_str("A background watcher detected the following event(s):\n\n");
    }
    for inj in injections {
        let (tag, source_attr) = match &inj.source {
            InjectionSource::User => match inj.level {
                InjectionLevel::L2CourseCorrect => ("user_correction", "user".to_string()),
                _ => ("user_nudge", "user".to_string()),
            },
            InjectionSource::Watcher {
                watcher_id,
                kind,
                handle,
            } => (
                "watcher_event",
                format!("{kind} '{handle}' watcher {watcher_id}"),
            ),
        };
        out.push_str(&format!(
            "<{tag} id=\"{}\" ts=\"{}\" source=\"{}\">\n{}\n</{tag}>\n",
            inj.id.0,
            inj.created_at.to_rfc3339(),
            source_attr,
            inj.text
        ));
    }
    out
}

fn guess_image_mime(path: &std::path::Path) -> Option<String> {
    let ext = path
        .extension()
        .and_then(|s| s.to_str())?
        .to_ascii_lowercase();
    Some(
        match ext.as_str() {
            "png" => "image/png",
            "jpg" | "jpeg" => "image/jpeg",
            "gif" => "image/gif",
            "webp" => "image/webp",
            _ => return None,
        }
        .to_string(),
    )
}

fn contract_allows_shell(contract: Option<&atman_dsl::ast::Contract>) -> bool {
    let Some(c) = contract else { return false };
    for block in &c.blocks {
        if block.name.name != "capabilities" {
            continue;
        }
        for (k, v) in &block.kwargs {
            if k.name != "shell" {
                continue;
            }
            if let atman_dsl::ast::Expr::Literal(atman_dsl::ast::Literal::Bool(true)) = v {
                return true;
            }
        }
    }
    false
}

pub struct TruncationStat {
    pub original_chars: usize,
    pub result_chars: usize,
    pub dropped_chars: usize,
    pub budget_tokens: u64,
}

pub(crate) fn resolve_tool_specs(
    expr: &Expr,
    tools: &crate::tool::ToolRegistry,
) -> Result<Vec<crate::tool::ToolSpec>, String> {
    let items = match expr {
        Expr::List(items) => items,
        _ => {
            return Err(
                "llm.tools: expected a list of tool references like [fs.read, bash.exec]".into(),
            );
        }
    };
    let mut out = Vec::with_capacity(items.len());
    for item in items {
        // Handle wildcard strings like "mcp.*" or "mcp.lark.*".
        // Expand against all registered tools whose names start with the prefix.
        if let Some(prefix) = wildcard_prefix(item) {
            for name in tools.names() {
                if name.starts_with(&prefix) {
                    if let Some(tool) = tools.get(&name) {
                        out.push(crate::tool::tool_spec(tool.as_ref()));
                    }
                }
            }
            continue;
        }
        let name = match tool_ref_name(item) {
            Some(n) => n,
            None => {
                return Err(format!(
                    "llm.tools: item is not a tool reference (want ident or ident.method, or a \"ns.*\" wildcard): {item:?}"
                ));
            }
        };
        let tool = tools
            .get(&name)
            .ok_or_else(|| format!("llm.tools: unknown tool `{name}`"))?;
        out.push(crate::tool::tool_spec(tool.as_ref()));
    }
    Ok(out)
}

/// If `expr` is a string literal ending with `.*`, return the prefix
/// (everything before the `.*`), e.g. `"mcp.*"` → `Some("mcp.")`.
fn wildcard_prefix(expr: &Expr) -> Option<String> {
    match expr {
        Expr::Literal(atman_dsl::ast::Literal::Str(s)) => s.strip_suffix(".*").map(|prefix| {
            if prefix.is_empty() {
                // ".*" means match everything — use empty prefix so all tools match.
                String::new()
            } else {
                // "mcp.*" → prefix = "mcp", but we want "mcp." so only mcp tools match.
                format!("{prefix}.")
            }
        }),
        _ => None,
    }
}

fn tool_ref_name(expr: &Expr) -> Option<String> {
    match expr {
        Expr::Ident(id) => Some(id.name.clone()),
        Expr::Member { base, field } => {
            let base = tool_ref_name(base)?;
            Some(format!("{base}.{}", field.name))
        }
        _ => None,
    }
}

pub(crate) fn parse_error_kind_list(
    expr: &Expr,
) -> Result<std::collections::HashSet<crate::error::ErrorKind>, String> {
    let items = match expr {
        Expr::List(items) => items,
        _ => {
            return Err(
                "retry_classified: expected a list literal like [timeout, rate_limit]".into(),
            );
        }
    };
    let mut out = std::collections::HashSet::new();
    for item in items {
        let name = match item {
            Expr::Ident(id) => id.name.clone(),
            Expr::Literal(atman_dsl::ast::Literal::Str(s)) => s.clone(),
            _ => {
                return Err(
                    "retry_classified: each item must be an identifier or string kind name".into(),
                );
            }
        };
        match crate::error::ErrorKind::from_name(&name) {
            Some(k) => {
                out.insert(k);
            }
            None => return Err(format!("retry_classified: unknown error kind `{name}`")),
        }
    }
    Ok(out)
}

fn sanitize_tool_pairs(messages: Vec<crate::message::Message>) -> Vec<crate::message::Message> {
    use crate::message::{Message, MessagePart, MessageRole};
    use std::collections::HashMap;
    let mut result_by_id: HashMap<String, Message> = HashMap::new();
    for m in &messages {
        for p in &m.parts {
            if let MessagePart::ToolResult { tool_use_id, .. } = p {
                result_by_id
                    .entry(tool_use_id.clone())
                    .or_insert_with(|| Message {
                        role: MessageRole::Tool,
                        parts: vec![p.clone()],
                        turn_id: m.turn_id.clone(),
                    });
            }
        }
    }
    let mut out: Vec<Message> = Vec::with_capacity(messages.len() + 4);
    for m in &messages {
        let uses: Vec<String> = m
            .parts
            .iter()
            .filter_map(|p| match p {
                MessagePart::ToolUse { id, .. } => Some(id.clone()),
                _ => None,
            })
            .collect();
        let is_pure_result = m
            .parts
            .iter()
            .all(|p| matches!(p, MessagePart::ToolResult { .. }));
        if is_pure_result {
            continue;
        }
        out.push(m.clone());
        if !uses.is_empty() {
            let next_has_all = match out.len().checked_sub(1) {
                Some(_) => false,
                None => false,
            };
            let _ = next_has_all;
            let mut filler_parts: Vec<MessagePart> = Vec::new();
            for u in &uses {
                if let Some(rm) = result_by_id.get(u) {
                    if let Some(MessagePart::ToolResult {
                        tool_use_id,
                        content,
                        is_error,
                    }) = rm.parts.first()
                    {
                        filler_parts.push(MessagePart::ToolResult {
                            tool_use_id: tool_use_id.clone(),
                            content: content.clone(),
                            is_error: *is_error,
                        });
                    }
                } else {
                    filler_parts.push(MessagePart::ToolResult {
                        tool_use_id: u.clone(),
                        content: "[tool execution interrupted — no result captured]".into(),
                        is_error: true,
                    });
                }
            }
            out.push(Message {
                role: MessageRole::Tool,
                parts: filler_parts,
                turn_id: m.turn_id.clone(),
            });
        }
    }
    out
}

pub fn truncate_prompt_to_budget(prompt: String, budget_tokens: u64) -> String {
    truncate_prompt_to_budget_tracked(prompt, budget_tokens).0
}

pub fn truncate_prompt_to_budget_tracked(
    prompt: String,
    budget_tokens: u64,
) -> (String, Option<TruncationStat>) {
    let budget_chars = budget_tokens.saturating_mul(4) as usize;
    if prompt.len() <= budget_chars {
        return (prompt, None);
    }
    let head_chars = budget_chars * 4 / 10;
    let tail_chars = budget_chars * 4 / 10;
    if head_chars + tail_chars >= prompt.len() {
        return (prompt, None);
    }
    let original_chars = prompt.len();
    let head_end = char_boundary(&prompt, head_chars, false);
    let tail_start = char_boundary(&prompt, prompt.len().saturating_sub(tail_chars), true);
    let head = &prompt[..head_end];
    let tail = &prompt[tail_start..];
    let dropped = original_chars - head.len() - tail.len();
    let result = format!("{head}\n\n[... truncated {dropped} chars ...]\n\n{tail}");
    let stat = TruncationStat {
        original_chars,
        result_chars: result.len(),
        dropped_chars: dropped,
        budget_tokens,
    };
    (result, Some(stat))
}

fn char_boundary(s: &str, target: usize, round_up: bool) -> usize {
    let mut idx = target.min(s.len());
    while idx > 0 && idx < s.len() && !s.is_char_boundary(idx) {
        if round_up {
            idx += 1;
        } else {
            idx -= 1;
        }
    }
    idx
}

// Bare primitive names inside `schema: { valid: bool, ... }` parse as tool calls; treat as Unit.
fn is_type_annotation(path: &[atman_dsl::ast::Ident]) -> bool {
    if path.len() != 1 {
        return false;
    }
    matches!(
        path[0].name.as_str(),
        "bool" | "int" | "float" | "string" | "path" | "bytes" | "duration"
    )
}

fn eval_literal(lit: &Literal) -> Value {
    match lit {
        Literal::Str(s) => Value::Str(s.clone()),
        Literal::Int(n) => Value::Int(*n),
        Literal::Float(f) => Value::Float(*f),
        Literal::Bool(b) => Value::Bool(*b),
    }
}

fn eval_binop(op: BinOp, l: &Value, r: &Value) -> Value {
    match op {
        BinOp::Eq => Value::Bool(value_eq(l, r)),
        BinOp::Ne => Value::Bool(!value_eq(l, r)),
        BinOp::Lt => value_cmp(l, r, |a, b| a < b, |a, b| a < b, |a, b| a < b),
        BinOp::Le => value_cmp(l, r, |a, b| a <= b, |a, b| a <= b, |a, b| a <= b),
        BinOp::Gt => value_cmp(l, r, |a, b| a > b, |a, b| a > b, |a, b| a > b),
        BinOp::Ge => value_cmp(l, r, |a, b| a >= b, |a, b| a >= b, |a, b| a >= b),
        BinOp::And => match (l, r) {
            (Value::Bool(a), Value::Bool(b)) => Value::Bool(*a && *b),
            _ => type_mismatch("bool && bool", l, r),
        },
        BinOp::Or => match (l, r) {
            (Value::Bool(a), Value::Bool(b)) => Value::Bool(*a || *b),
            _ => type_mismatch("bool || bool", l, r),
        },
        BinOp::Add => match (l, r) {
            (Value::Int(a), Value::Int(b)) => Value::Int(a + b),
            (Value::Float(a), Value::Float(b)) => Value::Float(a + b),
            (Value::Str(a), Value::Str(b)) => Value::Str(format!("{a}{b}")),
            (Value::Str(a), Value::Path(b)) => Value::Str(format!("{a}{}", b.display())),
            (Value::Path(a), Value::Str(b)) => Value::Str(format!("{}{b}", a.display())),
            _ => type_mismatch(
                "int+int | float+float | string+string | string+path | path+string",
                l,
                r,
            ),
        },
        BinOp::Sub => match (l, r) {
            (Value::Int(a), Value::Int(b)) => Value::Int(a - b),
            (Value::Float(a), Value::Float(b)) => Value::Float(a - b),
            _ => type_mismatch("int-int | float-float", l, r),
        },
        BinOp::Mul => match (l, r) {
            (Value::Int(a), Value::Int(b)) => Value::Int(a * b),
            (Value::Float(a), Value::Float(b)) => Value::Float(a * b),
            _ => type_mismatch("int*int | float*float", l, r),
        },
        BinOp::Div => match (l, r) {
            (Value::Int(_), Value::Int(0)) => {
                Value::Err(RuntimeError::ToolFailed("integer div by zero".into()))
            }
            (Value::Int(a), Value::Int(b)) => Value::Int(a / b),
            (Value::Float(a), Value::Float(b)) => Value::Float(a / b),
            _ => type_mismatch("int/int | float/float", l, r),
        },
        BinOp::Mod => match (l, r) {
            (Value::Int(_), Value::Int(0)) => {
                Value::Err(RuntimeError::ToolFailed("integer mod by zero".into()))
            }
            (Value::Int(a), Value::Int(b)) => Value::Int(a % b),
            (Value::Float(a), Value::Float(b)) => Value::Float(a % b),
            _ => type_mismatch("int%int | float%float", l, r),
        },
    }
}

fn eval_unop(op: UnOp, v: &Value) -> Value {
    match op {
        UnOp::Not => match v {
            Value::Bool(b) => Value::Bool(!b),
            other => Value::Err(RuntimeError::TypeMismatch {
                expected: "bool".into(),
                actual: other.kind_name().into(),
            }),
        },
        UnOp::Neg => match v {
            Value::Int(n) => Value::Int(-n),
            Value::Float(n) => Value::Float(-n),
            other => Value::Err(RuntimeError::TypeMismatch {
                expected: "int or float".into(),
                actual: other.kind_name().into(),
            }),
        },
    }
}

fn value_eq(l: &Value, r: &Value) -> bool {
    match (l, r) {
        (Value::Unit, Value::Unit) => true,
        (Value::Bool(a), Value::Bool(b)) => a == b,
        (Value::Int(a), Value::Int(b)) => a == b,
        (Value::Float(a), Value::Float(b)) => a == b,
        (Value::Str(a), Value::Str(b)) => a == b,
        (Value::Path(a), Value::Path(b)) => a == b,
        _ => false,
    }
}

fn value_cmp(
    l: &Value,
    r: &Value,
    int_cmp: fn(i64, i64) -> bool,
    float_cmp: fn(f64, f64) -> bool,
    str_cmp: fn(&str, &str) -> bool,
) -> Value {
    match (l, r) {
        (Value::Int(a), Value::Int(b)) => Value::Bool(int_cmp(*a, *b)),
        (Value::Float(a), Value::Float(b)) => Value::Bool(float_cmp(*a, *b)),
        (Value::Str(a), Value::Str(b)) => Value::Bool(str_cmp(a, b)),
        _ => type_mismatch("comparable pair", l, r),
    }
}

fn type_mismatch(expected: &str, l: &Value, r: &Value) -> Value {
    Value::Err(RuntimeError::TypeMismatch {
        expected: expected.into(),
        actual: format!("{} vs {}", l.kind_name(), r.kind_name()),
    })
}

fn input_with_cache_for_window(usage: &crate::provider::TokenUsage) -> u64 {
    usage.input + usage.cached_input
}

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

    #[test]
    fn parse_context_mode_handles_variants() {
        assert!(matches!(
            parse_context_mode("session"),
            ContextMode::Session
        ));
        assert!(matches!(parse_context_mode("none"), ContextMode::None));
        assert!(matches!(parse_context_mode(""), ContextMode::None));
        assert!(matches!(
            parse_context_mode(" session "),
            ContextMode::Session
        ));
        match parse_context_mode("session_recent(5)") {
            ContextMode::SessionRecent(n) => assert_eq!(n, 5),
            other => panic!("expected SessionRecent(5), got {other:?}"),
        }
        match parse_context_mode("session_recent") {
            ContextMode::SessionRecent(n) => assert_eq!(n, 10),
            other => panic!("expected SessionRecent(10), got {other:?}"),
        }
        assert!(matches!(parse_context_mode("garbage"), ContextMode::None));
    }

    #[test]
    fn input_with_cache_for_window_does_not_double_count_cache_write() {
        let usage = crate::provider::TokenUsage {
            input: 50_000,
            cached_input: 0,
            cache_write: 50_000,
            ..Default::default()
        };

        assert_eq!(input_with_cache_for_window(&usage), 50_000);
    }

    async fn eval_snippet(expr_src: &str) -> Value {
        let src = format!("flow t() {{\n    return {expr_src}\n}}\n");
        let file = parse_file(&src).expect("parse test snippet");
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };
        let stmt = &file.flows[0].body[0];
        if let atman_dsl::ast::Stmt::Return { value } = stmt {
            eval_expr(value, &Env::new(), &ctx).await
        } else {
            panic!("expected return statement");
        }
    }

    #[tokio::test]
    async fn literals_evaluate() {
        assert!(matches!(eval_snippet("42").await, Value::Int(42)));
        assert!(matches!(eval_snippet("true").await, Value::Bool(true)));
        assert!(matches!(
            eval_snippet(r#""hello""#).await,
            Value::Str(s) if s == "hello"
        ));
    }

    #[tokio::test]
    async fn undefined_ident_yields_err_value() {
        assert!(matches!(
            eval_snippet("missing").await,
            Value::Err(RuntimeError::UndefinedVar(name)) if name == "missing"
        ));
    }

    #[tokio::test]
    async fn binary_arithmetic_and_comparison() {
        assert!(matches!(eval_snippet("1 == 1").await, Value::Bool(true)));
        assert!(matches!(eval_snippet("2 < 3").await, Value::Bool(true)));
        assert!(matches!(
            eval_snippet(r#""a" + "b""#).await,
            Value::Str(s) if s == "ab"
        ));
    }

    #[tokio::test]
    async fn type_mismatch_bubbles_up() {
        assert!(matches!(
            eval_snippet(r#"1 + "x""#).await,
            Value::Err(RuntimeError::TypeMismatch { .. })
        ));
    }

    #[tokio::test]
    async fn err_short_circuits_binary() {
        assert!(matches!(
            eval_snippet("missing == 1").await,
            Value::Err(RuntimeError::UndefinedVar(name)) if name == "missing"
        ));
    }

    #[tokio::test]
    async fn list_evaluates_all_items() {
        let v = eval_snippet("[1, 2, 3]").await;
        if let Value::List(items) = v {
            assert_eq!(items.len(), 3);
            assert!(matches!(items[2], Value::Int(3)));
        } else {
            panic!("expected list");
        }
    }

    #[tokio::test]
    async fn struct_literal_evaluates_fields_in_order() {
        let v = eval_snippet(r#"{ severity: "critical", count: 3 }"#).await;
        if let Value::Struct(fields) = v {
            assert_eq!(fields[0].0, "severity");
            assert_eq!(fields[1].0, "count");
        } else {
            panic!("expected struct");
        }
    }

    #[tokio::test]
    async fn undefined_tool_returns_undefined_tool_err() {
        let src = r#"flow t() { return fs.readnope("/tmp") }"#;
        let file = parse_file(src).unwrap();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &Env::new(), &ctx).await;
            assert!(matches!(
                v,
                Value::Err(RuntimeError::UndefinedTool(name)) if name == "fs.readnope"
            ));
        }
    }

    #[tokio::test]
    async fn fanout_all_gathers_results_in_order() {
        use crate::tools::fs::FsRead;
        use std::sync::Arc;
        use tempfile::TempDir;

        let dir = TempDir::new().unwrap();
        let pa = dir.path().join("a.txt");
        let pb = dir.path().join("b.txt");
        tokio::fs::write(&pa, b"AAA").await.unwrap();
        tokio::fs::write(&pb, b"BBB").await.unwrap();

        let tools = ToolRegistry::new();
        tools.register(Arc::new(FsRead));
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };

        let mut env = Env::new();
        env.bind("a", Value::Path(pa));
        env.bind("b", Value::Path(pb));

        let src = r#"flow t() { return fanout [ fs.read(a), fs.read(b) ] collect: all }"#;
        let file = parse_file(src).unwrap();
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &env, &ctx).await;
            if let Value::List(items) = v {
                assert_eq!(items.len(), 2);
                assert!(matches!(&items[0], Value::Str(s) if s == "AAA"));
                assert!(matches!(&items[1], Value::Str(s) if s == "BBB"));
            } else {
                panic!("expected list");
            }
        }
    }

    #[tokio::test]
    async fn fanout_all_short_circuits_on_err() {
        let src = r#"flow t() { return fanout [ 1, missing, 3 ] collect: all }"#;
        let file = parse_file(src).unwrap();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &Env::new(), &ctx).await;
            assert!(matches!(
                v,
                Value::Err(RuntimeError::UndefinedVar(name)) if name == "missing"
            ));
        }
    }

    #[tokio::test]
    async fn llm_node_dispatches_to_mock_provider() {
        use crate::providers::mock::MockProvider;
        use std::sync::Arc;

        {
            let _lock = crate::model_registry::MODEL_CONFIG_LOCK.lock().unwrap();
            crate::model_registry::set_model_config(crate::model_registry::ModelConfig {
                models: std::collections::HashMap::from([(
                    "mock".into(),
                    crate::model_registry::ModelEntry {
                        model: "mock".into(),
                        context_budget: Some(8_192),
                        ..Default::default()
                    },
                )]),
                ..Default::default()
            });
        }

        let mut providers = crate::provider::ProviderRegistry::new();
        providers.register(Arc::new(MockProvider::new("mock").with_model(
            "mock",
            Value::Struct(vec![("severity".into(), Value::Str("info".into()))]),
        )));
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };

        let src = r#"flow t() {
    return llm {
        model: "mock"
        prompt: "review please"
        input: 1
    }
}
"#;
        let file = parse_file(src).unwrap();
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &Env::new(), &ctx).await;
            if let Value::Struct(fields) = v {
                assert_eq!(fields[0].0, "severity");
                assert!(matches!(&fields[0].1, Value::Str(s) if s == "info"));
            } else {
                panic!("expected struct");
            }
        }
    }

    #[tokio::test]
    async fn llm_missing_model_reports_missing_arg() {
        let providers = crate::provider::ProviderRegistry::new();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };
        let src = r#"flow t() { return llm { prompt: "hi" } }"#;
        let file = parse_file(src).unwrap();
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &Env::new(), &ctx).await;
            assert!(matches!(
                v,
                Value::Err(RuntimeError::MissingArg(name)) if name == "llm.model"
            ));
        }
    }

    #[tokio::test]
    async fn user_confirm_stub_returns_true() {
        let providers = crate::provider::ProviderRegistry::new();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };
        let src = r#"flow t() { return user_confirm("proceed?") }"#;
        let file = parse_file(src).unwrap();
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            assert!(matches!(
                eval_expr(value, &Env::new(), &ctx).await,
                Value::Bool(true)
            ));
        }
    }

    #[tokio::test]
    async fn subflow_calls_target_flow_with_positional_args() {
        let src = r#"flow child(n: Int) -> Int {
    return n + 100
}

flow parent(x: Int) -> Int {
    y = subflow(child, x)
    return y + 1
}
"#;
        let file = parse_file(src).unwrap();
        let flows_map: std::collections::HashMap<_, _> = file
            .flows
            .iter()
            .map(|f| (f.name.name.clone(), f.clone()))
            .collect();
        let parent = &file.flows[1];
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let out = crate::exec::exec_flow_with_siblings(
            parent,
            vec![("x".into(), Value::Int(5))],
            &tools,
            &tool_ctx,
            &providers,
            &flows_map,
            None,
            None,
            None,
            None,
            tokio_util::sync::CancellationToken::new(),
            None,
            None,
        )
        .await
        .unwrap();
        assert!(matches!(out, Value::Int(106)));
    }

    #[tokio::test]
    async fn subflow_missing_target_reports_undefined_tool() {
        let src = r#"flow parent() -> Int {
    return subflow(nope, 1)
}
"#;
        let file = parse_file(src).unwrap();
        let flows: std::collections::HashMap<_, _> = file
            .flows
            .iter()
            .map(|f| (f.name.name.clone(), f.clone()))
            .collect();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let err = crate::exec::exec_flow_with_siblings(
            &file.flows[0],
            vec![],
            &tools,
            &tool_ctx,
            &providers,
            &flows,
            None,
            None,
            None,
            None,
            tokio_util::sync::CancellationToken::new(),
            None,
            None,
        )
        .await
        .unwrap_err();
        assert!(matches!(err, RuntimeError::UndefinedTool(name) if name.contains("nope")));
    }

    #[tokio::test]
    async fn tool_call_dispatches_via_registry() {
        use crate::tools::fs::FsRead;
        use std::sync::Arc;
        use tempfile::TempDir;

        let dir = TempDir::new().unwrap();
        let path = dir.path().join("hi.txt");
        tokio::fs::write(&path, b"hello runtime").await.unwrap();

        let tools = ToolRegistry::new();
        tools.register(Arc::new(FsRead));
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: None,
            turn_id: None,
            flow_run_id: None,
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: None,
            source_dir: None,
        };

        let mut env = Env::new();
        env.bind("p", Value::Path(path));

        let src = r#"flow t() { return fs.read(p) }"#;
        let file = parse_file(src).unwrap();
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let v = eval_expr(value, &env, &ctx).await;
            assert!(matches!(v, Value::Str(s) if s == "hello runtime"));
        }
    }

    #[tokio::test]
    async fn fanout_emits_branch_start_end_events_with_parent_linkage() {
        let src = r#"flow t() { return fanout [1, 2, 3] collect: all }"#;
        let file = parse_file(src).unwrap();
        let tools = ToolRegistry::new();
        let tool_ctx = ToolCtx::new();
        let providers = crate::provider::ProviderRegistry::new();
        let flows = std::collections::HashMap::new();
        let events = crate::event::EventSink::new();
        let ctx = EvalCtx {
            tools: &tools,
            tool_ctx: &tool_ctx,
            providers: &providers,
            flows: &flows,
            contract: None,
            events: Some(&events),
            turn_id: None,
            flow_run_id: Some(crate::event::FlowRunId::now()),
            session_runtime: None,
            flow_cancel: tokio_util::sync::CancellationToken::new(),
            safety: None,
            current_node_id: Some("stmt_1".into()),
            source_dir: None,
        };
        if let atman_dsl::ast::Stmt::Return { value } = &file.flows[0].body[0] {
            let _ = eval_expr(value, &Env::new(), &ctx).await;
        }
        let snap = events.snapshot();
        let starts: Vec<_> = snap
            .iter()
            .filter_map(|e| match e {
                crate::event::Event::FlowNodeStart {
                    node_id,
                    parent_node_id,
                    ..
                } => Some((node_id.clone(), parent_node_id.clone())),
                _ => None,
            })
            .collect();
        assert_eq!(starts.len(), 3);
        assert_eq!(starts[0].0, "stmt_1.branch[0]");
        assert_eq!(starts[1].0, "stmt_1.branch[1]");
        assert_eq!(starts[2].0, "stmt_1.branch[2]");
        assert!(starts.iter().all(|(_, p)| p.as_deref() == Some("stmt_1")));
        let ends = snap
            .iter()
            .filter(|e| matches!(e, crate::event::Event::FlowNodeEnd { .. }))
            .count();
        assert_eq!(ends, 3);
    }

    #[test]
    fn resolve_tool_specs_wildcard_unknown_prefix_skips_silently() {
        let tools = crate::tool::ToolRegistry::new();
        let expr = Expr::List(vec![Expr::Literal(atman_dsl::ast::Literal::Str(
            "nonexistent.*".into(),
        ))]);
        let specs = resolve_tool_specs(&expr, &tools).unwrap();
        assert!(
            specs.is_empty(),
            "wildcard with no matches should return empty list"
        );
    }

    #[test]
    fn resolve_tool_specs_wildcard_matches_prefixed_tools() {
        let tools = crate::tool::ToolRegistry::new();
        struct FakeMcpTool {
            name: String,
            desc: String,
            schema: serde_json::Value,
        }
        impl crate::tool::Tool for FakeMcpTool {
            fn name(&self) -> &str {
                &self.name
            }
            fn description(&self) -> Option<&str> {
                Some(&self.desc)
            }
            fn input_schema(&self) -> serde_json::Value {
                self.schema.clone()
            }
            fn tier(&self) -> crate::tool::Tier {
                crate::tool::Tier::Zero
            }
            fn approval_level(
                &self,
                _args: &crate::tool::ToolArgs,
                _ctx: &crate::tool::ToolCtx,
            ) -> crate::tool::ApprovalLevel {
                crate::tool::ApprovalLevel::Auto
            }
            fn call<'a>(
                &'a self,
                _args: crate::tool::ToolArgs,
                _ctx: &'a crate::tool::ToolCtx,
            ) -> crate::tool::BoxFut<'a, crate::tool::ToolResult> {
                Box::pin(async { Ok(crate::value::Value::Unit) })
            }
        }
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "mcp.lark.send_mail".into(),
            desc: "send mail".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "mcp.lark.read_inbox".into(),
            desc: "read inbox".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "mcp.siyuan.search".into(),
            desc: "search notes".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));
        // Non-MCP tool should NOT be matched.
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "fs.read".into(),
            desc: "read file".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));

        // "mcp.*" matches all 3 mcp.* tools, but not fs.read
        let expr = Expr::List(vec![Expr::Literal(atman_dsl::ast::Literal::Str(
            "mcp.*".into(),
        ))]);
        let specs = resolve_tool_specs(&expr, &tools).unwrap();
        assert_eq!(specs.len(), 3, "mcp.* should match 3 MCP tools");
        let names: Vec<String> = specs.iter().map(|s| s.name.clone()).collect();
        assert!(names.contains(&"mcp.lark.send_mail".into()));
        assert!(names.contains(&"mcp.lark.read_inbox".into()));
        assert!(names.contains(&"mcp.siyuan.search".into()));

        // "mcp.lark.*" matches only the 2 lark tools
        let expr2 = Expr::List(vec![Expr::Literal(atman_dsl::ast::Literal::Str(
            "mcp.lark.*".into(),
        ))]);
        let specs2 = resolve_tool_specs(&expr2, &tools).unwrap();
        assert_eq!(specs2.len(), 2, "mcp.lark.* should match 2 lark tools");
    }

    #[test]
    fn resolve_tool_specs_mixed_concrete_and_wildcard() {
        let tools = crate::tool::ToolRegistry::new();
        struct FakeMcpTool {
            name: String,
            desc: String,
            schema: serde_json::Value,
        }
        impl crate::tool::Tool for FakeMcpTool {
            fn name(&self) -> &str {
                &self.name
            }
            fn description(&self) -> Option<&str> {
                Some(&self.desc)
            }
            fn input_schema(&self) -> serde_json::Value {
                self.schema.clone()
            }
            fn tier(&self) -> crate::tool::Tier {
                crate::tool::Tier::Zero
            }
            fn approval_level(
                &self,
                _args: &crate::tool::ToolArgs,
                _ctx: &crate::tool::ToolCtx,
            ) -> crate::tool::ApprovalLevel {
                crate::tool::ApprovalLevel::Auto
            }
            fn call<'a>(
                &'a self,
                _args: crate::tool::ToolArgs,
                _ctx: &'a crate::tool::ToolCtx,
            ) -> crate::tool::BoxFut<'a, crate::tool::ToolResult> {
                Box::pin(async { Ok(crate::value::Value::Unit) })
            }
        }
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "mcp.lark.send_mail".into(),
            desc: "send mail".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));
        tools.register(std::sync::Arc::new(FakeMcpTool {
            name: "bash.exec".into(),
            desc: "exec".into(),
            schema: serde_json::json!({"type":"object","properties":{}}),
        }));

        // Build tools list from DSL parse so we get valid spans.
        let src = r#"flow t() -> string {
    reply = llm {
        tools: [bash.exec, "mcp.*"]
    }
    return "ok"
}"#;
        let file = atman_dsl::parse::parse_file(src).unwrap();
        // Extract the tools list from the llm node
        let body = &file.flows[0].body;
        let tools_expr = match &body[0] {
            atman_dsl::ast::Stmt::Bind { value, .. } => match value {
                Expr::Node(atman_dsl::ast::Node::Llm { kwargs }) => kwargs
                    .iter()
                    .find(|(k, _)| k.name == "tools")
                    .map(|(_, v)| v.clone())
                    .unwrap(),
                _ => panic!("expected llm node"),
            },
            _ => panic!("expected bind stmt"),
        };
        let specs = resolve_tool_specs(&tools_expr, &tools).unwrap();
        assert_eq!(specs.len(), 2, "bash.exec + mcp.lark.send_mail = 2");
        let names: Vec<String> = specs.iter().map(|s| s.name.clone()).collect();
        assert!(names.contains(&"bash.exec".into()));
        assert!(names.contains(&"mcp.lark.send_mail".into()));
    }
}

#[cfg(test)]
mod sanitize_tests {
    use super::*;
    use crate::message::{Message, MessagePart, MessageRole};

    #[test]
    fn sanitize_fills_missing_tool_results() {
        let turn = crate::event::TurnId::now();
        let msgs = vec![
            Message {
                role: MessageRole::Assistant,
                parts: vec![MessagePart::ToolUse {
                    id: "call_orphan".into(),
                    name: "bash.exec".into(),
                    input: serde_json::json!({}),
                }],
                turn_id: turn.clone(),
            },
            Message {
                role: MessageRole::User,
                parts: vec![MessagePart::Text {
                    text: "user interrupt".into(),
                }],
                turn_id: turn.clone(),
            },
        ];
        let out = sanitize_tool_pairs(msgs);
        let has_filler = out.iter().any(|m| {
            m.parts.iter().any(|p| {
                matches!(p, MessagePart::ToolResult { tool_use_id, is_error: true, .. } if tool_use_id == "call_orphan")
            })
        });
        assert!(
            has_filler,
            "should append error tool_result for orphan tool_use"
        );
    }

    #[test]
    fn sanitize_noop_when_pairs_complete() {
        let turn = crate::event::TurnId::now();
        let msgs = vec![
            Message {
                role: MessageRole::Assistant,
                parts: vec![MessagePart::ToolUse {
                    id: "call_ok".into(),
                    name: "bash.exec".into(),
                    input: serde_json::json!({}),
                }],
                turn_id: turn.clone(),
            },
            Message {
                role: MessageRole::Tool,
                parts: vec![MessagePart::ToolResult {
                    tool_use_id: "call_ok".into(),
                    content: "done".into(),
                    is_error: false,
                }],
                turn_id: turn.clone(),
            },
        ];
        let out = sanitize_tool_pairs(msgs);
        assert_eq!(
            out.len(),
            2,
            "no filler should be added when pairs complete"
        );
    }

    // --- stall timeout tests ---
    use crate::providers::mock::MockProvider;

    fn stall_req(stall_secs: u64) -> crate::provider::LlmRequest {
        crate::provider::LlmRequest {
            model: "mock".into(),
            messages: vec![crate::provider::user_text_message("test")],
            system: None,
            input: crate::value::Value::Unit,
            schema: None,
            cache_prompt: false,
            tools: Vec::new(),
            thinking_enabled: false,
            stall_timeout_secs: stall_secs,
        }
    }

    #[tokio::test]
    async fn stall_timeout_fires_when_no_chunks_arrive() {
        // chunk_delay = 3s, stall_timeout = 1s → stall fires before 2nd chunk
        let provider = MockProvider::new("mock")
            .with_model("mock", Value::Str("hello world test".into()))
            .with_chunk_delay(std::time::Duration::from_secs(3));

        let (stream_tx, _rx) = tokio::sync::broadcast::channel(16);
        let result = call_and_maybe_stream(
            &provider,
            stall_req(1),
            StreamCallCtx {
                stream_tx: Some(stream_tx),
                ..Default::default()
            },
        )
        .await;
        match result {
            Err(RuntimeError::ToolFailed(msg)) => {
                assert!(
                    msg.contains("llm stall timeout after 1s"),
                    "expected stall message, got: {msg}"
                );
            }
            other => panic!("expected ToolFailed stall timeout, got: {other:?}"),
        }
    }

    #[tokio::test]
    async fn stall_timeout_does_not_fire_when_chunks_keep_coming() {
        // chunk_delay = 100ms, stall_timeout = 2s → all chunks within 300ms, no stall
        let provider = MockProvider::new("mock")
            .with_model("mock", Value::Str("hello world test".into()))
            .with_chunk_delay(std::time::Duration::from_millis(100));

        let (stream_tx, _rx) = tokio::sync::broadcast::channel(16);
        let result = call_and_maybe_stream(
            &provider,
            stall_req(2),
            StreamCallCtx {
                stream_tx: Some(stream_tx),
                ..Default::default()
            },
        )
        .await;
        match result {
            Ok(am) => {
                assert!(am.text_concat().contains("hello"));
            }
            other => panic!("expected Ok, got: {other:?}"),
        }
    }

    #[tokio::test]
    async fn stall_timeout_zero_disables_detection() {
        // chunk_delay = 3s, stall_timeout = 0 → disabled, all chunks arrive
        let provider = MockProvider::new("mock")
            .with_model("mock", Value::Str("hello world test".into()))
            .with_chunk_delay(std::time::Duration::from_secs(3));

        let (stream_tx, _rx) = tokio::sync::broadcast::channel(16);
        let result = call_and_maybe_stream(
            &provider,
            stall_req(0),
            StreamCallCtx {
                stream_tx: Some(stream_tx),
                ..Default::default()
            },
        )
        .await;
        match result {
            Ok(am) => {
                assert!(am.text_concat().contains("hello"));
            }
            other => panic!("expected Ok (stall disabled), got: {other:?}"),
        }
    }

    #[tokio::test]
    async fn stall_timeout_resets_on_each_chunk() {
        // 3 chunks at 800ms each. stall=1s. First chunk at t=0, second at t=800ms (<1s),
        // third at t=1.6s (>1s from start, but only 800ms from last chunk). Should pass.
        let provider = MockProvider::new("mock")
            .with_model("mock", Value::Str("hello world test".into()))
            .with_chunk_delay(std::time::Duration::from_millis(800));

        let (stream_tx, _rx) = tokio::sync::broadcast::channel(16);
        let result = call_and_maybe_stream(
            &provider,
            stall_req(1),
            StreamCallCtx {
                stream_tx: Some(stream_tx),
                ..Default::default()
            },
        )
        .await;
        match result {
            Ok(am) => {
                assert!(am.text_concat().contains("hello"));
            }
            other => panic!("expected Ok (timer reset each chunk), got: {other:?}"),
        }
    }

    #[tokio::test]
    async fn stall_timeout_fires_between_first_and_second_chunk() {
        // first chunk at t≈0, then 2s gap, stall=1s fires at t=1s
        let provider = MockProvider::new("mock")
            .with_model("mock", Value::Str("hello world test".into()))
            .with_chunk_delay(std::time::Duration::from_secs(2));

        let (stream_tx, _rx) = tokio::sync::broadcast::channel(16);
        let result = call_and_maybe_stream(
            &provider,
            stall_req(1),
            StreamCallCtx {
                stream_tx: Some(stream_tx),
                ..Default::default()
            },
        )
        .await;
        assert!(
            matches!(&result, Err(RuntimeError::ToolFailed(msg)) if msg.contains("stall timeout")),
            "expected stall timeout, got: {result:?}"
        );
    }
}