axon-lang 4.6.0

//! `axon run` native implementation — stub + real execution.
//!
//! Pipeline: Source → Lex → Parse → Type-check → IR → Execution Plan → Execute
//!
//! Execution modes:
//!   - stub (default): prints execution plan without API calls
//!   - real: sends each step to LLM backend (Anthropic Messages API)
//!
//! Exit codes:
//!   0 — success
//!   1 — compilation or execution error
//!   2 — I/O or configuration error
//!
//! # v1.24.0 — `crate::backend` deprecation
//!
//! This file is one of four callers of the deprecated synchronous
//! `crate::backend` mono-file (see `backend.rs` module docs).
//! The `#![allow(deprecated)]` below silences the deprecation
//! warnings on this file's call sites while the deeper async
//! migration progresses under followup step v1.24.0
//! (sync→async migration of the 4 callers, separate cycle).

#![allow(deprecated)]

use std::io::{self, IsTerminal};
use std::path::Path;

use crate::backend;
use crate::exec_context::ExecContext;
use crate::hooks::HookManager;
use crate::ir_generator::IRGenerator;
use crate::ir_nodes::*;
use crate::lexer::{Lexer, LexerError};
use crate::output::{OutputFormat, ReportBuilder, StepReport};
use crate::plan_export::{self, PlanBuilder, PlanUnit, PlanStep, PlanTools, PlanToolEntry, PlanDependencies, UnresolvedRef};
use crate::parser::{ParseError, Parser};
use crate::step_deps;
#[cfg(feature = "postgres")]
use crate::store::epistemic;
#[cfg(feature = "postgres")]
use crate::store::filter::SqlValue;
#[cfg(feature = "postgres")]
use crate::store::row_stream;
use crate::store::error::StoreError;
use crate::store::registry::StoreRegistry;
use crate::tool_registry::ToolRegistry;
use crate::type_checker::TypeChecker;

/// Single source of truth for the AXON version string.
///
/// v2.81.0 — **the definition moved to [`crate::version`]**, a leaf module
/// with no dependencies. It lived here because centralising it killed a real
/// drift bug (five files each declaring their own stale literal), but `runner`
/// is the flow executor: importing a constant from it dragged `sqlx`,
/// `reqwest`, `tokio`, `axum` and `axon-csys` into the reachable set of every
/// compiler-side subcommand. This re-export keeps `axon::runner::AXON_VERSION`
/// resolving for existing call sites; new code should name `crate::version`.
pub use crate::version::AXON_VERSION;

// ── ANSI helpers ─────────────────────────────────────────────────────────────

fn c(text: &str, code: &str, use_color: bool) -> String {
    if use_color {
        format!("{code}{text}\x1b[0m")
    } else {
        text.to_string()
    }
}

// ── Helpers ─────────────────────────────────────────────────────────────────

// ── Compiled execution plan ─────────────────────────────────────────────────

/// A compiled execution unit — one per `run` statement.
#[derive(Debug, serde::Serialize)]
struct ExecutionUnit {
    flow_name: String,
    persona_name: String,
    context_name: String,
    system_prompt: String,
    steps: Vec<CompiledStep>,
    anchor_instructions: Vec<String>,
    effort: String,
    #[serde(skip)]
    resolved_anchors: Vec<IRAnchor>,
    /// v1.32.0 (D1) — the Request Binding Contract bindings:
    /// `(flow parameter name, value)` pairs resolved from the HTTP
    /// request body. Seeded into the unit's `ExecContext` before the
    /// step walk so `${param}` interpolates. Empty for a caller with
    /// no request body (CLI / batch / pipeline) — D5 backwards-compat.
    #[serde(skip)]
    param_bindings: Vec<(String, String)>,
}

/// A compiled step ready for LLM dispatch.
#[derive(Debug, serde::Serialize)]
struct CompiledStep {
    step_name: String,
    step_type: String,
    system_prompt: String,
    user_prompt: String,
    /// For `use_tool` steps: the raw argument expression.
    #[serde(skip_serializing_if = "Option::is_none")]
    tool_argument: Option<String>,
    /// For memory steps: the expression/query/target.
    #[serde(skip_serializing_if = "Option::is_none")]
    memory_expression: Option<String>,
    /// v1.10.0 — for `lambda_data_apply` steps: the full payload
    /// (spec snapshot + target + output_type) so the runner can build
    /// ψ = ⟨T, V, E⟩ without reaching back into the IR.
    #[serde(skip_serializing_if = "Option::is_none")]
    lambda_apply_payload: Option<crate::lambda_runtime::LambdaApplyPayload>,
    /// v1.12.0 — for `let_binding` steps: the payload (target,
    /// value, value_kind) so the stub executor can perform the
    /// SSA binding without re-traversing the IR.
    #[serde(skip_serializing_if = "Option::is_none")]
    let_payload: Option<LetPayload>,
    /// v1.30.0 — for `persist` (INSERT columns) and `mutate`
    /// (UPDATE SET assignments) steps: the declared `{ col: value }`
    /// block. `Some` ⇒ the SQL row is built from exactly these columns
    /// (interpolated); `None` ⇒ no block was written and the runtime
    /// falls back to the flow's user bindings (v1.31.0).
    #[serde(skip_serializing_if = "Option::is_none")]
    store_fields: Option<Vec<(String, String)>>,
    /// v2.21.0 — for a `retrieve` step: the raw `order_by:` /
    /// `limit:` clauses. `None` = the clause was absent (the pre-67.b
    /// unordered / unbounded form). Carried so the executor builds the
    /// `ORDER BY … LIMIT …` suffix without reaching back into the IR.
    #[serde(skip_serializing_if = "Option::is_none")]
    retrieve_order_by: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    retrieve_limit: Option<String>,
    /// v2.33.0 — for a `retrieve` step: the raw `aggregate:` /
    /// `group_by:` clauses (closed catalog: `count` / `sum(col)` /
    /// `avg(col)` / `min(col)` / `max(col)`; comma-separated group
    /// columns). `None` = absent (the plain `SELECT *` retrieve).
    #[serde(skip_serializing_if = "Option::is_none")]
    retrieve_aggregate: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    retrieve_group_by: Option<String>,
    /// v2.8.0 — for a `use Tool(k = v, …)` dispatch: the bound keyword
    /// args `(name, raw value)`. Non-empty ⇒ the runtime assembles a STRUCTURED
    /// JSON request body (`{"query":"…","max_results":5}`) instead of the flat
    /// `{"input": …}`. Empty for the legacy single-`on <arg>` form (D5).
    /// v2.10.0 — each entry is `(name, raw value, value_kind)`; `value_kind`
    /// (`"literal"` / `"reference"`) drives runtime resolution: a reference is a
    /// binding lookup, a literal keeps `${…}` interpolation.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    tool_named_args: Vec<(String, String, String)>,
    /// v2.8.0 — the called tool's declared `(param, type)` schema, resolved
    /// from `ir.tools` at build time so the runtime coerces each arg value to
    /// its DECLARED JSON type (a `String` param stays a string even when its
    /// value is all-digits) without reaching back into the IR.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    tool_param_types: Vec<(String, String)>,
    /// v2.46.0 — the step's EFFECTIVE declared cognitive timezone
    /// (step-level `now:` ∨ the bound `context`'s `now:` — resolved at plan
    /// build). When present, the executor appends the unit's captured
    /// instant — rendered in this zone — to the step's system prompt
    /// (`time_is_an_explicit_input`). Elided from the compiled-plan JSON
    /// when absent (pre-v2.46.0 plans byte-identical).
    #[serde(skip_serializing_if = "Option::is_none")]
    now_tz: Option<String>,
}

/// v1.12.0 — payload carried inside a CompiledStep for `let X = value`
/// SSA bindings. `value_kind` ∈ {"literal", "reference", "expression"}
/// disambiguates a quoted literal from a dotted-identifier reference
/// resolved at runtime.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LetPayload {
    pub target: String,
    pub value: String,
    pub value_kind: String,
}

/// Trace event for execution recording.
#[derive(Debug, serde::Serialize)]
struct TraceEvent {
    event: String,
    unit: String,
    step: String,
    detail: String,
}

// ── Build execution plan from IR ────────────────────────────────────────────

fn build_execution_plan(ir: &IRProgram, backend: &str) -> Vec<ExecutionUnit> {
    let mut units = Vec::new();

    for run in &ir.runs {
        let system_prompt = build_system_prompt(run, backend);
        let anchor_instructions = build_anchor_instructions(run);
        let steps = build_compiled_steps(run, ir);

        units.push(ExecutionUnit {
            flow_name: run.flow_name.clone(),
            persona_name: run.persona_name.clone(),
            context_name: run.context_name.clone(),
            system_prompt,
            steps,
            anchor_instructions,
            effort: run.effort.clone(),
            resolved_anchors: run.resolved_anchors.clone(),
            // v1.32.0 — the CLI / `run`-statement plan builder has
            // no HTTP request body; the binding is empty (D5).
            param_bindings: Vec::new(),
        });
    }

    units
}

fn build_system_prompt(run: &IRRun, backend: &str) -> String {
    let mut parts: Vec<String> = Vec::new();

    // Persona block
    if let Some(ref persona) = run.resolved_persona {
        parts.push(format!("# Persona: {}", persona.name));
        if !persona.domain.is_empty() {
            parts.push(format!("Domain expertise: {}", persona.domain.join(", ")));
        }
        if !persona.tone.is_empty() {
            parts.push(format!("Communication tone: {}", persona.tone));
        }
        if !persona.language.is_empty() {
            parts.push(format!("Language: {}", persona.language));
        }
        if let Some(ct) = persona.confidence_threshold {
            parts.push(format!("Confidence threshold: {ct:.2}"));
        }
        if persona.cite_sources == Some(true) {
            parts.push("Always cite sources.".to_string());
        }
        if !persona.refuse_if.is_empty() {
            parts.push(format!("Refuse if: {}", persona.refuse_if.join(", ")));
        }
    }

    // Context block
    if let Some(ref ctx) = run.resolved_context {
        parts.push(format!("\n# Context: {}", ctx.name));
        if !ctx.depth.is_empty() {
            parts.push(format!("Analysis depth: {}", ctx.depth));
        }
        if !ctx.memory_scope.is_empty() {
            parts.push(format!("Memory scope: {}", ctx.memory_scope));
        }
        if let Some(t) = ctx.temperature {
            parts.push(format!("Temperature: {t:.1}"));
        }
        if let Some(mt) = ctx.max_tokens {
            parts.push(format!("Max tokens: {mt}"));
        }
    }

    // Anchor enforcement
    if !run.resolved_anchors.is_empty() {
        parts.push("\n# Constraints (Anchors)".to_string());
        for anchor in &run.resolved_anchors {
            let mut constraint = format!("- {}: {}", anchor.name, anchor.require);
            if let Some(cf) = anchor.confidence_floor {
                constraint.push_str(&format!(" (confidence ≥ {cf:.2})"));
            }
            if !anchor.on_violation.is_empty() {
                constraint.push_str(&format!(" [on_violation: {}]", anchor.on_violation));
            }
            parts.push(constraint);
        }
    }

    // Backend tag
    parts.push(format!("\n[Backend: {backend} | AXON {AXON_VERSION}]"));

    parts.join("\n")
}

fn build_anchor_instructions(run: &IRRun) -> Vec<String> {
    run.resolved_anchors
        .iter()
        .map(|a| {
            let mut s = format!("{}: {}", a.name, a.require);
            if let Some(cf) = a.confidence_floor {
                s.push_str(&format!(" (≥{cf:.2})"));
            }
            s
        })
        .collect()
}

fn build_compiled_steps(run: &IRRun, ir: &IRProgram) -> Vec<CompiledStep> {
    let flow = match &run.resolved_flow {
        Some(f) => f,
        None => return Vec::new(),
    };

    let mut steps = Vec::new();
    for node in &flow.steps {
        let (step_name, step_type, action) = extract_step_info(node);
        let system_prompt = format!(
            "You are executing step '{}' of flow '{}'.",
            step_name, flow.name
        );
        let user_prompt = if action.is_empty() {
            format!("Execute step: {step_name}")
        } else {
            action
        };

        // Extract tool argument for use_tool steps
        let tool_argument = match node {
            IRFlowNode::UseTool(s) => Some(s.argument.clone()),
            _ => None,
        };

        // v2.8.0 — the structured keyword args of a `use Tool(k = v, …)`
        // dispatch, plus the called tool's declared `(param, type)` schema
        // (resolved once from `ir.tools`) so the runtime coerces each value to
        // its declared JSON type. Both empty for the legacy single-arg form.
        let (tool_named_args, tool_param_types) = match node {
            IRFlowNode::UseTool(s) => {
                let named: Vec<(String, String, String)> = s
                    .named_args
                    .iter()
                    .map(|a| (a.name.clone(), a.value.clone(), a.value_kind.clone()))
                    .collect();
                let types: Vec<(String, String)> = ir
                    .tools
                    .iter()
                    .find(|t| t.name == s.tool_name)
                    .map(|t| {
                        t.parameters
                            .iter()
                            .map(|p| (p.name.clone(), p.type_name.clone()))
                            .collect()
                    })
                    .unwrap_or_default();
                (named, types)
            }
            _ => (Vec::new(), Vec::new()),
        };

        // Extract memory expression for remember/recall/persist/retrieve/mutate/purge
        let memory_expression = match node {
            IRFlowNode::Remember(s) => Some(s.expression.clone()),
            IRFlowNode::Recall(s) => Some(s.query.clone()),
            IRFlowNode::Persist(s) => Some(s.store_name.clone()),
            IRFlowNode::Retrieve(s) => Some(format!("{}:{}", s.store_name, s.where_expr)),
            IRFlowNode::Mutate(s) => Some(format!("{}:{}", s.store_name, s.where_expr)),
            IRFlowNode::Purge(s) => Some(format!("{}:{}", s.store_name, s.where_expr)),
            _ => None,
        };

        // v1.10.0 — materialise the lambda apply payload by looking
        // up the spec snapshot from ir.lambda_data_specs. The runner
        // needs the full snapshot at execute-time to construct ψ;
        // carrying it in the CompiledStep keeps the executor free of
        // IR back-references (mirrors Python's BaseBackend pattern).
        let lambda_apply_payload = match node {
            IRFlowNode::LambdaDataApply(s) => {
                let snap = ir
                    .lambda_data_specs
                    .iter()
                    .find(|spec| spec.name == s.lambda_data_name)
                    .map(|spec| crate::lambda_runtime::SpecSnapshot {
                        name: spec.name.clone(),
                        ontology: spec.ontology.clone(),
                        certainty: spec.certainty,
                        temporal_frame_start: spec.temporal_frame_start.clone(),
                        temporal_frame_end: spec.temporal_frame_end.clone(),
                        provenance: spec.provenance.clone(),
                        derivation: spec.derivation.clone(),
                    })
                    .unwrap_or_default();
                Some(crate::lambda_runtime::LambdaApplyPayload {
                    lambda_data_name: s.lambda_data_name.clone(),
                    target: s.target.clone(),
                    output_type: s.output_type.clone(),
                    spec_snapshot: snap,
                })
            }
            _ => None,
        };

        // v1.12.0 — materialise the let payload from the IR Let
        // node so the stub executor can bind without re-traversing
        // the IR. Same pattern as the lambda apply payload above.
        let let_payload = match node {
            IRFlowNode::Let(s) => Some(LetPayload {
                target: s.target.clone(),
                value: s.value.clone(),
                value_kind: s.value_kind.clone(),
            }),
            _ => None,
        };

        // v1.30.0 — materialise the declared `{ col: value }`
        // block of a `persist` (INSERT columns) or `mutate` (UPDATE SET
        // assignments) so `execute_sql_store_step` scopes the SQL row
        // to exactly those columns. No block ⇒ `None` → the v1.31.0
        // user-bindings fallback.
        let store_fields = match node {
            IRFlowNode::Persist(s) if !s.fields.is_empty() => {
                Some(s.fields.clone())
            }
            IRFlowNode::Mutate(s) if !s.fields.is_empty() => {
                Some(s.fields.clone())
            }
            _ => None,
        };

        // v2.21.0 — carry a `retrieve` step's `order_by:` / `limit:`
        // clauses so the executor renders the `ORDER BY … LIMIT …` suffix.
        // Empty source strings stay `None` (no clause written).
        // v2.33.0 — same carriage for `aggregate:` / `group_by:`.
        let (retrieve_order_by, retrieve_limit, retrieve_aggregate, retrieve_group_by) =
            match node {
                IRFlowNode::Retrieve(s) => (
                    Some(s.order_by.clone()).filter(|v| !v.is_empty()),
                    Some(s.limit_expr.clone()).filter(|v| !v.is_empty()),
                    Some(s.aggregate.clone()).filter(|v| !v.is_empty()),
                    Some(s.group_by.clone()).filter(|v| !v.is_empty()),
                ),
                _ => (None, None, None, None),
            };

        // v2.46.0 — the step's EFFECTIVE declared cognitive timezone:
        // its own `now:` overrides the bound `context` frame's (`run …
        // within <Context>`). Only real cognitive steps carry it.
        let now_tz = match node {
            IRFlowNode::Step(s) => s.now_tz.clone().or_else(|| {
                run.resolved_context
                    .as_ref()
                    .and_then(|c| c.now_tz.clone())
            }),
            _ => None,
        };

        steps.push(CompiledStep {
            step_name,
            step_type,
            system_prompt,
            user_prompt,
            tool_argument,
            memory_expression,
            lambda_apply_payload,
            let_payload,
            store_fields,
            retrieve_order_by,
            retrieve_limit,
            retrieve_aggregate,
            retrieve_group_by,
            tool_named_args,
            tool_param_types,
            now_tz,
        });
    }

    steps
}

/// v2.8.0 — assemble the STRUCTURED JSON request body for a `use Tool(k =
/// v, …)` dispatch from its ALREADY-INTERPOLATED `(name, value)` args. Each
/// value is coerced to its DECLARED parameter type so the tool backend receives
/// `{"query":"Acme","max_results":5,"safe":true}` — not a flat
/// `{"input": "…"}`. serde builds the object, so JSON escaping is correct.
pub(crate) fn build_structured_tool_body(
    interpolated_args: &[(String, String)],
    param_types: &[(String, String)],
) -> String {
    let mut map = serde_json::Map::new();
    for (name, value) in interpolated_args {
        let declared = param_types
            .iter()
            .find(|(p, _)| p == name)
            .map(|(_, t)| t.as_str());
        map.insert(name.clone(), coerce_tool_arg_value(value, declared));
    }
    serde_json::Value::Object(map).to_string()
}

/// v2.8.0 / v2.77.0 — coerce an interpolated arg value to JSON per its
/// DECLARED type. `Int`/`Float`/`Bool` parse into the matching JSON scalar;
/// `List<T>` materializes into a JSON ARRAY (see [`coerce_list_value`]); a value
/// that does not parse falls back to a JSON string (the v2.8.0 type-checker
/// already flags a literal mismatch at compile time — interpolated/runtime values
/// are coerced leniently rather than dropped). `String`, custom domain types, and
/// unknown / schema-less (`None`) stay JSON strings — so a `String` parameter
/// keeps its value verbatim even when it is all-digits.
pub(crate) fn coerce_tool_arg_value(value: &str, declared_type: Option<&str>) -> serde_json::Value {
    // Strip the optional `?` marker; the base is the head before `<` (`List` for
    // `List<String>`), the inner is the `<…>` payload (element type).
    let normalized = declared_type.map(|t| t.trim().trim_end_matches('?').trim());
    let base = normalized.map(|t| t.split('<').next().unwrap_or(t).trim());
    match base {
        Some("List") => {
            let inner = normalized
                .and_then(|t| t.split_once('<'))
                .map(|(_, rest)| rest.trim_end_matches('>').trim())
                .unwrap_or("String");
            coerce_list_value(value, inner)
        }
        Some(scalar) => coerce_scalar_value(value, scalar),
        None => serde_json::Value::String(value.to_string()),
    }
}

/// Coerce ONE scalar value to JSON per its base type. Non-parsing values fall
/// back to a JSON string (lenient — the compile-time checker owns rejection).
fn coerce_scalar_value(value: &str, base: &str) -> serde_json::Value {
    match base {
        "Int" => value
            .parse::<i64>()
            .map(|i| serde_json::Value::Number(i.into()))
            .unwrap_or_else(|_| serde_json::Value::String(value.to_string())),
        "Float" => value
            .parse::<f64>()
            .ok()
            .and_then(serde_json::Number::from_f64)
            .map(serde_json::Value::Number)
            .unwrap_or_else(|| serde_json::Value::String(value.to_string())),
        "Bool" => match value {
            "true" => serde_json::Value::Bool(true),
            "false" => serde_json::Value::Bool(false),
            _ => serde_json::Value::String(value.to_string()),
        },
        _ => serde_json::Value::String(value.to_string()),
    }
}

/// v2.77.0 — materialize a `List<T>` argument into a JSON array so a
/// `use Tool(media_urls = ["a", "b"])` reaches the connector as a real array
/// (pre-116.c.4 it stayed the opaque JSON STRING `"[a, b]"`, so EVERY list-typed
/// tool param — FB multi-photo, IG carousel — was unreachable from a flow).
///
/// Two forms are accepted:
///   1. Already-valid JSON (a step output, or a quoted-item literal `["a","b"]`)
///      whose parse is an array — passed through verbatim.
///   2. The parser's lossy surface rendering `[a, b, c]` (list items are
///      comma-joined and unquoted by `parse_let_list_literal`) — split at the top
///      level (quote/bracket-aware), surrounding quotes stripped, each element
///      coerced by the inner type `T`.
///
/// Residual limitation (named, not silent): because the surface rendering drops
/// the source quotes, an item that itself contains a top-level comma cannot be
/// recovered from form (2) — realistic string/URL/number lists do not hit this;
/// the deeper fix is structured list values carried through the IR (like v2.26.0's
/// `value_ast` for expressions).
fn coerce_list_value(value: &str, inner: &str) -> serde_json::Value {
    let trimmed = value.trim();
    // Form (1): already valid JSON array.
    if let Ok(v @ serde_json::Value::Array(_)) = serde_json::from_str::<serde_json::Value>(trimmed) {
        return v;
    }
    // Form (2): the `[…]` surface rendering.
    match trimmed.strip_prefix('[').and_then(|s| s.strip_suffix(']')) {
        Some(body) if body.trim().is_empty() => serde_json::Value::Array(Vec::new()),
        Some(body) => serde_json::Value::Array(
            split_top_level_commas(body)
                .into_iter()
                .map(|item| coerce_scalar_value(strip_matching_quotes(item.trim()), inner))
                .collect(),
        ),
        // Not a list surface — a lone value for a `List` param is a 1-element list.
        None => serde_json::Value::Array(vec![coerce_scalar_value(trimmed, inner)]),
    }
}

/// Split on commas at bracket/brace depth 0 and outside quotes, so a nested
/// structure or a quoted element containing a comma is not split mid-item.
fn split_top_level_commas(s: &str) -> Vec<String> {
    let mut out = Vec::new();
    let mut buf = String::new();
    let mut depth = 0i32;
    let mut quote: Option<char> = None;
    for c in s.chars() {
        match quote {
            Some(q) => {
                buf.push(c);
                if c == q {
                    quote = None;
                }
            }
            None => match c {
                '"' | '\'' => {
                    quote = Some(c);
                    buf.push(c);
                }
                '[' | '{' => {
                    depth += 1;
                    buf.push(c);
                }
                ']' | '}' => {
                    depth -= 1;
                    buf.push(c);
                }
                ',' if depth == 0 => out.push(std::mem::take(&mut buf)),
                _ => buf.push(c),
            },
        }
    }
    out.push(buf);
    out
}

/// Strip a single pair of matching surrounding quotes (`"…"` or `'…'`).
fn strip_matching_quotes(s: &str) -> &str {
    let b = s.as_bytes();
    if b.len() >= 2 && (b[0] == b'"' || b[0] == b'\'') && b[b.len() - 1] == b[0] {
        &s[1..s.len() - 1]
    } else {
        s
    }
}

fn extract_step_info(node: &IRFlowNode) -> (String, String, String) {
    match node {
        IRFlowNode::Step(s) => (s.name.clone(), "step".to_string(), s.ask.clone()),
        IRFlowNode::Declassify(s) => (s.output_type.clone(), "declassify".to_string(), format!("Declassify {} from {} via {}", s.class, s.source, s.shield)),
        IRFlowNode::Grad(s) => (s.output.clone(), "grad".to_string(), format!("Grad: d({})/d{:?}", s.target, s.wrt)),
        IRFlowNode::Probe(s) => (s.target.clone(), "probe".to_string(), format!("Probe: {}", s.target)),
        IRFlowNode::Reason(s) => (s.target.clone(), "reason".to_string(), format!("Reason about: {}", s.target)),
        IRFlowNode::Validate(s) => (s.target.clone(), "validate".to_string(), format!("Validate: {}", s.target)),
        IRFlowNode::Refine(s) => (s.target.clone(), "refine".to_string(), format!("Refine: {}", s.target)),
        IRFlowNode::Weave(s) => ("weave".to_string(), "weave".to_string(), format!("Weave {} sources into {}", s.sources.len(), s.target)),
        IRFlowNode::UseTool(s) => (s.tool_name.clone(), "use_tool".to_string(), format!("Use tool: {}", s.tool_name)),
        // v2.83.0 — the audit row names the AGENT, because that is the
        // unit an adopter reasons about; the loop's own iterations already
        // appear as their own `agent`-slug rows beneath it.
        IRFlowNode::AgentCall(s) => (s.agent_name.clone(), "agent_call".to_string(), format!("Run agent: {}({})", s.agent_name, s.arguments.join(", "))),
        // v2.46.0 — ephemeral-credential minting (dispatcher-routed).
        IRFlowNode::Mint(s) => (s.credential_ref.clone(), "mint".to_string(), format!("Mint credential: {} as {}", s.credential_ref, s.binding)),
        // v2.48.0 — mediated secret renewal (dispatcher-routed; the
        // summary names store + tool, NEVER a secret value).
        IRFlowNode::Rotate(s) => (s.store_ref.clone(), "rotate".to_string(), format!("Rotate secrets: {} with {}", s.store_ref, s.tool_ref)),
        IRFlowNode::Remember(s) => (s.memory_target.clone(), "remember".to_string(), format!("Remember: {}", s.expression)),
        IRFlowNode::Recall(s) => (s.memory_source.clone(), "recall".to_string(), format!("Recall: {}", s.query)),
        IRFlowNode::Conditional(s) => (s.condition.clone(), "conditional".to_string(), format!("If: {}", s.condition)),
        IRFlowNode::ForIn(s) => (s.variable.clone(), "for_in".to_string(), format!("For {} in {}", s.variable, s.iterable)),
        IRFlowNode::Let(s) => (s.target.clone(), "let".to_string(), format!("Let {} = {}", s.target, s.value)),
        IRFlowNode::Return(s) => ("return".to_string(), "return".to_string(), format!("Return: {}", s.value_expr)),
        IRFlowNode::Par(_) => ("parallel".to_string(), "parallel".to_string(), "Parallel block".to_string()),
        IRFlowNode::Hibernate(_) => ("hibernate".to_string(), "hibernate".to_string(), "Hibernate".to_string()),
        IRFlowNode::Deliberate(_) => ("deliberate".to_string(), "deliberate".to_string(), "Deliberate block".to_string()),
        IRFlowNode::Consensus(_) => ("consensus".to_string(), "consensus".to_string(), "Consensus block".to_string()),
        IRFlowNode::Forge(_) => ("forge".to_string(), "forge".to_string(), "Forge block".to_string()),
        IRFlowNode::Focus(s) => (s.expression.clone(), "focus".to_string(), format!("Focus: {}", s.expression)),
        IRFlowNode::Associate(s) => (s.left.clone(), "associate".to_string(), format!("Associate: {} ↔ {}", s.left, s.right)),
        IRFlowNode::Aggregate(s) => (s.target.clone(), "aggregate".to_string(), format!("Aggregate: {}", s.target)),
        IRFlowNode::Explore(s) => (s.target.clone(), "explore".to_string(), format!("Explore: {}", s.target)),
        IRFlowNode::Ingest(s) => (s.source.clone(), "ingest".to_string(), format!("Ingest: {}", s.source)),
        IRFlowNode::ShieldApply(s) => (s.shield_name.clone(), "shield_apply".to_string(), format!("Apply shield: {}", s.shield_name)),
        IRFlowNode::Stream(_) => ("stream".to_string(), "stream".to_string(), "Stream block".to_string()),
        // v2.87.0 — the algebraic-effect constructs. The slugs match
        // `flow_plan::ir_flow_node_kind` (drift-gated) and the wire `step_type`
        // the dispatcher emits, so one construct has one name everywhere.
        IRFlowNode::Handle(s) => (
            s.effect_names.join(","),
            "handle".to_string(),
            format!("Handle: {}", s.effect_names.join(", ")),
        ),
        IRFlowNode::Perform(s) => (
            s.operation_name.clone(),
            "perform".to_string(),
            format!("Perform: {}.{}", s.effect_name, s.operation_name),
        ),
        IRFlowNode::Resume(_) => ("resume".to_string(), "resume".to_string(), "Resume the continuation".to_string()),
        IRFlowNode::Abort(_) => ("abort".to_string(), "abort".to_string(), "Abort the handle".to_string()),
        IRFlowNode::Forward(s) => (
            s.operation_name.clone(),
            "forward".to_string(),
            format!("Forward: {}.{} to the outer handler", s.effect_name, s.operation_name),
        ),
        IRFlowNode::Navigate(s) => (s.pix_ref.clone(), "navigate".to_string(), format!("Navigate: {}", s.pix_ref)),
        IRFlowNode::Drill(s) => (s.pix_ref.clone(), "drill".to_string(), format!("Drill: {} → {}", s.pix_ref, s.subtree_path)),
        IRFlowNode::Trail(s) => (s.navigate_ref.clone(), "trail".to_string(), format!("Trail: {}", s.navigate_ref)),
        IRFlowNode::Corroborate(s) => (s.navigate_ref.clone(), "corroborate".to_string(), format!("Corroborate: {}", s.navigate_ref)),
        IRFlowNode::OtsApply(s) => (s.ots_name.clone(), "ots_apply".to_string(), format!("Apply OTS: {}", s.ots_name)),
        IRFlowNode::MandateApply(s) => (s.mandate_name.clone(), "mandate_apply".to_string(), format!("Apply mandate: {}", s.mandate_name)),
        IRFlowNode::ComputeApply(s) => (s.compute_name.clone(), "compute_apply".to_string(), format!("Apply compute: {}", s.compute_name)),
        IRFlowNode::Listen(s) => (s.channel.clone(), "listen".to_string(), format!("Listen: {}", s.channel)),
        IRFlowNode::DaemonStep(s) => (s.daemon_ref.clone(), "daemon".to_string(), format!("Daemon: {}", s.daemon_ref)),
        IRFlowNode::Persist(s) => (s.store_name.clone(), "persist".to_string(), format!("Persist to: {}", s.store_name)),
        IRFlowNode::Retrieve(s) => (s.store_name.clone(), "retrieve".to_string(), format!("Retrieve from: {}", s.store_name)),
        IRFlowNode::Mutate(s) => (s.store_name.clone(), "mutate".to_string(), format!("Mutate: {}", s.store_name)),
        IRFlowNode::Purge(s) => (s.store_name.clone(), "purge".to_string(), format!("Purge: {}", s.store_name)),
        IRFlowNode::Transact(_) => ("transact".to_string(), "transact".to_string(), "Transact block".to_string()),
        // v2.43.0 — `warden` adversarial-analysis block. step_type "warden"
        // matches `flow_plan::ir_flow_node_kind` (drift-gated).
        IRFlowNode::Warden(s) => (s.target.clone(), "warden".to_string(), format!("Warden: {}", s.target)),
        // v2.4.0 — `quant` cognitive block. step_type "quant" matches
        // `flow_plan::ir_flow_node_kind` (drift-gated).
        IRFlowNode::Quant(_) => ("quant".to_string(), "quant".to_string(), "Quant block".to_string()),
        // v2.4.0 — `yield` measurement point. step_type "yield" matches
        // `flow_plan::ir_flow_node_kind` (drift-gated).
        IRFlowNode::Yield(s) => (s.value_expr.clone(), "yield".to_string(), format!("Yield: {}", s.value_expr)),
        // v2.4.0 — `run <Flow>(args)` flow-step (invoke a flow from a body).
        IRFlowNode::Run(s) => (s.flow_name.clone(), "run".to_string(), format!("Run flow: {}", s.flow_name)),
        IRFlowNode::LambdaDataApply(s) => (s.lambda_data_name.clone(), "lambda_data_apply".to_string(), format!("Apply ΛD: {}", s.lambda_data_name)),
        // v1.6.0 — Mobile typed channel reductions.
        IRFlowNode::Emit(s) => (s.channel_ref.clone(), "emit".to_string(), format!("Emit on {}: {}", s.channel_ref, s.value_ref)),
        IRFlowNode::Publish(s) => (s.channel_ref.clone(), "publish".to_string(), format!("Publish {} within {}", s.channel_ref, s.shield_ref)),
        IRFlowNode::Discover(s) => (s.capability_ref.clone(), "discover".to_string(), format!("Discover {} as {}", s.capability_ref, s.alias)),
        // v1.14.0 — break / continue. Payload-free; the executor
        // raises sentinel exceptions caught by the enclosing for-in.
        IRFlowNode::Break(_) => ("break".to_string(), "break".to_string(), "Break out of for-in loop".to_string()),
        IRFlowNode::Continue(_) => ("continue".to_string(), "continue".to_string(), "Continue to next for-in iteration".to_string()),
    }
}

// ── v1.30.0 — axonstore SQL routing for the sync runner ──────────
//
// The sync runner is synchronous; `PostgresStoreBackend`'s operations
// are async. `block_on_store` bridges the two by running the future on
// a freshly-spawned OS thread that owns a current-thread Tokio runtime.
// A fresh thread never carries an ambient runtime, so this is safe
// whether `execute_real` runs on a server worker thread, a
// `spawn_blocking` thread, or a plain CLI thread — there is no
// "runtime within a runtime" hazard. `std::thread::scope` joins the
// thread before returning. One pool is created + used + dropped per
// store op; cross-request pooling is the streaming dispatcher's path
// (35.f, the production hot path).
// v2.89.0 — and a fresh thread never carries the ambient TENANT either.
//
// The paragraph above is right that a fresh thread has no ambient runtime, which
// is what makes this bridge safe against "runtime within a runtime". It is also
// the reason the tenant disappears: `CURRENT_TENANT_ID` is a `tokio::task_local!`,
// and a new OS thread starts with none. Measured:
//
// ```text
// same task = acme    spawn_blocking = default
// tokio::spawn = default    block_on_store = default
// ```
//
// That matters because `storage_postgres.rs` derives the RLS GUC from
// `current_tenant_id()` in **29 places** — `SET LOCAL axon.current_tenant` on
// every data method — and its module header presents that as the isolation of
// last resort: *"RLS isolation is enforced even if application code forgets to
// filter by tenant_id"*. Every store op the synchronous runner performs crosses
// THIS boundary, so that last resort would have resolved to `'default'`.
//
// (Those 29 are latent today — nothing in production calls `StorageBackend`
// through `ServerState.storage`. Fixed anyway, and fixed HERE: a defect that is
// unreachable by accident is still armed, and the accident that arms it is one
// call site away.)
//
// The fix is the boundary, not the readers. Capturing the tenant on the calling
// thread and re-binding it inside repairs every downstream ambient read at once
// the v2.87.0 discipline of fixing a class rather than adding the missing pass to
// each copy. Threading a tenant parameter through 29 storage methods would have
// produced 29 chances to forget.
fn block_on_store<F>(fut: F) -> F::Output
where
    F: std::future::Future + Send,
    F::Output: Send,
{
    let tenant = crate::tenant_context::current_tenant_id();
    std::thread::scope(|scope| {
        scope
            .spawn(|| {
                tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("failed to build the store-op Tokio runtime")
                    .block_on(crate::tenant_context::CURRENT_TENANT_ID.scope(tenant, fut))
            })
            .join()
            .expect("the store-op thread panicked")
    })
}

/// Execute one `persist`/`retrieve`/`mutate`/`purge` step against a
/// postgresql-backed `axonstore`, returning a human-readable result
/// summary or a typed [`StoreError`].
///
/// The store name doubles as the SQL table name (D12 — `IRAxonStore`
/// carries no schema, so v1.31.0 operates against existing tables).
/// v1.30.0 — `persist` (INSERT columns) and `mutate`
/// (UPDATE SET assignments) write the columns of their declared
/// `{ col: value }` block (`store_fields`, value expressions
/// interpolated); with no block they fall back to writing the flow's
/// user bindings as a row ([`ExecContext::user_bindings`]).
/// `retrieve`/`purge` are driven by the `where`-expression. D5 — the
/// SAME `PostgresStoreBackend` the streaming dispatcher uses, so the
/// two execution paths never diverge.
///
/// v1.32.0 (D3) — `memory_expr` is the RAW `store:where` expression
/// (NOT pre-interpolated). A `${name}` in the `where` clause is
/// resolved by the filter compiler against `ctx.vars()` into a `$N`
/// bind parameter — never string-spliced into the `where` source. The
/// pre-37.d path interpolated the whole expression first, which let a
/// request value carrying a `'` break a string-literal boundary.
/// v1.32.0 (POST-CLOSE HOTFIX 2026-05-21) — Async variant of
/// `execute_sql_store_step`. The pre-hotfix sync variant wrapped the
/// SQL dispatch in its OWN `block_on_store` (own temporary tokio
/// runtime per call). Combined with the eager pin acquisition (also
/// on its own temp runtime), this created a fatal cross-runtime
/// hazard: the pinned `PoolConnection<Postgres>` carries reactor
/// handles bound to the runtime that ACQUIRED it; awaiting on it from
/// a different runtime hangs indefinitely (the reactor that would
/// notify the I/O completion is already dropped).
///
/// 37.x.j.10 collapses the per-step runtime back into a SINGLE
/// outer-scope runtime owned by `execute_server_flow`. This async fn
/// runs on the caller's runtime, so the pin acquired at flow start
/// + every SQL dispatch + the implicit pin drop on flow exit ALL
/// live on the same runtime. Reactor handles stay valid.
///
/// The sync variant `execute_sql_store_step` is retained as a thin
/// wrapper for CLI tests + pre-async callers; it spins up a single
/// `block_on_store` and calls this async variant. New callers should
/// invoke the async variant directly from an async context.
#[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
async fn execute_sql_store_step_async(
    store_registry: &StoreRegistry,
    // v1.32.0 (D1) — pinned-connection map shared across the flow
    // execution. Keyed by axonstore name; when the entry exists the
    // store op routes its SQL through that exact physical Postgres
    // connection (held since `execute_server_flow` start). When the
    // entry is absent the op falls back to `StoreConn::Pool` (legacy
    // pre-37.x.j behavior) — keeping CLI tests + non-server callers
    // working unchanged.
    //
    // v1.32.0 — the `&mut` reference is held across `.await`
    // boundaries inside this fn. Safe because the function is the
    // unique &mut borrower of the map for its execution and the map
    // itself is owned by the outer scope (`execute_server_flow`'s
    // single block_on_store, or a test scope's single async wrapper).
    pinned_conns: &mut std::collections::HashMap<String, crate::pinned_conn::PinnedConn>,
    step_type: &str,
    store_name: &str,
    memory_expr: &str,
    store_fields: Option<&[(String, String)]>,
    // v2.21.0 — a `retrieve` step's `order_by:` / `limit:` clauses
    // (empty when absent). Rendered to the structural `ORDER BY … LIMIT
    // …` suffix by `stream_retrieve`. Ignored by persist/mutate/purge.
    order_by: &str,
    limit_expr: &str,
    // v2.33.0 — a `retrieve` step's `aggregate:` / `group_by:` clauses
    // (empty when absent). Closed catalog; rendered structurally by
    // `stream_retrieve`. Ignored by persist/mutate/purge.
    aggregate: &str,
    group_by: &str,
    ctx: &ExecContext,
) -> Result<String, StoreError> {
    // v2.81.0 — the REFUSAL. Without the `postgres` feature no driver is
    // linked, so a `backend: postgresql` axonstore cannot be read or written.
    // The signature is UNCHANGED across both profiles — the gate is one block,
    // not a second shape of this fifteen-parameter function (the design decision's rejected
    // option (ii)). Everything upstream — the flow, the dispatcher, the step
    // walk — compiles and runs identically; only the SQL itself is absent, and
    // it says so in writing.
    #[cfg(not(feature = "postgres"))]
    {
        return Err(crate::store::error::StoreError::Connect {
            source: format!(
                "axonstore `{store_name}`: this build was compiled without the `postgres`                  feature, so no PostgreSQL driver is linked and the `{step_type}` step cannot                  reach a database.
  Reinstall with: cargo install axon-lang --features postgres
                   (`axon check` type-checks the declaration in every build; an in-memory                  axonstore keeps working here.)"
            ),
        });
    }
    #[cfg(feature = "postgres")]
    {
        // The connection + confidence_floor live on the `IRAxonStore` the
        // registry validated.
        let spec = store_registry.spec(store_name);
        let _connection = spec.map(|s| s.connection.clone()).unwrap_or_default();
        let confidence_floor = spec.and_then(|s| s.confidence_floor);

        // v1.32.0 (D1) — the SHARED backend is resolved from the
        // registry cache INSIDE the `block_on_store` async block below
        // (the registry's `resolve` may need a tokio context when it
        // lazily builds the PgPool on first reference). Pre-37.x.j the
        // runner created a fresh `PgPool` per `connect_named` call — a
        // pre-existing inefficiency that 37.x.j fixes en passant by
        // routing through the cached pool.

        // `memory_expression` is `"store:where"` for retrieve/mutate/purge
        // and the bare store name for persist — the where-expr is whatever
        // follows the first colon (empty when absent).
        let where_expr = memory_expr
            .splitn(2, ':')
            .nth(1)
            .unwrap_or("")
            .to_string();
        // v1.32.0 (D3) — an OWNED copy of the flow's variable map, moved
        // into the store-op task; the filter compiler resolves `${name}`
        // in `where_expr` against it into `$N` bind parameters.
        let where_bindings: std::collections::HashMap<String, String> =
            ctx.vars().clone();

        // v1.30.0 — when the `persist` / `mutate` step declared a
        // `{ col: value }` block, the SQL row is exactly those columns with
        // their value expressions interpolated against the flow context.
        // With no block (`store_fields` is `None`) fall back to the v1.31.0
        // behaviour: every user binding as a text column. `store_fields` is
        // only materialised for `persist`/`mutate`, so `retrieve`/`purge`
        // (which ignore `data`) always take the fallback. Every value binds
        // as text (D12 — no column-type schema in v1.31).
        let data: Vec<(String, SqlValue)> = match store_fields {
            Some(fields) => fields
                .iter()
                .map(|(col, expr)| {
                    (col.clone(), SqlValue::Text(ctx.interpolate(expr)))
                })
                .collect(),
            None => ctx
                .user_bindings()
                .into_iter()
                .map(|(k, v)| (k, SqlValue::Text(v)))
                .collect(),
        };

        let store_name = store_name.to_string();
        let step_type = step_type.to_string();
        let store_name_for_reinsert = store_name.clone();

        // v1.32.0 (D1) — take the pin OUT of the shared map for the
        // duration of this dispatch. After the dispatch returns (success
        // OR error), the pin is re-inserted UNCONDITIONALLY so the next
        // store op against this same store reuses it.
        //
        // v1.32.0 — no longer wrapped in block_on_store. The
        // async fn runs on the caller's runtime, so the pin's reactor
        // handles stay valid for every `.await` below.
        let mut pin: Option<crate::pinned_conn::PinnedConn> =
            pinned_conns.remove(&store_name);

        // v1.32.0 (D1) — resolve the SHARED backend from the registry
        // cache. The registry caches `PostgresStoreBackend` by resolved
        // DSN; the backend's inner `PgPool` is `Arc<...>` so the clone
        // shares pool state with every other call AND with the eagerly-
        // acquired pin in `pinned_conns`.
        let backend = match store_registry.resolve(&store_name) {
            Ok(crate::store::registry::StoreHandle::Postgres(b)) => b,
            Ok(_) => {
                // Re-insert the pin if we removed one (we won't dispatch).
                if let Some(p) = pin {
                    pinned_conns.insert(store_name_for_reinsert, p);
                }
                return Err(StoreError::Connect {
                    source: format!(
                        "axonstore `{store_name}` expected to resolve to \
                         a postgresql backend but the registry returned \
                         `in_memory`. Routing bug — the SQL gate in \
                         `execute_real` should have skipped this step."
                    ),
                });
            }
            Err(e) => {
                if let Some(p) = pin {
                    pinned_conns.insert(store_name_for_reinsert, p);
                }
                return Err(e);
            }
        };

        // v1.32.0 — dispatch body inlined here. `pin` is `&mut`-
        // borrowed inside each match arm for the StoreConn::Pinned variant;
        // the borrow ends at the end of each arm so we can re-insert `pin`
        // unconditionally below regardless of result.
        let result: Result<String, StoreError> = async {
            match step_type.as_str() {
                "retrieve" => {
                    // v1.30.0 Pillar III — retrieve drains off a lazy cursor,
                    // bounded (never materializes a huge result set).
                    // v1.30.0 Pillar I — every tuple born Untrusted,
                    // confidence_floor filters sub-floor rows. The result
                    // is an epistemic envelope carrying both dispositions.
                    let cancel = crate::cancel_token::CancellationFlag::new();
                    // v1.32.0 (D1) — build `StoreConn::Pinned` when a
                    // pin is held for this store (the post-37.x.j default
                    // for server-driven flows), else `StoreConn::Pool`
                    // (legacy path for CLI / pre-server callers). The
                    // Pinned variant routes the SELECT through the exact
                    // physical Postgres backend connection acquired at
                    // flow start — Supavisor/PgBouncer cannot swap.
                    let mut store_conn = match &mut pin {
                        Some(p) => p.as_store_conn(),
                        None => crate::store::store_conn::StoreConn::Pool(backend.pool()),
                    };
                    let stream_outcome = row_stream::stream_retrieve(
                        &backend,
                        &mut store_conn,
                        &store_name,
                        &where_expr,
                        // v2.21.0 — the ORDER BY / LIMIT clauses.
                        order_by,
                        limit_expr,
                        // v2.33.0 — the aggregate / GROUP BY clauses.
                        aggregate,
                        group_by,
                        row_stream::DEFAULT_RETRIEVE_POLICY,
                        row_stream::DEFAULT_MAX_ROWS,
                        &cancel,
                        &where_bindings,
                    )
                    .await?;
                    let metadata = row_stream::stream_metadata(
                        row_stream::DEFAULT_RETRIEVE_POLICY,
                        &stream_outcome,
                    );
                    let outcome = epistemic::enforce_retrieve_floor(
                        epistemic::mark_retrieved(stream_outcome.rows),
                        confidence_floor,
                    );
                    let mut envelope =
                        epistemic::retrieve_envelope(&outcome, confidence_floor);
                    envelope["stream"] = metadata;
                    Ok(serde_json::to_string(&envelope)
                        .unwrap_or_else(|_| "{}".to_string()))
                }
                "purge" => {
                    // v1.32.0 (D1) — pinned/pool dispatch (see retrieve).
                    let mut store_conn = match &mut pin {
                        Some(p) => p.as_store_conn(),
                        None => crate::store::store_conn::StoreConn::Pool(backend.pool()),
                    };
                    let n = backend
                        .purge(&mut store_conn, &store_name, &where_expr, &where_bindings)
                        .await?;
                    Ok(format!("{n} row(s) purged"))
                }
                "persist" => {
                    // v1.30.0 Pillar I — a sub-floor or un-elevated write
                    // into a confidence-floored store is a typed error.
                    epistemic::enforce_persist_floor(
                        &data,
                        confidence_floor,
                        &store_name,
                    )?;
                    // v1.32.0 (D1) — pinned/pool dispatch.
                    let mut store_conn = match &mut pin {
                        Some(p) => p.as_store_conn(),
                        None => crate::store::store_conn::StoreConn::Pool(backend.pool()),
                    };
                    let n = backend.insert(&mut store_conn, &store_name, &data).await?;
                    Ok(format!("{n} row(s) persisted"))
                }
                "mutate" => {
                    // v1.32.0 (D1) — pinned/pool dispatch.
                    let mut store_conn = match &mut pin {
                        Some(p) => p.as_store_conn(),
                        None => crate::store::store_conn::StoreConn::Pool(backend.pool()),
                    };
                    let n = backend
                        .mutate(&mut store_conn, &store_name, &where_expr, &data, &where_bindings)
                        .await?;
                    Ok(format!("{n} row(s) mutated"))
                }
                // The caller only routes the four store-op step types here.
                other => Err(StoreError::Query {
                    op: "store",
                    source: format!("unsupported store step type `{other}`"),
                }),
            }
        }.await;

        // v1.32.0 (D1) — re-insert the pin (UNCONDITIONALLY — success
        // OR error path) so the next store op against this store reuses
        // the same physical Postgres backend connection. `pin` was taken
        // out at the top of this fn and the dispatch above only borrows
        // it `&mut`-wise inside each match arm — so it's still owned here
        // regardless of `result`'s Ok/Err outcome.
        if let Some(p) = pin {
            pinned_conns.insert(store_name_for_reinsert, p);
        }

        result
    }
}

/// v1.30.0 — Sync wrapper retained for CLI tests + pre-async callers.
///
/// v1.32.0 (POST-CLOSE HOTFIX) — wraps the new async fn
/// `execute_sql_store_step_async` in a SINGLE block_on_store so the
/// pin acquire (if any was pre-populated) + the SQL dispatch happen
/// on the SAME temporary tokio runtime. Pre-hotfix the sync variant
/// had its OWN block_on_store inside (per-step temp runtime); when
/// the caller's eager pin acquisition was ALSO on a separate temp
/// runtime, the cross-runtime hazard appeared. The wrapper here is
/// safe ONLY when the caller's pin map is empty (legacy Pool path)
/// — production callers MUST use the async variant directly inside
/// the OUTER block_on_store at `execute_server_flow`.
// The v1.32.0 source-gate (connection_pinning) pins this wrapper's
// pinned_conns threading; the legacy Pool path that called it is retired, so
// the fn is unreferenced in the lib build.
#[allow(dead_code)]
fn execute_sql_store_step(
    store_registry: &StoreRegistry,
    pinned_conns: &mut std::collections::HashMap<String, crate::pinned_conn::PinnedConn>,
    step_type: &str,
    store_name: &str,
    memory_expr: &str,
    store_fields: Option<&[(String, String)]>,
    ctx: &ExecContext,
) -> Result<String, StoreError> {
    block_on_store(execute_sql_store_step_async(
        store_registry,
        pinned_conns,
        step_type,
        store_name,
        memory_expr,
        store_fields,
        // v2.21.0 — the sync wrapper (CLI tests / pre-async callers)
        // does not carry retrieve bounds; default to none.
        "",
        "",
        // v2.33.0 — nor an aggregate.
        "",
        "",
        ctx,
    ))
}

/// v1.32.0 (POST-CLOSE HOTFIX 2026-05-21) — Async variant of
/// `execute_real`. Production callers MUST invoke this from inside
/// the OUTER `block_on_store` at `execute_server_flow` so the entire
/// flow execution (eager pin acquire + every store dispatch + implicit
/// pin drop on exit) lives on a SINGLE temporary tokio runtime. This
/// is the load-bearing structural property that prevents the cross-
/// runtime `PoolConnection<Postgres>` hazard the pre-hotfix code
/// exhibited.
///
/// The single store-op site (`execute_sql_store_step_async`) is now
/// awaited directly here — no nested `block_on_store`. Every other
/// operation in this fn is synchronous-style code; the async fn just
/// means the await of the SQL dispatch site is legal.
///
/// The sync variant `execute_real` retained as a thin wrapper for the
/// CLI path + pre-async callers that don't have a tokio context.
/// v2.15.0 — the dispatcher-shared state a STRUCTURAL `navigate` needs: the
/// axonstore registry (to read the corpus rows tenant-scoped) + the static MDN
/// corpus graphs (v2.13.0 `corpus { relations: }`) + the dynamic store-sourced
/// corpus specs (v2.14.0 `corpus … from axonstore`) + the adaptive set. Built once
/// per server flow from the IR — mirroring `run_streaming_via_dispatcher` — so a
/// NON-streaming `navigate` executes the SAME real MDN traversal as the SSE path
/// instead of the LLM fallthrough. `None` on the CLI path (its executor unifies
/// in a later step; navigate there keeps the legacy behavior for now).
struct NavDispatch {
    store_registry: std::sync::Arc<StoreRegistry>,
    corpora: std::sync::Arc<std::collections::HashMap<String, crate::mdn::Corpus>>,
    store_sources:
        std::sync::Arc<std::collections::HashMap<String, crate::ir_nodes::IRCorpusStoreSource>>,
    adaptive: std::sync::Arc<std::collections::HashSet<String>>,
    /// v2.63.0 — the columnar engine port, threaded from the caller's
    /// deployment (`None` ⇒ the data-plane verbs fail CLOSED in dispatch).
    dataspace_engine: Option<crate::dataspace_engine::SharedDataspaceEngine>,
    /// v2.67.0 — the compiled `scope` declarations, so a `warden(<t>) within
    /// <Scope>` can resolve its authorization envelope at dispatch. Empty ⇒
    /// every warden fails CLOSED (a scope that cannot be resolved authorises
    /// nothing — the v2.43.0 posture, now enforced in fact and not only in grammar).
    scopes: std::sync::Arc<Vec<crate::ir_nodes::IRScope>>,
    /// v2.67.0 — the compiled `observable` declarations (Pauli sums), so a
    /// `quant` block can resolve the `M` it measures. Empty ⇒ `quant` fails
    /// CLOSED: E = ⟨ψ|M|ψ⟩ with no M is not a weak result, it is a category
    /// error.
    observables: std::sync::Arc<Vec<crate::ir_nodes::IRObservable>>,
    /// v2.67.0 — the compiled `compute` declarations (pure v2.26.0 functions).
    compute_specs: std::sync::Arc<Vec<crate::ir_nodes::IRCompute>>,
    /// v2.83.0 — the compiled `mandate` declarations, so the enforcement
    /// loop can resolve the cage it is asked to close.
    mandate_specs: std::sync::Arc<Vec<crate::ir_nodes::IRMandate>>,
    /// v2.83.0 — the compiled ΛD + ots declarations (same doctrine).
    lambda_data_specs: std::sync::Arc<Vec<crate::ir_nodes::IRLambdaData>>,
    ots_specs: std::sync::Arc<Vec<crate::ir_nodes::IROts>>,
    /// v2.83.0 — the compiled `agent` declarations. Empty ⇒ an
    /// `<Agent>(args)` call fails CLOSED, because the declaration is where
    /// `max_iterations` lives and an unresolved agent is an unbounded one.
    agent_specs: std::sync::Arc<Vec<crate::ir_nodes::IRAgent>>,
    /// v2.89.0 — what this program memoises, resolved once from the IR.
    /// Empty (no `cache` declaration) ⇒ no runtime is attached and every call
    /// computes, byte-identical to pre-v2.89.0.
    cache_plan: std::sync::Arc<crate::cache_runtime::CachePlan>,
}

/// Build the dispatcher's catalogues from a compiled program: the MDN
/// corpora (declared + store-sourced + adaptive), the scopes / observables /
/// compute / agent / mandate / lambda / ots catalogues and the cache plan.
/// ONE builder for every door: the non-streaming server executor used to build
/// this inline while `axon run --tool-mode real` passed `None` — so on the CLI
/// every bridged verb (navigate, mint, grad, an agent call …) silently fell to
/// the LLM fallthrough. Same program, same catalogues, both doors.
fn build_nav_dispatch(
    ir: &crate::ir_nodes::IRProgram,
    store_registry: std::sync::Arc<StoreRegistry>,
    dataspace_engine: Option<crate::dataspace_engine::SharedDataspaceEngine>,
) -> NavDispatch {
    let mut corpora: std::collections::HashMap<String, crate::mdn::Corpus> =
        std::collections::HashMap::new();
    let mut store_sources: std::collections::HashMap<
        String,
        crate::ir_nodes::IRCorpusStoreSource,
    > = std::collections::HashMap::new();
    let mut adaptive: std::collections::HashSet<String> = std::collections::HashSet::new();
    for cspec in &ir.corpus_specs {
        if !cspec.relations.is_empty() {
            let rels: Vec<(String, String, String, f64)> = cspec
                .relations
                .iter()
                .map(|r| (r.etype.clone(), r.from.clone(), r.to.clone(), r.weight))
                .collect();
            if let Ok(corpus) = crate::mdn::Corpus::from_declaration(&cspec.documents, &rels) {
                corpora.insert(cspec.name.clone(), corpus);
            }
        }
        if let Some(src) = &cspec.store_source {
            store_sources.insert(cspec.name.clone(), src.clone());
        }
        if cspec.adaptive && (!cspec.relations.is_empty() || cspec.store_source.is_some()) {
            adaptive.insert(cspec.name.clone());
        }
    }
    NavDispatch {
        store_registry,
        corpora: std::sync::Arc::new(corpora),
        store_sources: std::sync::Arc::new(store_sources),
        adaptive: std::sync::Arc::new(adaptive),
        dataspace_engine,
        scopes: std::sync::Arc::new(ir.scopes.clone()),
        observables: std::sync::Arc::new(ir.observables.clone()),
        compute_specs: std::sync::Arc::new(ir.compute_specs.clone()),
        agent_specs: std::sync::Arc::new(ir.agents.clone()),
        cache_plan: std::sync::Arc::new(crate::cache_runtime::CachePlan::from_ir(ir)),
        mandate_specs: std::sync::Arc::new(ir.mandate_specs.clone()),
        lambda_data_specs: std::sync::Arc::new(ir.lambda_data_specs.clone()),
        ots_specs: std::sync::Arc::new(ir.ots_specs.clone()),
    }
}

fn truncate(s: &str, max: usize) -> String {
    let first_line = s.lines().next().unwrap_or(s);
    if first_line.len() > max {
        format!("{}…", &first_line[..max])
    } else {
        first_line.to_string()
    }
}

/// Build a plan export from compiled execution units.
fn build_plan_export(
    units: &[ExecutionUnit],
    source_file: &str,
    backend: &str,
    registry: &ToolRegistry,
) -> plan_export::PlanExport {
    let mut plan_units = Vec::new();
    let mut all_deps = PlanDependencies {
        max_depth: 0,
        parallel_groups: Vec::new(),
        unresolved_refs: Vec::new(),
    };

    for unit in units {
        // Build step infos for dependency analysis
        // v2.11.0 — fold `use Tool(k = v)` keyword-arg references into the
        // analysis argument so the plan reflects the real dependency edges.
        let step_name_set: std::collections::HashSet<&str> =
            unit.steps.iter().map(|s| s.step_name.as_str()).collect();
        let step_infos: Vec<step_deps::StepInfo> = unit.steps.iter().map(|s| {
            step_deps::StepInfo {
                name: s.step_name.clone(),
                step_type: s.step_type.clone(),
                user_prompt: s.user_prompt.clone(),
                argument: step_deps::use_tool_analysis_argument(
                    s.tool_argument.as_deref()
                        .or(s.memory_expression.as_deref())
                        .unwrap_or(""),
                    &s.tool_named_args,
                    &step_name_set,
                ),
            }
        }).collect();

        let dep_graph = step_deps::analyze(&step_infos);

        // Build plan steps with dependency info
        let plan_steps: Vec<PlanStep> = unit.steps.iter().zip(dep_graph.steps.iter()).map(|(s, d)| {
            PlanStep {
                name: s.step_name.clone(),
                step_type: s.step_type.clone(),
                prompt_preview: truncate(&s.user_prompt, 200),
                tool_argument: s.tool_argument.clone(),
                memory_expression: s.memory_expression.clone(),
                depends_on: d.depends_on.clone(),
                is_root: d.is_root,
            }
        }).collect();

        plan_units.push(PlanUnit {
            flow_name: unit.flow_name.clone(),
            persona_name: unit.persona_name.clone(),
            context_name: unit.context_name.clone(),
            effort: unit.effort.clone(),
            anchor_count: unit.resolved_anchors.len(),
            anchors: unit.anchor_instructions.clone(),
            steps: plan_steps,
        });

        // Merge dependency info
        if dep_graph.max_depth > all_deps.max_depth {
            all_deps.max_depth = dep_graph.max_depth;
        }
        all_deps.parallel_groups.extend(dep_graph.parallel_groups);
        all_deps.unresolved_refs.extend(
            dep_graph.unresolved_refs.into_iter().map(|(step, var)| {
                UnresolvedRef { step, variable: var }
            }),
        );
    }

    // Build tool info
    let tools = PlanTools {
        total: registry.len(),
        builtin: registry.builtin_names().into_iter().map(|s| s.to_string()).collect(),
        program: registry.program_names().into_iter().map(|s| s.to_string()).collect(),
        registered: registry.tool_names().into_iter().map(|name| {
            let entry = registry.get(name).unwrap();
            PlanToolEntry {
                name: entry.name.clone(),
                provider: entry.provider.clone(),
                source: format!("{:?}", entry.source).to_lowercase(),
                output_schema: entry.output_schema.clone(),
                effect_row: entry.effect_row.clone(),
            }
        }).collect(),
    };

    PlanBuilder::build(source_file, backend, &plan_units, tools, all_deps)
}

// ── Server execution entry point ─────────────────────────────────────────────

pub struct ServerRunnerMetrics {
    pub success: bool,
    pub steps_executed: usize,
    pub tokens_input: u64,
    pub tokens_output: u64,
    pub anchor_breaches: usize,
    pub step_names: Vec<String>,
    pub step_results: Vec<String>,
    /// Per-step token chunks for streaming (simulated from step results).
    pub per_step_chunks: Vec<Vec<String>>,
    /// v2.0.0 — semantic provenance events captured during
    /// flow execution. Each entry is a `kind:identifier` slug
    /// (closed taxonomy enforced by producer sites):
    ///   - `retrieve:<store>`         — Pillar II store read
    ///   - `persist:<store>`          — Pillar II store insert
    ///   - `mutate:<store>`           — Pillar II store update
    ///   - `purge:<store>`            — Pillar II store delete
    ///   - `shield:<name>@<step>`     — Pillar I shield invocation
    ///   - `ots:<name>@<step>`        — OTS apply
    ///   - `mandate:<name>@<step>`    — mandate apply
    ///   - `compute:<name>@<step>`    — compute apply
    ///   - `lambda_apply:<name>@<step>` — lambda data apply
    /// The wire envelope's `provenance_chain` is built from
    /// `[flow:F, …events…, step:S0, step:S1, …, backend:B]`.
    /// Empty for trivial flows; populated by [`emit_provenance_event`]
    /// at the runtime sites.
    pub provenance_events: Vec<String>,
    /// v2.0.0 — closed-catalog blame attribution from runtime
    /// degradation events. Populated when:
    ///   - an anchor with severity != "error" fires (degraded path
    ///     proceeds)
    ///   - a shield flags content but flow proceeds
    ///   - a store mutation chain verification fails AND flow
    ///     proceeds with prior-state read
    ///   - a backend returns truncated / partial response
    ///   - D5 detects a recoverable type mismatch
    /// `None` on the clean happy path. The first surfaced blame
    /// wins (subsequent events are recorded in audit_log but do
    /// not overwrite the primary attribution).
    pub blame_attribution: Option<crate::wire_envelope::BlameContext>,
    /// v2.7.0 — the Theorem 5.1 `(base, scope, confidence)` triple of
    /// every flow-level `use <Tool>` dispatch whose tool declares an
    /// `epistemic:<level>` effect. Derived from the IR via
    /// [`crate::epistemic_capture::collect_for_flow`] — the same function
    /// the streaming path calls, so both transports surface byte-identical
    /// envelopes (v2.7.0 parity). Empty for flows with no epistemic tool.
    pub epistemic_envelopes: Vec<crate::epistemic_capture::EpistemicEnvelope>,
    /// v2.15.0 — the HONEST hard-failure detail when a node's
    /// `DispatchError` aborted the flow (a failing `persist`/`mutate`/`purge`
    /// store write, a backend error, etc.). `Some(detail)` names the FAILING
    /// NODE + the underlying cause — byte-parity with the streaming
    /// dispatcher's `FlowError.error` (v1.32.0/D6 honest-failure doctrine).
    /// `None` on the clean path. The v2.15.0 cutover regressed this: the
    /// non-streaming driver swallowed the error (`success:false`, empty result,
    /// no log, no wire detail), so a pre-insert store failure presented as a
    /// silent 0-SQL abort. This field restores parity — a `BlameContext` is
    /// the WRONG surface (it is soft-degradation-on-success only, per
    /// `wire_envelope::BlameContext` docs); a hard fail needs its own slot.
    pub error: Option<String>,
    /// v2.21.0 — observable per-run store row counts (closing brief
    /// #34 Q3: a daemon run `completed, duration 0` is no longer
    /// indistinguishable from "found no work"). Aggregated over every
    /// store op in the run (par-branch counts merged). The enterprise
    /// daemon run report / `daemon_runs` ledger / status API + audit
    /// surface these per the v2.4.0 / v2.4.0 contract.
    pub rows_retrieved: u64,
    pub rows_persisted: u64,
    pub rows_mutated: u64,
    pub rows_purged: u64,
    /// v2.46.0 — the run's temporal record when any step rendered a
    /// declared `now:` (`captured_utc` + `tzdb_version` + zones, the
    /// replayability triple of `time_is_an_explicit_input`). `None` — and
    /// elided from the wire envelope — for every `now:`-less flow. NOTE:
    /// the `AXON_LEGACY_EXECUTOR` kill-switch path does not inject temporal
    /// context (the unified dispatcher is the production engine, v2.15.0);
    /// it reports `None` here.
    pub temporal_context: Option<crate::temporal_context::TemporalRecord>,
}

/// v2.7.0 — derive a flow's epistemic envelopes from the IR. This is
/// the SINGLE site both transports funnel through — the synchronous runner
/// calls it directly with its in-hand `ir`; the streaming
/// `axon_server::resolve_epistemic_envelopes_for_flow` re-derives the IR
/// from source and calls THIS function — so the sync `FlowEnvelope` and the
/// streaming `axon.complete` carry byte-identical `(base, scope, confidence)`
/// triples by construction (the v2.7.0 parity invariant: there is exactly
/// one derivation, never two that could drift). `input_confidence = 1.0`:
/// a top-level flow's ψ is clean before any tool degrades it.
pub fn derive_epistemic_envelopes_for_flow(
    ir: &crate::ir_nodes::IRProgram,
    flow_name: &str,
) -> Vec<crate::epistemic_capture::EpistemicEnvelope> {
    ir.flows
        .iter()
        .find(|f| f.name == flow_name)
        .map(|f| crate::epistemic_capture::collect_for_flow(f, &ir.tools, 1.0))
        .unwrap_or_default()
}

/// v2.15.0 — the collected result of running a flow through the dispatcher.
struct CollectedRun {
    success: bool,
    /// v4.6.0 — whether each step's `StepComplete` reported success, in the
    /// same order as `step_names`.
    ///
    /// The events always carried this; nothing read it, because the legacy
    /// executor printed the CLI's per-step ✓/✗ from its own bookkeeping. With
    /// that executor gone the CLI prints from these events, and printing a
    /// green check over a step whose result text is an error is the exact shape
    /// this release exists to remove.
    step_success: Vec<bool>,
    steps_executed: usize,
    tokens_output: u64,
    step_names: Vec<String>,
    step_results: Vec<String>,
    // The dispatcher records per-step anchor breaches in
    // `ctx.step_audit_records` (`StepAuditRecord.anchor_breaches: Vec<String>`);
    // the collector sums them and derives the `blame_attribution` from them.
    // Anchors are evaluated on EVERY backend including stub.
    anchor_breaches: usize,
    blame_attribution: Option<crate::wire_envelope::BlameContext>,
    /// v2.15.0 — `Some(detail)` when a node's `DispatchError` aborted the
    /// walk: the failing NODE named + the cause, mirroring the streaming
    /// dispatcher's `FlowError.error`. Pre-v2.15.0 this error was swallowed
    /// (the `Err(_) => { success = false; break; }` arm), so a failed store
    /// write presented as a silent abort with no diagnostic.
    flow_error: Option<String>,
    /// v2.21.0 — per-run store row counts, read from the dispatcher
    /// ctx's shared (par-branch-merged) counter after the walk.
    store_row_counts: crate::flow_dispatcher::StoreRowCounts,
    /// v2.46.0 — the run's temporal record (`now:` capture + zones),
    /// read from the dispatcher ctx's shared state after the walk. `None`
    /// when the flow declares no `now:` (zero wire drift).
    temporal_context: Option<crate::temporal_context::TemporalRecord>,
}

/// v2.15.0 — run a flow through the DISPATCHER with a BUFFER sink and collect
/// the result (the non-streaming half of the unified driver). This is the SAME
/// engine the SSE path uses (`dispatch_node` over the flow's IR nodes), so the
/// step results are equivalent — gated by the 50-flow parity corpus. The buffer
/// channel queues every wire event; after the walk we project them into
/// `step_names` + `step_results` (the v2.15.0 projection captures structural-verb
/// outputs from `StepComplete.full_output`, not just per-token `StepToken`).
///
/// v2.15.0 — anchor breaches + blame attribution are projected from the
/// dispatcher's per-step audit records (`ctx.step_audit_records`). Provenance
/// events are IR-derived by the caller (pure walk) and epistemic envelopes
/// likewise, so both are execution-independent. Eager connection-pinning is
/// handled by the caller (the v1.32.0 discipline).
///
/// v4.6.0 — the "at parity with the legacy executor" these lines used to claim
/// has no referent: that executor is retired. What the fields carry is stated
/// above on its own terms.
async fn collect_via_dispatcher(
    flow: &crate::ir_nodes::IRFlow,
    backend: &str,
    // v2.49.0 — the VERIFIED tenant this flow runs under. Set on the
    // DispatchCtx so every tenant-scoped runtime seam that takes an EXPLICIT
    // tenant — the `axon::secret_custody` port (`rotate` enumeration/reveal,
    // `retrieve` over a `backend: secrets` store, `tool { secret: }` /
    // `secret_partition:` dispatch injection), the `mint` minter, session
    // state — receives the caller's scoped tenant, never ambient state (the
    // v2.47.0 explicit-tenant posture). Empty ⇒ those custody surfaces fail
    // CLOSED (`require_tenant` refuses an unscoped call), which is correct
    // for a CLI/test with no tenant scope. The endpoint passes the
    // request-verified `route.tenant_id`; the daemon supervisor passes the
    // daemon's scoped tenant.
    tenant_id: &str,
    system_prompt: &str,
    // v2.46.0 — the frame-level declared cognitive timezone (the program's
    // first `context` declaration's `now:` — the same first-context convention
    // the system-prompt composer uses). `None` ⇒ only step-level `now:` injects.
    default_now_tz: Option<String>,
    api_key: Option<&str>,
    // v1.18.0 (Kivi brief #37) — per-tenant LLM endpoint override.
    llm_base_url: Option<&str>,
    llm_chat_path: Option<&str>,
    anchors: std::sync::Arc<Vec<crate::ir_nodes::IRAnchor>>,
    nav_dispatch: &NavDispatch,
    registry: std::sync::Arc<ToolRegistry>,
    param_bindings: &[(String, String)],
    // v2.15.0 — the eager-acquired flow-scoped pins (one per postgresql
    // axonstore), shared with the dispatcher's store handlers for pooler
    // coherence (v1.32.0) — the same discipline the streaming path applies.
    pinned: std::sync::Arc<
        std::sync::Mutex<
            std::collections::HashMap<String, crate::pinned_conn::PinnedConn>,
        >,
    >,
    // v2.28.0 — the active `budget { … }` gate when a budgeted daemon runs
    // this flow. `None` ⇒ unbudgeted (tool dispatch unconditional, pre-v2.28.0).
    budget: Option<std::sync::Arc<std::sync::Mutex<crate::runtime::budget_kernel::BudgetGate>>>,
    // v2.69.0 — the per-channel concurrency semaphores (`resource.capacity`),
    // held on ServerState across requests. `None` ⇒ no channel is bounded.
    channel_semaphores: Option<std::sync::Arc<crate::channel_semaphore::ChannelSemaphores>>,
    // v2.69.0 — the tool-lease guard.
    tool_leases: Option<std::sync::Arc<crate::resource_lease::ResourceLeaseGuard>>,
    // v2.31.0 — the typed event BUS (built from the program's `channel`
    // definitions) + the durable event OUTBOX (the v2.31.0 `EventOutbox` seam),
    // attached as a pair on the daemon path. The bus carries each channel's
    // `persistence` metadata; `run_emit` consults it to route a
    // `persistent_axonstore` emit to `outbox.append` (durable — survives the
    // consumer being down + a process restart) vs an ephemeral emit to the bus.
    // `Some` only when the enterprise supervisor injects its per-tenant Pg outbox;
    // `None` for every other caller (HTTP, CLI, tests) ⇒ `emit` keeps its prior
    // in-process buffer behavior (byte-identical to pre-v2.31.0).
    event_bus: Option<std::sync::Arc<crate::runtime::channels::TypedEventBus>>,
    event_outbox: Option<std::sync::Arc<dyn crate::event_outbox::EventOutbox>>,
    // v2.46.0 — the compiled `credential` contracts (from `ir.credentials`)
    // + the minter port. `None` minter ⇒ a reached `mint` fails CLOSED with
    // `MissingDependency` (no silent stub); the enterprise executor injects
    // its PASETO minter (v2.46.0) the same way it injects the event outbox.
    credentials: std::sync::Arc<
        std::collections::HashMap<String, crate::ir_nodes::IRCredential>,
    >,
    credential_minter: Option<std::sync::Arc<dyn crate::credential_minter::CredentialMinter>>,
    // v2.48.0 — the secret-custody port behind `backend: secrets` /
    // `rotate` / `tool { secret: }`. `None` ⇒ all three fail CLOSED
    // (`MissingDependency`); the enterprise executor injects its
    // envelope-encrypted Pg custody (v2.48.0) — the v2.46.0 injection shape.
    secret_custody: Option<std::sync::Arc<dyn crate::secret_custody::SecretCustody>>,
) -> CollectedRun {
    use crate::flow_dispatcher::{dispatch_node, DispatchCtx, NodeOutcome};
    use crate::flow_execution_event::FlowExecutionEvent;

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
    let mut ctx = DispatchCtx::new(
        flow.name.clone(),
        backend.to_string(),
        system_prompt.to_string(),
        crate::cancel_token::CancellationFlag::new(),
        tx,
    )
    // v2.49.0 — thread the verified tenant onto the ctx (was defaulted to
    // empty, silently unscoping every explicit-tenant custody call).
    .with_tenant_id(tenant_id)
    .with_store_registry(nav_dispatch.store_registry.clone())
    .with_mdn_corpora(nav_dispatch.corpora.clone())
    .with_mdn_adaptive(nav_dispatch.adaptive.clone())
    .with_mdn_store_sources(nav_dispatch.store_sources.clone())
    .with_api_key(api_key.map(|s| s.to_string()))
    .with_llm_endpoint(
        llm_base_url.map(|s| s.to_string()),
        llm_chat_path.map(|s| s.to_string()),
    )
    .with_anchors(anchors)
    .with_tool_registry(registry)
    .with_pinned_conns(pinned);
    // v2.63.0 — attach the deployment's columnar engine so the five
    // data-plane verbs execute against the DECLARED stores; absent, each
    // fails CLOSED in its handler (the v2.63.0 honesty floor).
    if let Some(engine) = &nav_dispatch.dataspace_engine {
        ctx = ctx.with_dataspace_engine(engine.clone());
    }
    // v2.67.0 — mount the adversarial-analysis engine + the deployment's
    // scope catalog, so a `warden` block RUNS instead of silently completing.
    // OSS mounts the deterministic `ReferenceStaticWarden` (depth
    // `static_artifact`; deeper depths fail closed with `DepthNotSupported`);
    // enterprise mounts its abduction engine behind the same trait. With no
    // scope catalog the block still fails CLOSED — the `within <Scope>` clause
    // is mandatory in fact, not just in grammar.
    ctx = ctx.with_warden(
        std::sync::Arc::new(crate::warden::ReferenceStaticWarden),
        nav_dispatch.scopes.clone(),
    );
    // v2.89.0 — mount the memoisation tier, so a declared `cache` memoises.
    //
    // Attached only when the program declares one: with no `cache` block there
    // is nothing to memoise, and leaving the port `None` keeps this path
    // byte-identical to pre-v2.89.0 rather than adding a lookup that can only
    // miss.
    //
    // `process_local` shares the BACKEND across flow runs (entries surviving
    // between runs is what makes it a cache) and takes the TENANT from this run
    // (the design decision — the tenant is a key component, so isolation holds even in a
    // single process). The enterprise v2.40.0 Redis tier replaces the backend
    // here; it is not a second call site.
    //
    // The SSE path in `streaming_via_dispatcher` mounts the same tier the same
    // way. Wiring one and not the other is the "real-on-one-path,
    // dead-on-the-other" defect v2.67.0 exists to end, and the v2.87.0 both-doors
    // rule after it; `both_dispatch_paths_mount_the_cache` pins them together.
    if !nav_dispatch.cache_plan.is_empty() {
        let cache_tenant = ctx.tenant_id.clone();
        ctx = ctx.with_cache(
            std::sync::Arc::new(crate::cache_runtime::CacheRuntime::process_local(
                cache_tenant,
            )),
            &nav_dispatch.cache_plan,
        );
    }
    // v2.67.0 — mount the Hilbert-space simulator + the observable catalog,
    // so a `quant` block MEASURES instead of silently skipping its body. OSS
    // mounts the capped dense-statevector reference simulator (a register above
    // the cap fails closed with axon-E0783, never a silent truncation);
    // enterprise mounts its Q32Simulator behind the same trait.
    ctx = ctx.with_quant(
        std::sync::Arc::new(crate::quant::ReferenceSimulator::new()),
        nav_dispatch.observables.clone(),
    );
    // v2.67.0 — attach the compute catalog so `compute … on …` evaluates its
    // declared v2.26.0 expression NATIVELY instead of binding "compute:Name(args)".
    ctx = ctx.with_computes(nav_dispatch.compute_specs.clone());
    // v2.83.0 — attach the mandate catalog so the cage can close.
    ctx = ctx.with_mandates(nav_dispatch.mandate_specs.clone());
    // v2.83.0 — attach the ΛD + ots catalogs.
    ctx = ctx.with_lambdas(nav_dispatch.lambda_data_specs.clone());
    ctx = ctx.with_ots(nav_dispatch.ots_specs.clone());
    // The agent catalog. Measured absent on THIS engine (the non-streaming
    // server path, the one the enterprise executor calls): the SSE twin
    // attached it, the bridge attached it, and the default engine did not —
    // so every `<Agent>(…)` on a deployed endpoint failed closed with "0
    // agent(s) in this catalog". Caught by the fixture-driven gate in
    // tests/agent_loop_closes_its_promises.rs, which drives the README's own
    // agent-in-a-step shape through this exact function.
    ctx = ctx.with_agents(nav_dispatch.agent_specs.clone());
    // v2.28.0 — attach the linear-effect budget gate (daemon path only).
    if let Some(gate) = budget {
        ctx = ctx.with_budget(gate);
    }
    // v2.69.0 — attach the cross-request channel semaphores so `capacity:`
    // bounds simultaneous in-flight calls (v2.69.0 derived the number; this makes
    // it a bound).
    if let Some(sems) = channel_semaphores {
        ctx = ctx.with_channel_semaphores(sems);
    }
    // v2.69.0 — attach the tool-lease guard so a post-expiry vendor call breaches.
    if let Some(leases) = tool_leases {
        ctx = ctx.with_tool_leases(leases);
    }
    // v2.31.0 — attach the typed event bus + durable outbox as a pair (daemon
    // path only). The bus supplies the channel's `persistence` so `run_emit`
    // routes a `persistent_axonstore` emit to `outbox.append` (durable); the
    // enterprise supervisor then drains + delivers it across replicas / restarts
    // (the v2.31.0 Pg outbox). Without the bus, the durability metadata is unknown
    // and the emit would fall back to the legacy in-process buffer.
    if let Some(bus) = event_bus {
        ctx = ctx.with_event_bus(bus);
    }
    if let Some(outbox) = event_outbox {
        ctx = ctx.with_event_outbox(outbox);
    }
    // v1.32.0 — seed the request-bound flow params so `${param}` resolves.
    for (k, v) in param_bindings {
        ctx.let_bindings.insert(k.clone(), v.clone());
    }
    // v2.46.0 — the frame-level declared zone; a step's own `now:` overrides.
    ctx.default_now_tz = default_now_tz;
    // v2.46.0 — the credential contracts + (optionally) the minter port.
    ctx.credentials = credentials;
    ctx.credential_minter = credential_minter;
    // v2.48.0 — the secret-custody port (None ⇒ fail-closed).
    ctx.secret_custody = secret_custody;

    // v2.15.0 — share the audit-record sink so we can read the per-step
    // anchor breaches AFTER the walk (the `drop(ctx)` below releases the ctx's
    // own handle; this Arc clone keeps the records alive for the projection).
    let audit_records = ctx.step_audit_records.clone();
    // v2.46.0 — share the temporal state (capture + rendered zones) so the
    // envelope record is readable after the walk (the same Arc discipline).
    let temporal = ctx.temporal.clone();
    // v2.21.0 — clone the shared row-count Arc up front (like
    // `audit_records`); the store handlers increment the SAME Mutex
    // during the walk, so reading it after the walk (via this clone)
    // yields the final per-run totals even once `ctx` has been consumed.
    let row_counts = ctx.store_row_counts.clone();

    let mut success = true;
    let mut tokens_output: u64 = 0;
    let mut flow_return: Option<String> = None;
    // v2.15.0 — the HONEST hard-failure detail. Pre-v2.15.0 a node's
    // `DispatchError` hit `Err(_) => { success = false; break; }` — swallowed
    // with no log + no wire surface, so a failing `persist` (most often a
    // pre-insert gate: v1.30.0 confidence-floor, registry resolve, or a
    // connection error — all BEFORE any SQL reaches the DB) presented to the
    // adopter as a silent abort: `success:false`, empty step result, zero
    // diagnostic. The streaming dispatcher never had this gap — it emits a
    // `FlowError` naming the failing node + the cause (v1.32.0/D6). This restores
    // that parity on the non-streaming path.
    let mut flow_error: Option<String> = None;
    for node in &flow.steps {
        match dispatch_node(node, &mut ctx).await {
            Ok(NodeOutcome::Completed { tokens_emitted, .. }) => tokens_output += tokens_emitted,
            // `run_return` is a flow-terminating SENTINEL that emits no wire
            // step — capture its value as the flow's output (below).
            Ok(NodeOutcome::Return { value }) => {
                flow_return = Some(value);
                break;
            }
            Ok(_) => {} // stray Break/LoopContinue → no-op
            Err(crate::flow_dispatcher::DispatchError::UpstreamCancelled) => break,
            Err(e) => {
                success = false;
                // v2.15.0 — name the FAILING NODE (the four store ops + a
                // `step` carry a meaningful name; any other variant is named by
                // its flow position) so the diagnostic pinpoints WHERE + WHY,
                // byte-for-byte with `streaming_via_dispatcher`'s v1.32.0/D6
                // `node_label`. The detail reaches BOTH the structured server
                // log (parity with the streaming `tracing::error!`) AND the
                // wire envelope (the new `ServerRunnerMetrics.error` slot).
                use crate::ir_nodes::IRFlowNode;
                let node_label = match node {
                    IRFlowNode::Step(s) if !s.name.is_empty() => {
                        format!("step '{}'", s.name)
                    }
                    IRFlowNode::Retrieve(r) => format!("retrieve from '{}'", r.store_name),
                    IRFlowNode::Persist(p) => format!("persist into '{}'", p.store_name),
                    IRFlowNode::Mutate(m) => format!("mutate '{}'", m.store_name),
                    IRFlowNode::Purge(p) => format!("purge '{}'", p.store_name),
                    _ => "node".to_string(),
                };
                let detail = format!("flow '{}' failed at {node_label}: {e:?}", flow.name);
                tracing::error!(
                    flow = %flow.name,
                    node = %node_label,
                    detail = %detail,
                    "axon non-streaming flow failed — node dispatch error"
                );
                flow_error = Some(detail);
                break;
            }
        }
    }
    drop(ctx); // close `tx` so the drain below terminates

    let mut step_names: Vec<String> = Vec::new();
    let mut step_results: Vec<String> = Vec::new();
    let mut step_success: Vec<bool> = Vec::new();
    let mut cur: Option<usize> = None;
    while let Ok(ev) = rx.try_recv() {
        match ev {
            FlowExecutionEvent::StepStart { step_name, .. } => {
                step_names.push(step_name);
                step_results.push(String::new());
                // A step that starts and never completes did not succeed.
                step_success.push(false);
                cur = Some(step_results.len() - 1);
            }
            FlowExecutionEvent::StepToken { content, .. } => {
                if let Some(i) = cur {
                    if let Some(a) = step_results.get_mut(i) {
                        a.push_str(&content);
                    }
                }
            }
            FlowExecutionEvent::StepComplete {
                full_output,
                success,
                ..
            } => {
                if let Some(i) = cur {
                    if let Some(a) = step_results.get_mut(i) {
                        if a.is_empty() {
                            *a = full_output;
                        }
                    }
                    if let Some(ok) = step_success.get_mut(i) {
                        *ok = success;
                    }
                }
                cur = None;
            }
            _ => {}
        }
    }

    // v2.15.0 — surface the `return` value as the flow's final result. The
    // dispatcher's `run_return` emits no wire step, so we append it here as a
    // `return` step carrying the resolved binding — matching the non-streaming
    // runner's return-step (v2.15.0) so the adopter's FlowEnvelope output is the
    // returned value, not the prior step's output.
    if let Some(value) = flow_return {
        step_names.push("return".to_string());
        step_results.push(value);
    }

    // v2.15.0 — project anchor breaches + blame from the per-step audit
    // records, closing the v2.15.0 cutover gap. `anchor_breaches` is the count;
    // `blame_attribution` carries AnchorBreach attribution (location
    // `step:<name>`, structural message, AnchorBreach kind), coalesced by
    // `merge_blame`'s first-emitted-wins discipline. Anchors are evaluated on
    // EVERY backend including stub — which the retired executor did not do, so
    // this surfaces breaches that used to be silently dropped there.
    let mut anchor_breaches = 0usize;
    let mut blame_attribution: Option<crate::wire_envelope::BlameContext> = None;
    for rec in audit_records.lock().await.iter() {
        if rec.anchor_breaches.is_empty() {
            continue;
        }
        anchor_breaches += rec.anchor_breaches.len();
        let blame = crate::wire_envelope::BlameContext {
            kind: crate::wire_envelope::BlameKind::AnchorBreach,
            // v2.83.0 — negative blame, assigned by the paper by name:
            // an anchor breach is the sub-agent reaching outside its granted
            // containment. This is one of only two sites in the workspace that
            // production actually emits a `BlameContext` from, so the axis is
            // live on the path adopters see, not just on the helpers.
            party: Some(crate::wire_envelope::BlameParty::Server),
            location: format!("step:{}", rec.step_name),
            message: format!(
                "{} anchor breach(es) on step '{}' — flow \
                 proceeded on degraded posture",
                rec.anchor_breaches.len(),
                rec.step_name
            ),
            d_letter: Some("39.c.z".to_string()),
        };
        blame_attribution =
            crate::wire_envelope_producers::merge_blame(blame_attribution, Some(blame));
    }

    // v2.21.0 — read the shared per-run row totals the store handlers
    // folded in (par-branch counts merged via the shared Arc). Bound to a
    // local so the transient `MutexGuard` doesn't outlive the struct
    // construction's tail expression.
    let store_row_counts = *row_counts.lock().unwrap();

    // v2.46.0 — project the run's temporal state into the envelope record.
    // `None` when no `now:`-bearing step rendered (zero wire drift).
    let temporal_context = crate::temporal_context::record_of(&temporal.lock().unwrap());

    CollectedRun {
        success,
        steps_executed: step_names.len(),
        tokens_output,
        step_names,
        step_results,
        anchor_breaches,
        blame_attribution,
        flow_error,
        store_row_counts,
        temporal_context,
        step_success,
    }
}

/// v4.6.0 — `axon run`, on the same engine as everything else.
///
/// # Why this exists
///
/// The CLI was the last door still driven by `execute_real_async` (and, for
/// `--tool-mode stub`, by `execute_stub` — a THIRD executor, which bridged
/// nothing but agent calls). v2.15.0 moved the server to the dispatcher and
/// left both behind a kill-switch "for one release"; the release came and went.
/// What that cost was measurable and specific: twenty verbs — `shield`,
/// `warden`, `quant`, `emit`, `publish`, the algebraic-effect family — reached
/// a language model on this door and their real handler on every other one.
/// Not because anyone decided they should. Because the route table's job was to
/// list what had been ported, and nobody finished porting.
///
/// Deleting the second engine is what makes that number zero, and it is the
/// only thing that makes it *stay* zero: there is no longer a second place
/// where a verb can be forgotten.
///
/// # What changed for someone running `axon run`
///
/// The step walk is LINEAR. The legacy executor had an auto-parallel wave
/// scheduler that inferred concurrency from data dependencies between steps;
/// the dispatcher runs steps in source order and concurrency is written, with
/// `par`. That is the one intentional behavioural difference v2.15.0 recorded
/// when the server crossed over, and it arrives here for the same reason.
///
/// Per-step output prints after each unit rather than during it, because the
/// collector drains the event channel once the flow completes. The lines
/// themselves say the same things.
#[allow(clippy::too_many_arguments)]
fn execute_cli_via_dispatcher(
    units: &[ExecutionUnit],
    ir: &crate::ir_nodes::IRProgram,
    backend: &str,
    use_color: bool,
    trace: bool,
    json: bool,
    report: &mut ReportBuilder,
    registry: std::sync::Arc<ToolRegistry>,
    nav_dispatch: &NavDispatch,
) -> (bool, Vec<TraceEvent>, usize) {
    let mut events: Vec<TraceEvent> = Vec::new();
    let mut all_ok = true;
    // What RAN, not what compiled. An agent loop expands into a step per
    // deliberation plus one per tool call, so the compiled count under-reports
    // it — and a summary line reading "1 step" after four ran is the same false
    // green in a smaller font.
    let mut steps_run = 0usize;

    for (i, unit) in units.iter().enumerate() {
        if !json {
            println!(
                "\n{}",
                c(
                    &format!(
                        "▶ Execution Unit {}/{}: {} as {}",
                        i + 1,
                        units.len(),
                        unit.flow_name,
                        unit.persona_name
                    ),
                    "\x1b[1;36m",
                    use_color,
                )
            );
        }
        if trace {
            events.push(TraceEvent {
                event: "unit_start".to_string(),
                unit: unit.flow_name.clone(),
                step: String::new(),
                detail: format!(
                    "persona={}, context={}",
                    unit.persona_name, unit.context_name
                ),
            });
        }

        report.begin_unit(&unit.flow_name, &unit.persona_name);
        let mut hooks = HookManager::new();
        hooks.on_unit_start(&unit.flow_name, &unit.persona_name);

        // The unit names a flow the program declared, or `build_execution_units`
        // would not have produced it. If that stops being true, say so instead
        // of running nothing and reporting success.
        let Some(flow) = ir.flows.iter().find(|f| f.name == unit.flow_name) else {
            let detail = format!(
                "execution unit names flow '{}', which this program does not declare",
                unit.flow_name
            );
            if !json {
                eprintln!("{}", c(&format!("✗ {detail}"), "\x1b[1;31m", use_color));
            }
            all_ok = false;
            hooks.on_unit_end();
            report.end_unit(&hooks);
            continue;
        };

        let collected = block_on_store(collect_via_dispatcher(
            flow,
            backend,
            // The CLI has no tenant scope. Empty, not `"default"`: the custody
            // seams then refuse an unscoped call, which is the honest answer
            // for a one-shot local run.
            "",
            &unit.system_prompt,
            ir.contexts.first().and_then(|c| c.now_tz.clone()),
            // No override — the dispatcher resolves the key from the
            // environment, which is what a CLI run has.
            None,
            None,
            None,
            std::sync::Arc::new(ir.anchors.clone()),
            nav_dispatch,
            registry.clone(),
            &unit.param_bindings,
            std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
            // A CLI run has no daemon budget, no cross-request channel guards,
            // no durable outbox and no minter/custody port. Each of those verbs
            // fails closed when reached, which is the same answer the server
            // gives an undeployed capability — and a great deal better than the
            // sentence the legacy path bound in its place.
            None,
            None,
            None,
            None,
            None,
            std::sync::Arc::new(
                ir.credentials
                    .iter()
                    .map(|cr| (cr.name.clone(), cr.clone()))
                    .collect(),
            ),
            None,
            None,
        ));

        for ((name, result), ok) in collected
            .step_names
            .iter()
            .zip(collected.step_results.iter())
            .zip(collected.step_success.iter().copied())
        {
            steps_run += 1;
            if !json {
                // The mark comes from the step's own `StepComplete`, not from
                // this loop having reached the line. A ✓ printed over a step
                // whose result text is an error message is the small, everyday
                // version of the defect this release exists to remove.
                let (mark, colour) = if ok {
                    ("✓", "\x1b[32m")
                } else {
                    ("✗", "\x1b[31m")
                };
                println!(
                    "  {} {}",
                    c(mark, colour, use_color),
                    c(&format!("{name} → {}", truncate(result, 100)), "\x1b[2m", use_color),
                );
            }
            if trace {
                events.push(TraceEvent {
                    event: "step_complete".to_string(),
                    unit: unit.flow_name.clone(),
                    step: name.clone(),
                    detail: format!("{} char(s)", result.len()),
                });
            }
            report.record_step(StepReport {
                name: name.clone(),
                step_type: unit
                    .steps
                    .iter()
                    .find(|s| &s.step_name == name)
                    .map(|s| s.step_type.clone())
                    .unwrap_or_default(),
                result: result.clone(),
                duration_ms: 0,
                input_tokens: 0,
                output_tokens: 0,
                anchor_breaches: 0,
                chain_activations: 0,
                was_retried: false,
            });
        }

        // A node that refused names itself and why. The legacy executor had no
        // slot for this on the CLI at all — a failing flow printed nothing and
        // exited 0.
        if let Some(detail) = &collected.flow_error {
            if !json {
                eprintln!("{}", c(&format!("✗ {detail}"), "\x1b[1;31m", use_color));
            }
            if trace {
                events.push(TraceEvent {
                    event: "flow_error".to_string(),
                    unit: unit.flow_name.clone(),
                    step: String::new(),
                    detail: detail.clone(),
                });
            }
        }
        all_ok &= collected.success;

        hooks.on_unit_end();
        report.end_unit(&hooks);
    }

    (all_ok, events, steps_run)
}

pub fn execute_server_flow(
    ir: &crate::ir_nodes::IRProgram,
    flow_name: &str,
    backend: &str,
    // v2.49.0 — the VERIFIED tenant this flow executes under, set on the
    // DispatchCtx so every explicit-tenant runtime seam (the
    // `axon::secret_custody` port for `rotate`/`retrieve`-secrets/`secret:`
    // injection, the `mint` minter, session state) runs under the caller's
    // scoped tenant — never ambient (the v2.47.0 posture). The endpoint passes
    // the request-verified `route.tenant_id`; the daemon supervisor passes
    // the daemon's scoped tenant; a CLI/test with no scope passes `""`
    // (custody then fails CLOSED — `require_tenant` refuses an unscoped
    // call — which is correct). Before v2.49.0 this was silently empty, so
    // every custody call was refused as "unscoped" in real deployments.
    tenant_id: &str,
    // v4.6.0 — read by nothing since the legacy executor was retired:
    // it named the file for that executor's `SessionStore` and its
    // `ReportBuilder`, and the dispatcher anchors its session on the flow name.
    // Kept so the enterprise call site does not have to change in a minor, and
    // named here rather than left to look load-bearing. It goes with the next
    // signature change this function takes.
    _source_file: &str,
    api_key_override: Option<&str>,
    // v1.32.0 (D1) — the parsed HTTP request body. The flow's
    // declared parameters bind from its same-named fields (the Request
    // Binding Contract) and seed each `ExecContext` before the step
    // walk. `None` for a caller with no request body (D5).
    request_body: Option<&serde_json::Value>,
    // v1.32.0 (D3) — the URL path captures (e.g. for
    // `/api/tenants/{tenant_id}` the map is `{tenant_id: "acme"}`).
    // Empty map for callers without a dynamic route (D5 backwards-
    // compat). Passed to `bind_request` alongside `request_body`.
    request_path: &std::collections::HashMap<String, String>,
    // v1.32.0 (D3) — the URL query string parsed into name → value.
    // Single-value semantics in v1.38.5 (multi-value query keys
    // deferred per plan vivo section 7); axum's `Query<HashMap>` extractor
    // provides this shape.
    request_query: &std::collections::HashMap<String, String>,
    // v2.8.0 (D7) — optional per-tenant / per-server tool base URL.
    // When `Some`, every URL-dispatched program tool with a RELATIVE
    // `runtime` is resolved against it (`{base}/{slug}`) so the adopter
    // wires their tool-server via config without touching the program;
    // absolute runtimes stay verbatim (D5). `None` → no resolution.
    tool_base_url: Option<&str>,
    // v1.18.0 (Kivi brief #37) — optional per-tenant LLM endpoint
    // override: the base URL + chat-completions path the resolved backend
    // hits, threaded into the dispatcher's backend factory. `None` (D5
    // back-compat) → the provider's env/default endpoint. Mirrors the
    // per-tenant `tool_base_url` shape, applied to the LLM call.
    llm_base_url: Option<&str>,
    llm_chat_path: Option<&str>,
    // v2.28.0 — the active `budget { … }` linear-effect gate when a budgeted
    // `daemon` runs this flow (built by the daemon supervisor from the daemon's
    // compiled budget). `None` for every other caller (HTTP, CLI, tests) — tool
    // dispatch is then unconditional, byte-identical to pre-v2.28.0.
    budget: Option<std::sync::Arc<std::sync::Mutex<crate::runtime::budget_kernel::BudgetGate>>>,
    // v2.69.0 — the per-channel concurrency semaphores (`resource.capacity`),
    // held on ServerState across requests. `None` for every non-server caller.
    channel_semaphores: Option<std::sync::Arc<crate::channel_semaphore::ChannelSemaphores>>,
    // v2.69.0 — the tool-lease guard (held on ServerState). `None` off-server.
    tool_leases: Option<std::sync::Arc<crate::resource_lease::ResourceLeaseGuard>>,
    // v2.31.0 — the durable event outbox (the v2.31.0 `EventOutbox` seam). `Some`
    // only when a `daemon` runs this flow AND the deployment configures a durable
    // channel sink (the enterprise supervisor passes its per-tenant Postgres
    // outbox); a daemon `emit` to a `persistent_axonstore` channel then APPENDS
    // durably instead of buffering in-process. `None` for every other caller
    // (HTTP, CLI, tests) ⇒ pre-v2.31.0 in-process `emit` behavior.
    event_outbox: Option<std::sync::Arc<dyn crate::event_outbox::EventOutbox>>,
    // v2.46.0 — the ephemeral-credential minter port behind the `mint`
    // flow verb. `Some` only when the deployment owns a minter (the
    // enterprise executor injects its PASETO minter, v2.46.0 — the same
    // injection shape as `event_outbox`); `None` for every other caller
    // (HTTP, CLI, tests) ⇒ a reached `mint` fails CLOSED with a loud
    // missing-dependency error, never a silent stub (v2.41.0 lesson).
    credential_minter: Option<std::sync::Arc<dyn crate::credential_minter::CredentialMinter>>,
    // v2.48.0 — the secret-custody port behind the `backend: secrets`
    // metadata store, the `rotate` verb, and `tool { secret: }` dispatch
    // injection. `Some` only when the deployment owns a custody (the
    // enterprise executor injects its envelope-encrypted Pg custody,
    // v2.48.0 — the same injection shape as `credential_minter`); `None`
    // for every other caller (HTTP, CLI, tests) ⇒ each of those surfaces
    // fails CLOSED with a loud missing-dependency error, never a silent
    // stub and never a fabricated result.
    secret_custody: Option<std::sync::Arc<dyn crate::secret_custody::SecretCustody>>,
    // v2.63.0 — the deterministic columnar engine port behind the five
    // data-plane verbs. `Some` only when the deployment instantiated the
    // declared dataspaces (the OSS deploy hook / the enterprise executor —
    // the same injection shape as `credential_minter`); `None` for every
    // other caller ⇒ each data-plane verb fails CLOSED with a loud
    // missing-dependency error (the v2.63.0 honesty floor), never an LLM
    // narration.
    dataspace_engine: Option<crate::dataspace_engine::SharedDataspaceEngine>,
    // v2.56.0 — per-tenant scrape overrides (proxy / crawl
    // concurrency) resolved by the deployed executor's `SecretResolver`, applied
    // to the request-scoped registry before any scrape dispatch (the same
    // per-tenant-rewrite shape as `tool_base_url`). `None` for every caller that
    // does no per-tenant scrape config (HTTP/CLI/tests/daemon) — the
    // source-declared scrape config then stands. The tenant is stamped onto the
    // scrape entries regardless (from `tenant_id`), so the v2.56.0 selector
    // memory is keyed even without overrides.
    scrape_overrides: Option<&crate::tool_registry::ScrapeOverrides>,
) -> Result<ServerRunnerMetrics, String> {
    let mut target_run = None;
    for run in &ir.runs {
        if run.flow_name == flow_name {
            target_run = Some(run);
            break;
        }
    }

    let mut execution_units = Vec::new();

    if let Some(run) = target_run {
        execution_units.push(ExecutionUnit {
            flow_name: run.flow_name.clone(),
            persona_name: run.persona_name.clone(),
            context_name: run.context_name.clone(),
            system_prompt: build_system_prompt(run, backend),
            steps: build_compiled_steps(run, ir),
            anchor_instructions: build_anchor_instructions(run),
            effort: run.effort.clone(),
            resolved_anchors: run.resolved_anchors.clone(),
            // v1.32.0 (D1) — bind the request body to the resolved
            // flow's declared parameters.
            // v1.32.0 (D3) — extended to bind from path + query
            // sources too; the runtime merge respects the D4
            // compile-time collision rejection (axon-T901).
            param_bindings: run
                .resolved_flow
                .as_ref()
                .map(|f| crate::request_binding::bind_request(
                    f,
                    request_path,
                    request_query,
                    request_body,
                ))
                .unwrap_or_default(),
        });
    } else {
        let target_flow: &crate::ir_nodes::IRFlow = ir
            .flows
            .iter()
            .find(|f| f.name == flow_name)
            .ok_or_else(|| format!("flow '{}' not found in compiled IR", flow_name))?;

        let default_persona = ir.personas.first().cloned().unwrap_or_else(|| crate::ir_nodes::IRPersona {
            node_type: "Persona",
            source_line: 0,
            source_column: 0,
            name: "Default".to_string(),
            domain: vec![],
            tone: "".to_string(),
            confidence_threshold: None,
            cite_sources: None,
            refuse_if: vec![],
            language: "".to_string(),
            description: "".to_string(),
        });
        let default_context = ir.contexts.first().cloned().unwrap_or_else(|| crate::ir_nodes::IRContext {
            node_type: "Context",
            source_line: 0,
            source_column: 0,
            name: "Default".to_string(),
            memory_scope: "".to_string(),
            language: "".to_string(),
            depth: "".to_string(),
            max_tokens: None,
            temperature: None,
            cite_sources: None,
            now_tz: None,
        });

        let run = crate::ir_nodes::IRRun {
            node_type: "Run",
            source_line: 0,
            source_column: 0,
            flow_name: flow_name.to_string(),
            arguments: vec![],
            persona_name: default_persona.name.clone(),
            context_name: default_context.name.clone(),
            anchor_names: vec![],
            on_failure: "".to_string(),
            on_failure_params: vec![],
            output_to: "".to_string(),
            effort: "low".to_string(),
            resolved_flow: Some(target_flow.clone()),
            resolved_persona: Some(default_persona),
            resolved_context: Some(default_context),
            resolved_anchors: ir.anchors.clone(),
        };

        execution_units.push(ExecutionUnit {
            flow_name: run.flow_name.clone(),
            persona_name: run.persona_name.clone(),
            context_name: run.context_name.clone(),
            system_prompt: build_system_prompt(&run, backend),
            steps: build_compiled_steps(&run, ir),
            anchor_instructions: build_anchor_instructions(&run),
            effort: run.effort.clone(),
            resolved_anchors: run.resolved_anchors.clone(),
            // v1.32.0 (D1) — bind the request body to the flow's
            // declared parameters (the dynamic-route execution path).
            // v1.32.0 (D3) — extended to bind from path + query
            // sources too.
            param_bindings: crate::request_binding::bind_request(
                target_flow,
                request_path,
                request_query,
                request_body,
            ),
        });
    }

    let mut registry = crate::tool_registry::ToolRegistry::new();
    // v2.8.0 — register the program's declared tools on the SERVER path
    // (the CLI path already does this in `run_run`). Without this, every
    // program-declared `tool { provider: http … }` missed the registry and the
    // step silently degraded to an LLM call (the brief #22 / #17 finding). This
    // `registry` is a per-call local (built fresh above for THIS request), so
    // registration is request-scoped — no cross-tenant tool contamination
    // between concurrent flows (v2.8.0 D10). Provider→URL resolves via each tool's
    // declared `runtime:` field (D7); the v2.8.0 structured body then POSTs to it.
    registry.register_from_ir(&ir.tools);
    // v2.8.0 (D7) — resolve relative tool runtimes against the
    // caller-supplied per-tenant / per-server base URL. Request-scoped
    // (this `registry` is a per-call local) → no cross-tenant leakage.
    if let Some(base) = tool_base_url {
        registry.resolve_relative_endpoints(base);
    }
    // v2.69.0 — a tool on a `resource` derives its endpoint + concurrency from
    // it (via the config resolver), governed exactly as an `axonstore` does (v2.67.0).
    // Runs AFTER the base-URL pass so a resource-backed tool overrides the legacy
    // slug resolution. `refused` tools (unresolvable endpoint) are dropped, so a
    // dispatch of one fails honestly rather than reaching a phantom.
    let _refused_tools = registry.resolve_from_resources_within(
        &ir.resources,
        &crate::resource_resolver::EnvResourceResolver,
        &ir.fabrics,
    );
    // v2.56.0 — stamp the dispatching tenant onto every scrape entry
    // (keys the v2.56.0 adaptive-selector memory) + apply the per-tenant
    // proxy/concurrency overrides (v2.56.0). Request-scoped registry → no
    // cross-tenant leakage. Runs even without overrides so the tenant is always
    // stamped for the memory key.
    registry.apply_scrape_tenant_context(tenant_id, scrape_overrides);

    // v1.30.0 — build the axonstore registry from the program's
    // declarations. The D2 closed-catalog gate runs here: an unknown
    // backend fails fast, at deploy, with a named error.
    // v2.15.0 — Arc the registry so it can be shared (by clone) into the
    // structural-navigate `DispatchCtx` while still being borrowed (via Deref)
    // by the eager-pin walk + `execute_real_async`'s own store path.
    // v2.67.0 — build the registry GOVERNED: the store derives its DSN and its
    // POOL SIZE from the `resource:` it names, and the `lease`s over those
    // resources are acquired so a store operation is a *use* of the resource.
    //
    // 🔴 THIS LINE IS THE FASE. It used to read `StoreRegistry::build(&ir
    // .axonstore_specs)` — the legacy entry, which passes NO resources and NO
    // leases. With that call, `build_with_resources` and `build_governed` would
    // have existed and been reachable from nothing: a real engine with a dead
    // wire, in the very cycle whose purpose is deleting them. `capacity: 20` would
    // have produced a pool of 10 in every deployed flow, and the gate proving
    // otherwise would have been testing a code path production never took.
    let store_registry = std::sync::Arc::new(
        StoreRegistry::build_governed(
            &ir.axonstore_specs,
            &ir.resources,
            &ir.leases,
            &crate::resource_resolver::EnvResourceResolver,
        )
        .map_err(|e| format!("axonstore registry: {e}"))?,
    );

    // v2.15.0 — build the dispatcher's corpus state from the IR exactly as
    // `run_streaming_via_dispatcher` does, so a NON-streaming `navigate` runs the
    // SAME structural MDN traversal as the SSE path (instead of hallucinating via
    // the LLM). Static v2.13.0 graphs + dynamic v2.14.0 store-sourced corpora + the
    // adaptive set are all wired.
    let nav_dispatch = build_nav_dispatch(
        ir,
        store_registry.clone(),
        dataspace_engine.clone(),
    );

    // v1.32.0 (D1) — Eager acquire one PoolConnection per
    // postgresql-backed axonstore referenced in the flow body BEFORE
    // executing any step. Each pin is held for the whole flow
    // execution and released on `pinned_conns` drop at the end of
    // this function (Rust handles the drop order: HashMap drops →
    // every PoolConnection drops → the per-conn `after_release
    // DEALLOCATE ALL` hook from v1.31.0 D2 runs → conn returns
    // to the pool clean).
    //
    // The discovery walk filters `step.step_type` to the four SQL
    // store ops + checks the registry's `backend_kind` to skip
    // in_memory stores (no race, no pin needed). The set is
    // deduplicated by store_name — multiple steps against the same
    // store share ONE pin (the D1 invariant).
    //
    // Acquire failure is non-fatal: the flow proceeds with an empty
    // pin map, which falls back to the legacy `StoreConn::Pool`
    // path. This preserves resilience against transient pool
    // saturation (a deploy-time `verify_postgres_schemas` failure
    // is the right gate for "store unreachable", not flow-time).
    // v1.32.0 (POST-CLOSE HOTFIX) — Compute the set of
    // postgresql-backed axonstores referenced by ANY execution unit's
    // body. This walks the IR purely SYNCHRONOUSLY — no .await, no
    // tokio runtime needed. The actual pin acquisition happens INSIDE
    // the single outer `block_on_store` below so the pins acquire on
    // the SAME runtime that later dispatches their SQL.
    // v4.6.0 — postgres-only. The legacy executor read this set
    // unconditionally; the dispatcher's eager-pin loop is already behind
    // `#[cfg(feature = "postgres")]`, and this is its only remaining reader.
    #[cfg(feature = "postgres")]
    let needed_pg_stores: std::collections::HashSet<String> = {
        let mut needed = std::collections::HashSet::new();
        for unit in &execution_units {
            for step in &unit.steps {
                if matches!(
                    step.step_type.as_str(),
                    "persist" | "retrieve" | "mutate" | "purge"
                ) && store_registry.backend_kind(&step.step_name)
                    == Some(crate::store::registry::StoreBackendKind::Postgresql)
                {
                    needed.insert(step.step_name.clone());
                }
            }
        }
        needed
    };

    // v4.6.0 — THIS IS THE DEFAULT ENGINE, and the comment that used
    // to sit here said the opposite.
    //
    // It read: *"When `AXON_UNIFIED_DRIVER` is set … OFF by default → the legacy
    // executors below stay the engine."* That described the v2.15.0 opt-in shape.
    // The cutover landed in v2.15.0 — [`unified_driver_enabled`] was inverted into
    // a kill-switch reading `AXON_LEGACY_EXECUTOR` — and this comment was never
    // updated. `AXON_UNIFIED_DRIVER` is set by nothing and read by nothing; it
    // does not exist. So the block below is entered on EVERY call unless an
    // operator sets the kill-switch, and it RETURNS — `execute_real_async` and
    // `execute_stub` further down are the fallback, not the engine.
    //
    // The cost of the stale sentence was a whole release cycle aimed at the wrong door:
    // the plan for this cycle opened by asserting that the verbs `execute_real_async` narrates
    // are narrated "en cada endpoint no-SSE de ENT", and they are not — ENT's only
    // runner entry is `execute_server_flow`, which lands HERE. Measured, not
    // read: `the_default_engine_is_the_dispatcher.rs` drives a program through
    // this function and reads back the answer only the real handler can give.
    //
    // Run the NON-streaming path through the SAME dispatcher the SSE path uses
    // (one engine, two sinks) + assemble the envelope from the collected events
    // and the IR-derived epistemics.
    {
        let flow = match ir.flows.iter().find(|f| f.name == flow_name) {
            Some(f) => f,
            // v4.6.0 — this used to fall THROUGH, into the legacy
            // executor below. In practice an unknown flow name is refused
            // earlier, on the way to building the execution units, so nothing
            // wrong was executed — but the refusal depended on that other guard
            // holding, and the branch here said nothing about it. There is no
            // longer anything below to fall into, and the arm says so.
            None => {
                return Err(format!(
                    "flow '{flow_name}' is not declared in this program — nothing to execute"
                ))
            }
        };
        {
            let system_prompt = execution_units
                .first()
                .map(|u| u.system_prompt.clone())
                .unwrap_or_default();
            let param_bindings = execution_units
                .first()
                .map(|u| u.param_bindings.clone())
                .unwrap_or_default();
            let anchors = std::sync::Arc::new(ir.anchors.clone());
            let registry_arc = std::sync::Arc::new(registry);
            let collected = block_on_store(async {
                // v2.15.0 — eager pin acquisition ON THIS runtime (the v1.32.0
                // discipline): one PoolConnection per postgresql axonstore, held
                // for the flow + shared with the dispatcher's store handlers so
                // every store op routes through the same physical backend
                // (transaction-mode-pooler safe). Acquire failure is non-fatal
                // (falls back to lazy per-op pool acquisition).
                let pinned: std::sync::Arc<
                    std::sync::Mutex<
                        std::collections::HashMap<
                            String,
                            crate::pinned_conn::PinnedConn,
                        >,
                    >,
                > = std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
                // v2.51.0 — skip the eager pin entirely under a session/direct
                // pooler (store ops then acquire per-op, releasing across
                // cognition). `acquire_pin` also refuses in that mode, but
                // skipping here avoids the loop's work + a misleading warn.
                // v2.81.0 — no driver, no pins to acquire. The map stays empty,
                // which is exactly what an uninhabited `PinnedConn` already guarantees.
                #[cfg(feature = "postgres")]
                if crate::store::pooler_mode::connection_pinning_enabled() {
                    for store_name in &needed_pg_stores {
                        if let Ok(crate::store::registry::StoreHandle::Postgres(backend_pool)) =
                            store_registry.resolve(store_name)
                        {
                            if let Ok(conn) = backend_pool.acquire_pin().await {
                                pinned.lock().unwrap().insert(store_name.clone(), conn);
                            }
                        }
                    }
                }
                collect_via_dispatcher(
                    flow,
                    backend,
                    tenant_id,
                    &system_prompt,
                    // v2.46.0 — the frame-level `now:` (first-context convention).
                    ir.contexts.first().and_then(|c| c.now_tz.clone()),
                    api_key_override,
                    llm_base_url,
                    llm_chat_path,
                    anchors,
                    &nav_dispatch,
                    registry_arc,
                    &param_bindings,
                    pinned,
                    budget.clone(),
                    channel_semaphores.clone(),
                    tool_leases.clone(),
                    // v2.31.0 — when a durable outbox is injected (the
                    // enterprise daemon path), build the typed event bus from the
                    // program's `channel` definitions so `run_emit` knows each
                    // channel's `persistence` + routes a `persistent_axonstore`
                    // emit to the outbox. Paired: no outbox ⇒ no bus ⇒ pre-v2.31.0
                    // in-process `emit`.
                    event_outbox.as_ref().map(|_| {
                        std::sync::Arc::new(
                            crate::runtime::channels::TypedEventBus::from_ir_program(ir),
                        )
                    }),
                    event_outbox.clone(),
                    // v2.46.0 — the compiled credential contracts + the
                    // minter port (None ⇒ mint fails closed).
                    std::sync::Arc::new(
                        ir.credentials
                            .iter()
                            .map(|c| (c.name.clone(), c.clone()))
                            .collect(),
                    ),
                    credential_minter.clone(),
                    // v2.48.0 — the custody port (None ⇒ fail-closed).
                    secret_custody.clone(),
                )
                .await
            });
            let per_step_chunks: Vec<Vec<String>> = collected
                .step_results
                .iter()
                .map(|r| if r.is_empty() { Vec::new() } else { vec![r.clone()] })
                .collect();
            // v2.15.0 — `provenance_events` is a PURE IR walk in the legacy
            // path (`execution_units` → `(step_type, step_name)` → closed-catalog
            // slugs), zero execution dependency — so deriving it here from the
            // same `execution_units` is byte-identical with the legacy + SSE
            // paths, closing the v2.15.0 provenance regression.
            let provenance_walk: Vec<(String, String)> = execution_units
                .iter()
                .flat_map(|u| {
                    u.steps
                        .iter()
                        .map(|s| (s.step_type.clone(), s.step_name.clone()))
                })
                .collect();
            let provenance_events =
                crate::wire_envelope_producers::collect_provenance_events_from(
                    &provenance_walk,
                );
            return Ok(ServerRunnerMetrics {
                success: collected.success,
                steps_executed: collected.steps_executed,
                tokens_input: 0,
                tokens_output: collected.tokens_output,
                // v2.15.0 — projected from the dispatcher's per-step audit
                // records.
                anchor_breaches: collected.anchor_breaches,
                step_names: collected.step_names,
                step_results: collected.step_results,
                per_step_chunks,
                // IR-derived (provenance) + audit-derived (blame). The
                // envelope carries the run's epistemic lineage; `provenance_blame_parity.rs`
                // is the gate that keeps these three fields POPULATED rather
                // than quietly empty.
                provenance_events,
                blame_attribution: collected.blame_attribution,
                // IR-derived → byte-identical with the SSE path.
                epistemic_envelopes: derive_epistemic_envelopes_for_flow(ir, flow_name),
                // v2.15.0 — the honest hard-failure detail (named node +
                // cause), or `None` on the clean path. Closes the v2.15.0
                // silent-abort regression for store writes + every other node.
                error: collected.flow_error,
                // v2.21.0 — the dispatcher (default engine) per-run row
                // counts the store handlers folded in.
                rows_retrieved: collected.store_row_counts.retrieved,
                rows_persisted: collected.store_row_counts.persisted,
                rows_mutated: collected.store_row_counts.mutated,
                rows_purged: collected.store_row_counts.purged,
                // v2.46.0 — the run's temporal record (capture + zones),
                // `None` for flows with no `now:` (zero wire drift).
                temporal_context: collected.temporal_context,
            });
        }
    }

}

// ── Public entry point ───────────────────────────────────────────────────────

pub fn run_run(
    file: &str,
    backend: &str,
    trace: bool,
    tool_mode: &str,
    stream: bool,
    output: &str,
    export_plan: bool,
) -> i32 {
    let output_fmt = match OutputFormat::from_str(output) {
        Some(f) => f,
        None => {
            eprintln!("✗ Invalid output format '{}'. Use 'text' or 'json'.", output);
            return 2;
        }
    };
    let json = output_fmt.is_json();
    let use_color = if json { false } else { io::stdout().is_terminal() };
    let path = Path::new(file);
    let filename = path
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| file.to_string());

    // ── 1. Read source ───────────────────────────────────────────
    let source = match std::fs::read_to_string(path) {
        Ok(s) => s,
        Err(_) => {
            eprintln!(
                "{}",
                c(&format!("✗ File not found: {file}"), "\x1b[1;31m", use_color)
            );
            return 2;
        }
    };

    // ── 2–5. Compile: EMS when imports are declared (v2.76.0),
    // the classic single-file pipeline otherwise.
    let ir_program = if axon_frontend::ems::source_declares_imports(&source, file) {
        let opts = axon_frontend::ems::EmsOptions {
            modules_root: std::env::var("AXON_MODULES_ROOT").ok().map(Into::into),
            use_cache: true,
            cache_dir: None,
        };
        let base = |origin: &str| -> String {
            Path::new(origin)
                .file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_else(|| origin.to_string())
        };
        match axon_frontend::ems::compile_project(path, &opts) {
            Err(fail) => {
                eprintln!(
                    "{}  {} error(s)",
                    c(&format!("✗ {filename}"), "\x1b[1;31m", use_color),
                    fail.errors.len()
                );
                for e in &fail.errors {
                    eprintln!("  error [{} line {}]: {}", base(&e.file), e.line, e.message);
                }
                return 1;
            }
            Ok(out) => {
                for w in &out.warnings {
                    eprintln!("  warning [{} line {}]: {}", base(&w.file), w.line, w.message);
                }
                out.ir
            }
        }
    } else {
        // ── 2. Lex ───────────────────────────────────────────────
        let tokens = match Lexer::new(&source, file).tokenize() {
            Ok(t) => t,
            Err(LexerError { message, line, column }) => {
                let loc = if column > 0 {
                    format!(":{line}:{column}")
                } else {
                    format!(":{line}")
                };
                eprintln!(
                    "{}  {message}",
                    c(&format!("✗ {filename}{loc}"), "\x1b[1;31m", use_color)
                );
                return 1;
            }
        };

        // ── 3. Parse ─────────────────────────────────────────────
        let mut parser = Parser::new(tokens);
        let program = match parser.parse() {
            Ok(p) => p,
            Err(ParseError { message, line, column, .. }) => {
                let loc = if column > 0 {
                    format!(":{line}:{column}")
                } else {
                    format!(":{line}")
                };
                eprintln!(
                    "{}  {message}",
                    c(&format!("✗ {filename}{loc}"), "\x1b[1;31m", use_color)
                );
                return 1;
            }
        };

        // ── 4. Type check ────────────────────────────────────────
        let type_errors = TypeChecker::new(&program).check();
        if !type_errors.is_empty() {
            eprintln!(
                "{}  {} type error(s)",
                c(&format!("✗ {filename}"), "\x1b[1;31m", use_color),
                type_errors.len()
            );
            for te in &type_errors {
                eprintln!("  error [line {}]: {}", te.line, te.message);
            }
            return 1;
        }

        // ── 5. Generate IR ───────────────────────────────────────
        IRGenerator::new().generate(&program)
    };

    // ── 6. Build execution plan ──────────────────────────────────
    let units = build_execution_plan(&ir_program, backend);

    if units.is_empty() {
        eprintln!(
            "{}",
            c("⚠ No run statements found — nothing to execute.", "\x1b[1;33m", use_color)
        );
        return 0;
    }

    // ── 7. Execute ───────────────────────────────────────────────
    let mode_label = if tool_mode == "real" {
        if stream { "real+stream" } else { "real" }
    } else {
        "stub"
    };

    if !json {
        println!(
            "{}",
            c(
                &format!(
                    "═══ AXON Run: {filename} ({} unit{}, backend={backend}, mode={tool_mode}) ═══",
                    units.len(),
                    if units.len() == 1 { "" } else { "s" }
                ),
                "\x1b[1;36m",
                use_color,
            )
        );
    }

    let mut report = ReportBuilder::new(file, backend, mode_label);

    // Build tool registry from IR + builtins
    let mut registry = ToolRegistry::new();
    registry.register_from_ir(&ir_program.tools);

    // v1.30.0 — build the axonstore registry (D2 closed-catalog
    // gate). An unknown `backend:` fails fast, before execution.
    // v2.67.0 — governed: the store derives DSN + pool size from its `resource:`,
    // and leases over those resources are acquired (see `execute_server_flow`).
    let store_registry = match StoreRegistry::build_governed(
        &ir_program.axonstore_specs,
        &ir_program.resources,
        &ir_program.leases,
        &crate::resource_resolver::EnvResourceResolver,
    ) {
        Ok(r) => std::sync::Arc::new(r),
        Err(e) => {
            eprintln!(
                "{}  {e}",
                c(&format!("✗ {filename}"), "\x1b[1;31m", use_color)
            );
            return 1;
        }
    };
    // The dispatcher bridge for this program — the SAME catalogues the server
    // executor builds, so `axon run` (real and stub) reaches the real handler
    // for every bridged verb instead of the LLM fallthrough.
    let nav_dispatch = build_nav_dispatch(
        &ir_program,
        store_registry.clone(),
        None,
    );

    if !json && !registry.program_names().is_empty() {
        println!(
            "  {}",
            c(
                &format!(
                    "Tools: {} registered ({} builtin + {} program)",
                    registry.len(),
                    registry.builtin_names().len(),
                    registry.program_names().len(),
                ),
                "\x1b[2m",
                use_color,
            )
        );
    }

    // ── Export plan and exit (no execution) ──────────────────────
    if export_plan {
        let plan = build_plan_export(&units, file, backend, &registry);
        println!("{}", PlanBuilder::to_json(&plan));
        return 0;
    }

    // v4.6.0 — one engine. `--tool-mode stub` selects the stub
    // BACKEND rather than a second executor: the step results read `(stub)`
    // exactly as before, but every non-cognitive verb now reaches the handler
    // that computes it instead of a walk that skipped it.
    let engine_backend = if tool_mode == "real" { backend } else { "stub" };

    // Resolve the credential BEFORE the walk, and exit 2 — a CONFIG error, not
    // an execution failure.
    //
    // The retired executor did this on its first line and the distinction is
    // load-bearing at the shell: exit 1 means the program ran and something in
    // it failed; exit 2 means it never should have started. Left to the
    // dispatcher, an unset key surfaces per-step, the run "completes", and a CI
    // job that greps for a non-zero code learns the wrong thing about why.
    // An unknown backend name fails here for the same reason.
    if engine_backend != "stub" {
        if let Err(e) = backend::get_api_key(engine_backend) {
            eprintln!(
                "{}",
                c(&format!("✗ Backend error: {e:?}"), "\x1b[1;31m", use_color)
            );
            return 2;
        }
    }

    let (success, events, steps_run) = execute_cli_via_dispatcher(
        &units,
        &ir_program,
        engine_backend,
        use_color,
        trace,
        json,
        &mut report,
        std::sync::Arc::new(registry),
        &nav_dispatch,
    );

    // ── 8. JSON output or text summary ─────────────────────────
    if json {
        // Build report with a dummy HookManager for stub mode
        // (real mode already populated hooks inside execute_real)
        let stub_hooks = crate::hooks::HookManager::new();
        let execution_report = report.build(success, &stub_hooks);
        println!("{}", ReportBuilder::to_json(&execution_report));
    } else {
        let total_steps = steps_run;
        println!(
            "\n{}",
            c(
                &format!(
                    "═══ {} unit{}, {} step{} — {mode_label} execution complete ═══",
                    units.len(),
                    if units.len() == 1 { "" } else { "s" },
                    total_steps,
                    if total_steps == 1 { "" } else { "s" },
                ),
                "\x1b[1;32m",
                use_color,
            )
        );
    }

    // ── 9. Save trace ────────────────────────────────────────────
    if trace && !events.is_empty() {
        let trace_path = Path::new(file).with_extension("trace.json");
        let trace_json = serde_json::json!({
            "_meta": {
                "source": file,
                "backend": backend,
                "tool_mode": tool_mode,
                "axon_version": AXON_VERSION,
                "mode": "stub",
            },
            "events": events,
        });
        match serde_json::to_string_pretty(&trace_json) {
            Ok(json_str) => match std::fs::write(&trace_path, json_str) {
                Ok(_) => {
                    if !json {
                        println!(
                            "{}",
                            c(
                                &format!("📋 Trace saved → {}", trace_path.display()),
                                "\x1b[1;35m",
                                use_color,
                            )
                        );
                    }
                }
                Err(e) => eprintln!("⚠ Could not save trace: {e}"),
            },
            Err(e) => eprintln!("⚠ Could not serialize trace: {e}"),
        }
    }

    if success { 0 } else { 1 }
}

// ── v1.30.0 — sync-runner axonstore wiring tests ─────────────────

#[cfg(test)]
mod tests_2 {
    use super::*;

    #[test]
    fn coerce_respects_declared_int_float_bool() {
        assert_eq!(coerce_tool_arg_value("5", Some("Int")), serde_json::json!(5));
        assert_eq!(
            coerce_tool_arg_value("3.14", Some("Float")),
            serde_json::json!(3.14)
        );
        assert_eq!(
            coerce_tool_arg_value("true", Some("Bool")),
            serde_json::json!(true)
        );
        assert_eq!(
            coerce_tool_arg_value("false", Some("Bool")),
            serde_json::json!(false)
        );
    }

    #[test]
    fn coerce_keeps_string_param_verbatim_even_if_all_digits() {
        // Robustness invariant: a `String` param is NEVER numified.
        assert_eq!(
            coerce_tool_arg_value("12345", Some("String")),
            serde_json::json!("12345")
        );
        assert_eq!(
            coerce_tool_arg_value("Acme Corp", Some("String")),
            serde_json::json!("Acme Corp")
        );
    }

    #[test]
    fn coerce_optional_and_generic_types_use_base() {
        assert_eq!(coerce_tool_arg_value("7", Some("Int?")), serde_json::json!(7));
        // v2.77.0 — `List<String>` → base `List` → a JSON ARRAY. A lone value
        // for a List param is a 1-element list (never a bare string).
        assert_eq!(
            coerce_tool_arg_value("x", Some("List<String>")),
            serde_json::json!(["x"])
        );
    }

    #[test]
    fn coerce_list_materializes_the_surface_form_into_a_json_array() {
        // v2.77.0 — the parser's lossy surface rendering `[a, b]` (items unquoted,
        // comma-joined) becomes a real JSON array — the FB multi-photo / IG carousel
        // unblocker. Pre-116.c.4 this stayed the opaque string "[a, b]".
        assert_eq!(
            coerce_tool_arg_value(
                "[https://a/1.png, https://a/2.png]",
                Some("List<String>?")
            ),
            serde_json::json!(["https://a/1.png", "https://a/2.png"])
        );
        // Empty list.
        assert_eq!(
            coerce_tool_arg_value("[]", Some("List<String>")),
            serde_json::json!([])
        );
        // Inner element type coercion (List<Int>).
        assert_eq!(
            coerce_tool_arg_value("[1, 2, 3]", Some("List<Int>")),
            serde_json::json!([1, 2, 3])
        );
    }

    #[test]
    fn coerce_list_passes_through_valid_json_and_strips_quotes() {
        // Already-valid JSON (e.g. a step output) passes through unchanged.
        assert_eq!(
            coerce_tool_arg_value(r#"["x","y"]"#, Some("List<String>")),
            serde_json::json!(["x", "y"])
        );
        // A quoted element containing a comma is NOT split mid-item.
        assert_eq!(
            coerce_tool_arg_value(r#"["a, b", c]"#, Some("List<String>")),
            serde_json::json!(["a, b", "c"])
        );
    }

    #[test]
    fn build_body_emits_a_list_param_as_a_json_array_alongside_scalars() {
        // v2.77.0 — the EXACT assembly the runner runs at dispatch: a
        // `use facebook_publish_post(body = "album", media_urls = [a, b, c])`
        // must reach the connector with `media_urls` as a JSON ARRAY (so
        // build_publish_request materializes the multi-photo flow), while a
        // sibling `String` param stays a string.
        let interpolated = vec![
            ("body".to_string(), "album".to_string()),
            (
                "media_urls".to_string(),
                "[https://a/1.png, https://a/2.png, https://a/3.png]".to_string(),
            ),
        ];
        let param_types = vec![
            ("body".to_string(), "String".to_string()),
            ("media_urls".to_string(), "List<String>?".to_string()),
        ];
        let body = build_structured_tool_body(&interpolated, &param_types);
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(v["body"], serde_json::json!("album"));
        assert_eq!(
            v["media_urls"],
            serde_json::json!([
                "https://a/1.png",
                "https://a/2.png",
                "https://a/3.png"
            ])
        );
    }

    #[test]
    fn coerce_unparseable_scalar_falls_back_to_string_not_dropped() {
        // Declared Int/Bool but the (interpolated) value isn't one → lenient
        // string rather than a drop. The v2.8.0 type-checker already flags a
        // literal mismatch at compile time.
        assert_eq!(
            coerce_tool_arg_value("not-a-number", Some("Int")),
            serde_json::json!("not-a-number")
        );
        assert_eq!(
            coerce_tool_arg_value("maybe", Some("Bool")),
            serde_json::json!("maybe")
        );
    }

    #[test]
    fn coerce_unknown_or_schemaless_param_is_string() {
        assert_eq!(coerce_tool_arg_value("5", None), serde_json::json!("5"));
        assert_eq!(
            coerce_tool_arg_value("5", Some("SearchResults")),
            serde_json::json!("5")
        );
    }

    #[test]
    fn build_body_assembles_typed_structured_object() {
        let args = vec![
            ("query".to_string(), "Acme Corp".to_string()),
            ("max_results".to_string(), "5".to_string()),
            ("safesearch".to_string(), "true".to_string()),
        ];
        let types = vec![
            ("query".to_string(), "String".to_string()),
            ("max_results".to_string(), "Int".to_string()),
            ("safesearch".to_string(), "Bool".to_string()),
        ];
        let v: serde_json::Value =
            serde_json::from_str(&build_structured_tool_body(&args, &types)).unwrap();
        assert_eq!(v["query"], serde_json::json!("Acme Corp"));
        assert_eq!(v["max_results"], serde_json::json!(5));
        assert_eq!(v["safesearch"], serde_json::json!(true));
        // NOT the flat `{"input": …}` legacy shape.
        assert!(v.get("input").is_none());
    }

    #[test]
    fn build_body_escapes_special_characters_via_serde() {
        let args = vec![("query".to_string(), "a\"b\nc".to_string())];
        let types = vec![("query".to_string(), "String".to_string())];
        let v: serde_json::Value =
            serde_json::from_str(&build_structured_tool_body(&args, &types)).unwrap();
        assert_eq!(v["query"], serde_json::json!("a\"b\nc"));
    }

    #[test]
    fn build_body_empty_args_is_empty_object() {
        assert_eq!(build_structured_tool_body(&[], &[]), "{}");
    }
}

#[cfg(test)]
mod tests_3 {
    use super::*;

    // Every caller is `#[cfg(feature = "postgres")]`; so is this.
    #[cfg(feature = "postgres")]
    fn pg_store(name: &str, connection: &str) -> IRAxonStore {
        IRAxonStore {
            node_type: "axonstore",
            source_line: 0,
            source_column: 0,
            name: name.to_string(),
            backend: "postgresql".to_string(),
            connection: connection.to_string(),
            confidence_floor: None,
            isolation: String::new(),
            on_breach: String::new(),
            capability: String::new(),
            class: String::new(),
            column_schema: None,
            resource_ref: String::new(),
        }
    }

    #[test]
    fn block_on_store_runs_a_future_from_a_plain_thread() {
        // The CLI path: `execute_real` runs with no ambient runtime.
        let n = block_on_store(async { 20 + 15 });
        assert_eq!(n, 35);
    }

    /// v2.89.0 — the bridge carries the AMBIENT TENANT across its thread.
    ///
    /// `block_on_store` runs its future on a freshly-spawned OS thread, and a
    /// fresh thread starts with no task-local. Every store op the synchronous
    /// runner performs crosses this boundary, and `storage_postgres.rs` derives
    /// `SET LOCAL axon.current_tenant` from `current_tenant_id()` in 30 places —
    /// so before v2.89.0 that RLS scope resolved to `'default'`, silently,
    /// because the fallback is a plausible string rather than an error.
    ///
    /// This test lives HERE, in the crate, because `block_on_store` is private:
    /// the boundary gate in `tests/tenant_survives_task_boundaries.rs` can only reach the public
    /// primitives, which means it proves `scope_tenant_blocking` works and
    /// proves nothing about whether this function calls it. That gap is the
    /// exact shape of the defect being fixed — a correct mechanism with no
    /// caller — so leaving it would have been ironic and untested.
    #[tokio::test]
    async fn block_on_store_carries_the_ambient_tenant_across_its_thread() {
        let seen = crate::tenant_context::scope_tenant("acme".to_string(), async {
            block_on_store(async { crate::tenant_context::current_tenant_id() })
        })
        .await;
        assert_eq!(
            seen, "acme",
            "the store-op thread must inherit the caller's tenant; `default` here means \
             every RLS scope on the synchronous runner's store path is wrong"
        );
    }

    #[tokio::test]
    async fn block_on_store_runs_a_future_from_within_a_runtime() {
        // The server path: `execute_real` runs on a Tokio worker
        // thread. `block_on_store` must NOT panic with "runtime within
        // a runtime" — it spawns a fresh OS thread that owns its own
        // runtime.
        let n = block_on_store(async { 7 * 6 });
        assert_eq!(n, 42);
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn sql_store_step_surfaces_missing_env_var_never_a_kv_fallback() {
        // The SQL path is reached (routing works) and fails honestly:
        // a postgresql store whose `env:` var is unset yields a typed
        // StoreError — D2's "never a silent KV fallback", proven
        // end-to-end through the sync runner's helper.
        let registry = StoreRegistry::build(&[pg_store(
            "logs",
            "env:AXON_NONEXISTENT_VAR_FASE35E",
        )])
        .unwrap();
        let ctx = ExecContext::new("F", "P", 0);
        let mut pin_map = std::collections::HashMap::new();
        let result = execute_sql_store_step(
            &registry,
            &mut pin_map,
            "retrieve",
            "logs",
            "logs:id = 1",
            None,
            &ctx,
        );
        assert!(matches!(result, Err(StoreError::MissingEnvVar { .. })));
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn sql_persist_below_confidence_floor_is_blocked() {
        // v1.30.0 Pillar I — a store declaring confidence_floor rejects
        // an un-elevated persist (no `_confidence` binding) with a
        // typed epistemic error, before any row is written.
        let mut store = pg_store("ledger", "postgresql://u:p@localhost:5432/db");
        store.confidence_floor = Some(0.8);
        let registry = StoreRegistry::build(&[store]).unwrap();
        let mut ctx = ExecContext::new("F", "P", 0);
        ctx.set("amount", "100"); // a user binding, but no `_confidence`
        let mut pin_map = std::collections::HashMap::new();
        let result =
            execute_sql_store_step(&registry, &mut pin_map, "persist", "ledger", "ledger", None, &ctx);
        assert!(matches!(result, Err(StoreError::Epistemic(_))));
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn sql_store_step_persist_builds_a_row_from_user_bindings() {
        // persist into a postgresql store writes the flow's user
        // bindings as a row. With a malformed DSN the connect fails
        // (typed PoolInit error) — proving persist reaches the SQL
        // path with the bindings-as-row data assembled, not the KV
        // path.
        let registry =
            StoreRegistry::build(&[pg_store("events", "not a dsn")]).unwrap();
        let mut ctx = ExecContext::new("F", "P", 0);
        ctx.set("event_kind", "login");
        let mut pin_map = std::collections::HashMap::new();
        let result =
            execute_sql_store_step(&registry, &mut pin_map, "persist", "events", "events", None, &ctx);
        assert!(matches!(result, Err(StoreError::PoolInit { .. })));
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn sql_persist_scopes_the_row_to_the_declared_field_block() {
        // v1.30.0 — a `persist` carrying a `{ col: value }` block
        // writes EXACTLY those columns (value expressions interpolated
        // against the flow context), ignoring every other binding the
        // flow holds. The malformed DSN fails at connect (typed
        // PoolInit) — proving the field-scoped row was assembled and
        // reached the SQL path. The pre-35.o behaviour would have
        // dumped `message`/`channel_kind`/… into the INSERT.
        let registry =
            StoreRegistry::build(&[pg_store("chat_history", "not a dsn")]).unwrap();
        let mut ctx = ExecContext::new("F", "P", 0);
        ctx.set("message", "hello");
        ctx.set("channel_kind", "whatsapp");
        ctx.set("tenant_id", "acme");
        let fields = vec![
            ("sender".to_string(), "user".to_string()),
            ("content".to_string(), "${message}".to_string()),
            ("tenant_id".to_string(), "${tenant_id}".to_string()),
        ];
        let mut pin_map = std::collections::HashMap::new();
        let result = execute_sql_store_step(
            &registry,
            &mut pin_map,
            "persist",
            "chat_history",
            "chat_history",
            Some(&fields),
            &ctx,
        );
        assert!(matches!(result, Err(StoreError::PoolInit { .. })));
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn sql_mutate_scopes_the_set_to_the_declared_field_block() {
        // v1.30.0 — a `mutate` carrying a `{ col: value }` block
        // builds the UPDATE SET from EXACTLY those columns (value
        // expressions interpolated), ignoring every other binding the
        // flow holds. The malformed DSN fails at connect (typed
        // PoolInit) — proving the field-scoped SET row was assembled
        // and reached the SQL path. The pre-35.p behaviour would have
        // SET `tenant_id` (a flow param, not a column).
        let registry =
            StoreRegistry::build(&[pg_store("accounts", "not a dsn")]).unwrap();
        let mut ctx = ExecContext::new("F", "P", 0);
        ctx.set("tenant_id", "acme"); // a flow param, NOT a column
        ctx.set("new_balance", "500");
        let fields = vec![
            ("balance".to_string(), "${new_balance}".to_string()),
            ("status".to_string(), "active".to_string()),
        ];
        let mut pin_map = std::collections::HashMap::new();
        let result = execute_sql_store_step(
            &registry,
            &mut pin_map,
            "mutate",
            "accounts",
            "accounts:id = 1",
            Some(&fields),
            &ctx,
        );
        assert!(matches!(result, Err(StoreError::PoolInit { .. })));
    }
}


// ─────────────────────────────────────────────────────────────────────────────
// v2.15.0 — unified executor: structural navigate bridge
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod navigate_bridge {
    use super::*;

    /// A dispatcher context shaped like the one `collect_via_dispatcher` builds
    /// for a store-sourced corpus, with nothing behind the stores.
    ///
    /// The two tests below used to drive `dispatch_structural` — the bridge the
    /// legacy executor used to reach a real handler for the handful of verbs
    /// somebody had ported. That bridge is gone with the executor it served
    /// (v4.6.0): there is one engine now, so a verb does not need carrying
    /// across to it. What the tests were protecting is unchanged and still
    /// worth a gate — **an unreachable corpus binds empty rather than
    /// producing documents** — so they now assert it where the work actually
    /// happens.
    fn dispatch_ctx(
        store_sources: std::collections::HashMap<String, crate::ir_nodes::IRCorpusStoreSource>,
    ) -> (
        crate::flow_dispatcher::DispatchCtx,
        tokio::sync::mpsc::UnboundedReceiver<crate::flow_execution_event::FlowExecutionEvent>,
    ) {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let ctx = crate::flow_dispatcher::DispatchCtx::new(
            "F",
            "kimi",
            "",
            crate::cancel_token::CancellationFlag::new(),
            tx,
        )
        .with_store_registry(std::sync::Arc::new(
            crate::store::registry::StoreRegistry::empty(),
        ))
        .with_mdn_corpora(std::sync::Arc::new(std::collections::HashMap::new()))
        .with_mdn_adaptive(std::sync::Arc::new(std::collections::HashSet::new()))
        .with_mdn_store_sources(std::sync::Arc::new(store_sources));
        (ctx, rx)
    }

    /// THE anti-hallucination guarantee (Kivi acceptance, unit scope): a
    /// store-sourced `navigate` whose backing store is NOT Postgres (here an
    /// empty registry) binds an EMPTY result — the honest degrade — instead of
    /// fabricating documents. The full real-rows → real-hits E2E runs in the
    /// Postgres CI lane.
    #[tokio::test]
    async fn store_sourced_navigate_without_postgres_binds_empty_not_hallucinated() {
        let src = crate::ir_nodes::IRCorpusStoreSource {
            doc_store: "LtmSummaries".into(),
            doc_id: "id".into(),
            doc_title: "summary".into(),
            edge_store: "LtmEdges".into(),
            edge_from: "from_id".into(),
            edge_to: "to_id".into(),
            edge_type: "etype".into(),
            edge_weight: "weight".into(),
        };
        let mut store_sources = std::collections::HashMap::new();
        store_sources.insert("LtmGraph".to_string(), src);
        let (mut ctx, _rx) = dispatch_ctx(store_sources);
        let nav = crate::ir_nodes::IRFlowNode::Navigate(crate::ir_nodes::IRNavigateStep {
            depth: None,
            node_type: "navigate",
            source_line: 0,
            source_column: 0,
            pix_ref: "LtmGraph".into(),
            corpus_ref: "LtmGraph".into(),
            query: "prueba de recall".into(),
            trail_enabled: false,
            output_name: "hits".into(),
            seed: String::new(),
            budget: Some(5),
            where_expr: String::new(),
        });
        crate::flow_dispatcher::dispatch_node(&nav, &mut ctx)
            .await
            .expect("an unreachable corpus is an honest empty, not an error");
        assert_eq!(
            ctx.let_bindings.get("hits").map(String::as_str),
            Some(""),
            "empty corpus must bind empty, never fabricate hits"
        );
    }

    /// A `drill` with no indexable PIX source in scope degrades to its
    /// structural placeholder, bound under the drill's `output:` name — a
    /// deterministic value the handler produced, not a sentence about one.
    #[tokio::test]
    async fn drill_without_source_degrades_structurally() {
        let (mut ctx, _rx) = dispatch_ctx(std::collections::HashMap::new());
        let drill = crate::ir_nodes::IRFlowNode::Drill(crate::ir_nodes::IRDrillStep {
            node_type: "drill",
            source_line: 0,
            source_column: 0,
            pix_ref: "Unknown".into(),
            subtree_path: "A.B".into(),
            query: "q".into(),
            output_name: "section".into(),
        });
        let outcome = crate::flow_dispatcher::dispatch_node(&drill, &mut ctx)
            .await
            .expect("drill degrades, it does not error");
        let crate::flow_dispatcher::NodeOutcome::Completed { output, .. } = outcome else {
            panic!("drill completes")
        };
        assert_eq!(
            ctx.let_bindings.get("section").map(String::as_str),
            Some(output.as_str())
        );
    }
    /// v2.15.0 — the FULL Kivi flow shape (navigate → `return hits`) through
    /// the PRODUCTION non-streaming executor (`execute_server_flow` →
    /// `execute_real_async`, backend ≠ "stub"). v2.15.0 fixed the `navigate` step;
    /// this asserts the `return hits` that FOLLOWS resolves to the REAL navigate
    /// output instead of re-HALLUCINATING via the LLM — the second half of the
    /// Kivi gap (its envelope showed `step_names: ["LtmGraph","return"]`, BOTH
    /// LLM-fabricated). A static v2.13.0 corpus keeps it DB-free; neither navigate
    /// (structural) nor return (control flow) calls the LLM, so a DUMMY key
    /// suffices — and if `return` regressed to an LLM step, the dummy key would
    /// surface a backend error instead of the resolved value.
    #[test]
    fn kivi_flow_navigate_then_return_yields_real_hits_not_hallucination() {
        let source = r#"
type DocA { content: Text }
type DocB { content: Text }
corpus G {
    documents: [DocA, DocB]
    relations: [ elaborate(DocA, DocB, 0.9) ]
}
flow Recall(q: Text) -> Text {
    navigate G { query: "${q}", budget: 5, output: hits }
    return hits
}
"#;
        let (_program, ir) =
            crate::flow_plan::compile_source_to_ir(source, "kivi.axon").expect("compile");
        // Bind `q` to a document title so the lexical navigation selects it
        // (deterministic, embeddings-free) — a non-empty REAL result to compare.
        let body = serde_json::json!({ "q": "DocA" });
        let metrics = execute_server_flow(
            &ir,
            "Recall",
            "anthropic", // a real backend name (NOT "stub" → execute_real_async)
            "acme", // v2.49.0 — tenant scope
            "kivi.axon",
            Some("dummy-key"), // no step actually calls the LLM
            Some(&body),
            &std::collections::HashMap::new(),
            &std::collections::HashMap::new(),
            None,
            None,
            None,
            None, // v2.28.0 — budget (test: unbudgeted)
            None, // v2.69.0 — channel semaphores (test: none)
            None, // v2.69.0 — tool leases (test: none)
            None, // v2.31.0 — event outbox (test: no durable sink)
            None, // v2.46.0 — credential minter (test: none)
            None, // v2.48.0 — secret custody (test: none)
                None, // v2.63.0 dataspace_engine (tests: fail closed)
                None, // v2.56.0 scrape_overrides
)
        .expect("flow runs");

        assert!(metrics.success, "the flow succeeds with zero LLM calls");
        assert_eq!(metrics.step_results.len(), 2, "navigate + return");
        let ret = metrics.step_results.last().expect("a return result");
        // THE load-bearing assertions: `return hits` resolves to the navigate
        // step's output (binding lookup) — NOT the LLM "(stub)" hallucination it
        // produced before v2.15.0 (the Kivi envelope's fabricated second step).
        // `step_results[0] == ret` is the proof: it holds iff `return` carried
        // the navigate output, and FAILS if `return` hit the LLM (then the
        // return result would be "(stub)" while the navigate output is its own
        // value). The navigate producing REAL content from real rows is covered
        // separately by the Postgres lane (`navigate_pg_integration::t1`).
        assert_ne!(ret, "(stub)", "`return` must NOT be dispatched to the LLM");
        assert_eq!(
            metrics.step_results[0], *ret,
            "the returned value IS the navigate step's output (hits propagated, \
             not re-fabricated by the LLM)"
        );
    }

    /// v2.15.0 — the REVERSIBLE unified driver: `collect_via_dispatcher` runs
    /// the Kivi flow (navigate → return) through the SAME dispatcher the SSE path
    /// uses, with a buffer sink, and produces the real result. The `return`
    /// carries the navigate output (not "(stub)"); navigate is structural + return
    /// is control flow, so no LLM is called.
    #[tokio::test]
    async fn unified_collector_runs_kivi_flow_via_dispatcher() {
        let source = r#"
type DocA { content: Text }
type DocB { content: Text }
corpus G { documents: [DocA, DocB] relations: [ elaborate(DocA, DocB, 0.9) ] }
flow Recall(q: Text) -> Text {
    navigate G { query: "${q}", budget: 5, output: hits }
    return hits
}
"#;
        let (_p, ir) =
            crate::flow_plan::compile_source_to_ir(source, "k.axon").expect("compile");
        let flow = ir.flows.iter().find(|f| f.name == "Recall").expect("flow");

        let mut corpora = std::collections::HashMap::new();
        for c in &ir.corpus_specs {
            if !c.relations.is_empty() {
                let rels: Vec<(String, String, String, f64)> = c
                    .relations
                    .iter()
                    .map(|r| (r.etype.clone(), r.from.clone(), r.to.clone(), r.weight))
                    .collect();
                if let Ok(corpus) = crate::mdn::Corpus::from_declaration(&c.documents, &rels) {
                    corpora.insert(c.name.clone(), corpus);
                }
            }
        }
        let nd = NavDispatch {
            store_registry: std::sync::Arc::new(crate::store::registry::StoreRegistry::empty()),
            corpora: std::sync::Arc::new(corpora),
            store_sources: std::sync::Arc::new(std::collections::HashMap::new()),
            adaptive: std::sync::Arc::new(std::collections::HashSet::new()),
            dataspace_engine: None,
            scopes: std::sync::Arc::new(Vec::new()),
            observables: std::sync::Arc::new(Vec::new()),
            compute_specs: std::sync::Arc::new(Vec::new()),
            agent_specs: std::sync::Arc::new(Vec::new()),
            mandate_specs: std::sync::Arc::new(Vec::new()),
            lambda_data_specs: std::sync::Arc::new(Vec::new()),
            ots_specs: std::sync::Arc::new(Vec::new()),
            // v2.89.0 — no `cache` declaration in these fixtures, so an
            // empty plan: nothing is memoised and dispatch is unchanged.
            cache_plan: std::sync::Arc::new(crate::cache_runtime::CachePlan::default()),
        };
        let pb = vec![("q".to_string(), "DocA".to_string())];
        let collected = collect_via_dispatcher(
            flow,
            "stub",
            "", // v2.49.0 — tenant_id (test: no custody, empty scope ok)
            "",
            None, // v2.46.0 — default_now_tz (test: no frame zone)
            None,
            None,
            None,
            std::sync::Arc::new(Vec::new()),
            &nd,
            std::sync::Arc::new(ToolRegistry::new()),
            &pb,
            std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
            None, // v2.28.0 — budget (test: unbudgeted)
            None, // v2.69.0 — channel semaphores (test: none)
            None, // v2.69.0 — tool leases (test: none)
            None, // v2.31.0 — event bus (test: no durable sink)
            None, // v2.31.0 — event outbox (test: no durable sink)
            std::sync::Arc::new(std::collections::HashMap::new()), // v2.46.0 — credentials
            None, // v2.46.0 — credential minter (test: none)
            None, // v2.48.0 — secret custody (test: none)
        )
        .await;

        assert!(collected.success, "the flow runs through the dispatcher");
        // navigate (1) + the appended `return` (1).
        assert_eq!(collected.step_names.last().map(String::as_str), Some("return"));
        let ret = collected.step_results.last().expect("a return result");
        assert_ne!(ret, "(stub)", "`return hits` resolves the binding, not the LLM");
        // `return hits` == the navigate step's output (hits propagated).
        assert_eq!(collected.step_results.first(), collected.step_results.last());
    }

    /// v2.60.0 — a flow-body `let` is MATERIALISED (not dropped to the LLM):
    /// its value binds under the target name so `${x}` interpolation AND a downstream
    /// reference resolve it. Pre-105.h the step_type check was `"let_binding"` while
    /// the runtime emits `"let"` (extract_step_info), so a `let` silently fell to the
    /// model. A `let`'s step_name IS its target, so it lands in `step_results` under
    /// the target name — exactly what the v2.60.0 `deliver` resolver reads.
    #[test]
    fn execute_server_flow_materialises_a_let_binding() {
        let source = r#"
flow Lead() -> Text {
    let email = "ada@acme.com"
    return email
}
"#;
        let (_p, ir) =
            crate::flow_plan::compile_source_to_ir(source, "lead.axon").expect("compile");
        let metrics = execute_server_flow(
            &ir,
            "Lead",
            "stub",
            "acme",
            "lead.axon",
            None,                                 // api_key_override
            None,                                 // request_body
            &std::collections::HashMap::new(),    // request_path
            &std::collections::HashMap::new(),    // request_query
            None,                                 // tool_base_url
            None,                                 // llm_base_url
            None,                                 // llm_chat_path
            None,                                 // budget
            None, // v2.69.0 channel semaphores
            None, // v2.69.0 — tool leases (test: none)
            None,                                 // event_outbox
            None,                                 // credential_minter
            None,                                 // secret_custody
            None,                                 // (schemas / etc.)
            None, // v2.63.0 dataspace_engine (tests: fail closed)
        )
        .expect("flow runs");
        assert!(metrics.success, "the let flow runs");
        // The behavioural proof: `return email` resolves the materialised `let` —
        // NOT the LLM stub. (Pre-105.h the `let` fell through to the model because
        // the step_type check was `"let_binding"` ≠ the runtime `"let"`, so `email`
        // was unbound and the return could not resolve it.) The DEPLOYED executor's
        // arm additionally `report.record_step`s the let under its target name, so a
        // `deliver` ref reads it from `step_results` (v2.60.0 resolver).
        let ret = metrics.step_results.last().expect("return value");
        assert_eq!(
            ret, "ada@acme.com",
            "return resolves the materialised let, not the stub (got {ret:?})"
        );
    }

    /// v2.31.0 (Kivi brief #44) — DIRECT repro of the event-consumer binding:
    /// `execute_server_flow` for a flow invoked BY NAME (no top-level `run`, the
    /// daemon-listener path) MUST bind the flow's params from the request body and
    /// make them resolvable as `${param}`. This is exactly what the event drain
    /// does (`deliver_event` passes the event payload as the body). If this fails,
    /// `${session_id_generic}` is empty in the consumer → `where … == ''` → 0 rows.
    #[test]
    fn execute_server_flow_binds_a_param_from_the_body_for_a_named_flow() {
        let source = r#"
flow Echo(p: Text) -> Text {
    return "got ${p}"
}
"#;
        let (_p, ir) =
            crate::flow_plan::compile_source_to_ir(source, "echo.axon").expect("compile");
        let body = serde_json::json!({ "p": "VALUE", "extra": "ignored" });
        let metrics = execute_server_flow(
            &ir,
            "Echo",
            "stub",
            "acme", // v2.49.0 — tenant scope
            "echo.axon",
            None,
            Some(&body),
            &std::collections::HashMap::new(),
            &std::collections::HashMap::new(),
            None,
            None,
            None,
            None, // budget
            None, // v2.69.0 — channel semaphores (test: none)
            None, // v2.69.0 — tool leases (test: none)
            None, // outbox
            None, // v2.46.0 — credential minter (test: none)
            None, // v2.48.0 — secret custody (test: none)
                None, // v2.63.0 dataspace_engine (tests: fail closed)
                None, // v2.56.0 scrape_overrides
)
        .expect("flow runs");
        assert!(metrics.success, "the flow runs");
        let ret = metrics.step_results.last().expect("a return value");
        assert_eq!(
            ret, "got VALUE",
            "the body param `p` must bind + interpolate into the return (got: {ret:?})"
        );
    }

    /// v2.31.0 (Kivi brief #44) — the EXACT consumer shape: TWO params, the
    /// SECOND used downstream, fed the FULL event payload (the row `s`, with extra
    /// fields). Mirrors `LearnFromHibernation(tenant_id, session_id_generic)` over
    /// the `SessionHibernated` payload, to catch a param-order / extra-field /
    /// multi-param binding bug the single-param Echo would miss.
    #[test]
    fn execute_server_flow_binds_the_second_param_from_a_full_event_payload() {
        let source = r#"
flow Learn(tenant_id: Text, session_id_generic: Text) -> Text {
    return "sid=${session_id_generic} tid=${tenant_id}"
}
"#;
        let (_p, ir) =
            crate::flow_plan::compile_source_to_ir(source, "learn.axon").expect("compile");
        // The full row `s` the producer emits — extra fields + the two the flow needs.
        let payload = serde_json::json!({
            "session_id_generic": "e2e-il-test-5",
            "tenant_id": "0e2e51",
            "conversation_id": "fb8659ea",
            "status": "ACTIVE"
        });
        let metrics = execute_server_flow(
            &ir, "Learn", "stub", "acme", "learn.axon", None, Some(&payload),
            &std::collections::HashMap::new(), &std::collections::HashMap::new(),
            None, // tool_base_url
            None, // llm_base_url
            None, // llm_chat_path
            None, // budget
            None, // v2.69.0 channel semaphores
            None, // v2.69.0 — tool leases (test: none)
            None, // event_outbox
            None, // v2.46.0 — credential minter (test: none)
            None, // v2.48.0 — secret custody (test: none)
                None, // v2.63.0 dataspace_engine (tests: fail closed)
                None, // v2.56.0 scrape_overrides
)
        .expect("flow runs");
        assert!(metrics.success);
        let ret = metrics.step_results.last().expect("a return value");
        assert_eq!(
            ret, "sid=e2e-il-test-5 tid=0e2e51",
            "BOTH params must bind from the full payload by name (got: {ret:?})"
        );
    }

    /// v2.31.0 — the headline of the OSS producer wiring: when a durable
    /// `event_outbox` is injected into `execute_server_flow` (the enterprise
    /// daemon path), a flow's `emit` to a `persistent_axonstore` channel APPENDS
    /// to that outbox — durably — instead of buffering in-process. This proves the
    /// runner builds the typed bus from the program's `channel` defs + attaches
    /// the bus+outbox pair so `run_emit` routes the persistent emit to the outbox.
    /// Without the outbox (`None`) the emit stays in-process (the pre-v2.31.0 path,
    /// covered by the other runner tests).
    #[test]
    fn execute_server_flow_appends_a_persistent_emit_to_the_injected_outbox() {
        use crate::event_outbox::{EventOutbox, InMemoryEventOutbox};
        let source = r#"
type Hib { tenant_id: Text }
channel HibCh { message: Hib  qos: at_least_once  persistence: persistent_axonstore }
flow Producer(tenant_id: Text) -> Text {
    emit HibCh(tenant_id)
    return "emitted"
}
"#;
        let (_p, ir) =
            crate::flow_plan::compile_source_to_ir(source, "producer.axon").expect("compile");
        let body = serde_json::json!({ "tenant_id": "acme" });
        // Keep a concrete handle (`probe`) to read the outbox back after the run;
        // inject a trait-object clone of the SAME outbox into the runner.
        let probe = std::sync::Arc::new(InMemoryEventOutbox::new());

        let metrics = execute_server_flow(
            &ir,
            "Producer",
            "stub",
            "acme", // v2.49.0 — tenant scope
            "producer.axon",
            None,
            Some(&body),
            &std::collections::HashMap::new(),
            &std::collections::HashMap::new(),
            None,
            None,
            None,
            None, // v2.28.0 — budget (unbudgeted)
            None, // v2.69.0 — channel semaphores (test: none)
            None, // v2.69.0 — tool leases (test: none)
            // v2.31.0 — inject the durable outbox (the enterprise daemon path).
            Some(probe.clone() as std::sync::Arc<dyn EventOutbox>),
            None, // v2.46.0 — credential minter (test: none)
            None, // v2.48.0 — secret custody (test: none)
                None, // v2.63.0 dataspace_engine (tests: fail closed)
                None, // v2.56.0 scrape_overrides
)
        .expect("flow runs");

        assert!(metrics.success, "the producer flow runs to completion");
        // THE assertion: the persistent-channel emit landed in the durable outbox
        // (1 unprocessed event), not the in-process buffer.
        assert_eq!(
            probe.pending_total(),
            1,
            "a `persistent_axonstore` emit must APPEND to the injected outbox"
        );
        let tail = probe.unprocessed("HibCh");
        assert_eq!(tail.len(), 1, "the event is on HibCh's redelivery tail");
        assert_eq!(
            tail[0].payload,
            serde_json::Value::String("acme".to_string()),
            "the emitted payload (the bound `tenant_id`) is recorded verbatim"
        );
    }

}