flow_execution/

executor.rs

use chrono::Utc;
use flow_adapter_ai::{CloudAiRegistry, CloudAiRequest, LlmStreamSink};
#[cfg(test)]
use flow_adapter_ai::NullStreamSink;
use flow_domain::contract::{AiNodeContract, AiOutputEnvelope, RouteDecision};
use flow_domain::graph::{EdgeOutcome, FlowEdge, FlowGraph, FlowNode, Position, SubFlow};
use flow_security::{CredentialError, CredentialResolver, PiiSanitizer};
use std::collections::{HashMap, HashSet, VecDeque};
use std::path::PathBuf;
use std::sync::Arc;
use thiserror::Error;
use uuid::Uuid;

use crate::adapter::AdapterRegistry;
use crate::events::{EventSink, ExecutionEvent, LogStream};

#[derive(Debug, Error)]
pub enum ExecutorError {
    #[error("graph contains a cycle involving node {0}")]
    Cycle(String),
    #[error("node {node_id} references unknown adapter")]
    MissingAdapter { node_id: String },
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ExecutionSummary {
    pub execution_id: String,
    pub status: String,
    pub succeeded: usize,
    pub failed: usize,
    pub skipped: usize,
}

pub struct Executor {
    pub adapters: Arc<AdapterRegistry>,
    pub sanitizer: Arc<PiiSanitizer>,
    pub events: Arc<dyn EventSink>,
    pub cloud_providers: Arc<CloudAiRegistry>,
    pub credentials: Arc<dyn CredentialResolver>,
    pub allow_cloud_ai: bool,
    /// Gate for the `local` provider (on-device OpenAI-compatible server).
    /// Separate from `allow_cloud_ai` because a localhost call is not
    /// network egress.
    pub allow_local_ai: bool,
    /// Endpoint for the `local` provider, sourced from settings. Passed
    /// through as `CloudAiRequest.base_url` for local-provider nodes.
    pub local_ai_base_url: Option<String>,
    /// Fan-out target for streaming LLM tokens. Wired by `FlowApp` so
    /// every AI (local/cloud) and agentic node invocation feeds the
    /// header chip ticker via Tauri's `flow:llm_token` event. Headless
    /// tests use the default `NullStreamSink`.
    pub stream_sink: Arc<dyn LlmStreamSink>,
    /// Session-scoped key→JSON working memory, carried across runs by `FlowApp`
    /// so re-plan iterations see prior results. Succeeded `set-variable` utility
    /// nodes write it; any field (and `when` conditions) read it via
    /// `{{memory.<key>}}`. Headless tests pass a fresh empty store.
    pub working_memory: WorkingMemory,
    /// Out-of-band pause/resume/cancel signal, polled at node boundaries.
    /// Shared with `FlowApp` so the Tauri pause/resume/stop commands can steer
    /// an in-progress run. Headless tests use the default (always `Running`).
    pub control: SharedRunControl,
    /// Per-step destructive-action confirmation gate (roadmap E1). When `true`,
    /// the executor pauses before running a node that performs a destructive
    /// operation (file delete, `rm`, `git push`, …), emits
    /// [`ExecutionEvent::AwaitingConfirmation`], and blocks until the user
    /// confirms (resume) or cancels (stop). Headless runners leave it `false`
    /// so a destructive node never blocks on a confirmation that can't arrive.
    pub confirm_destructive: bool,
    /// Whether a host capable of resolving an AI review gate is attached.
    /// Headless runners (CLI / TUI) set this `false`: a contract-bound output
    /// in the human review band then fails onto its fallback path instead of
    /// holding the run on a gate no one can resolve - RAO's deterministic
    /// fallback, never a silent auto-approve.
    pub review_gate_available: bool,
    /// When `Some`, AI-node tool calls that mutate the filesystem (`fs`
    /// write/edit/delete) are *staged* into this buffer - computed but not
    /// written - so a coding-agent turn can be reviewed before its edits land.
    /// Read-only tools and every `shell` tool still run for real, and a staged
    /// file is visible to later `read-file` calls in the same turn.
    /// [`FlowApp::run_agent_turn`] installs it; every other run leaves it `None`.
    pub edit_staging: Option<crate::ai_tools::StagedEdits>,
    /// Resolved workspace root for this run, forwarded to every adapter via
    /// [`AdapterCtx::workspace_root`]. `FlowApp` resolves it (per-run override,
    /// stored per-flow path, or edition default) before the run starts.
    pub workspace_root: PathBuf,
}

/// Session-scoped store backing `{{memory.<key>}}`. Shared (`Arc`) so it
/// survives across `Executor::run` calls within a `FlowApp` session, and
/// interior-mutable (`Mutex`) so a node can write it through `&self`.
pub type WorkingMemory = Arc<std::sync::Mutex<HashMap<String, serde_json::Value>>>;

/// Run phase used by [`RunControl`]. Pause and cancel are honored at node
/// boundaries (the executor checkpoints between node executions), so an
/// in-flight node finishes before the request takes effect.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunPhase {
    Running,
    Paused,
    Cancelling,
}

/// Out-of-band pause/resume/cancel control for a run, shared (`Arc`) between
/// `FlowApp` (which the host's pause/resume/stop commands signal) and the
/// `Executor` (which polls it at node boundaries). One instance is created per
/// run and registered under that run's execution id, so several flows (e.g.
/// multiple canvas tabs) can run and be steered independently.
pub struct RunControl {
    phase: std::sync::atomic::AtomicU8,
    /// Woken on `resume`/`cancel` so a paused checkpoint stops awaiting.
    notify: tokio::sync::Notify,
    /// Pending human decision for an AI review gate (RAO constraint 5). Set by
    /// [`RunControl::resolve_review`] just before the resume that releases the
    /// gate; taken by the executor when the gate wakes.
    review: std::sync::Mutex<Option<ReviewDecision>>,
}

/// Human verdict on a contract-bound AI output held at a review gate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReviewDecision {
    Approved,
    Rejected,
}

impl Default for RunControl {
    fn default() -> Self {
        Self {
            phase: std::sync::atomic::AtomicU8::new(Self::RUNNING),
            notify: tokio::sync::Notify::new(),
            review: std::sync::Mutex::new(None),
        }
    }
}

impl RunControl {
    const RUNNING: u8 = 0;
    const PAUSED: u8 = 1;
    const CANCELLING: u8 = 2;

    fn set(&self, v: u8) {
        self.phase.store(v, std::sync::atomic::Ordering::SeqCst);
    }

    pub fn phase(&self) -> RunPhase {
        match self.phase.load(std::sync::atomic::Ordering::SeqCst) {
            Self::PAUSED => RunPhase::Paused,
            Self::CANCELLING => RunPhase::Cancelling,
            _ => RunPhase::Running,
        }
    }

    /// Reset to `Running` at the start of a run so a stale cancel/pause from a
    /// prior run can't leak into the next one.
    pub fn reset(&self) {
        self.set(Self::RUNNING);
        // Drain any pending permit so a fresh run doesn't skip its first wait.
        self.notify.notify_waiters();
    }

    /// Request a pause. No-op if already cancelling.
    pub fn pause(&self) {
        let _ = self.phase.compare_exchange(
            Self::RUNNING,
            Self::PAUSED,
            std::sync::atomic::Ordering::SeqCst,
            std::sync::atomic::Ordering::SeqCst,
        );
    }

    /// Release a pause. No-op if cancelling.
    pub fn resume(&self) {
        let _ = self.phase.compare_exchange(
            Self::PAUSED,
            Self::RUNNING,
            std::sync::atomic::Ordering::SeqCst,
            std::sync::atomic::Ordering::SeqCst,
        );
        self.notify.notify_waiters();
    }

    /// Request cancellation; takes precedence over pause and wakes any waiter.
    pub fn cancel(&self) {
        self.set(Self::CANCELLING);
        self.notify.notify_waiters();
    }

    pub fn is_cancelling(&self) -> bool {
        self.phase() == RunPhase::Cancelling
    }

    /// Record a human verdict for the AI review gate and release the pause.
    /// The executor picks the decision up via [`RunControl::take_review`].
    pub fn resolve_review(&self, approved: bool) {
        if let Ok(mut slot) = self.review.lock() {
            *slot = Some(if approved {
                ReviewDecision::Approved
            } else {
                ReviewDecision::Rejected
            });
        }
        self.resume();
    }

    /// Take the pending review decision, if any, clearing the slot.
    pub fn take_review(&self) -> Option<ReviewDecision> {
        self.review.lock().ok().and_then(|mut slot| slot.take())
    }

    /// Block while the phase is `Paused`, returning once resumed or cancelling.
    pub async fn wait_while_paused(&self) {
        while self.phase() == RunPhase::Paused {
            self.notify.notified().await;
        }
    }
}

/// Shared run control, mirroring [`WorkingMemory`]'s `Arc` sharing pattern.
pub type SharedRunControl = Arc<RunControl>;

/// Hard caps that guarantee a graph with cycles terminates. The per-node cap
/// bounds re-entries of any single node; the global cap is a backstop across
/// all node executions. Exceeding either ends the run as `partial` (an
/// `IterationCapped` event marks the node) rather than aborting, so prior
/// outputs are preserved.
const PER_NODE_VISIT_CAP: u32 = 25;
const GLOBAL_STEP_CAP: u32 = 500;

impl Executor {
    pub async fn run(&self, graph: &FlowGraph) -> Result<ExecutionSummary, ExecutorError> {
        self.run_with_id(Uuid::new_v4().to_string(), graph).await
    }

    /// Like [`Executor::run`] but uses a caller-supplied execution id. The host
    /// provides the id so it can associate the run with a specific tab/flow and
    /// steer it (pause / resume / cancel) immediately - before the `Started`
    /// event arrives - which is what makes concurrent multi-tab runs routable.
    pub async fn run_with_id(
        &self,
        execution_id: String,
        graph: &FlowGraph,
    ) -> Result<ExecutionSummary, ExecutorError> {
        // Clear any stale pause/cancel from a prior run before scheduling.
        self.control.reset();
        self.events.emit(ExecutionEvent::Started {
            execution_id: execution_id.clone(),
            at: Utc::now(),
        });

        // Per-node terminal status (drives outgoing-edge fire decisions) and
        // succeeded-node output payloads (consumed by `{{...}}` interpolation
        // for inter-node data flow). On a loop re-run, both are overwritten so
        // references resolve to the latest pass.
        let mut terminal: HashMap<String, NodeStatus> = HashMap::new();
        let mut outputs: HashMap<String, serde_json::Value> = HashMap::new();

        // Acyclic graphs take the proven single-pass path (behaviour
        // unchanged); a detected cycle switches to the bounded-iteration
        // scheduler. `topo_sort` itself still reports cycles as an error so it
        // stays a reusable pure helper.
        let capped = if graph.subflows.is_empty() {
            match topo_sort(graph) {
                Ok(order) => {
                    self.run_acyclic(graph, &order, &execution_id, &mut outputs, &mut terminal)
                        .await;
                    false
                }
                Err(ExecutorError::Cycle(_)) => {
                    self.run_iterative(graph, &execution_id, &mut outputs, &mut terminal)
                        .await
                }
                Err(e) => return Err(e),
            }
        } else {
            // Sub-flows: schedule over a collapsed outer graph (each sub-flow is
            // one virtual node) so each unit runs atomically and in dependency
            // order, then runs its members as a retry-able unit.
            let outer = collapse_subflows(graph);
            match topo_sort(&outer) {
                Ok(order) => {
                    self.run_acyclic_units(
                        graph,
                        &order,
                        &execution_id,
                        &mut outputs,
                        &mut terminal,
                    )
                    .await;
                    false
                }
                Err(ExecutorError::Cycle(_)) => {
                    // Cyclic outer graph: fall back to the flat iterative
                    // scheduler (sub-flows run as ordinary nodes, no unit retry).
                    self.run_iterative(graph, &execution_id, &mut outputs, &mut terminal)
                        .await
                }
                Err(e) => return Err(e),
            }
        };

        let mut succeeded = 0usize;
        let mut failed = 0usize;
        let mut skipped = 0usize;
        for status in terminal.values() {
            match status {
                NodeStatus::Succeeded => succeeded += 1,
                NodeStatus::Failed => failed += 1,
                NodeStatus::Skipped => skipped += 1,
            }
        }

        let status = if self.control.is_cancelling() {
            "cancelled"
        } else if capped {
            "partial"
        } else if failed > 0 {
            if succeeded > 0 || skipped > 0 {
                "partial"
            } else {
                "failed"
            }
        } else if skipped > 0 && succeeded == 0 {
            "skipped"
        } else {
            "succeeded"
        };

        self.events.emit(ExecutionEvent::Done {
            execution_id: execution_id.clone(),
            at: Utc::now(),
            status: status.into(),
        });

        Ok(ExecutionSummary {
            execution_id,
            status: status.into(),
            succeeded,
            failed,
            skipped,
        })
    }

    /// Honor a pending pause/cancel at a node boundary. While paused, blocks
    /// until resumed or cancelled (emitting `Paused`/`Resumed` so the UI can
    /// confirm the in-flight node finished). Returns `true` when the run should
    /// stop scheduling because cancellation was requested.
    async fn checkpoint(&self, execution_id: &str) -> bool {
        match self.control.phase() {
            RunPhase::Running => false,
            RunPhase::Cancelling => true,
            RunPhase::Paused => {
                self.events.emit(ExecutionEvent::Paused {
                    execution_id: execution_id.to_string(),
                    at: Utc::now(),
                });
                self.control.wait_while_paused().await;
                if self.control.is_cancelling() {
                    true
                } else {
                    self.events.emit(ExecutionEvent::Resumed {
                        execution_id: execution_id.to_string(),
                        at: Utc::now(),
                    });
                    false
                }
            }
        }
    }

    /// Snapshot the shared working memory for read-only interpolation. Cloned
    /// per use so a node always sees writes made by earlier nodes this run.
    fn snapshot_memory(&self) -> HashMap<String, serde_json::Value> {
        self.working_memory
            .lock()
            .map(|m| m.clone())
            .unwrap_or_default()
    }

    /// If a succeeded node was a `set-variable` utility action, persist its
    /// `name`→`value` into working memory. The adapter wraps its result as
    /// `{ status, payload }`, so the action fields live under `payload`.
    fn store_set_variable(&self, output: &serde_json::Value) {
        let inner = output.get("payload").unwrap_or(output);
        if inner.get("actionId").and_then(|v| v.as_str()) != Some("set-variable") {
            return;
        }
        let Some(name) = inner.get("name").and_then(|v| v.as_str()) else {
            return;
        };
        let value = inner.get("value").cloned().unwrap_or(serde_json::Value::Null);
        if let Ok(mut mem) = self.working_memory.lock() {
            mem.insert(name.to_string(), value);
        }
    }

    /// Single forward pass over a topologically sorted DAG. A node runs iff it
    /// has no incoming edges or at least one incoming edge is active; otherwise
    /// it is skipped. Identical in behaviour to the pre-iteration executor.
    async fn run_acyclic(
        &self,
        graph: &FlowGraph,
        order: &[String],
        execution_id: &str,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) {
        let nodes_by_id: HashMap<&str, &FlowNode> =
            graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();

        for node_id in order {
            // Honor pause/cancel between nodes. On cancel, stop scheduling and
            // fall through to the skip sweep below.
            if self.checkpoint(execution_id).await {
                break;
            }
            let Some(node) = nodes_by_id.get(node_id.as_str()).copied() else {
                continue;
            };
            self.run_one_node(graph, node, execution_id, 0, outputs, terminal)
                .await;
        }

        // Nodes never reached (e.g. a cancelled run broke out of the loop) are
        // marked skipped, matching the iterative path's reachability sweep.
        for node in &graph.nodes {
            if !terminal.contains_key(&node.id) {
                terminal.insert(node.id.clone(), NodeStatus::Skipped);
                self.events.emit(ExecutionEvent::NodeSkipped {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    reason: "run cancelled before this node was scheduled".into(),
                    iteration: 0,
                });
            }
        }
    }

    /// Run a single node: evaluate reachability from its incoming edges and
    /// either execute it or mark it skipped. Shared by the flat acyclic pass
    /// and each sub-flow's inner pass.
    async fn run_one_node(
        &self,
        graph: &FlowGraph,
        node: &FlowNode,
        execution_id: &str,
        iteration: u32,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) {
        let in_edges: Vec<_> = graph.edges.iter().filter(|e| e.target == node.id).collect();
        let memory = self.snapshot_memory();
        let reachable = in_edges.is_empty()
            || in_edges.iter().any(|e| edge_active(e, terminal, outputs, &memory));

        if !reachable {
            terminal.insert(node.id.clone(), NodeStatus::Skipped);
            self.events.emit(ExecutionEvent::NodeSkipped {
                execution_id: execution_id.to_string(),
                node_id: node.id.clone(),
                at: Utc::now(),
                reason: "no incoming edge fired (upstream did not pass/fail to here)".into(),
                iteration,
            });
            return;
        }

        self.execute_node(graph, node, execution_id, iteration, outputs, terminal)
            .await;
    }

    /// Forward pass over an outer topo order whose entries are free node ids and
    /// collapsed sub-flow ids. Free nodes run individually; each sub-flow runs
    /// as one unit (see [`Self::run_subflow_unit`]).
    async fn run_acyclic_units(
        &self,
        graph: &FlowGraph,
        outer_order: &[String],
        execution_id: &str,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) {
        let nodes_by_id: HashMap<&str, &FlowNode> =
            graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();

        for outer_id in outer_order {
            if self.checkpoint(execution_id).await {
                break;
            }
            if let Some(sf) = graph.subflows.iter().find(|s| &s.id == outer_id) {
                self.run_subflow_unit(graph, sf, execution_id, outputs, terminal)
                    .await;
            } else if let Some(node) = nodes_by_id.get(outer_id.as_str()).copied() {
                self.run_one_node(graph, node, execution_id, 0, outputs, terminal)
                    .await;
            }
        }

        // Anything never reached (cancelled run, or an unreachable branch).
        for node in &graph.nodes {
            if !terminal.contains_key(&node.id) {
                terminal.insert(node.id.clone(), NodeStatus::Skipped);
                self.events.emit(ExecutionEvent::NodeSkipped {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    reason: "no incoming edge fired (upstream did not pass/fail to here)".into(),
                    iteration: 0,
                });
            }
        }
    }

    /// Run one sub-flow as an execution unit: gate on its boundary-crossing
    /// incoming edges, then run its members in inner-topo order. If any member
    /// fails, re-run the whole unit up to `retry` times. Member terminal/outputs
    /// land in the shared maps, so the surrounding flat edges (which reference
    /// member ids directly) fire normally.
    async fn run_subflow_unit(
        &self,
        graph: &FlowGraph,
        sf: &SubFlow,
        execution_id: &str,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) {
        let members: HashSet<&str> = sf.node_ids.iter().map(|s| s.as_str()).collect();

        // Reachability: a boundary-crossing edge into the unit must fire (or
        // there are none, making it an entry unit).
        let cross_in: Vec<&FlowEdge> = graph
            .edges
            .iter()
            .filter(|e| members.contains(e.target.as_str()) && !members.contains(e.source.as_str()))
            .collect();
        let memory = self.snapshot_memory();
        let reachable = cross_in.is_empty()
            || cross_in
                .iter()
                .any(|e| edge_active(e, terminal, outputs, &memory));
        if !reachable {
            for id in &sf.node_ids {
                if !terminal.contains_key(id) {
                    terminal.insert(id.clone(), NodeStatus::Skipped);
                    self.events.emit(ExecutionEvent::NodeSkipped {
                        execution_id: execution_id.to_string(),
                        node_id: id.clone(),
                        at: Utc::now(),
                        reason: format!("sub-flow \"{}\" not reached", sf.label),
                        iteration: 0,
                    });
                }
            }
            return;
        }

        // Inner topo order over members + their internal edges.
        let inner_graph = FlowGraph {
            id: sf.id.clone(),
            name: sf.label.clone(),
            version: "1.0.0".into(),
            description: None,
            nodes: graph
                .nodes
                .iter()
                .filter(|n| members.contains(n.id.as_str()))
                .cloned()
                .collect(),
            edges: graph
                .edges
                .iter()
                .filter(|e| {
                    members.contains(e.source.as_str()) && members.contains(e.target.as_str())
                })
                .cloned()
                .collect(),
            subflows: Vec::new(),
        };
        let inner_order = topo_sort(&inner_graph).unwrap_or_else(|_| sf.node_ids.clone());

        let nodes_by_id: HashMap<&str, &FlowNode> =
            graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
        let max_attempts = 1 + sf.retry.unwrap_or(0);
        for attempt in 0..max_attempts {
            for nid in &inner_order {
                if self.checkpoint(execution_id).await {
                    return;
                }
                if let Some(node) = nodes_by_id.get(nid.as_str()).copied() {
                    self.run_one_node(graph, node, execution_id, attempt, outputs, terminal)
                        .await;
                }
            }
            let failed = sf
                .node_ids
                .iter()
                .any(|id| matches!(terminal.get(id), Some(NodeStatus::Failed)));
            if !failed || attempt + 1 >= max_attempts {
                break;
            }
            // Retry the whole unit: clear member state so the next pass re-runs
            // from the unit's entry.
            for id in &sf.node_ids {
                terminal.remove(id);
                outputs.remove(id);
            }
        }
    }

    /// Bounded worklist scheduler for graphs with cycles. A node re-runs each
    /// time an active edge re-arrives; per-node and global caps guarantee
    /// termination. Returns `true` if either cap was hit.
    async fn run_iterative(
        &self,
        graph: &FlowGraph,
        execution_id: &str,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) -> bool {
        let nodes_by_id: HashMap<&str, &FlowNode> =
            graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();

        // Seed with entry nodes (no incoming edges), in stable graph order.
        let mut queue: VecDeque<String> = graph
            .nodes
            .iter()
            .filter(|n| !graph.edges.iter().any(|e| e.target == n.id))
            .map(|n| n.id.clone())
            .collect();

        let mut visits: HashMap<String, u32> = HashMap::new();
        let mut global_steps = 0u32;
        let mut capped = false;

        while let Some(node_id) = queue.pop_front() {
            // Honor pause/cancel between nodes. On cancel, stop scheduling; the
            // skip sweep below marks everything still pending.
            if self.checkpoint(execution_id).await {
                queue.clear();
                break;
            }
            if global_steps >= GLOBAL_STEP_CAP {
                capped = true;
                break;
            }
            let Some(node) = nodes_by_id.get(node_id.as_str()).copied() else {
                continue;
            };

            let visit = visits.entry(node_id.clone()).or_insert(0);
            if *visit >= PER_NODE_VISIT_CAP {
                self.events.emit(ExecutionEvent::IterationCapped {
                    execution_id: execution_id.to_string(),
                    node_id: node_id.clone(),
                    at: Utc::now(),
                    visits: *visit,
                });
                capped = true;
                continue;
            }
            let iteration = *visit;
            *visit += 1;
            global_steps += 1;

            self.execute_node(graph, node, execution_id, iteration, outputs, terminal)
                .await;

            // Enqueue every target whose edge from this node is now active.
            let memory = self.snapshot_memory();
            for edge in graph.edges.iter().filter(|e| e.source == node_id) {
                if edge_active(edge, terminal, outputs, &memory) {
                    queue.push_back(edge.target.clone());
                }
            }
        }

        // Nodes the worklist never reached are skipped, matching the acyclic
        // path's reachability semantics.
        for node in &graph.nodes {
            if !terminal.contains_key(&node.id) {
                terminal.insert(node.id.clone(), NodeStatus::Skipped);
                self.events.emit(ExecutionEvent::NodeSkipped {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    reason: "no incoming edge fired (upstream did not pass/fail to here)".into(),
                    iteration: 0,
                });
            }
        }

        capped
    }

    /// Interpolate this node's data against prior outputs, dispatch by node
    /// type, and record the outcome - emitting lifecycle events tagged with
    /// `iteration`. Shared by both schedulers.
    async fn execute_node(
        &self,
        graph: &FlowGraph,
        node: &FlowNode,
        execution_id: &str,
        iteration: u32,
        outputs: &mut HashMap<String, serde_json::Value>,
        terminal: &mut HashMap<String, NodeStatus>,
    ) {
        // Per-step destructive-action confirmation gate (roadmap E1). Inspected
        // before the node runs (and before NodeStarted) so a destructive op
        // never executes without an explicit confirm. A cancel here skips the
        // node and the next checkpoint stops scheduling.
        if self.confirm_destructive {
            if let Some(action) = destructive_reason(node) {
                if !self.control.is_cancelling() {
                    self.events.emit(ExecutionEvent::AwaitingConfirmation {
                        execution_id: execution_id.to_string(),
                        node_id: node.id.clone(),
                        action,
                        at: Utc::now(),
                    });
                    self.control.pause();
                    self.control.wait_while_paused().await;
                }
                if self.control.is_cancelling() {
                    terminal.insert(node.id.clone(), NodeStatus::Skipped);
                    self.events.emit(ExecutionEvent::NodeSkipped {
                        execution_id: execution_id.to_string(),
                        node_id: node.id.clone(),
                        at: Utc::now(),
                        reason: "destructive action not confirmed".into(),
                        iteration,
                    });
                    return;
                }
            }
        }

        self.events.emit(ExecutionEvent::NodeStarted {
            execution_id: execution_id.to_string(),
            node_id: node.id.clone(),
            at: Utc::now(),
            iteration,
        });

        // Inter-node data flow: `{{input}}` resolves to the output(s) of the
        // upstream node(s) whose edge fired into this node; `{{<id>}}` /
        // `{{<id>.<field>}}` reference any already-run node by id;
        // `{{memory.<key>}}` reads session working memory.
        let memory = self.snapshot_memory();
        let fired_sources: Vec<String> = graph
            .edges
            .iter()
            .filter(|e| e.target == node.id && edge_active(e, terminal, outputs, &memory))
            .map(|e| e.source.clone())
            .collect();
        let mut resolved = node.clone();
        resolved.data = interpolate_value(&resolved.data, outputs, &fired_sources, &memory);
        let node = &resolved;

        let outcome = match node.node_type.as_str() {
            // Unified AI node: `provider` selects local (on-device) vs cloud.
            "ai" => self.run_ai_node(node, execution_id).await,
            // Agentic nodes are design-time only - they generate a flow for review
            // and never execute as a runtime node. Skip one if it reaches here.
            "agentic" => NodeOutcome::Skipped("agentic nodes are design-time only".into()),
            // action / utility / service all dispatch through their adapter.
            _ => self.run_adapter_node(node, execution_id).await,
        };

        match outcome {
            NodeOutcome::Succeeded(payload) => {
                terminal.insert(node.id.clone(), NodeStatus::Succeeded);
                self.store_set_variable(&payload);
                outputs.insert(node.id.clone(), payload.clone());
                self.events.emit(ExecutionEvent::NodeSucceeded {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    output: payload,
                    iteration,
                });
            }
            NodeOutcome::Skipped(reason) => {
                terminal.insert(node.id.clone(), NodeStatus::Skipped);
                self.events.emit(ExecutionEvent::NodeSkipped {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    reason,
                    iteration,
                });
            }
            NodeOutcome::Failed(error) => {
                terminal.insert(node.id.clone(), NodeStatus::Failed);
                self.events.emit(ExecutionEvent::NodeFailed {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    error,
                    iteration,
                });
            }
        }
    }

    /// Unified AI node. The `provider` field selects the backend: `local`
    /// (default) runs an LLM on the managed `llama-server` (zero egress); any
    /// other value is a cloud provider, gated by `allow_cloud_ai` + an API key.
    async fn run_ai_node(&self, node: &FlowNode, execution_id: &str) -> NodeOutcome {
        let outcome = self.run_ai_node_once(node, execution_id).await;
        // Per-node failover (roadmap E1 model routing & fallback): on a hard
        // failure, retry once with the node's configured fallback provider /
        // model. Only `Failed` triggers it - a `Skipped` (policy gate) is not a
        // transient error worth retrying.
        if let NodeOutcome::Failed(err) = &outcome {
            if let Some(fb) = fallback_node(node) {
                let fb_provider = fb
                    .data
                    .get("provider")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
                    .to_string();
                self.events.emit(ExecutionEvent::NodeLog {
                    execution_id: execution_id.to_string(),
                    node_id: node.id.clone(),
                    at: Utc::now(),
                    stream: LogStream::Stderr,
                    line: format!("AI provider failed ({err}); falling back to {fb_provider}"),
                });
                return self.run_ai_node_once(&fb, execution_id).await;
            }
        }
        outcome
    }

    /// One AI-node attempt: resolve the provider (`local` vs cloud) and dispatch.
    /// [`run_ai_node`] wraps this with per-node failover.
    async fn run_ai_node_once(&self, node: &FlowNode, execution_id: &str) -> NodeOutcome {
        let model_id = node
            .data
            .get("modelId")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        if model_id.is_empty() {
            return NodeOutcome::Failed(
                "no model selected on ai node (set modelId to a loaded model)".into(),
            );
        }

        let provider = node
            .data
            .get("provider")
            .and_then(|v| v.as_str())
            .unwrap_or("local");
        if provider.is_empty() || provider == "local" {
            self.run_local_llm_ai_node(node, &model_id, execution_id).await
        } else {
            self.run_cloud_ai_node(node, execution_id).await
        }
    }

    /// AI-node path for LLM chat models hosted by the managed
    /// `llama-server` subprocess. Mirrors the local branch of
    /// `run_cloud_ai_node` but reads `input` (the AI-node convention)
    /// instead of `prompt`, and opts in to advanced sampling params
    /// (`top_p`/`top_k`/`stop`) sourced from the inspector.
    ///
    /// The AI node has no `provider` field - the implicit provider is
    /// `local`, so we always gate on `allow_local_ai` and route through
    /// `LocalOpenAiProvider`. Confidence thresholds are ignored
    /// (LLM chat responses don't carry one); the outcome is `Succeeded`
    /// with the assistant text wrapped in a small JSON envelope so the
    /// History panel can display it without parsing.
    async fn run_local_llm_ai_node(
        &self,
        node: &FlowNode,
        model_id: &str,
        execution_id: &str,
    ) -> NodeOutcome {
        if !self.allow_local_ai {
            return NodeOutcome::Skipped(
                "local AI disabled by policy (settings.allow_local_ai = false)".into(),
            );
        }
        let provider = match self.cloud_providers.get("local") {
            Some(p) => p,
            None => {
                return NodeOutcome::Skipped(
                    "local AI provider not registered in CloudAiRegistry".into(),
                );
            }
        };
        if self.local_ai_base_url.is_none() {
            return NodeOutcome::Failed(
                "local AI base URL not set; load a model from the Model Hub first".into(),
            );
        }

        let input = match node
            .data
            .get("input")
            .and_then(|v| v.as_str())
            .filter(|s| !s.trim().is_empty())
        {
            Some(s) => s.to_string(),
            None => return NodeOutcome::Failed("missing 'input' on ai node (local LLM)".into()),
        };
        let system = node
            .data
            .get("system")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(str::to_owned);
        let max_tokens = node
            .data
            .get("maxTokens")
            .and_then(|v| v.as_u64())
            .map(|v| v as u32);
        let temperature = node
            .data
            .get("temperature")
            .and_then(|v| v.as_f64())
            .map(|v| v as f32);
        let top_p = node
            .data
            .get("topP")
            .and_then(|v| v.as_f64())
            .map(|v| v as f32);
        let top_k = node
            .data
            .get("topK")
            .and_then(|v| v.as_u64())
            .map(|v| v as u32);
        let stop = node.data.get("stop").and_then(|v| v.as_array()).map(|arr| {
            arr.iter()
                .filter_map(|x| x.as_str().map(str::to_owned))
                .collect::<Vec<_>>()
        });

        // Optional credential lookup mirrors `run_cloud_ai_node`: local
        // servers usually need no auth, but if the user stored a key
        // (e.g. for a remote OpenAI-compat endpoint) we forward it.
        let api_key = match self
            .credentials
            .resolve_cloud_ai_key("local", provider.env_var())
        {
            Ok(k) => k,
            Err(CredentialError::NotFound { .. }) => String::new(),
            Err(e) => return NodeOutcome::Failed(format!("credential lookup failed: {e}")),
        };

        // RAO node contract (docs/rao-spec): an unreadable contract is a node
        // failure before any inference happens.
        let contract = match AiNodeContract::from_node_data(&node.data) {
            Ok(c) => c,
            Err(e) => return NodeOutcome::Failed(e.to_string()),
        };

        let sanitized = self.sanitizer.sanitize(&input);
        self.events.emit(ExecutionEvent::AiInvocation {
            execution_id: execution_id.to_string(),
            node_id: node.id.clone(),
            at: Utc::now(),
            provider: "local".into(),
            model: model_id.to_string(),
            input: sanitized.text.clone(),
            contract_version: contract.as_ref().map(|c| c.contract_version.clone()),
        });

        let images = match crate::ai_tools::resolve_node_images(node) {
            Ok(v) => v,
            Err(e) => return NodeOutcome::Failed(e),
        };
        let workspace = node
            .data
            .get("workspace")
            .and_then(|v| v.as_str())
            .filter(|s| !s.trim().is_empty())
            .map(str::to_owned);

        let req = CloudAiRequest {
            model: model_id.to_string(),
            prompt: sanitized.text,
            max_tokens,
            temperature,
            api_key,
            system,
            base_url: self.local_ai_base_url.clone(),
            top_p,
            top_k,
            stop,
            stream: true,
            call_id: Some(format!("ai-node-{}", Uuid::new_v4())),
            reasoning: crate::ai_tools::ai_reasoning(node),
            images,
            tools: Vec::new(),
            response_schema: None,
        };

        self.dispatch_ai_node(
            &provider,
            req,
            &sanitized.map,
            &crate::ai_tools::ai_task(node),
            &crate::ai_tools::ai_labels(node),
            &crate::ai_tools::ai_tool_adapters(node),
            workspace,
            true,
            "",
            execution_id,
            &node.id,
            crate::ai_tools::ai_max_tool_iters(node),
            crate::ai_tools::ai_output_schema(node),
            crate::ai_tools::ai_expect(node),
            contract.as_ref(),
        )
        .await
    }

    /// Capability-driven dispatch shared by the local + cloud AI nodes. `req`
    /// carries the sanitized prompt + params + reasoning/images. `task` selects
    /// the path: `"embedding"` → vector output; `"classify"` → a constrained
    /// label (branchable as `{{node.label}}`); `"structured"` → a JSON object
    /// matching the node's `outputSchema`, spread into the output so each field
    /// is branchable as `{{node.<field>}}`; otherwise `"generate"`, which runs a
    /// tool loop when adapters are bound, else a streamed completion.
    /// `rehydrate` restores PII placeholders in surfaced text.
    #[allow(clippy::too_many_arguments)]
    async fn dispatch_ai_node(
        &self,
        provider: &Arc<dyn flow_adapter_ai::CloudAiProvider>,
        mut req: CloudAiRequest,
        rehydrate: &flow_security::RehydrationMap,
        task: &str,
        labels: &[String],
        tool_adapters: &[String],
        workspace: Option<String>,
        audit: bool,
        original_prompt: &str,
        execution_id: &str,
        node_id: &str,
        max_tool_iters: usize,
        schema: Option<String>,
        expect: Option<String>,
        contract: Option<&AiNodeContract>,
    ) -> NodeOutcome {
        // Contract-bound output is defined for the generate path; the other
        // tasks have their own output shapes. Failing loudly beats silently
        // running an ungoverned node the flow author believed was governed.
        if let Some(contract) = contract {
            match task {
                "embedding" | "classify" | "structured" => {
                    return NodeOutcome::Failed(format!(
                        "contract-bound output is not supported for the `{task}` task \
                         (remove `contract: true` or use the generate task)"
                    ));
                }
                _ => {
                    return self
                        .run_contract_bound(
                            provider,
                            req,
                            rehydrate,
                            tool_adapters,
                            workspace,
                            execution_id,
                            node_id,
                            max_tool_iters,
                            contract,
                        )
                        .await;
                }
            }
        }
        match task {
            "embedding" => {
                req.stream = false;
                match provider.embed(&req).await {
                    Ok(e) => NodeOutcome::Succeeded(serde_json::json!({
                        "provider": e.provider,
                        "model": e.model,
                        "embedding": e.embedding,
                        "dims": e.dims,
                        "latency_ms": e.latency_ms,
                    })),
                    Err(err) => NodeOutcome::Failed(err.to_string()),
                }
            }
            "classify" => {
                if labels.is_empty() {
                    return NodeOutcome::Failed("classify task needs at least one label".into());
                }
                req.stream = false;
                let constraint = format!(
                    "You are a classifier. Read the input and reply with EXACTLY one of these labels and nothing else: {}.",
                    labels.join(", ")
                );
                req.system = Some(match req.system {
                    Some(s) if !s.is_empty() => format!("{constraint}\n\n{s}"),
                    _ => constraint,
                });
                match provider.invoke(&req).await {
                    Ok(resp) => {
                        let raw = rehydrate.rehydrate(&resp.text);
                        let label = crate::ai_tools::match_label(&raw, labels);
                        NodeOutcome::Succeeded(serde_json::json!({
                            "provider": resp.provider,
                            "model": resp.model,
                            "label": label,
                            "raw": raw,
                            "labels": labels,
                            "input_tokens": resp.input_tokens,
                            "output_tokens": resp.output_tokens,
                        }))
                    }
                    Err(err) => NodeOutcome::Failed(err.to_string()),
                }
            }
            "structured" => {
                let Some(schema) = schema.filter(|s| !s.trim().is_empty()) else {
                    return NodeOutcome::Failed(
                        "structured task needs a non-empty `outputSchema`".into(),
                    );
                };
                req.stream = false;
                // Hard enforcement where supported: a parseable schema rides as
                // `response_format` on the OpenAI-compatible providers (the
                // prompt below still constrains Claude/Gemini and is the fallback
                // for servers that ignore `response_format`).
                req.response_schema = serde_json::from_str(&schema).ok();
                let constraint = format!(
                    "Reply with ONLY a single JSON object that conforms to this JSON Schema. \
                     No prose, no explanation, no markdown code fences:\n{schema}"
                );
                req.system = Some(match req.system {
                    Some(s) if !s.is_empty() => format!("{constraint}\n\n{s}"),
                    _ => constraint,
                });
                match provider.invoke(&req).await {
                    Ok(resp) => {
                        let raw = rehydrate.rehydrate(&resp.text);
                        let mut out = serde_json::json!({
                            "provider": resp.provider,
                            "model": resp.model,
                            "raw": raw,
                            "input_tokens": resp.input_tokens,
                            "output_tokens": resp.output_tokens,
                        });
                        match crate::ai_tools::parse_json_lenient(&raw) {
                            Some(value) => {
                                // Spread the structured fields into the output so
                                // each is branchable as `{{node.<field>}}`, and
                                // keep the whole object under `output`. Never
                                // clobber the envelope's own keys.
                                if let (Some(obj), Some(fields)) =
                                    (out.as_object_mut(), value.as_object())
                                {
                                    for (k, v) in fields {
                                        obj.entry(k.clone()).or_insert_with(|| v.clone());
                                    }
                                }
                                out["output"] = value;
                            }
                            None => {
                                out["parseError"] =
                                    serde_json::json!("model did not return valid JSON");
                            }
                        }
                        NodeOutcome::Succeeded(out)
                    }
                    Err(err) => NodeOutcome::Failed(err.to_string()),
                }
            }
            _ => {
                if !tool_adapters.is_empty() {
                    let specs = crate::ai_tools::build_tool_specs(&self.adapters, tool_adapters);
                    if specs.is_empty() {
                        return NodeOutcome::Failed(
                            "tool use requested but the bound adapters expose no tools".into(),
                        );
                    }
                    req.stream = false;
                    req.tools = specs.clone();
                    // Emit a ToolCall event per dispatched call so the UI gets
                    // live tool-step granularity during the loop (the dispatcher
                    // otherwise produces no per-call events).
                    let observer = crate::ai_tools::ToolObserver::new(
                        self.events.clone(),
                        execution_id.to_string(),
                        node_id.to_string(),
                    );
                    // A coding-agent turn stages fs mutations for review; every
                    // other run dispatches tool calls straight to the adapters.
                    let result = if let Some(staging) = self.edit_staging.clone() {
                        let dispatcher = crate::ai_tools::StagingToolDispatcher::new(
                            self.adapters.clone(),
                            workspace,
                            staging,
                        )
                        .with_observer(observer);
                        provider
                            .invoke_tools(&req, &specs, &dispatcher, max_tool_iters)
                            .await
                    } else {
                        let dispatcher = crate::ai_tools::AdapterToolDispatcher::new(
                            self.adapters.clone(),
                            workspace,
                        )
                        .with_observer(observer);
                        provider
                            .invoke_tools(&req, &specs, &dispatcher, max_tool_iters)
                            .await
                    };
                    match result {
                        Ok(resp) => {
                            // Per-node AI evaluation: if the node carries an
                            // `expect` assertion, self-check the output and fail
                            // the node when unmet so the monitor/fix path engages
                            // (roadmap E1, beyond hard-failure-triggered).
                            if let Some(exp) = expect.as_deref() {
                                if matches!(
                                    evaluate_expectation(provider, &req, &resp.text, exp).await,
                                    Some(false)
                                ) {
                                    return NodeOutcome::Failed(format!(
                                        "node output did not meet expectation: {exp}"
                                    ));
                                }
                            }
                            NodeOutcome::Succeeded(ai_generate_envelope(
                                &resp,
                                rehydrate,
                                audit,
                                original_prompt,
                            ))
                        }
                        Err(err) => NodeOutcome::Failed(err.to_string()),
                    }
                } else {
                    req.stream = true;
                    match provider.invoke_stream(&req, self.stream_sink.as_ref()).await {
                        Ok(resp) => {
                            // Per-node AI evaluation: if the node carries an
                            // `expect` assertion, self-check the output and fail
                            // the node when unmet so the monitor/fix path engages
                            // (roadmap E1, beyond hard-failure-triggered).
                            if let Some(exp) = expect.as_deref() {
                                if matches!(
                                    evaluate_expectation(provider, &req, &resp.text, exp).await,
                                    Some(false)
                                ) {
                                    return NodeOutcome::Failed(format!(
                                        "node output did not meet expectation: {exp}"
                                    ));
                                }
                            }
                            NodeOutcome::Succeeded(ai_generate_envelope(
                                &resp,
                                rehydrate,
                                audit,
                                original_prompt,
                            ))
                        }
                        Err(err) => NodeOutcome::Failed(err.to_string()),
                    }
                }
            }
        }
    }

    /// Contract-bound AI invocation (RAO Spec v1.0). The model must return an
    /// [`AiOutputEnvelope`]; anything else fails the node. The engine - never
    /// the model - routes the validated output by the contract's thresholds:
    /// auto-approve, human review gate, or suppression onto the `.fail`
    /// fallback path. The envelope schema rides in the system instruction (and
    /// as `response_schema` when no tools are bound); the thresholds do not -
    /// they are applied here, structurally (RAO constraint 5).
    #[allow(clippy::too_many_arguments)]
    async fn run_contract_bound(
        &self,
        provider: &Arc<dyn flow_adapter_ai::CloudAiProvider>,
        mut req: CloudAiRequest,
        rehydrate: &flow_security::RehydrationMap,
        tool_adapters: &[String],
        workspace: Option<String>,
        execution_id: &str,
        node_id: &str,
        max_tool_iters: usize,
        contract: &AiNodeContract,
    ) -> NodeOutcome {
        let schema = AiOutputEnvelope::json_schema();
        let constraint = format!(
            "Reply with ONLY a single JSON object conforming to this JSON Schema. \
             No prose, no markdown code fences:\n{schema}\n\
             `primary_output` is your full answer. `confidence` is your confidence in it, \
             0.0 to 1.0. `confidence_type` is \"{}\". Set `escalate` to true only if a \
             human should review this output regardless of confidence. Put any \
             explanation for a human reviewer in `reasoning`.",
            contract
                .confidence_type
                .unwrap_or(flow_domain::contract::ConfidenceType::Verbalized)
                .as_str()
        );
        req.system = Some(match req.system.take() {
            Some(s) if !s.is_empty() => format!("{constraint}\n\n{s}"),
            _ => constraint,
        });
        req.stream = false;

        let result = if tool_adapters.is_empty() {
            req.response_schema = Some(schema);
            budgeted(contract.max_inference_ms, provider.invoke(&req)).await
        } else {
            let specs = crate::ai_tools::build_tool_specs(&self.adapters, tool_adapters);
            if specs.is_empty() {
                return NodeOutcome::Failed(
                    "tool use requested but the bound adapters expose no tools".into(),
                );
            }
            req.tools = specs.clone();
            let observer = crate::ai_tools::ToolObserver::new(
                self.events.clone(),
                execution_id.to_string(),
                node_id.to_string(),
            );
            if let Some(staging) = self.edit_staging.clone() {
                let dispatcher = crate::ai_tools::StagingToolDispatcher::new(
                    self.adapters.clone(),
                    workspace,
                    staging,
                )
                .with_observer(observer);
                budgeted(
                    contract.max_inference_ms,
                    provider.invoke_tools(&req, &specs, &dispatcher, max_tool_iters),
                )
                .await
            } else {
                let dispatcher =
                    crate::ai_tools::AdapterToolDispatcher::new(self.adapters.clone(), workspace)
                        .with_observer(observer);
                budgeted(
                    contract.max_inference_ms,
                    provider.invoke_tools(&req, &specs, &dispatcher, max_tool_iters),
                )
                .await
            }
        };

        let resp = match result {
            Ok(r) => r,
            Err(e) => return NodeOutcome::Failed(e),
        };

        let raw = rehydrate.rehydrate(&resp.text);
        let parsed = crate::ai_tools::parse_json_lenient(&raw)
            .ok_or_else(|| {
                flow_domain::contract::ContractError::SchemaViolation(
                    "model did not return a JSON object".into(),
                )
            })
            .and_then(|v| AiOutputEnvelope::from_value(&v))
            .and_then(|env| match contract.confidence_type {
                Some(expected) if env.confidence_type != expected => {
                    Err(flow_domain::contract::ContractError::SchemaViolation(format!(
                        "confidence_type `{}` does not match the contract's `{}`",
                        env.confidence_type.as_str(),
                        expected.as_str()
                    )))
                }
                _ => Ok(env),
            });

        let envelope = match parsed {
            Ok(env) => env,
            Err(violation) => {
                self.events.emit(ExecutionEvent::AiRoutingDecision {
                    execution_id: execution_id.to_string(),
                    node_id: node_id.to_string(),
                    at: Utc::now(),
                    decision: "contract_violation".into(),
                    confidence: None,
                    threshold: "expected_output schema".into(),
                    contract_version: contract.contract_version.clone(),
                });
                return NodeOutcome::Failed(format!("contract violation: {violation}"));
            }
        };

        let thresholds = &contract.thresholds;
        let decision = thresholds.route(envelope.confidence, envelope.escalate);
        let threshold_desc = match decision {
            RouteDecision::AutoApprove => format!(
                "confidence {} > autoApproveAbove {}",
                envelope.confidence, thresholds.auto_approve_above
            ),
            RouteDecision::Suppress => format!(
                "confidence {} < suppressBelow {}",
                envelope.confidence, thresholds.suppress_below
            ),
            RouteDecision::HumanReview if envelope.escalate => "model set escalate".into(),
            RouteDecision::HumanReview => format!(
                "confidence {} within review band [{}, {}]",
                envelope.confidence, thresholds.human_review_band.0, thresholds.human_review_band.1
            ),
        };
        self.events.emit(ExecutionEvent::AiRoutingDecision {
            execution_id: execution_id.to_string(),
            node_id: node_id.to_string(),
            at: Utc::now(),
            decision: decision.as_str().into(),
            confidence: Some(envelope.confidence),
            threshold: threshold_desc,
            contract_version: contract.contract_version.clone(),
        });

        match decision {
            RouteDecision::AutoApprove => {
                NodeOutcome::Succeeded(contract_payload(&envelope, &resp, contract, decision))
            }
            RouteDecision::Suppress => NodeOutcome::Failed(format!(
                "output suppressed: confidence {:.2} below suppressBelow {:.2}",
                envelope.confidence, thresholds.suppress_below
            )),
            RouteDecision::HumanReview => {
                self.await_review(&envelope, &resp, contract, execution_id, node_id)
                    .await
            }
        }
    }

    /// Hold a contract-bound output at the human review gate. The run
    /// self-pauses (like the destructive-step gate) until the verdict arrives
    /// via [`RunControl::resolve_review`] or the run is cancelled. A plain
    /// resume releases the gate as an approval, mirroring the destructive
    /// gate's resume-confirms semantics.
    async fn await_review(
        &self,
        envelope: &AiOutputEnvelope,
        resp: &flow_adapter_ai::CloudAiResponse,
        contract: &AiNodeContract,
        execution_id: &str,
        node_id: &str,
    ) -> NodeOutcome {
        if self.control.is_cancelling() {
            return NodeOutcome::Skipped("run cancelled before the review gate".into());
        }
        if !self.review_gate_available {
            return NodeOutcome::Failed(format!(
                "output requires human review (confidence {:.2}) but no review gate is \
                 attached to this run; routing to the fallback path",
                envelope.confidence
            ));
        }
        // Drop any stale verdict so only a fresh one can release this gate.
        let _ = self.control.take_review();
        self.events.emit(ExecutionEvent::AiReviewRequired {
            execution_id: execution_id.to_string(),
            node_id: node_id.to_string(),
            at: Utc::now(),
            primary_output: envelope.primary_output.clone(),
            confidence: envelope.confidence,
            reasoning: envelope.reasoning.clone(),
        });
        self.control.pause();
        self.control.wait_while_paused().await;
        if self.control.is_cancelling() {
            return NodeOutcome::Skipped("ai output not reviewed (run cancelled)".into());
        }
        let approved = !matches!(self.control.take_review(), Some(ReviewDecision::Rejected));
        self.events.emit(ExecutionEvent::AiReviewResolved {
            execution_id: execution_id.to_string(),
            node_id: node_id.to_string(),
            at: Utc::now(),
            approved,
        });
        if approved {
            NodeOutcome::Succeeded(contract_payload(
                envelope,
                resp,
                contract,
                RouteDecision::HumanReview,
            ))
        } else {
            NodeOutcome::Failed(format!(
                "output rejected at human review gate (confidence {:.2})",
                envelope.confidence
            ))
        }
    }

    /// Cloud branch of the unified AI node: invokes a frontier provider's API.
    /// Reached from `run_ai_node` when `provider` is not `local`.
    async fn run_cloud_ai_node(&self, node: &FlowNode, execution_id: &str) -> NodeOutcome {
        let provider_name = match node.data.get("provider").and_then(|v| v.as_str()) {
            Some(p) => p.to_string(),
            None => return NodeOutcome::Failed("missing 'provider' on ai node".into()),
        };

        // Policy gate. The `local` provider hits a localhost endpoint (no
        // egress) so it's governed by `allow_local_ai`; every other
        // provider is true cloud egress, gated by `allow_cloud_ai`.
        let is_local = provider_name == "local";
        if is_local {
            if !self.allow_local_ai {
                return NodeOutcome::Skipped(
                    "local AI disabled by policy (settings.allow_local_ai = false)".into(),
                );
            }
        } else if !self.allow_cloud_ai {
            return NodeOutcome::Skipped(
                "cloud AI disabled by policy (settings.allow_cloud_ai = false)".into(),
            );
        }

        let model = match node.data.get("modelId").and_then(|v| v.as_str()) {
            Some(m) => m.to_string(),
            None => return NodeOutcome::Failed("missing 'modelId' on ai node".into()),
        };
        // The AI node's canonical field is `input`; `prompt` is accepted as a
        // deliberate leniency for model-generated DSL that emits it.
        let prompt = match node
            .data
            .get("prompt")
            .and_then(|v| v.as_str())
            .or_else(|| node.data.get("input").and_then(|v| v.as_str()))
        {
            Some(p) => p.to_string(),
            None => {
                return NodeOutcome::Failed("missing 'input' on ai node (cloud provider)".into())
            }
        };
        let max_tokens = node
            .data
            .get("maxTokens")
            .and_then(|v| v.as_u64())
            .map(|v| v as u32);
        let temperature = node
            .data
            .get("temperature")
            .and_then(|v| v.as_f64())
            .map(|v| v as f32);
        let audit_content = node
            .data
            .get("auditContent")
            .and_then(|v| v.as_bool())
            .unwrap_or(false);

        let provider = match self.cloud_providers.get(&provider_name) {
            Some(p) => p,
            None => {
                return NodeOutcome::Skipped(format!(
                    "cloud provider '{provider_name}' not enabled"
                ));
            }
        };

        let env_var = provider.env_var();
        let api_key = match self
            .credentials
            .resolve_cloud_ai_key(&provider_name, env_var)
        {
            Ok(k) => k,
            // Local servers (Ollama / LM Studio / llama.cpp) typically need
            // no auth - a missing key is not fatal; the provider simply omits
            // the bearer header. Cloud providers still require one.
            Err(CredentialError::NotFound { .. }) if is_local => String::new(),
            Err(CredentialError::NotFound { .. }) => {
                return NodeOutcome::Failed(format!(
                    "no API key for provider '{provider_name}'. Set it in Settings, or export {env_var} in your environment."
                ));
            }
            Err(e) => return NodeOutcome::Failed(format!("credential lookup failed: {e}")),
        };

        let contract = match AiNodeContract::from_node_data(&node.data) {
            Ok(c) => c,
            Err(e) => return NodeOutcome::Failed(e.to_string()),
        };

        let sanitized = self.sanitizer.sanitize(&prompt);
        // Without `auditContent`, record only a preview of the input - the
        // same privacy posture as the surfaced `text_preview` output.
        let recorded_input = if audit_content {
            sanitized.text.clone()
        } else {
            sanitized.text.chars().take(200).collect()
        };
        self.events.emit(ExecutionEvent::AiInvocation {
            execution_id: execution_id.to_string(),
            node_id: node.id.clone(),
            at: Utc::now(),
            provider: provider_name.clone(),
            model: model.clone(),
            input: recorded_input,
            contract_version: contract.as_ref().map(|c| c.contract_version.clone()),
        });

        let system = node
            .data
            .get("system")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(str::to_owned);
        let images = match crate::ai_tools::resolve_node_images(node) {
            Ok(v) => v,
            Err(e) => return NodeOutcome::Failed(e),
        };
        let workspace = node
            .data
            .get("workspace")
            .and_then(|v| v.as_str())
            .filter(|s| !s.trim().is_empty())
            .map(str::to_owned);

        let req = CloudAiRequest {
            model: model.clone(),
            prompt: sanitized.text,
            max_tokens,
            temperature,
            api_key,
            system,
            // Local provider reads its endpoint from settings; cloud
            // providers ignore this and use their fixed const.
            base_url: if is_local {
                self.local_ai_base_url.clone()
            } else {
                None
            },
            stream: true,
            call_id: Some(format!("cloud-ai-{}", Uuid::new_v4())),
            reasoning: crate::ai_tools::ai_reasoning(node),
            images,
            ..Default::default()
        };

        // `auditContent` is the cloud privacy gate: when off, only a short
        // preview of the response is surfaced (no full text / prompt echo).
        self.dispatch_ai_node(
            &provider,
            req,
            &sanitized.map,
            &crate::ai_tools::ai_task(node),
            &crate::ai_tools::ai_labels(node),
            &crate::ai_tools::ai_tool_adapters(node),
            workspace,
            audit_content,
            &prompt,
            execution_id,
            &node.id,
            crate::ai_tools::ai_max_tool_iters(node),
            crate::ai_tools::ai_output_schema(node),
            crate::ai_tools::ai_expect(node),
            contract.as_ref(),
        )
        .await
    }

    async fn run_adapter_node(&self, node: &FlowNode, execution_id: &str) -> NodeOutcome {
        let adapter_name = node
            .data
            .get("adapter")
            .and_then(|v| v.as_str())
            .unwrap_or("mock");

        let Some(adapter) = self.adapters.get(adapter_name) else {
            return NodeOutcome::Failed(format!("adapter '{adapter_name}' not registered"));
        };

        let ctx = crate::adapter::AdapterCtx {
            events: self.events.clone(),
            execution_id: execution_id.to_string(),
            node_id: node.id.clone(),
            workspace_root: self.workspace_root.clone(),
        };

        match adapter.execute_with_events(node, &ctx).await {
            Ok(out) => {
                NodeOutcome::Succeeded(serde_json::to_value(out).unwrap_or(serde_json::Value::Null))
            }
            Err(e) => NodeOutcome::Failed(e.to_string()),
        }
    }
}

/// If `node` performs a destructive operation, return a short human-readable
/// description of it; otherwise `None`. Used by the per-step confirmation gate
/// (roadmap E1). Detection is intentionally conservative - it flags the clear
/// data-loss cases (deleting files, `rm`, `git push`/`reset --hard`/`clean`,
/// `kubectl delete`, `drop table/database`, `truncate`, `mkfs`, `dd if=`,
/// `shutdown`/`reboot`) rather than guessing at every possible command.
/// A pre-apply advisory about a node in a proposed flow, surfaced in the review
/// + interception modals before the flow runs (roadmap E1 pre-apply
/// verification). `severity` is `"destructive"` or `"sandbox-escape"`.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FlowWarning {
    pub node_id: String,
    pub severity: String,
    pub message: String,
}

/// Static pre-apply scan of a proposed graph: flags nodes that perform a
/// destructive operation (reusing [`destructive_reason`]) or reference a path
/// that would escape the workspace jail (`..` parent traversal). Advisory only -
/// it does not block; the executor's runtime gates still apply.
pub fn verify_graph(graph: &FlowGraph) -> Vec<FlowWarning> {
    let mut out = Vec::new();
    for node in &graph.nodes {
        if let Some(message) = destructive_reason(node) {
            out.push(FlowWarning {
                node_id: node.id.clone(),
                severity: "destructive".into(),
                message,
            });
        }
        if let Some(message) = sandbox_escape_reason(node) {
            out.push(FlowWarning {
                node_id: node.id.clone(),
                severity: "sandbox-escape".into(),
                message,
            });
        }
    }
    out
}

/// Flags a node whose path-like fields contain a `..` parent-traversal segment -
/// an attempt to leave the workspace jail (`flow_security::confine_path` rejects
/// these at runtime; surfacing it pre-apply catches an AI-generated escape before
/// the run starts). Whole-segment match, so `a..b` is not a false positive.
fn sandbox_escape_reason(node: &FlowNode) -> Option<String> {
    let data = &node.data;
    for key in ["path", "cwd", "workspaceRoot", "command", "args"] {
        let candidate = match data.get(key) {
            Some(serde_json::Value::String(s)) => s.clone(),
            Some(serde_json::Value::Array(arr)) => arr
                .iter()
                .filter_map(|v| v.as_str())
                .collect::<Vec<_>>()
                .join(" "),
            _ => continue,
        };
        if candidate
            .split(|c: char| c == '/' || c == '\\' || c == ' ')
            .any(|seg| seg == "..")
        {
            let preview: String = candidate.chars().take(120).collect();
            return Some(format!(
                "Path may escape the workspace (contains `..`): {preview}"
            ));
        }
    }
    None
}

/// Build the failover variant of an `ai` node: a clone with `provider` (and
/// `modelId`, if a `fallbackModelId` is set) swapped to the node's configured
/// fallback, and the `fallbackProvider` / `fallbackModelId` fields stripped so
/// the retry can't fall over a second time. `None` when no fallback is set.
fn fallback_node(node: &FlowNode) -> Option<FlowNode> {
    let fb_provider = node
        .data
        .get("fallbackProvider")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())?
        .to_string();
    let mut fb = node.clone();
    if let Some(obj) = fb.data.as_object_mut() {
        obj.insert("provider".into(), serde_json::Value::String(fb_provider));
        if let Some(m) = node
            .data
            .get("fallbackModelId")
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|s| !s.is_empty())
        {
            obj.insert("modelId".into(), serde_json::Value::String(m.to_string()));
        }
        obj.remove("fallbackProvider");
        obj.remove("fallbackModelId");
    }
    Some(fb)
}

pub fn destructive_reason(node: &FlowNode) -> Option<String> {
    let data = &node.data;
    let adapter = data.get("adapter").and_then(|v| v.as_str()).unwrap_or("");
    let action = data.get("actionId").and_then(|v| v.as_str()).unwrap_or("");

    // Filesystem adapter: deleting a file is unambiguously destructive.
    if adapter == "fs" && action == "delete-file" {
        let path = data
            .get("path")
            .and_then(|v| v.as_str())
            .filter(|s| !s.trim().is_empty())
            .unwrap_or("a file");
        return Some(format!("Delete file: {path}"));
    }

    // Shell / CLI adapters: scan the program + arguments for destructive
    // commands. Tokens are matched whole-word to avoid substring false
    // positives (e.g. `confirm` containing `rm`).
    if adapter == "shell" || adapter == "cli" {
        let mut parts: Vec<String> = Vec::new();
        // The shell adapter's actionId names the program for curated tools
        // (`git`, `npm`, `kubectl`, …); only `run-command` carries a free-form
        // program in `command`. Include it so `git push` etc. are detected.
        if adapter == "shell" && action != "run-command" && !action.is_empty() {
            parts.push(action.to_string());
        }
        for key in ["bin", "command"] {
            if let Some(s) = data.get(key).and_then(|v| v.as_str()) {
                parts.push(s.to_string());
            }
        }
        match data.get("args") {
            Some(serde_json::Value::String(s)) => parts.push(s.clone()),
            Some(serde_json::Value::Array(arr)) => {
                for a in arr {
                    if let Some(s) = a.as_str() {
                        parts.push(s.to_string());
                    }
                }
            }
            _ => {}
        }
        let cmd = parts.join(" ");
        if let Some(label) = destructive_command(&cmd) {
            let preview: String = cmd.chars().take(120).collect();
            return Some(format!("Runs a destructive command ({label}): {preview}"));
        }
    }

    None
}

/// Whole-word scan of a command line for known destructive operations. Returns
/// a short label for the first match, or `None`.
fn destructive_command(cmd: &str) -> Option<&'static str> {
    let lower = cmd.to_lowercase();
    let tokens: Vec<&str> = lower.split_whitespace().collect();
    let has = |w: &str| tokens.iter().any(|t| *t == w);

    if has("rm") || has("rmdir") {
        return Some("rm");
    }
    if has("mkfs") || tokens.iter().any(|t| t.starts_with("mkfs.")) {
        return Some("mkfs");
    }
    if has("shutdown") {
        return Some("shutdown");
    }
    if has("reboot") {
        return Some("reboot");
    }
    if has("truncate") {
        return Some("truncate");
    }
    if has("dd") && tokens.iter().any(|t| t.starts_with("if=")) {
        return Some("dd");
    }
    // Multi-token tool subcommands.
    let after = |tool: &str, sub: &str| -> bool {
        tokens
            .iter()
            .position(|t| *t == tool)
            .map(|i| tokens[i + 1..].iter().any(|t| *t == sub))
            .unwrap_or(false)
    };
    if after("git", "push") {
        return Some("git push");
    }
    if after("git", "clean") {
        return Some("git clean");
    }
    if has("git") && has("reset") && has("--hard") {
        return Some("git reset --hard");
    }
    if after("kubectl", "delete") {
        return Some("kubectl delete");
    }
    if has("drop") && (has("table") || has("database")) {
        return Some("drop");
    }
    None
}

/// Wrap an LLM text response in the AI-node output envelope. With `audit` the
/// full (rehydrated) text is surfaced - and the original prompt echoed when
/// non-empty; without it (the cloud privacy posture) only a short preview is
/// kept.
/// Self-evaluate an AI node's output against its `expect` assertion (roadmap E1
/// per-node evaluation): a second, constrained call to the same provider/model
/// returns PASS/FAIL. `Some(false)` = the output fails the assertion (the node
/// is then failed so the monitor/fix path engages); `None` = the eval itself
/// errored, treated as a pass so a flaky check never blocks a good run.
async fn evaluate_expectation(
    provider: &Arc<dyn flow_adapter_ai::CloudAiProvider>,
    base: &CloudAiRequest,
    output: &str,
    expect: &str,
) -> Option<bool> {
    let preview: String = output.chars().take(4000).collect();
    let eval_req = CloudAiRequest {
        model: base.model.clone(),
        prompt: format!("Output:\n{preview}\n\nExpectation: {expect}"),
        api_key: base.api_key.clone(),
        base_url: base.base_url.clone(),
        system: Some(
            "You are a strict evaluator. Decide whether the Output satisfies the \
             Expectation. Reply with EXACTLY one word: PASS or FAIL."
                .into(),
        ),
        max_tokens: Some(8),
        ..Default::default()
    };
    let resp = provider.invoke(&eval_req).await.ok()?;
    Some(!resp.text.trim().to_uppercase().contains("FAIL"))
}

fn ai_generate_envelope(
    resp: &flow_adapter_ai::CloudAiResponse,
    rehydrate: &flow_security::RehydrationMap,
    audit: bool,
    original_prompt: &str,
) -> serde_json::Value {
    let text = rehydrate.rehydrate(&resp.text);
    let mut payload = serde_json::json!({
        "provider": resp.provider,
        "model": resp.model,
        "finish_reason": resp.finish_reason,
        "input_tokens": resp.input_tokens,
        "output_tokens": resp.output_tokens,
        "latency_ms": resp.latency_ms,
    });
    if audit {
        payload["text"] = serde_json::Value::String(text);
        if !original_prompt.is_empty() {
            payload["prompt"] = serde_json::Value::String(original_prompt.to_string());
        }
    } else {
        let preview: String = text.chars().take(200).collect();
        payload["text_preview"] = serde_json::Value::String(preview);
    }
    payload
}

/// Run an inference future under the contract's `maxInferenceMs` budget.
/// Exceeding the budget is a node failure (RAO node-contract
/// `max_inference_ms`), which routes to the flow's fallback path.
async fn budgeted<F>(budget_ms: Option<u64>, fut: F) -> Result<flow_adapter_ai::CloudAiResponse, String>
where
    F: std::future::Future<
        Output = Result<flow_adapter_ai::CloudAiResponse, flow_adapter_ai::CloudAiError>,
    >,
{
    match budget_ms {
        Some(ms) => match tokio::time::timeout(std::time::Duration::from_millis(ms), fut).await {
            Ok(r) => r.map_err(|e| e.to_string()),
            Err(_) => Err(format!("inference exceeded the contract budget ({ms} ms)")),
        },
        None => fut.await.map_err(|e| e.to_string()),
    }
}

/// Output payload of a contract-bound AI node. The envelope fields are
/// top-level so downstream nodes can branch on `{{node.primary_output}}`,
/// `{{node.confidence}}`, etc.; `route` records how the engine routed it.
fn contract_payload(
    envelope: &AiOutputEnvelope,
    resp: &flow_adapter_ai::CloudAiResponse,
    contract: &AiNodeContract,
    route: RouteDecision,
) -> serde_json::Value {
    let mut payload = serde_json::json!({
        "provider": resp.provider,
        "model": resp.model,
        "input_tokens": resp.input_tokens,
        "output_tokens": resp.output_tokens,
        "latency_ms": resp.latency_ms,
        "contract_version": contract.contract_version,
        "route": route.as_str(),
        "primary_output": envelope.primary_output,
        "confidence": envelope.confidence,
        "confidence_type": envelope.confidence_type.as_str(),
        "escalate": envelope.escalate,
    });
    if let Some(reasoning) = &envelope.reasoning {
        payload["reasoning"] = serde_json::Value::String(reasoning.clone());
    }
    payload
}

enum NodeOutcome {
    Succeeded(serde_json::Value),
    Skipped(String),
    Failed(String),
}

#[derive(Debug, Clone, Copy)]
enum NodeStatus {
    Succeeded,
    Failed,
    Skipped,
}

/// Recursively substitute `{{...}}` references in every string leaf of a
/// node's `data` value. Objects and arrays are walked; non-string scalars are
/// returned unchanged. `outputs` holds prior succeeded-node payloads; `upstream`
/// is the list of node ids whose edge fired into the current node (the source
/// for `{{input}}`).
fn interpolate_value(
    value: &serde_json::Value,
    outputs: &HashMap<String, serde_json::Value>,
    upstream: &[String],
    memory: &HashMap<String, serde_json::Value>,
) -> serde_json::Value {
    match value {
        serde_json::Value::String(s) => {
            serde_json::Value::String(interpolate_str(s, outputs, upstream, memory))
        }
        serde_json::Value::Array(items) => serde_json::Value::Array(
            items
                .iter()
                .map(|v| interpolate_value(v, outputs, upstream, memory))
                .collect(),
        ),
        serde_json::Value::Object(map) => serde_json::Value::Object(
            map.iter()
                .map(|(k, v)| (k.clone(), interpolate_value(v, outputs, upstream, memory)))
                .collect(),
        ),
        other => other.clone(),
    }
}

/// Replace each `{{token}}` occurrence in `s`. Unmatched braces and tokens
/// that resolve to nothing are left as the empty string. Strings with no `{{`
/// are returned as-is.
fn interpolate_str(
    s: &str,
    outputs: &HashMap<String, serde_json::Value>,
    upstream: &[String],
    memory: &HashMap<String, serde_json::Value>,
) -> String {
    if !s.contains("{{") {
        return s.to_string();
    }
    let mut out = String::with_capacity(s.len());
    let mut rest = s;
    while let Some(open) = rest.find("{{") {
        out.push_str(&rest[..open]);
        let after = &rest[open + 2..];
        match after.find("}}") {
            Some(close) => {
                let token = after[..close].trim();
                out.push_str(&resolve_token(token, outputs, upstream, memory));
                rest = &after[close + 2..];
            }
            // No closing braces - emit the literal `{{` and stop scanning.
            None => {
                out.push_str("{{");
                rest = after;
                break;
            }
        }
    }
    out.push_str(rest);
    out
}

/// Resolve one `{{...}}` token to a string. `input` joins the upstream node
/// outputs; `memory.<key>[.<path>]` reads session working memory; `<id>` yields
/// that node's primary text; `<id>.<a>.<b>` walks into the node's output JSON.
fn resolve_token(
    token: &str,
    outputs: &HashMap<String, serde_json::Value>,
    upstream: &[String],
    memory: &HashMap<String, serde_json::Value>,
) -> String {
    if token == "input" {
        return upstream
            .iter()
            .filter_map(|id| outputs.get(id))
            .map(primary_text)
            .collect::<Vec<_>>()
            .join("\n\n");
    }
    if let Some(rest) = token.strip_prefix("memory.") {
        let (key, path) = match rest.split_once('.') {
            Some((key, path)) => (key, Some(path)),
            None => (rest, None),
        };
        let Some(value) = memory.get(key) else {
            return String::new();
        };
        return match path {
            Some(path) => value_at(value, path).map(primary_text).unwrap_or_default(),
            None => primary_text(value),
        };
    }
    let (id, path) = match token.split_once('.') {
        Some((id, path)) => (id, Some(path)),
        None => (token, None),
    };
    let Some(value) = outputs.get(id) else {
        return String::new();
    };
    match path {
        Some(path) => value_at(value, path).map(primary_text).unwrap_or_default(),
        None => primary_text(value),
    }
}

/// Walk a dotted path (`a.b.c`) into a JSON object, returning the nested value.
fn value_at<'a>(value: &'a serde_json::Value, path: &str) -> Option<&'a serde_json::Value> {
    let mut cur = value;
    for segment in path.split('.') {
        cur = cur.get(segment)?;
    }
    Some(cur)
}

/// Render a node-output value as the string a downstream field consumes. Plain
/// strings pass through; scalars use their natural string form. For objects:
/// adapter outputs nest their result under `payload`, so descend into it first;
/// then expose a common text-bearing key (`text`/`stdout`/`output`/`result`/
/// `value`) when present, else compact JSON.
fn primary_text(value: &serde_json::Value) -> String {
    match value {
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Null => String::new(),
        serde_json::Value::Bool(b) => b.to_string(),
        serde_json::Value::Number(n) => n.to_string(),
        serde_json::Value::Object(map) => {
            // Adapter results are wrapped as `{ status, payload }`; the useful
            // content lives in `payload`.
            if let Some(payload) = map.get("payload") {
                if !payload.is_null() {
                    return primary_text(payload);
                }
            }
            for key in ["text", "stdout", "output", "result", "value"] {
                if let Some(serde_json::Value::String(s)) = map.get(key) {
                    return s.clone();
                }
            }
            serde_json::to_string(value).unwrap_or_default()
        }
        serde_json::Value::Array(_) => serde_json::to_string(value).unwrap_or_default(),
    }
}

/// An outgoing edge is active when its routing `outcome` matches the source's
/// terminal status AND its optional `when` condition evaluates true. The
/// condition's `{{...}}` references are resolved against prior node outputs
/// (the edge source seeds `{{input}}`) before evaluation; a malformed or
/// unresolved condition fails closed - the edge does not fire.
fn edge_active(
    edge: &FlowEdge,
    terminal: &HashMap<String, NodeStatus>,
    outputs: &HashMap<String, serde_json::Value>,
    memory: &HashMap<String, serde_json::Value>,
) -> bool {
    let Some(status) = terminal.get(&edge.source) else {
        return false;
    };
    if !edge_fires(edge.outcome, *status) {
        return false;
    }
    match edge.condition.as_deref() {
        None => true,
        Some(expr) if expr.trim().is_empty() => true,
        Some(expr) => {
            let resolved =
                interpolate_str(expr, outputs, std::slice::from_ref(&edge.source), memory);
            crate::condition::eval(&resolved).unwrap_or(false)
        }
    }
}

fn edge_fires(outcome: EdgeOutcome, status: NodeStatus) -> bool {
    matches!(
        (outcome, status),
        (EdgeOutcome::Always, _)
            | (EdgeOutcome::Pass, NodeStatus::Succeeded)
            | (EdgeOutcome::Fail, NodeStatus::Failed | NodeStatus::Skipped)
    )
}

/// Build an outer graph where each sub-flow collapses to a single virtual node
/// (id = the sub-flow id). Edges are remapped to the virtual node; edges
/// internal to a sub-flow (or self-loops after remapping) are dropped. Lets the
/// scheduler topo-order free nodes and sub-flow units together.
fn collapse_subflows(graph: &FlowGraph) -> FlowGraph {
    let mut member_to_sf: HashMap<&str, &str> = HashMap::new();
    for sf in &graph.subflows {
        for id in &sf.node_ids {
            member_to_sf.insert(id.as_str(), sf.id.as_str());
        }
    }
    let map = |id: &str| -> String {
        member_to_sf
            .get(id)
            .map(|s| s.to_string())
            .unwrap_or_else(|| id.to_string())
    };

    let mut nodes: Vec<FlowNode> = graph
        .nodes
        .iter()
        .filter(|n| !member_to_sf.contains_key(n.id.as_str()))
        .cloned()
        .collect();
    for sf in &graph.subflows {
        nodes.push(FlowNode {
            id: sf.id.clone(),
            node_type: "subflow".into(),
            position: Position { x: 0.0, y: 0.0 },
            data: serde_json::Value::Null,
        });
    }

    let mut edges: Vec<FlowEdge> = Vec::new();
    for (i, e) in graph.edges.iter().enumerate() {
        let s = map(&e.source);
        let t = map(&e.target);
        if s == t {
            continue; // internal edge, or a self-loop after collapsing
        }
        edges.push(FlowEdge {
            id: format!("outer-{i}"),
            source: s,
            target: t,
            label: None,
            condition: None,
            outcome: EdgeOutcome::Always,
        });
    }

    FlowGraph {
        id: graph.id.clone(),
        name: graph.name.clone(),
        version: graph.version.clone(),
        description: None,
        nodes,
        edges,
        subflows: Vec::new(),
    }
}

fn topo_sort(graph: &FlowGraph) -> Result<Vec<String>, ExecutorError> {
    let mut indegree: HashMap<&str, usize> =
        graph.nodes.iter().map(|n| (n.id.as_str(), 0)).collect();
    let mut adj: HashMap<&str, Vec<&str>> = HashMap::new();

    for edge in &graph.edges {
        adj.entry(edge.source.as_str())
            .or_default()
            .push(edge.target.as_str());
        *indegree.entry(edge.target.as_str()).or_insert(0) += 1;
    }

    let mut queue: VecDeque<&str> = indegree
        .iter()
        .filter(|(_, &d)| d == 0)
        .map(|(id, _)| *id)
        .collect();

    let mut order = Vec::with_capacity(graph.nodes.len());
    let mut visited = HashSet::new();

    while let Some(id) = queue.pop_front() {
        if !visited.insert(id) {
            continue;
        }
        order.push(id.to_string());

        if let Some(children) = adj.get(id) {
            for &c in children {
                let entry = indegree.entry(c).or_insert(0);
                if *entry > 0 {
                    *entry -= 1;
                    if *entry == 0 {
                        queue.push_back(c);
                    }
                }
            }
        }
    }

    if order.len() < graph.nodes.len() {
        let unresolved = graph
            .nodes
            .iter()
            .find(|n| !visited.contains(n.id.as_str()))
            .map(|n| n.id.clone())
            .unwrap_or_default();
        return Err(ExecutorError::Cycle(unresolved));
    }

    Ok(order)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::adapter::MockAdapter;
    use crate::events::CapturingSink;
    use flow_domain::graph::{FlowEdge, FlowNode, Position};

    fn make_node(id: &str, kind: &str, data: serde_json::Value) -> FlowNode {
        FlowNode {
            id: id.into(),
            node_type: kind.into(),
            position: Position { x: 0.0, y: 0.0 },
            data,
        }
    }

    fn make_edge(id: &str, src: &str, dst: &str) -> FlowEdge {
        FlowEdge {
            id: id.into(),
            source: src.into(),
            target: dst.into(),
            label: None,
            condition: None,
            outcome: EdgeOutcome::Always,
        }
    }

    fn make_edge_with(id: &str, src: &str, dst: &str, outcome: EdgeOutcome) -> FlowEdge {
        FlowEdge {
            id: id.into(),
            source: src.into(),
            target: dst.into(),
            label: None,
            condition: None,
            outcome,
        }
    }

    /// Build an executor with a single mock adapter (1ms delay) that always succeeds.
    async fn build_executor() -> (Executor, CapturingSink) {
        let mut adapters = AdapterRegistry::new();
        adapters.register(Arc::new(MockAdapter::with_delay(
            "mock",
            std::time::Duration::from_millis(1),
        )));
        let sink = CapturingSink::default();
        let credentials: Arc<dyn CredentialResolver> =
            Arc::new(flow_security::EnvFallbackResolver::new(Arc::new(
                flow_security::InMemoryCredentialStore::new(),
            )));
        let exec = Executor {
            adapters: Arc::new(adapters),
            sanitizer: Arc::new(PiiSanitizer::new()),
            events: Arc::new(sink.clone()),
            cloud_providers: Arc::new(CloudAiRegistry::new()),
            credentials,
            allow_cloud_ai: false,
            allow_local_ai: false,
            local_ai_base_url: None,
            stream_sink: Arc::new(NullStreamSink),
            working_memory: Default::default(),
            control: Default::default(),
            confirm_destructive: false,
            review_gate_available: true,
            edit_staging: None,
            workspace_root: PathBuf::from("."),
        };
        (exec, sink)
    }

    /// A node with `actionId="forced-fail"` and `adapter="missing"` so the
    /// executor reports `Failed` (no such adapter registered).
    fn make_failing_action(id: &str) -> FlowNode {
        make_node(
            id,
            "action",
            serde_json::json!({ "adapter": "missing-adapter" }),
        )
    }

    #[tokio::test]
    async fn topological_order_is_respected() {
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("c", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge("e1", "a", "b"), make_edge("e2", "b", "c")],
        };

        let order = topo_sort(&graph).unwrap();
        let pos = |id: &str| order.iter().position(|x| x == id).unwrap();
        assert!(pos("a") < pos("b"));
        assert!(pos("b") < pos("c"));
    }

    #[tokio::test]
    async fn cycle_detected() {
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge("e1", "a", "b"), make_edge("e2", "b", "a")],
        };
        assert!(matches!(topo_sort(&graph), Err(ExecutorError::Cycle(_))));
    }

    #[tokio::test]
    async fn happy_path_emits_events_and_returns_succeeded() {
        let (executor, sink) = build_executor().await;

        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "demo".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge("e1", "a", "b")],
        };

        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.status, "succeeded");
        assert_eq!(summary.succeeded, 2);

        let events = sink.events.lock().unwrap();
        assert!(matches!(
            events.first(),
            Some(ExecutionEvent::Started { .. })
        ));
        assert!(matches!(events.last(), Some(ExecutionEvent::Done { .. })));
    }

    #[tokio::test]
    async fn subflow_runs_as_unit_and_downstream_fires() {
        let (executor, _sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: vec![SubFlow {
                id: "setup".into(),
                label: "Setup".into(),
                node_ids: vec!["a".into(), "b".into()],
                retry: None,
            }],
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("start", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("done", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![
                make_edge("e1", "start", "a"), // boundary-crossing into the unit
                make_edge("e2", "a", "b"),     // internal to the unit
                make_edge("e3", "b", "done"),  // boundary-crossing out of the unit
            ],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.status, "succeeded");
        assert_eq!(summary.succeeded, 4, "start + a + b + done all run");
        assert_eq!(summary.failed, 0);
        assert_eq!(summary.skipped, 0);
    }

    #[tokio::test]
    async fn subflow_retries_as_a_unit_on_failure() {
        let (executor, sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: vec![SubFlow {
                id: "unit".into(),
                label: "Unit".into(),
                node_ids: vec!["boom".into()],
                retry: Some(2),
            }],
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![make_failing_action("boom")],
            edges: vec![],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.failed, 1, "the unit ultimately fails");
        // retry: Some(2) → 1 initial attempt + 2 retries = 3 runs of the member.
        let events = sink.events.lock().unwrap();
        let starts = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::NodeStarted { node_id, .. } if node_id == "boom"))
            .count();
        assert_eq!(starts, 3, "the unit re-ran the failing member 1 + retry(2) times");
    }

    #[tokio::test]
    async fn subflow_skipped_when_entry_not_reached() {
        let (executor, _sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: vec![SubFlow {
                id: "unit".into(),
                label: "Unit".into(),
                node_ids: vec!["a".into(), "b".into()],
                retry: None,
            }],
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_failing_action("gate"),
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            // gate fails; the `pass` edge into the unit does not fire → unit skipped.
            edges: vec![
                make_edge_with("e1", "gate", "a", EdgeOutcome::Pass),
                make_edge("e2", "a", "b"),
            ],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.failed, 1, "gate fails");
        assert_eq!(summary.skipped, 2, "both unit members are skipped");
        assert_eq!(summary.succeeded, 0);
    }

    /// A `Pass` edge from a failing source must NOT fire; downstream is skipped.
    #[tokio::test]
    async fn pass_edge_skips_node_on_failure() {
        let (executor, sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_failing_action("a"),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge_with("e1", "a", "b", EdgeOutcome::Pass)],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.failed, 1, "a must fail");
        assert_eq!(
            summary.skipped, 1,
            "b must be skipped (no incoming edge fired)"
        );
        assert_eq!(summary.succeeded, 0);
        let events = sink.events.lock().unwrap();
        let b_skipped = events.iter().any(|e| {
            matches!(e,
            ExecutionEvent::NodeSkipped { node_id, .. } if node_id == "b")
        });
        assert!(b_skipped, "b should have a skipped event");
    }

    /// A `Fail` edge fires when the source is skipped (skipped is treated as
    /// "did not pass", so it propagates Fail).
    #[tokio::test]
    async fn fail_edge_fires_on_skip() {
        let (executor, _) = build_executor().await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_failing_action("a"),
                make_node(
                    "recover",
                    "action",
                    serde_json::json!({ "adapter": "mock" }),
                ),
            ],
            edges: vec![make_edge_with("e1", "a", "recover", EdgeOutcome::Fail)],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.failed, 1);
        assert_eq!(
            summary.succeeded, 1,
            "recover runs because a Fail edge fired"
        );
    }

    /// Build an executor wired with the real `local` provider so we can
    /// exercise the local-gate + optional-key path. `allow_cloud_ai` is
    /// false throughout to prove the local path is governed independently.
    async fn build_executor_with_local(
        allow_local_ai: bool,
        local_ai_base_url: Option<String>,
    ) -> Executor {
        let adapters = AdapterRegistry::new();
        let mut cloud = CloudAiRegistry::new();
        cloud.register(Arc::new(flow_adapter_ai::LocalOpenAiProvider::new()));
        let credentials: Arc<dyn CredentialResolver> =
            Arc::new(flow_security::EnvFallbackResolver::new(Arc::new(
                flow_security::InMemoryCredentialStore::new(),
            )));
        Executor {
            adapters: Arc::new(adapters),
            sanitizer: Arc::new(PiiSanitizer::new()),
            events: Arc::new(CapturingSink::default()),
            cloud_providers: Arc::new(cloud),
            credentials,
            allow_cloud_ai: false,
            allow_local_ai,
            local_ai_base_url,
            stream_sink: Arc::new(NullStreamSink),
            working_memory: Default::default(),
            control: Default::default(),
            confirm_destructive: false,
            review_gate_available: true,
            edit_staging: None,
            workspace_root: PathBuf::from("."),
        }
    }

    fn local_ai_graph() -> FlowGraph {
        FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![make_node(
                "llm",
                "ai",
                serde_json::json!({
                    "provider": "local",
                    "modelId": "local-llm",
                    "input": "hello",
                }),
            )],
            edges: vec![],
        }
    }

    /// `allow_local_ai = false` skips the node even though `allow_cloud_ai`
    /// is irrelevant - the local provider has its own gate.
    #[tokio::test]
    async fn local_provider_skipped_when_local_gate_off() {
        let exec = build_executor_with_local(false, None).await;
        let summary = exec.run(&local_ai_graph()).await.unwrap();
        assert_eq!(summary.skipped, 1);
        assert_eq!(summary.succeeded, 0);
        assert_eq!(summary.failed, 0);
    }

    /// With the local gate ON and no stored API key, the node must NOT fail
    /// for a missing key (local servers need none). With no base_url set it
    /// fails inside the provider on a Shape error - proving we got past both
    /// the gate and the credential check.
    #[tokio::test]
    async fn local_provider_no_key_required_reaches_provider() {
        let exec = build_executor_with_local(true, None).await;
        let summary = exec.run(&local_ai_graph()).await.unwrap();
        // Failed (no base_url), not Skipped (gate) and not a key error.
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.skipped, 0);
    }

    // --- RAO contract-bound AI nodes (docs/rao-spec) -------------------------

    /// Stub provider replaying fixed responses, registered under the `local`
    /// name so the contract tests drive the real local-AI node path.
    struct ScriptedLocalProvider {
        responses: std::sync::Mutex<VecDeque<String>>,
    }

    impl ScriptedLocalProvider {
        fn new(responses: Vec<&str>) -> Arc<Self> {
            Arc::new(Self {
                responses: std::sync::Mutex::new(
                    responses.into_iter().map(String::from).collect(),
                ),
            })
        }
    }

    #[async_trait::async_trait]
    impl flow_adapter_ai::CloudAiProvider for ScriptedLocalProvider {
        fn name(&self) -> &str {
            "local"
        }
        fn env_var(&self) -> &str {
            "STUB_KEY"
        }
        fn default_models(&self) -> &[&str] {
            &["stub-model"]
        }
        async fn invoke(
            &self,
            req: &CloudAiRequest,
        ) -> Result<flow_adapter_ai::CloudAiResponse, flow_adapter_ai::CloudAiError> {
            let text = self
                .responses
                .lock()
                .unwrap()
                .pop_front()
                .expect("scripted provider ran out of responses");
            Ok(flow_adapter_ai::CloudAiResponse {
                provider: "local".into(),
                model: req.model.clone(),
                text,
                finish_reason: "stop".into(),
                input_tokens: 0,
                output_tokens: 0,
                latency_ms: 0,
            })
        }
    }

    async fn build_contract_executor(responses: Vec<&str>) -> (Executor, CapturingSink) {
        let sink = CapturingSink::default();
        let mut cloud = CloudAiRegistry::new();
        cloud.register(ScriptedLocalProvider::new(responses));
        let mut adapters = AdapterRegistry::new();
        adapters.register(Arc::new(MockAdapter::with_delay(
            "mock",
            std::time::Duration::from_millis(1),
        )));
        let credentials: Arc<dyn CredentialResolver> =
            Arc::new(flow_security::EnvFallbackResolver::new(Arc::new(
                flow_security::InMemoryCredentialStore::new(),
            )));
        let executor = Executor {
            adapters: Arc::new(adapters),
            sanitizer: Arc::new(PiiSanitizer::new()),
            events: Arc::new(sink.clone()),
            cloud_providers: Arc::new(cloud),
            credentials,
            allow_cloud_ai: false,
            allow_local_ai: true,
            local_ai_base_url: Some("http://127.0.0.1:9".into()),
            stream_sink: Arc::new(NullStreamSink),
            working_memory: Default::default(),
            control: Default::default(),
            confirm_destructive: false,
            review_gate_available: true,
            edit_staging: None,
            workspace_root: PathBuf::from("."),
        };
        (executor, sink)
    }

    /// One contract-bound AI node with a `.fail` fallback edge to a mock
    /// recovery action - the RAO reference shape.
    fn contract_graph() -> FlowGraph {
        FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node(
                    "interpret",
                    "ai",
                    serde_json::json!({
                        "provider": "local",
                        "modelId": "stub-model",
                        "input": "interpret this log",
                        "contract": true,
                    }),
                ),
                make_node("fallback", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge_with("e1", "interpret", "fallback", EdgeOutcome::Fail)],
        }
    }

    fn envelope_json(confidence: f64, escalate: bool) -> String {
        serde_json::json!({
            "primary_output": "step 3 failed: missing dataset",
            "confidence": confidence,
            "confidence_type": "verbalized",
            "escalate": escalate,
            "reasoning": "matched the abend code"
        })
        .to_string()
    }

    fn routing_decisions(sink: &CapturingSink) -> Vec<String> {
        sink.events
            .lock()
            .unwrap()
            .iter()
            .filter_map(|e| match e {
                ExecutionEvent::AiRoutingDecision { decision, .. } => Some(decision.clone()),
                _ => None,
            })
            .collect()
    }

    /// Output outside the envelope schema is a failed node (RAO constraint 3),
    /// which routes onto the `.fail` fallback path (constraint 4).
    #[tokio::test]
    async fn contract_violation_fails_node_onto_fallback() {
        let (executor, sink) = build_contract_executor(vec!["sure, here's my analysis"]).await;
        let summary = executor.run(&contract_graph()).await.unwrap();
        assert_eq!(summary.failed, 1, "schema violation fails the ai node");
        assert_eq!(summary.succeeded, 1, "fallback runs via the .fail edge");
        assert_eq!(routing_decisions(&sink), vec!["contract_violation"]);
    }

    /// Confidence above `autoApproveAbove` routes the output downstream with
    /// the envelope fields branchable on the payload.
    #[tokio::test]
    async fn contract_auto_approves_high_confidence() {
        let (executor, sink) =
            build_contract_executor(vec![&envelope_json(0.95, false)]).await;
        let summary = executor.run(&contract_graph()).await.unwrap();
        assert_eq!(summary.succeeded, 1);
        assert_eq!(summary.skipped, 1, "fallback not reached");
        assert_eq!(routing_decisions(&sink), vec!["auto_approve"]);
        let events = sink.events.lock().unwrap();
        let output = events
            .iter()
            .find_map(|e| match e {
                ExecutionEvent::NodeSucceeded { node_id, output, .. }
                    if node_id == "interpret" =>
                {
                    Some(output.clone())
                }
                _ => None,
            })
            .expect("ai node succeeded");
        assert_eq!(
            output.get("primary_output").and_then(|v| v.as_str()),
            Some("step 3 failed: missing dataset")
        );
        assert_eq!(output.get("route").and_then(|v| v.as_str()), Some("auto_approve"));
    }

    /// Confidence below `suppressBelow` suppresses the output: the node fails
    /// and the fallback path runs.
    #[tokio::test]
    async fn contract_suppresses_low_confidence_onto_fallback() {
        let (executor, sink) =
            build_contract_executor(vec![&envelope_json(0.2, false)]).await;
        let summary = executor.run(&contract_graph()).await.unwrap();
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.succeeded, 1, "fallback runs via the .fail edge");
        assert_eq!(routing_decisions(&sink), vec!["suppress"]);
    }

    /// Review-band output with no attached review gate (headless run) fails
    /// deterministically onto the fallback path - never a silent auto-approve.
    #[tokio::test]
    async fn contract_review_band_falls_back_when_gate_unavailable() {
        let (mut executor, sink) =
            build_contract_executor(vec![&envelope_json(0.7, false)]).await;
        executor.review_gate_available = false;
        let summary = executor.run(&contract_graph()).await.unwrap();
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.succeeded, 1, "fallback runs via the .fail edge");
        assert_eq!(routing_decisions(&sink), vec!["human_review"]);
    }

    /// `escalate: true` forces the human gate even at high confidence - the
    /// model can request review but can never bypass it (RAO checklist 5.4).
    #[tokio::test]
    async fn contract_escalate_forces_review_at_high_confidence() {
        let (mut executor, sink) =
            build_contract_executor(vec![&envelope_json(0.99, true)]).await;
        executor.review_gate_available = false;
        let summary = executor.run(&contract_graph()).await.unwrap();
        assert_eq!(summary.failed, 1, "escalated output held back without a gate");
        assert_eq!(routing_decisions(&sink), vec!["human_review"]);
    }

    /// Drive the review gate to a verdict: watch for `AiReviewRequired`, then
    /// resolve via the shared `RunControl` the way the host command does.
    async fn run_with_review_verdict(approved: bool) -> (ExecutionSummary, CapturingSink) {
        let (executor, sink) = build_contract_executor(vec![&envelope_json(0.7, false)]).await;
        let control = executor.control.clone();
        let watcher_sink = sink.clone();
        let watcher = tokio::spawn(async move {
            loop {
                let gated = watcher_sink
                    .events
                    .lock()
                    .unwrap()
                    .iter()
                    .any(|e| matches!(e, ExecutionEvent::AiReviewRequired { .. }));
                if gated {
                    control.resolve_review(approved);
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        });
        let summary = executor.run(&contract_graph()).await.unwrap();
        watcher.await.unwrap();
        (summary, sink)
    }

    /// Approval at the gate passes the contract-bound output downstream, and
    /// the human decision lands in the event trail (RAO checklist 6.4).
    #[tokio::test]
    async fn contract_review_gate_approval_passes_output() {
        let (summary, sink) = run_with_review_verdict(true).await;
        assert_eq!(summary.succeeded, 1);
        assert_eq!(summary.failed, 0);
        let events = sink.events.lock().unwrap();
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::AiReviewResolved { approved: true, .. }
        )));
    }

    /// Rejection at the gate fails the node onto its fallback path.
    #[tokio::test]
    async fn contract_review_gate_rejection_routes_fallback() {
        let (summary, sink) = run_with_review_verdict(false).await;
        assert_eq!(summary.failed, 1);
        assert_eq!(summary.succeeded, 1, "fallback runs via the .fail edge");
        let events = sink.events.lock().unwrap();
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::AiReviewResolved { approved: false, .. }
        )));
    }

    /// Every AI invocation is recorded as a system event with its input
    /// (RAO checklist 6.1), contract-bound or not.
    #[tokio::test]
    async fn ai_invocation_is_recorded() {
        let (executor, sink) = build_contract_executor(vec!["plain text answer"]).await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![make_node(
                "llm",
                "ai",
                serde_json::json!({
                    "provider": "local",
                    "modelId": "stub-model",
                    "input": "hello",
                }),
            )],
            edges: vec![],
        };
        executor.run(&graph).await.unwrap();
        let events = sink.events.lock().unwrap();
        let recorded = events.iter().any(|e| matches!(
            e,
            ExecutionEvent::AiInvocation { provider, input, .. }
                if provider == "local" && input == "hello"
        ));
        assert!(recorded, "AiInvocation event with the sanitized input");
    }

    /// `Always` edges fire regardless of the source's terminal status, so
    /// the target runs even when the source failed.
    #[tokio::test]
    async fn always_edges_run_target_unconditionally() {
        let (executor, _) = build_executor().await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_failing_action("a"),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![make_edge_with("e1", "a", "b", EdgeOutcome::Always)],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.failed, 1, "a still fails");
        assert_eq!(
            summary.succeeded, 1,
            "b runs despite a's failure (Always edge)"
        );
    }

    // --- inter-node data flow (`{{...}}` interpolation) ---------------------

    fn output_for(sink: &CapturingSink, node_id: &str) -> Option<serde_json::Value> {
        sink.events.lock().unwrap().iter().find_map(|e| match e {
            ExecutionEvent::NodeSucceeded {
                node_id: nid,
                output,
                ..
            } if nid == node_id => Some(output.clone()),
            _ => None,
        })
    }

    #[test]
    fn primary_text_extracts_common_keys_and_scalars() {
        assert_eq!(primary_text(&serde_json::json!("hi")), "hi");
        assert_eq!(primary_text(&serde_json::json!({ "text": "abc" })), "abc");
        assert_eq!(primary_text(&serde_json::json!({ "stdout": "out" })), "out");
        assert_eq!(primary_text(&serde_json::json!(42)), "42");
        assert_eq!(primary_text(&serde_json::json!(true)), "true");
        assert_eq!(primary_text(&serde_json::Value::Null), "");
        // No text-bearing key → compact JSON.
        assert_eq!(
            primary_text(&serde_json::json!({ "a": 1 })),
            "{\"a\":1}".to_string()
        );
    }

    #[test]
    fn resolve_token_handles_input_id_and_path() {
        let mut outputs = HashMap::new();
        outputs.insert("a".to_string(), serde_json::json!({ "text": "hello" }));
        outputs.insert(
            "b".to_string(),
            serde_json::json!({ "nested": { "k": "deep" } }),
        );
        let upstream = vec!["a".to_string()];
        let mut memory = HashMap::new();
        memory.insert("region".to_string(), serde_json::json!("us-east-1"));
        memory.insert("cfg".to_string(), serde_json::json!({ "retries": 3 }));

        assert_eq!(resolve_token("input", &outputs, &upstream, &memory), "hello");
        assert_eq!(resolve_token("a", &outputs, &upstream, &memory), "hello");
        assert_eq!(resolve_token("a.text", &outputs, &upstream, &memory), "hello");
        assert_eq!(resolve_token("b.nested.k", &outputs, &upstream, &memory), "deep");
        // Working-memory reads.
        assert_eq!(resolve_token("memory.region", &outputs, &upstream, &memory), "us-east-1");
        assert_eq!(resolve_token("memory.cfg.retries", &outputs, &upstream, &memory), "3");
        // Unknown id / missing field / missing memory key → empty.
        assert_eq!(resolve_token("missing", &outputs, &upstream, &memory), "");
        assert_eq!(resolve_token("a.nope", &outputs, &upstream, &memory), "");
        assert_eq!(resolve_token("memory.nope", &outputs, &upstream, &memory), "");
    }

    #[test]
    fn interpolate_str_replaces_tokens_and_preserves_literals() {
        let mut outputs = HashMap::new();
        outputs.insert("a".to_string(), serde_json::json!({ "text": "world" }));
        let upstream = vec!["a".to_string()];
        let memory = HashMap::new();

        assert_eq!(
            interpolate_str("hi {{a.text}}!", &outputs, &upstream, &memory),
            "hi world!"
        );
        assert_eq!(
            interpolate_str("no tokens here", &outputs, &upstream, &memory),
            "no tokens here"
        );
        // Unterminated braces are left literal.
        assert_eq!(
            interpolate_str("oops {{a.text", &outputs, &upstream, &memory),
            "oops {{a.text"
        );
    }

    #[test]
    fn interpolate_value_recurses_into_arrays_and_objects() {
        let mut outputs = HashMap::new();
        outputs.insert("a".to_string(), serde_json::json!({ "text": "X" }));
        let upstream = vec!["a".to_string()];
        let memory = HashMap::new();
        let data = serde_json::json!({
            "command": "echo {{a.text}}",
            "args": ["--msg", "{{a.text}}"],
            "count": 3,
        });
        let resolved = interpolate_value(&data, &outputs, &upstream, &memory);
        assert_eq!(resolved["command"], serde_json::json!("echo X"));
        assert_eq!(resolved["args"][1], serde_json::json!("X"));
        assert_eq!(resolved["count"], serde_json::json!(3));
    }

    #[tokio::test]
    async fn upstream_output_flows_into_downstream_field() {
        // The mock adapter echoes `data.label` into its output payload (under
        // the `{status, payload}` wrapper). Node B references node A's label via
        // `{{a.payload.label}}`, so B's resolved label - and thus B's echoed
        // output - must carry A's value.
        let (executor, sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("a", "action", serde_json::json!({ "adapter": "mock", "label": "hello" })),
                make_node(
                    "b",
                    "action",
                    serde_json::json!({ "adapter": "mock", "label": "got: {{a.payload.label}}" }),
                ),
            ],
            edges: vec![make_edge("e1", "a", "b")],
        };
        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(summary.succeeded, 2);
        let b_out = output_for(&sink, "b").expect("b succeeded");
        assert_eq!(b_out["payload"]["label"], serde_json::json!("got: hello"));
    }

    #[tokio::test]
    async fn input_token_resolves_to_upstream_output() {
        // `{{input}}` resolves to the upstream node's primary text. The mock
        // payload has no text-bearing key, so `primary_text` descends into the
        // payload and serializes it - assert the downstream label embeds the
        // upstream label.
        let (executor, sink) = build_executor().await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("a", "action", serde_json::json!({ "adapter": "mock", "label": "seed" })),
                make_node(
                    "b",
                    "action",
                    serde_json::json!({ "adapter": "mock", "label": "{{input}}" }),
                ),
            ],
            edges: vec![make_edge("e1", "a", "b")],
        };
        executor.run(&graph).await.unwrap();
        let b_out = output_for(&sink, "b").expect("b succeeded");
        let label = b_out["payload"]["label"].as_str().unwrap();
        assert!(
            label.contains("\"label\":\"seed\""),
            "b's label embeds a's serialized output: {label}"
        );
    }

    // --- executor iteration: when-conditions + bounded loops ----------------

    use crate::adapter::{Adapter, AdapterError, NodeOutput};
    use std::sync::atomic::{AtomicI64, Ordering};

    /// Edge with a `when` condition, otherwise an `Always` Pass-equivalent.
    fn make_edge_when(id: &str, src: &str, dst: &str, outcome: EdgeOutcome, when: &str) -> FlowEdge {
        FlowEdge {
            id: id.into(),
            source: src.into(),
            target: dst.into(),
            label: None,
            condition: Some(when.into()),
            outcome,
        }
    }

    /// Adapter that returns `payload.remaining` decreasing by 1 each call,
    /// so a back-edge `when {{<id>.payload.remaining}} > 0` exits after N runs.
    struct CountdownAdapter {
        name: String,
        remaining: AtomicI64,
    }

    impl CountdownAdapter {
        fn new(name: &str, start: i64) -> Self {
            Self {
                name: name.into(),
                remaining: AtomicI64::new(start),
            }
        }
    }

    #[async_trait::async_trait]
    impl Adapter for CountdownAdapter {
        fn name(&self) -> &str {
            &self.name
        }
        async fn execute(&self, _node: &FlowNode) -> Result<NodeOutput, AdapterError> {
            let after = self.remaining.fetch_sub(1, Ordering::SeqCst) - 1;
            Ok(NodeOutput {
                status: "succeeded".into(),
                payload: serde_json::json!({ "remaining": after }),
            })
        }
    }

    async fn build_executor_with_adapters(extra: Vec<Arc<dyn Adapter>>) -> (Executor, CapturingSink) {
        build_executor_with(extra, WorkingMemory::default()).await
    }

    async fn build_executor_with(
        extra: Vec<Arc<dyn Adapter>>,
        working_memory: WorkingMemory,
    ) -> (Executor, CapturingSink) {
        let mut adapters = AdapterRegistry::new();
        adapters.register(Arc::new(MockAdapter::with_delay(
            "mock",
            std::time::Duration::from_millis(1),
        )));
        for a in extra {
            adapters.register(a);
        }
        let sink = CapturingSink::default();
        let credentials: Arc<dyn CredentialResolver> =
            Arc::new(flow_security::EnvFallbackResolver::new(Arc::new(
                flow_security::InMemoryCredentialStore::new(),
            )));
        let exec = Executor {
            adapters: Arc::new(adapters),
            sanitizer: Arc::new(PiiSanitizer::new()),
            events: Arc::new(sink.clone()),
            cloud_providers: Arc::new(CloudAiRegistry::new()),
            credentials,
            allow_cloud_ai: false,
            allow_local_ai: false,
            local_ai_base_url: None,
            stream_sink: Arc::new(NullStreamSink),
            working_memory,
            control: Default::default(),
            confirm_destructive: false,
            review_gate_available: true,
            edit_staging: None,
            workspace_root: PathBuf::from("."),
        };
        (exec, sink)
    }

    /// Test adapter mirroring the utility `set-variable` payload contract:
    /// emits `{ actionId: "set-variable", name, value }` so the executor's
    /// `store_set_variable` writes working memory. Avoids depending on the
    /// `flow-adapter-utility` crate (layering).
    struct SetVarAdapter;

    #[async_trait::async_trait]
    impl Adapter for SetVarAdapter {
        fn name(&self) -> &str {
            "setvar"
        }
        async fn execute(&self, node: &FlowNode) -> Result<NodeOutput, AdapterError> {
            let name = node
                .data
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            let value = node.data.get("value").cloned().unwrap_or(serde_json::Value::Null);
            Ok(NodeOutput {
                status: "succeeded".into(),
                payload: serde_json::json!({
                    "actionId": "set-variable",
                    "name": name,
                    "value": value,
                }),
            })
        }
    }

    fn count_succeeded(sink: &CapturingSink, node_id: &str) -> usize {
        sink.events
            .lock()
            .unwrap()
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::NodeSucceeded { node_id: nid, .. } if nid == node_id))
            .count()
    }

    fn was_skipped(sink: &CapturingSink, node_id: &str) -> bool {
        sink.events
            .lock()
            .unwrap()
            .iter()
            .any(|e| matches!(e, ExecutionEvent::NodeSkipped { node_id: nid, .. } if nid == node_id))
    }

    fn count_capped(sink: &CapturingSink, node_id: &str) -> usize {
        sink.events
            .lock()
            .unwrap()
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::IterationCapped { node_id: nid, .. } if nid == node_id))
            .count()
    }

    #[tokio::test]
    async fn when_condition_gates_edge_on_dag() {
        let (executor, sink) = build_executor().await;
        // src emits label "go"; the edge whose condition matches fires, the
        // other is gated off and its target is skipped.
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("src", "action", serde_json::json!({ "adapter": "mock", "label": "go" })),
                make_node("a", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("b", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![
                make_edge_when("e1", "src", "a", EdgeOutcome::Pass, "{{src.payload.label}} == 'go'"),
                make_edge_when("e2", "src", "b", EdgeOutcome::Pass, "{{src.payload.label}} == 'stop'"),
            ],
        };

        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(count_succeeded(&sink, "a"), 1, "matching condition runs a");
        assert!(was_skipped(&sink, "b"), "non-matching condition skips b");
        // src+a succeeded, b skipped via routing - a skip alongside successes
        // is not a failure, so the run is still "succeeded".
        assert_eq!(summary.status, "succeeded");
    }

    #[tokio::test]
    async fn bounded_loop_converges_on_when_exit() {
        let countdown = Arc::new(CountdownAdapter::new("countdown", 3));
        let (executor, sink) = build_executor_with_adapters(vec![countdown]).await;
        // start -> work -> check, with check looping back to work while
        // remaining > 0. Countdown starts at 3, so check exits after 3 runs.
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("start", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("work", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("check", "action", serde_json::json!({ "adapter": "countdown" })),
            ],
            edges: vec![
                make_edge("e1", "start", "work"),
                make_edge("e2", "work", "check"),
                make_edge_when(
                    "e3",
                    "check",
                    "work",
                    EdgeOutcome::Always,
                    "{{check.payload.remaining}} > 0",
                ),
            ],
        };

        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(count_succeeded(&sink, "check"), 3, "loop runs until remaining hits 0");
        assert_eq!(count_succeeded(&sink, "work"), 3);
        assert_eq!(count_capped(&sink, "check"), 0, "converged before the cap");
        assert_eq!(summary.status, "succeeded");
    }

    #[tokio::test]
    async fn runaway_loop_hits_cap_and_finishes_partial() {
        let (executor, sink) = build_executor().await;
        // A self-loop with an always-active back-edge never converges; the
        // per-node cap stops it and the run ends `partial` rather than hanging.
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("start", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("loop", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![
                make_edge("e1", "start", "loop"),
                make_edge("e2", "loop", "loop"),
            ],
        };

        let summary = executor.run(&graph).await.unwrap();
        assert_eq!(
            count_succeeded(&sink, "loop"),
            PER_NODE_VISIT_CAP as usize,
            "node runs exactly up to the cap"
        );
        assert_eq!(count_capped(&sink, "loop"), 1, "one cap diagnostic emitted");
        assert_eq!(summary.status, "partial");
    }

    // --- working memory: set-variable writes, {{memory.x}} reads ------------

    fn set_var_node(id: &str, name: &str, value: &str) -> FlowNode {
        make_node(
            id,
            "utility",
            serde_json::json!({ "adapter": "setvar", "name": name, "value": value }),
        )
    }

    #[tokio::test]
    async fn set_variable_writes_memory_and_downstream_reads_it() {
        let (executor, sink) = build_executor_with_adapters(vec![Arc::new(SetVarAdapter)]).await;
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                set_var_node("set", "region", "us-east-1"),
                make_node(
                    "reader",
                    "action",
                    serde_json::json!({ "adapter": "mock", "label": "{{memory.region}}" }),
                ),
            ],
            edges: vec![make_edge("e1", "set", "reader")],
        };

        executor.run(&graph).await.unwrap();
        let reader = output_for(&sink, "reader").expect("reader succeeded");
        assert_eq!(
            reader["payload"]["label"].as_str(),
            Some("us-east-1"),
            "downstream node reads the value written via set-variable"
        );
    }

    #[tokio::test]
    async fn working_memory_persists_across_runs() {
        // A shared store mirrors FlowApp owning memory across execute_graph
        // calls (i.e. across re-plan iterations).
        let memory = WorkingMemory::default();
        let (executor, sink) =
            build_executor_with(vec![Arc::new(SetVarAdapter)], memory.clone()).await;

        // Run 1: write the variable.
        let write_graph = FlowGraph {
            subflows: Vec::new(),
            id: "g1".into(),
            name: "g1".into(),
            version: "1".into(),
            description: None,
            nodes: vec![set_var_node("set", "token", "abc123")],
            edges: vec![],
        };
        executor.run(&write_graph).await.unwrap();
        assert_eq!(memory.lock().unwrap().get("token").and_then(|v| v.as_str()), Some("abc123"));

        // Run 2: a separate graph reads it - the store survived the first run.
        let read_graph = FlowGraph {
            subflows: Vec::new(),
            id: "g2".into(),
            name: "g2".into(),
            version: "1".into(),
            description: None,
            nodes: vec![make_node(
                "reader",
                "action",
                serde_json::json!({ "adapter": "mock", "label": "{{memory.token}}" }),
            )],
            edges: vec![],
        };
        executor.run(&read_graph).await.unwrap();
        let reader = output_for(&sink, "reader").expect("reader succeeded");
        assert_eq!(reader["payload"]["label"].as_str(), Some("abc123"));
    }

    #[tokio::test]
    async fn when_condition_reads_memory() {
        let (executor, sink) = build_executor_with_adapters(vec![Arc::new(SetVarAdapter)]).await;
        // set flag=go, then a Pass edge gated on memory routes to `go_node`
        // while the mismatched edge skips `stop_node`.
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                set_var_node("set", "flag", "go"),
                make_node("go_node", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("stop_node", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![
                make_edge_when("e1", "set", "go_node", EdgeOutcome::Pass, "{{memory.flag}} == 'go'"),
                make_edge_when(
                    "e2",
                    "set",
                    "stop_node",
                    EdgeOutcome::Pass,
                    "{{memory.flag}} == 'stop'",
                ),
            ],
        };

        executor.run(&graph).await.unwrap();
        assert_eq!(count_succeeded(&sink, "go_node"), 1, "memory-matched edge fires");
        assert!(was_skipped(&sink, "stop_node"), "memory-mismatched edge skips");
    }

    #[test]
    fn destructive_reason_flags_data_loss_only() {
        let del = make_node(
            "d",
            "action",
            serde_json::json!({"adapter":"fs","actionId":"delete-file","path":"build/"}),
        );
        assert!(destructive_reason(&del).is_some(), "fs delete-file");

        let rm = make_node(
            "s",
            "action",
            serde_json::json!({"adapter":"shell","actionId":"run-command","command":"rm","args":["-rf","dist"]}),
        );
        assert!(destructive_reason(&rm).is_some(), "shell rm");

        let push = make_node(
            "g",
            "action",
            serde_json::json!({"adapter":"shell","actionId":"git","args":["push","origin","main"]}),
        );
        assert!(destructive_reason(&push).is_some(), "git push");

        let read = make_node(
            "r",
            "action",
            serde_json::json!({"adapter":"fs","actionId":"read-file","path":"x"}),
        );
        assert!(destructive_reason(&read).is_none(), "read is safe");

        // No substring false positive: `confirm` must not match the `rm` token.
        let confirm = make_node(
            "c",
            "action",
            serde_json::json!({"adapter":"shell","actionId":"run-command","command":"./confirm","args":["build"]}),
        );
        assert!(destructive_reason(&confirm).is_none(), "no rm substring match");
    }

    #[test]
    fn sandbox_escape_flags_parent_traversal() {
        let esc = make_node(
            "e",
            "action",
            serde_json::json!({"adapter":"fs","actionId":"read-file","path":"../../etc/passwd"}),
        );
        assert!(sandbox_escape_reason(&esc).is_some(), "`..` path flagged");
        let safe = make_node(
            "s",
            "action",
            serde_json::json!({"adapter":"fs","actionId":"read-file","path":"src/main.rs"}),
        );
        assert!(sandbox_escape_reason(&safe).is_none(), "in-jail path is fine");
        // No false positive on `..` as a substring (not a whole segment).
        let dots = make_node(
            "d",
            "action",
            serde_json::json!({"adapter":"fs","actionId":"read-file","path":"a..b.txt"}),
        );
        assert!(
            sandbox_escape_reason(&dots).is_none(),
            "`..` substring not flagged"
        );
    }

    #[test]
    fn verify_graph_collects_destructive_and_escape() {
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node(
                    "del",
                    "action",
                    serde_json::json!({"adapter":"fs","actionId":"delete-file","path":"x"}),
                ),
                make_node(
                    "esc",
                    "action",
                    serde_json::json!({"adapter":"fs","actionId":"read-file","path":"../x"}),
                ),
                make_node(
                    "ok",
                    "action",
                    serde_json::json!({"adapter":"fs","actionId":"read-file","path":"x"}),
                ),
            ],
            edges: vec![],
        };
        let w = verify_graph(&graph);
        assert!(w.iter().any(|x| x.node_id == "del" && x.severity == "destructive"));
        assert!(w
            .iter()
            .any(|x| x.node_id == "esc" && x.severity == "sandbox-escape"));
        assert!(!w.iter().any(|x| x.node_id == "ok"));
    }

    #[test]
    fn fallback_node_swaps_provider_and_strips_fallback_fields() {
        let n = make_node(
            "ai",
            "ai",
            serde_json::json!({
                "provider": "claude",
                "modelId": "m1",
                "fallbackProvider": "local",
                "fallbackModelId": "m2",
                "input": "x"
            }),
        );
        let fb = fallback_node(&n).expect("has fallback");
        assert_eq!(fb.data.get("provider").unwrap(), "local");
        assert_eq!(fb.data.get("modelId").unwrap(), "m2");
        assert!(fb.data.get("fallbackProvider").is_none());
        assert!(fb.data.get("fallbackModelId").is_none());
        assert_eq!(fb.data.get("input").unwrap(), "x"); // other fields preserved

        // No fallbackProvider → no failover variant.
        let no_fb = make_node(
            "ai",
            "ai",
            serde_json::json!({"provider":"claude","modelId":"m1"}),
        );
        assert!(fallback_node(&no_fb).is_none());
    }

    #[test]
    fn run_control_phase_transitions() {
        let c = RunControl::default();
        assert_eq!(c.phase(), RunPhase::Running);
        c.pause();
        assert_eq!(c.phase(), RunPhase::Paused);
        c.resume();
        assert_eq!(c.phase(), RunPhase::Running);
        c.cancel();
        assert!(c.is_cancelling());
        // pause/resume are no-ops once cancelling; cancel wins.
        c.pause();
        c.resume();
        assert!(c.is_cancelling());
        // reset clears back to running for the next run.
        c.reset();
        assert_eq!(c.phase(), RunPhase::Running);
    }

    #[tokio::test]
    async fn wait_while_paused_returns_immediately_when_not_paused() {
        let c = RunControl::default();
        // Running: returns at once.
        c.wait_while_paused().await;
        // Cancelling (not Paused): also returns at once rather than blocking.
        c.cancel();
        c.wait_while_paused().await;
    }

    #[tokio::test]
    async fn wait_while_paused_unblocks_on_resume() {
        let c = Arc::new(RunControl::default());
        c.pause();
        let c2 = c.clone();
        let waiter = tokio::spawn(async move { c2.wait_while_paused().await });
        // Let the waiter park inside `notified()` before releasing it.
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        c.resume();
        tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("waiter did not wake within timeout")
            .expect("waiter task panicked");
        assert_eq!(c.phase(), RunPhase::Running);
    }

    /// Test adapter that cancels the shared run control the first time it runs,
    /// so the next node-boundary checkpoint observes the cancellation.
    struct CancelTriggerAdapter {
        control: SharedRunControl,
    }

    #[async_trait::async_trait]
    impl Adapter for CancelTriggerAdapter {
        fn name(&self) -> &str {
            "cancel-trigger"
        }
        async fn execute(&self, _node: &FlowNode) -> Result<NodeOutput, AdapterError> {
            self.control.cancel();
            Ok(NodeOutput {
                status: "succeeded".into(),
                payload: serde_json::Value::Null,
            })
        }
    }

    #[tokio::test]
    async fn cancel_skips_remaining_nodes_and_reports_cancelled() {
        let (mut exec, sink) = build_executor().await;
        let control = exec.control.clone();
        let trigger: Arc<dyn Adapter> = Arc::new(CancelTriggerAdapter { control });
        // Re-register the adapter set with the cancel trigger added.
        let mut adapters = AdapterRegistry::new();
        adapters.register(Arc::new(MockAdapter::with_delay(
            "mock",
            std::time::Duration::from_millis(1),
        )));
        adapters.register(trigger);
        exec.adapters = Arc::new(adapters);

        // n1 cancels on execution; n2 and n3 should never run.
        let graph = FlowGraph {
            subflows: Vec::new(),
            id: "g".into(),
            name: "g".into(),
            version: "1".into(),
            description: None,
            nodes: vec![
                make_node("n1", "action", serde_json::json!({ "adapter": "cancel-trigger" })),
                make_node("n2", "action", serde_json::json!({ "adapter": "mock" })),
                make_node("n3", "action", serde_json::json!({ "adapter": "mock" })),
            ],
            edges: vec![
                make_edge("e1", "n1", "n2"),
                make_edge("e2", "n2", "n3"),
            ],
        };

        let summary = exec.run(&graph).await.unwrap();
        assert_eq!(summary.status, "cancelled", "cancel yields cancelled status");
        assert_eq!(count_succeeded(&sink, "n1"), 1, "the triggering node still completes");
        assert!(was_skipped(&sink, "n2"), "downstream node is skipped after cancel");
        assert!(was_skipped(&sink, "n3"), "all remaining nodes are skipped after cancel");
    }
}