zebflow 0.9.0

//! Real framework engine with graph traversal and built-in node dispatch.
//!
//! # Engine-level common config
//!
//! The engine reads certain keys from each node's resolved config before
//! dispatching execution.  These are injected as DSL flags into every node
//! definition by [`builtin_node_definitions()`] so they are available in both
//! DSL and the UI pipeline editor.
//!
//! | Config key      | DSL flag      | Description |
//! |-----------------|---------------|-------------|
//! | `timeout_secs`  | `--timeout`   | Per-node execution timeout in seconds (5–3600). Overrides project-level `pipeline_node_timeout_secs`. |

use std::collections::hash_map::DefaultHasher;
use std::collections::{HashMap, HashSet, VecDeque};
use std::hash::{Hash, Hasher};
use std::sync::{Arc, RwLock};

use async_trait::async_trait;
use base64::Engine as _;
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};

use super::composite_host::execute_installed_node;
use crate::infra::io::state::{DynStateBus, MemStateBus};
use crate::infra::mem::MemHub;
use crate::infra::transport::ws::WsHub;
use crate::infra::ws_client::WsClientManager;
use crate::language::{DenoSandboxEngine, LanguageEngine};
use crate::pipeline::expr::{resolve_config_expressions, scanner::scan as scan_exprs};
use crate::pipeline::interface::PipelineEngine;
use crate::pipeline::model::{
    ExecuteOptions, ExecutionBus, NodeTraceEntry, PipelineContext, PipelineError, PipelineGraph,
    PipelineNode, PipelineOutput, Signal,
};
use crate::pipeline::nodes::shared::file_ref::{BACKEND_ZEBFS, FILE_REF_TYPE, LIFECYCLE_DURABLE};
use crate::pipeline::nodes::basic::{
    ai, auth, browser, concept, crypto, fs, function, geo, http, input, kv, logic, mail, ms, pg,
    script, sekejap, sqlite, table,
    trigger::{
        function as trigger_function, kv_subscribe, manual, mcp_trigger, schedule, weberror,
        webhook, ws_client as trigger_ws_client,
    },
    web, ws,
};
use crate::pipeline::nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler};
use crate::pipeline::trace_capture::{TraceCapture, redact_literal_secrets};
#[cfg(test)]
use crate::pipeline::trace_capture::TraceCaptureSettings;
use crate::platform::services::CredentialService;
use crate::platform::services::PlatformService;
use crate::rwe::{ReactiveWebEngine, TemplateSource, resolve_engine_or_default};

/// A single entry in the template compile cache.
/// Pairs the compiled page artifact with the set of component files it depends on,
/// enabling dependency-aware eviction when a component is edited.
pub struct CacheEntry {
    /// The compiled page artifact.
    pub page: Arc<web::response::CompiledPage>,
    /// Absolute filesystem paths of all component files inlined during compilation.
    /// Populated from `CompiledTemplate.dependency_paths` (the `visited` set of
    /// `collect_inlined_module`). When any of these paths change on disk,
    /// this entry is evicted so the next request recompiles with the updated component.
    pub dependencies: std::collections::HashSet<String>,
}

/// In-memory compile cache for template nodes.
///
/// Key: hash of the entry-page markup string.
/// Value: compiled page + dependency set for targeted eviction.
///
/// Entry-page changes → new hash → automatic cache miss.
/// Component changes → `evict_template_cache_by_path` removes affected entries.
///
/// Uses `RwLock` so concurrent reads (cache hits) never block each other;
/// only cache-miss writes take an exclusive lock.
pub type TemplateCache = Arc<RwLock<HashMap<u64, CacheEntry>>>;

/// Create a new empty template compile cache.
pub fn new_template_cache() -> TemplateCache {
    Arc::new(RwLock::new(HashMap::new()))
}

/// Evict all cache entries whose dependency set includes `abs_path`.
/// Called whenever any template or component file is written (UI, MCP, API).
/// Entry-page saves don't need this — they already cause a hash miss automatically.
pub fn evict_template_cache_by_path(cache: &TemplateCache, abs_path: &str) {
    cache
        .write()
        .unwrap_or_else(|e| e.into_inner())
        .retain(|_, entry| !entry.dependencies.contains(abs_path));
}

#[derive(Debug, Clone, Default)]
struct ReducePendingState {
    acc: Option<Value>,
    received: usize,
    expected: Option<usize>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum NodesAccess {
    None,
    Exact(HashSet<String>),
}

impl Default for NodesAccess {
    fn default() -> Self {
        Self::None
    }
}

#[derive(Debug, Clone, Default)]
struct NodesRetentionPlan {
    consumer_access: HashMap<String, NodesAccess>,
    retained_nodes: HashSet<String>,
}

const RETRY_STATE_KEY: &str = "__zf_retry";
const MAX_NODE_OUTPUT_FILE_BYTES: usize = 256 * 1024 * 1024;

fn hash_markup(s: &str) -> u64 {
    let mut h = DefaultHasher::new();
    s.hash(&mut h);
    h.finish()
}

fn build_nodes_retention_plan(graph: &PipelineGraph) -> Result<NodesRetentionPlan, PipelineError> {
    let mut plan = NodesRetentionPlan::default();

    for node in &graph.nodes {
        let mut access = scan_node_nodes_access(node)?;
        // A retry node counts its own attempts from its last output
        // (`$nodes.<self>.__zf_retry`), so it is always retained and always
        // in its own scope. Without this, a loop whose payload is replaced
        // on the way round (`http.request` answers with a fresh body) would
        // sit at attempt 1 for ever — the poll loops needed a script whose
        // only job was to copy the count back.
        if node.kind == logic::retry::NODE_KIND {
            plan.retained_nodes.insert(node.id.clone());
            access = match access {
                NodesAccess::None => NodesAccess::Exact(HashSet::from([node.id.clone()])),
                NodesAccess::Exact(mut ids) => {
                    ids.insert(node.id.clone());
                    NodesAccess::Exact(ids)
                }
            };
        }
        match &access {
            NodesAccess::None => {}
            NodesAccess::Exact(ids) => {
                plan.retained_nodes.extend(ids.iter().cloned());
            }
        }
        plan.consumer_access.insert(node.id.clone(), access);
    }

    Ok(plan)
}

fn scan_node_nodes_access(node: &PipelineNode) -> Result<NodesAccess, PipelineError> {
    let mut access = NodesAccessAccumulator::default();
    scan_value_nodes_access(node, &node.config, &mut access)?;
    Ok(access.finish())
}

#[derive(Debug, Default)]
struct NodesAccessAccumulator {
    refs: HashSet<String>,
}

impl NodesAccessAccumulator {
    fn mark_ref(&mut self, id: String) {
        self.refs.insert(id);
    }

    fn finish(self) -> NodesAccess {
        if self.refs.is_empty() {
            NodesAccess::None
        } else {
            NodesAccess::Exact(self.refs)
        }
    }
}

fn scan_value_nodes_access(
    node: &PipelineNode,
    value: &Value,
    access: &mut NodesAccessAccumulator,
) -> Result<(), PipelineError> {
    match value {
        Value::String(text) => scan_text_nodes_access(node, text, access)?,
        Value::Array(items) => {
            for item in items {
                scan_value_nodes_access(node, item, access)?;
            }
        }
        Value::Object(map) => {
            for item in map.values() {
                scan_value_nodes_access(node, item, access)?;
            }
        }
        _ => {}
    }
    Ok(())
}

fn scan_text_nodes_access(
    node: &PipelineNode,
    text: &str,
    access: &mut NodesAccessAccumulator,
) -> Result<(), PipelineError> {
    scan_nodes_marker(node, text, "$nodes", access)?;
    scan_nodes_marker(node, text, "ctx.nodes", access)?;
    Ok(())
}

fn scan_nodes_marker(
    node: &PipelineNode,
    text: &str,
    marker: &str,
    access: &mut NodesAccessAccumulator,
) -> Result<(), PipelineError> {
    let mut offset = 0;
    while let Some(found) = text[offset..].find(marker) {
        let marker_start = offset + found;
        let marker_end = marker_start + marker.len();
        if !marker_boundary_ok(text, marker_start, marker_end) {
            offset = marker_end;
            continue;
        }

        match parse_nodes_reference_after_marker(text, marker_end) {
            NodesReferenceScan::Exact(id, next) => {
                access.mark_ref(id);
                offset = next;
            }
            NodesReferenceScan::Dynamic(_) => {
                return Err(PipelineError::new(
                    "FW_NODES_SCOPE_DYNAMIC",
                    format!(
                        "node '{}' uses dynamic {} access; use a literal node id like {}.foo or {}[\"foo\"]",
                        node.id, marker, marker, marker
                    ),
                ));
            }
            NodesReferenceScan::NoReference(next) => {
                offset = next;
            }
        }
    }
    Ok(())
}

fn marker_boundary_ok(text: &str, start: usize, end: usize) -> bool {
    let before_ok = text[..start]
        .chars()
        .next_back()
        .is_none_or(|ch| !is_js_ident_continue(ch));
    let after_ok = text[end..]
        .chars()
        .next()
        .is_none_or(|ch| !is_js_ident_continue(ch));
    before_ok && after_ok
}

#[derive(Debug, PartialEq, Eq)]
enum NodesReferenceScan {
    Exact(String, usize),
    Dynamic(usize),
    NoReference(usize),
}

fn parse_nodes_reference_after_marker(text: &str, marker_end: usize) -> NodesReferenceScan {
    let mut idx = skip_ascii_ws(text, marker_end);
    let Some(ch) = text[idx..].chars().next() else {
        return NodesReferenceScan::NoReference(idx);
    };

    if ch == '.' {
        idx += ch.len_utf8();
        let start = idx;
        while let Some(next) = text[idx..].chars().next() {
            if is_js_ident_continue(next) || next == '-' {
                idx += next.len_utf8();
            } else {
                break;
            }
        }
        if idx == start {
            return NodesReferenceScan::Dynamic(idx);
        }
        return NodesReferenceScan::Exact(text[start..idx].to_string(), idx);
    }

    if ch == '[' {
        return parse_bracket_nodes_reference(text, idx);
    }

    NodesReferenceScan::Dynamic(idx)
}

fn parse_bracket_nodes_reference(text: &str, bracket_start: usize) -> NodesReferenceScan {
    let mut idx = skip_ascii_ws(text, bracket_start + 1);
    let Some(quote) = text[idx..].chars().next() else {
        return NodesReferenceScan::Dynamic(idx);
    };
    if quote != '\'' && quote != '"' {
        return NodesReferenceScan::Dynamic(idx);
    }
    idx += quote.len_utf8();
    let value_start = idx;
    let mut escaped = false;
    let mut out = String::new();
    while let Some(ch) = text[idx..].chars().next() {
        idx += ch.len_utf8();
        if escaped {
            out.push(ch);
            escaped = false;
            continue;
        }
        if ch == '\\' {
            escaped = true;
            continue;
        }
        if ch == quote {
            let after_quote = skip_ascii_ws(text, idx);
            if text[after_quote..].starts_with(']') {
                let id = if out.is_empty() {
                    text[value_start..idx - quote.len_utf8()].to_string()
                } else {
                    out
                };
                if id.trim().is_empty() {
                    return NodesReferenceScan::Dynamic(after_quote + 1);
                }
                return NodesReferenceScan::Exact(id, after_quote + 1);
            }
            return NodesReferenceScan::Dynamic(after_quote);
        }
        out.push(ch);
    }
    NodesReferenceScan::Dynamic(idx)
}

fn skip_ascii_ws(text: &str, mut idx: usize) -> usize {
    while let Some(ch) = text[idx..].chars().next() {
        if ch.is_ascii_whitespace() {
            idx += ch.len_utf8();
        } else {
            break;
        }
    }
    idx
}

fn is_js_ident_continue(ch: char) -> bool {
    ch == '_' || ch == '$' || ch.is_ascii_alphanumeric()
}

fn nodes_scope_for_target(
    target_node_id: &str,
    nodes_output: &serde_json::Map<String, Value>,
    retention: &NodesRetentionPlan,
) -> Value {
    match retention
        .consumer_access
        .get(target_node_id)
        .unwrap_or(&NodesAccess::None)
    {
        NodesAccess::None => json!({}),
        NodesAccess::Exact(ids) => {
            let mut scoped = serde_json::Map::new();
            for id in ids {
                if let Some(value) = nodes_output.get(id) {
                    scoped.insert(id.clone(), value.clone());
                }
            }
            Value::Object(scoped)
        }
    }
}

fn should_retain_node_output(node_id: &str, retention: &NodesRetentionPlan) -> bool {
    retention.retained_nodes.contains(node_id)
}

fn execution_metadata(
    ctx: &PipelineContext,
    nodes_scope: Value,
    placeholder: Option<Value>,
) -> Value {
    json!({
        "owner": ctx.owner,
        "project": ctx.project,
        "pipeline": ctx.pipeline,
        "request_id": ctx.request_id,
        "route": ctx.route,
        "trigger": ctx.trigger,
        "nodes": nodes_scope,
        "placeholder": placeholder,
    })
}

fn take_private_tokens(payload: &mut Value, key: &str) -> Vec<String> {
    let Some(map) = payload.as_object_mut() else {
        return Vec::new();
    };
    let Some(Value::Array(items)) = map.remove(key) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for item in items {
        let Some(text) = item.as_str() else {
            continue;
        };
        let trimmed = text.trim();
        if trimmed.is_empty() || out.iter().any(|existing| existing == trimmed) {
            continue;
        }
        out.push(trimmed.to_string());
    }
    out
}

fn take_private_paths(payload: &mut Value, key: &str) -> Vec<Vec<String>> {
    let Some(map) = payload.as_object_mut() else {
        return Vec::new();
    };
    let Some(Value::Array(items)) = map.remove(key) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for item in items {
        let parts = match item {
            Value::String(path) => path
                .split('.')
                .map(str::trim)
                .filter(|segment| !segment.is_empty())
                .map(ToString::to_string)
                .collect::<Vec<_>>(),
            Value::Array(segments) => segments
                .into_iter()
                .filter_map(|segment| segment.as_str().map(str::trim).map(ToString::to_string))
                .filter(|segment| !segment.is_empty())
                .collect::<Vec<_>>(),
            _ => Vec::new(),
        };
        if parts.is_empty() || out.iter().any(|existing| existing == &parts) {
            continue;
        }
        out.push(parts);
    }
    out
}

fn take_private_redact_tokens(payload: &mut Value) -> Vec<String> {
    take_private_tokens(payload, "__zf_private_redact")
}

fn take_private_redact_except_paths(payload: &mut Value) -> Vec<Vec<String>> {
    take_private_paths(payload, "__zf_private_redact_except_paths")
}

fn retry_attempt_from_payload(payload: &Value) -> usize {
    payload
        .get(RETRY_STATE_KEY)
        .and_then(|value| value.get("attempt"))
        .and_then(Value::as_u64)
        .map(|n| n as usize)
        .unwrap_or(0)
}

/// The attempt a retry node has already counted to, read from its last
/// output (`nodes_output[<retry id>].__zf_retry.attempt`); 0 before it has
/// run. Kept beside the payload's own count because a payload that was
/// replaced on the way round the loop has forgotten it.
fn retry_last_attempt(nodes_output: &serde_json::Map<String, Value>, retry_node_id: &str) -> usize {
    nodes_output
        .get(retry_node_id)
        .map(retry_attempt_from_payload)
        .unwrap_or(0)
}

/// A retry node's `--max-attempts` as the graph declares it — the DSL coerces
/// the scalar to a number, an edited JSON may still hold a string.
fn retry_max_attempts(node: &PipelineNode) -> Option<u64> {
    let raw = node.config.get("max_attempts")?;
    raw.as_u64()
        .or_else(|| raw.as_str().and_then(|s| s.trim().parse().ok()))
}

/// The failure envelope an `:error` edge carries: the failing node's input
/// under `input`, the error, and the retry state. `last_attempt` is what the
/// consuming retry node counted to on the previous round (0 when there is
/// none, or it has not run yet); the payload's own count wins when higher.
fn build_retry_error_payload(
    input_payload: &Value,
    error: &PipelineError,
    last_attempt: usize,
) -> Value {
    let attempt = retry_attempt_from_payload(input_payload).max(last_attempt) + 1;
    json!({
        "input": input_payload,
        "error": {
            "code": error.code,
            "message": error.message,
            "node_id": error.node_id,
            "node_kind": error.node_kind,
        },
        RETRY_STATE_KEY: {
            "attempt": attempt,
            "failing_node_id": error.node_id,
            "failing_node_kind": error.node_kind,
        }
    })
}

pub(crate) fn is_sensitive_trace_config_key(key: &str) -> bool {
    let normalized = key
        .chars()
        .filter(|ch| ch.is_ascii_alphanumeric())
        .collect::<String>()
        .to_ascii_lowercase();
    matches!(
        normalized.as_str(),
        "password"
            | "passwd"
            | "passphrase"
            | "secret"
            | "clientsecret"
            | "accesstoken"
            | "refreshtoken"
            | "idtoken"
            | "authorization"
            | "apikey"
            // `--header X-API-Key=…` typed on an http.request is a key in the
            // node's config, and the recorded config is the snapshot the
            // record keeps of it.
            | "xapikey"
            // A node outside this tree may name a field plainly. `accesstoken`
            // was matched and `token` was not, which is the gap a third-party
            // node would fall into.
            | "token"
            | "apitoken"
            | "authtoken"
            | "sessiontoken"
            | "bearertoken"
            | "privatekey"
            | "signingkey"
            | "jwtsecret"
            | "cookiesecret"
            | "webhooksecret"
    )
}

#[cfg(test)]
fn trace_config_snapshot(value: &Value) -> Option<Value> {
    let mut capture = TraceCapture::new(TraceCaptureSettings::default().resolve(None));
    capture.begin_node();
    capture.config(value)
}

fn redact_string(value: &str, tokens: &[String]) -> String {
    redact_literal_secrets(value, tokens)
}

fn redact_json_value(
    value: &Value,
    tokens: &[String],
    except_paths: &[Vec<String>],
    current_path: &[String],
) -> Value {
    if except_paths.iter().any(|path| path == current_path) {
        return value.clone();
    }
    match value {
        Value::String(text) => Value::String(redact_string(text, tokens)),
        Value::Array(items) => Value::Array(
            items
                .iter()
                .map(|item| redact_json_value(item, tokens, except_paths, current_path))
                .collect(),
        ),
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(key, item)| {
                    let mut next_path = current_path.to_vec();
                    next_path.push(key.clone());
                    (
                        key.clone(),
                        redact_json_value(item, tokens, except_paths, &next_path),
                    )
                })
                .collect(),
        ),
        other => other.clone(),
    }
}

#[cfg(test)]
const TRACE_SUMMARY_MAX_DEPTH: usize = 8;
#[cfg(test)]
const TRACE_SUMMARY_MAX_OBJECT_KEYS: usize = 48;
#[cfg(test)]
const TRACE_SUMMARY_MAX_ARRAY_ITEMS: usize = 16;
#[cfg(test)]
const TRACE_SUMMARY_PREVIEW_ITEMS: usize = 6;
#[cfg(test)]
const TRACE_SUMMARY_NUMERIC_ARRAY_THRESHOLD: usize = 64;
#[cfg(test)]
const TRACE_SUMMARY_MAX_STRING_CHARS: usize = 8192;
#[cfg(test)]
const TRACE_SUMMARY_STRING_PREVIEW_CHARS: usize = 512;

#[cfg(test)]
fn summarize_trace_value(value: &Value) -> Value {
    summarize_trace_value_inner(value, 0)
}

#[cfg(test)]
fn summarize_trace_value_inner(value: &Value, depth: usize) -> Value {
    if depth >= TRACE_SUMMARY_MAX_DEPTH {
        return match value {
            Value::Object(map) => json!({
                "__zf_trace_summary": "object",
                "keys": map.len(),
            }),
            Value::Array(items) => json!({
                "__zf_trace_summary": "array",
                "len": items.len(),
            }),
            Value::String(text) => summarize_trace_string(text),
            other => other.clone(),
        };
    }

    match value {
        Value::String(text) if text.chars().count() > TRACE_SUMMARY_MAX_STRING_CHARS => {
            summarize_trace_string(text)
        }
        Value::Array(items)
            if items.len() >= TRACE_SUMMARY_NUMERIC_ARRAY_THRESHOLD
                && items.iter().all(Value::is_number) =>
        {
            let preview = items
                .iter()
                .take(TRACE_SUMMARY_PREVIEW_ITEMS)
                .cloned()
                .collect::<Vec<_>>();
            let tail = items
                .last()
                .cloned()
                .map(|item| vec![item])
                .unwrap_or_default();
            json!({
                "__zf_trace_summary": "numeric_array",
                "len": items.len(),
                "preview": preview,
                "tail": tail,
            })
        }
        Value::Array(items) if items.len() > TRACE_SUMMARY_MAX_ARRAY_ITEMS => {
            let preview = items
                .iter()
                .take(TRACE_SUMMARY_PREVIEW_ITEMS)
                .map(|item| summarize_trace_value_inner(item, depth + 1))
                .collect::<Vec<_>>();
            json!({
                "__zf_trace_summary": "array",
                "len": items.len(),
                "preview": preview,
            })
        }
        Value::Array(items) => Value::Array(
            items
                .iter()
                .map(|item| summarize_trace_value_inner(item, depth + 1))
                .collect(),
        ),
        Value::Object(map) if map.len() > TRACE_SUMMARY_MAX_OBJECT_KEYS => {
            let preview = map
                .iter()
                .take(TRACE_SUMMARY_MAX_OBJECT_KEYS)
                .map(|(key, item)| (key.clone(), summarize_trace_value_inner(item, depth + 1)))
                .collect::<serde_json::Map<_, _>>();
            json!({
                "__zf_trace_summary": "object",
                "keys": map.len(),
                "preview": Value::Object(preview),
            })
        }
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(key, item)| (key.clone(), summarize_trace_value_inner(item, depth + 1)))
                .collect(),
        ),
        other => other.clone(),
    }
}

#[cfg(test)]
fn summarize_trace_string(text: &str) -> Value {
    let preview = text
        .chars()
        .take(TRACE_SUMMARY_STRING_PREVIEW_CHARS)
        .collect::<String>();
    json!({
        "__zf_trace_summary": "string",
        "chars": text.chars().count(),
        "preview": preview,
    })
}

/// Blanks any value sitting under a key that names a secret.
///
/// The token mechanism above strikes out *values a node declared*; this strikes
/// out values *nobody declared* — the payload a cron, WS, or script-built run
/// carries, where no node publishes redaction tokens at all. The two are
/// complementary: tokens catch a secret that was copied somewhere else in the
/// trace, key names catch a secret nobody thought to declare.
///
/// `except_paths` is honoured here too, so a pipeline that deliberately traces
/// a field named `token` keeps the same one escape hatch it already had.
#[cfg(test)]
fn blank_sensitive_keys(value: &Value, except_paths: &[Vec<String>], path: &[String]) -> Value {
    if except_paths.iter().any(|candidate| candidate == path) {
        return value.clone();
    }
    match value {
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(key, item)| {
                    let mut next = path.to_vec();
                    next.push(key.clone());
                    let sanitized = if except_paths.iter().any(|candidate| candidate == &next) {
                        item.clone()
                    } else if crate::pipeline::nodes::basic::trigger::webhook::is_sensitive_payload_key(key)
                    {
                        blanked_like(item)
                    } else {
                        blank_sensitive_keys(item, except_paths, &next)
                    };
                    (key.clone(), sanitized)
                })
                .collect(),
        ),
        Value::Array(items) => Value::Array(
            items
                .iter()
                .map(|item| blank_sensitive_keys(item, except_paths, path))
                .collect(),
        ),
        other => other.clone(),
    }
}

/// Replaces a secret with a mask of the same *shape*, so a trace still shows
/// whether the field was present, a string, or a list — just not what it said.
///
/// Booleans and nulls pass through: one bit carries no secret, and a masked
/// `"authorization": true` would cost a reader real information for nothing.
/// Numbers do not, because a `cvv`, `otp`, or `pin` is a number.
#[cfg(test)]
fn blanked_like(value: &Value) -> Value {
    match value {
        Value::Null | Value::Bool(_) => value.clone(),
        Value::Array(items) => Value::Array(items.iter().map(blanked_like).collect()),
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(key, item)| (key.clone(), blanked_like(item)))
                .collect(),
        ),
        Value::String(_) | Value::Number(_) => Value::String("••••••".to_string()),
    }
}

/// Lifts every redaction marker out of a payload, wherever it sits, and
/// returns the literal values they name.
///
/// The markers used to be read at the top level only. That is where a node
/// puts one, but not where it stays: `n.web.response` nests the whole upstream
/// payload under `__zf_response.body`, so by the time that node's *output* was
/// traced the marker was one level down -- unread, unremoved, and printed
/// verbatim into the run history with the secrets inside it.
#[cfg(test)]
fn sweep_private_markers(value: &mut Value, tokens: &mut Vec<String>) {
    match value {
        Value::Object(map) => {
            for key in ["__zf_private_trace_redact", "__zf_private_redact"] {
                let Some(Value::Array(items)) = map.remove(key) else {
                    continue;
                };
                for item in items {
                    let Some(text) = item.as_str() else { continue };
                    let trimmed = text.trim();
                    if !trimmed.is_empty() && !tokens.iter().any(|seen| seen == trimmed) {
                        tokens.push(trimmed.to_string());
                    }
                }
            }
            map.remove("__zf_private_redact_except_paths");
            for item in map.values_mut() {
                sweep_private_markers(item, tokens);
            }
        }
        Value::Array(items) => {
            for item in items {
                sweep_private_markers(item, tokens);
            }
        }
        _ => {}
    }
}

#[cfg(test)]
fn legacy_sanitized_trace_value(value: &Value) -> Value {
    let mut payload = value.clone();
    // Only a top-level exception list defines paths, because a path is written
    // from the payload root and a nested copy names nothing meaningful.
    let except_paths = take_private_redact_except_paths(&mut payload);
    let mut tokens = Vec::new();
    sweep_private_markers(&mut payload, &mut tokens);
    let redacted = if tokens.is_empty() {
        payload
    } else {
        redact_json_value(&payload, &tokens, &except_paths, &[])
    };
    let blanked = blank_sensitive_keys(&redacted, &except_paths, &[]);
    summarize_trace_value(&blanked)
}

#[cfg(test)]
fn sanitized_trace_value(value: &Value) -> Value {
    let mut capture = TraceCapture::new(TraceCaptureSettings::default().resolve(None));
    capture.begin_node();
    capture.capture(value)
}

fn materialize_node_output_files(
    platform: Option<&Arc<PlatformService>>,
    ctx: &PipelineContext,
    node_kind: &str,
    mut payload: Value,
) -> Result<Value, PipelineError> {
    let Some(files) = take_node_output_files(&mut payload) else {
        return Ok(payload);
    };
    if files.is_empty() {
        return Ok(payload);
    }
    let platform = platform.ok_or_else(|| {
        PipelineError::new(
            "FW_NODE_OUTPUT_FILE_CONTEXT",
            "node output files require platform storage context",
        )
    })?;
    let layout = platform
        .file
        .ensure_project_layout(&ctx.owner, &ctx.project)
        .map_err(|err| PipelineError::new("FW_NODE_OUTPUT_FILE", err.to_string()))?;
    let zebfs = layout.open_files();
    let request_id = sanitize_path_part(&ctx.request_id);
    let kind_part = sanitize_path_part(node_kind);
    let mut refs = Map::new();

    for (idx, file) in files.iter().enumerate() {
        let descriptor = file.as_object().ok_or_else(|| {
            PipelineError::new(
                "FW_NODE_OUTPUT_FILE",
                format!("node output file {idx} must be an object"),
            )
        })?;
        let name = descriptor
            .get("name")
            .and_then(Value::as_str)
            .map(sanitize_filename)
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| format!("file-{idx}.json"));
        let content_type = descriptor
            .get("content_type")
            .or_else(|| descriptor.get("mime"))
            .and_then(Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("application/octet-stream");
        let encoding = descriptor
            .get("encoding")
            .and_then(Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("json");
        let data = descriptor.get("data").ok_or_else(|| {
            PipelineError::new(
                "FW_NODE_OUTPUT_FILE",
                format!("node output file '{name}' is missing data"),
            )
        })?;
        let bytes = node_output_file_bytes(node_kind, &name, encoding, data)?;
        if bytes.len() > MAX_NODE_OUTPUT_FILE_BYTES {
            return Err(PipelineError::new(
                "FW_NODE_OUTPUT_FILE_TOO_LARGE",
                format!(
                    "node '{node_kind}' output file '{name}' is {} bytes; limit is {MAX_NODE_OUTPUT_FILE_BYTES}",
                    bytes.len()
                ),
            ));
        }
        let rel_path = format!("artifacts/nodes/{kind_part}/{request_id}/{name}");
        let stat = zebfs
            .put(&rel_path, &bytes)
            .map_err(|err| PipelineError::new("FW_NODE_OUTPUT_FILE_WRITE", err.to_string()))?;
        let file_ref = json!({
            "__zf_type": FILE_REF_TYPE,
            "backend": BACKEND_ZEBFS,
            "ref": stat.path,
            "filename": name,
            "mime": content_type,
            "kind": crate::pipeline::nodes::shared::file_ref::infer_kind(content_type, &rel_path),
            "size": bytes.len(),
            "sha256": format!("sha256:{:x}", Sha256::digest(&bytes)),
            "lifecycle": LIFECYCLE_DURABLE,
            "origin": "node-output",
            "trust": "generated",
        });
        refs.insert(name, file_ref);
    }

    let refs_value = Value::Object(refs);
    replace_file_placeholders(&mut payload, &refs_value);
    if let Some(obj) = payload.as_object_mut() {
        obj.insert("file_refs".to_string(), refs_value);
    }
    Ok(payload)
}

fn take_node_output_files(payload: &mut Value) -> Option<Vec<Value>> {
    let map = payload.as_object_mut()?;
    let raw = map
        .remove("__zf_files")
        .or_else(|| map.remove("artifacts"))?;
    match raw {
        Value::Array(files) => Some(files),
        other => Some(vec![other]),
    }
}

fn node_output_file_bytes(
    node_kind: &str,
    name: &str,
    encoding: &str,
    data: &Value,
) -> Result<Vec<u8>, PipelineError> {
    match encoding {
        "json" => serde_json::to_vec(data).map_err(|err| {
            PipelineError::new(
                "FW_NODE_OUTPUT_FILE_JSON",
                format!("node '{node_kind}' output file '{name}' JSON serialization failed: {err}"),
            )
        }),
        "text" | "utf8" => Ok(data
            .as_str()
            .map(str::as_bytes)
            .map(ToOwned::to_owned)
            .unwrap_or_else(|| data.to_string().into_bytes())),
        "base64" => {
            let encoded = data.as_str().ok_or_else(|| {
                PipelineError::new(
                    "FW_NODE_OUTPUT_FILE_BASE64",
                    format!("node '{node_kind}' output file '{name}' base64 data must be a string"),
                )
            })?;
            base64::engine::general_purpose::STANDARD
                .decode(encoded.trim())
                .map_err(|err| {
                    PipelineError::new(
                        "FW_NODE_OUTPUT_FILE_BASE64",
                        format!(
                            "node '{node_kind}' output file '{name}' base64 decode failed: {err}"
                        ),
                    )
                })
        }
        other => Err(PipelineError::new(
            "FW_NODE_OUTPUT_FILE_ENCODING",
            format!("node '{node_kind}' output file '{name}' uses unsupported encoding '{other}'"),
        )),
    }
}

fn replace_file_placeholders(value: &mut Value, refs: &Value) {
    match value {
        Value::String(text) => {
            if let Some(name) = exact_file_placeholder(text)
                && let Some(file_ref) = refs.get(name)
            {
                *value = file_ref.clone();
                return;
            }
            let mut replaced = text.clone();
            if let Some(map) = refs.as_object() {
                for (name, file_ref) in map {
                    let url = file_ref
                        .get("url")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    replaced = replaced.replace(&format!("{{{{file:{name}}}}}"), url);
                    replaced = replaced.replace(&format!("{{{{artifact:{name}}}}}"), url);
                }
            }
            *text = replaced;
        }
        Value::Array(items) => {
            for item in items {
                replace_file_placeholders(item, refs);
            }
        }
        Value::Object(map) => {
            for child in map.values_mut() {
                replace_file_placeholders(child, refs);
            }
        }
        _ => {}
    }
}

fn exact_file_placeholder(value: &str) -> Option<&str> {
    value
        .strip_prefix("{{file:")
        .or_else(|| value.strip_prefix("{{artifact:"))
        .and_then(|rest| rest.strip_suffix("}}"))
        .map(str::trim)
        .filter(|value| !value.is_empty())
}

fn sanitize_filename(raw: &str) -> String {
    let name = raw.rsplit(['/', '\\']).next().unwrap_or(raw).trim();
    let mut out = String::new();
    for ch in name.chars() {
        if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_') {
            out.push(ch);
        } else if ch.is_whitespace() {
            out.push('-');
        }
    }
    let trimmed = out.trim_matches(['.', '-']).to_string();
    if trimmed.is_empty() {
        "file.bin".to_string()
    } else {
        trimmed
    }
}

fn sanitize_path_part(raw: &str) -> String {
    let mut out = String::new();
    for ch in raw.chars() {
        if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') {
            out.push(ch);
        } else {
            out.push('-');
        }
    }
    let trimmed = out.trim_matches('-');
    if trimmed.is_empty() {
        "run".to_string()
    } else {
        trimmed.to_string()
    }
}

/// Main framework engine used for real pipeline execution.
pub struct BasicPipelineEngine {
    language: Arc<dyn LanguageEngine>,
    rwe: Arc<dyn ReactiveWebEngine>,
    credentials: Option<Arc<CredentialService>>,
    template_cache: Option<TemplateCache>,
    ws_hub: Option<Arc<WsHub>>,
    ws_client_manager: Option<Arc<WsClientManager>>,
    state_bus: Option<DynStateBus>,
    platform: Option<Arc<PlatformService>>,
    /// Filesystem root for resolving `@/` alias imports in TSX templates.
    ///
    /// Derived from `repo_layout` and never set on its own, so the alias root
    /// and the layout that names it cannot disagree.
    template_root: Option<std::path::PathBuf>,
    /// The project's resolved repository layout, when the engine runs for one.
    ///
    /// Nodes that need a repository directory other than the source root read
    /// it from here instead of walking up from the template root, which only
    /// works for a source root exactly one segment deep.
    repo_layout: Option<crate::platform::model::ProjectFileLayout>,
    /// Platform data root — used by SQLite nodes to locate the project DB.
    data_root: Option<std::path::PathBuf>,
    /// The bundle policy governing this engine's nodes.
    ///
    /// Set only on an engine built to run a bundle's function pipeline, so it
    /// covers the whole inner subtree. A project's own pipeline runs with
    /// `None` and is unrestricted. `Some` means "inside a bundle" whether or
    /// not that bundle declared a host; the declared hosts are what it carries,
    /// not what makes it present.
    bundle_egress: Option<Arc<crate::pipeline::security::BundleEgress>>,
}

impl Default for BasicPipelineEngine {
    fn default() -> Self {
        let rwe_engine_id = std::env::var("ZEBFLOW_RWE_ENGINE_ID").ok();
        Self {
            language: Arc::new(DenoSandboxEngine::default()),
            rwe: resolve_engine_or_default(rwe_engine_id.as_deref()),
            credentials: None,
            template_cache: None,
            ws_hub: None,
            ws_client_manager: None,
            state_bus: None,
            platform: None,
            template_root: None,
            repo_layout: None,
            data_root: None,
            bundle_egress: None,
        }
    }
}

impl BasicPipelineEngine {
    pub fn new(
        language: Arc<dyn LanguageEngine>,
        rwe: Arc<dyn ReactiveWebEngine>,
        credentials: Option<Arc<CredentialService>>,
    ) -> Self {
        Self {
            language,
            rwe,
            credentials,
            template_cache: None,
            ws_hub: None,
            ws_client_manager: None,
            state_bus: None,
            platform: None,
            template_root: None,
            repo_layout: None,
            data_root: None,
            bundle_egress: None,
        }
    }

    /// Attach the project's repository layout.
    ///
    /// This also sets the template root, because the template root *is* the
    /// layout's source directory; there is no second way to set it.
    pub fn with_project_layout(
        mut self,
        layout: Option<crate::platform::model::ProjectFileLayout>,
    ) -> Self {
        self.template_root = layout.as_ref().map(|value| value.repo_source_dir());
        self.repo_layout = layout;
        self
    }

    /// Source libraries (`zeb/ui`, …) the compiler resolves by file. They ship
    /// with the platform, so an engine without one has none.
    fn library_roots(&self) -> std::collections::BTreeMap<String, std::path::PathBuf> {
        self.platform
            .as_ref()
            .map(|p| p.library.source_roots())
            .unwrap_or_default()
    }

    /// Attach a shared template compile cache to this engine.
    /// Same cache instance should be passed on every request so hits accumulate.
    pub fn with_template_cache(mut self, cache: TemplateCache) -> Self {
        self.template_cache = Some(cache);
        self
    }

    /// Attach the WS hub so ws_sync_state and ws_emit nodes can access rooms.
    pub fn with_ws_hub(mut self, hub: Arc<WsHub>) -> Self {
        self.ws_hub = Some(hub);
        self
    }

    /// Attach the WS client manager so n.ws.client.send nodes can send through outbound connections.
    pub fn with_ws_client_manager(mut self, mgr: Arc<WsClientManager>) -> Self {
        self.ws_client_manager = Some(mgr);
        self
    }

    /// Attach the state bus so `n.kv.*` nodes can access the shared project-scoped KV/pubsub layer.
    pub fn with_state_bus(mut self, bus: DynStateBus) -> Self {
        self.state_bus = Some(bus);
        self
    }

    /// Attach the legacy mem hub convenience wrapper.
    ///
    /// This keeps current call sites simple while routing the live mem node surface through the
    /// stronger `StateBus` abstraction.
    pub fn with_mem_hub(mut self, hub: Arc<MemHub>) -> Self {
        self.state_bus = Some(Arc::new(MemStateBus::from_hub((*hub).clone())));
        self
    }

    /// Attach the platform service so n.function.call nodes can invoke sub-pipelines.
    pub fn with_platform(mut self, platform: Arc<PlatformService>) -> Self {
        self.platform = Some(platform);
        self
    }

    /// Attach the platform data root so SQLite nodes can locate the project DB.
    pub fn with_data_root(mut self, root: std::path::PathBuf) -> Self {
        self.data_root = Some(root);
        self
    }

    /// Confine this engine's nodes to the hosts a node bundle declared.
    ///
    /// Attached where a bundle's function pipeline is dispatched, which is what
    /// makes the declaration cover every node in that pipeline rather than only
    /// the bundle node the outer graph names.
    pub fn with_bundle_egress(
        mut self,
        egress: Option<Arc<crate::pipeline::security::BundleEgress>>,
    ) -> Self {
        self.bundle_egress = egress;
        self
    }

    fn build_node(&self, node: &PipelineNode) -> Result<NodeDispatch, PipelineError> {
        if let Some(egress) = &self.bundle_egress {
            refuse_uncheckable_egress_node(egress, &node.kind, self.language.grants_network())?;
        }
        match node.kind.as_str() {
            webhook::NODE_KIND => Ok(NodeDispatch::Webhook(webhook::Node::new(
                serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_WEBHOOK_CONFIG", err.to_string()))?,
            ))),
            schedule::NODE_KIND => Ok(NodeDispatch::Schedule(schedule::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_SCHEDULE_CONFIG", err.to_string())
                })?,
            ))),
            manual::NODE_KIND => Ok(NodeDispatch::Manual(manual::Node::new(
                serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_MANUAL_CONFIG", err.to_string()))?,
            ))),
            script::NODE_KIND => Ok(NodeDispatch::Script(script::Node::new(
                &node.id,
                serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_SCRIPT_CONFIG", err.to_string()))?,
                self.language.clone(),
            )?)),
            http::request::NODE_KIND => Ok(NodeDispatch::HttpRequest(http::request::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_HTTP_REQUEST_CONFIG", err.to_string())
                })?,
                self.language.clone(),
                self.credentials.clone(),
                self.platform.clone(),
                self.bundle_egress.clone(),
            )?)),
            sqlite::query::NODE_KIND => {
                let Some(data_root) = &self.data_root else {
                    return Err(PipelineError::new(
                        "FW_NODE_SQLITE_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::SqliteQuery(sqlite::query::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_SQLITE_QUERY_CONFIG", err.to_string())
                    })?,
                    data_root.clone(),
                )?))
            }
            sekejap::query::NODE_KIND => {
                let Some(data_root) = &self.data_root else {
                    return Err(PipelineError::new(
                        "FW_NODE_SEKEJAP_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::SekejapQuery(sekejap::query::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_SEKEJAP_QUERY_CONFIG", err.to_string())
                    })?,
                    data_root.clone(),
                )?))
            }
            sekejap::insert::NODE_KIND => {
                let Some(data_root) = &self.data_root else {
                    return Err(PipelineError::new(
                        "FW_NODE_SEKEJAP_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::SekejapInsert(sekejap::insert::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_SEKEJAP_INSERT_CONFIG", err.to_string())
                    })?,
                    data_root.clone(),
                )?))
            }
            sqlite::mutate::NODE_KIND => {
                let Some(data_root) = &self.data_root else {
                    return Err(PipelineError::new(
                        "FW_NODE_SQLITE_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::SqliteMutate(sqlite::mutate::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_SQLITE_MUTATE_CONFIG", err.to_string())
                    })?,
                    data_root.clone(),
                )?))
            }
            browser::run::NODE_KIND => {
                let Some(credentials) = &self.credentials else {
                    return Err(PipelineError::new(
                        "FW_NODE_BROWSER_RUN_UNAVAILABLE",
                        "credential service is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::BrowserRun(browser::run::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_BROWSER_RUN_CONFIG", e.to_string())
                    })?,
                    credentials.clone(),
                    self.bundle_egress.clone(),
                )?))
            }
            pg::query::NODE_KIND => {
                let Some(credentials) = &self.credentials else {
                    return Err(PipelineError::new(
                        "FW_NODE_PG_UNAVAILABLE",
                        "credential service is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::Postgres(pg::query::Node::new(
                    serde_json::from_value(node.config.clone())
                        .map_err(|err| PipelineError::new("FW_NODE_PG_CONFIG", err.to_string()))?,
                    credentials.clone(),
                )?))
            }
            table::query::NODE_KIND => {
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_TABLE_QUERY_UNAVAILABLE",
                        "platform service is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::TableQuery(table::query::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_TABLE_QUERY_CONFIG", err.to_string())
                    })?,
                    platform.clone(),
                    self.language.clone(),
                )?))
            }
            web::response::NODE_KIND => {
                let config: web::response::Config = serde_json::from_value(node.config.clone())
                    .map_err(|err| {
                        PipelineError::new("FW_NODE_WEB_RESPONSE_CONFIG", err.to_string())
                    })?;
                if config.template.is_some() {
                    Ok(NodeDispatch::InlineWebResponse {
                        node_id: node.id.clone(),
                        config,
                    })
                } else {
                    Ok(NodeDispatch::WebResponse(web::response::Node::new(config, self.template_root.clone())))
                }
            }
            web::static_generate::NODE_KIND => {
                if self.data_root.is_none() {
                    return Err(PipelineError::new(
                        "FW_NODE_WEB_STATIC_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                }
                let config: web::static_generate::Config =
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_WEB_STATIC_CONFIG", err.to_string())
                    })?;
                Ok(NodeDispatch::InlineWebStaticGenerate {
                    node_id: node.id.clone(),
                    config,
                })
            }
            web::docs_generate::NODE_KIND => {
                if self.data_root.is_none() {
                    return Err(PipelineError::new(
                        "FW_NODE_WEB_DOCS_UNAVAILABLE",
                        "data_root is not configured on this pipeline engine",
                    ));
                }
                let config: web::docs_generate::Config = serde_json::from_value(node.config.clone())
                    .map_err(|err| {
                        PipelineError::new("FW_NODE_WEB_DOCS_CONFIG", err.to_string())
                    })?;
                Ok(NodeDispatch::InlineWebDocsGenerate {
                    node_id: node.id.clone(),
                    config,
                })
            }
            ai::agent::NODE_KIND => {
                let config: ai::agent::Config = serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_AGENT_CONFIG", err.to_string()))?;
                Ok(NodeDispatch::Agent(ai::agent::Node::new(
                    config,
                    self.credentials.clone(),
                    self.platform.clone(),
                )))
            }
            ai::tts::NODE_KIND => {
                let config: ai::tts::Config = serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_AI_TTS_CONFIG", err.to_string()))?;
                Ok(NodeDispatch::AiTts(ai::tts::Node::new(
                    config,
                    self.credentials.clone(),
                    self.platform.clone(),
                )))
            }
            logic::if_::NODE_KIND => Ok(NodeDispatch::LogicIf(logic::if_::Node::new(
                &node.id,
                serde_json::from_value(node.config.clone())
                    .map_err(|e| PipelineError::new("FW_NODE_LOGIC_IF_CONFIG", e.to_string()))?,
                self.language.clone(),
            )?)),
            logic::match_::NODE_KIND => Ok(NodeDispatch::LogicMatch(logic::match_::Node::new(
                &node.id,
                serde_json::from_value(node.config.clone())
                    .map_err(|e| PipelineError::new("FW_NODE_LOGIC_MATCH_CONFIG", e.to_string()))?,
                self.language.clone(),
            )?)),
            logic::collect::NODE_KIND => Ok(NodeDispatch::LogicCollect(logic::collect::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|e| {
                    PipelineError::new("FW_NODE_LOGIC_COLLECT_CONFIG", e.to_string())
                })?,
            ))),
            logic::foreach_::NODE_KIND => {
                Ok(NodeDispatch::LogicForeach(logic::foreach_::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_LOGIC_FOREACH_CONFIG", e.to_string())
                    })?,
                    self.language.clone(),
                )?))
            }
            logic::reduce::NODE_KIND => Ok(NodeDispatch::LogicReduce(logic::reduce::Node::new(
                &node.id,
                serde_json::from_value(node.config.clone()).map_err(|e| {
                    PipelineError::new("FW_NODE_LOGIC_REDUCE_CONFIG", e.to_string())
                })?,
                self.language.clone(),
            )?)),
            logic::retry::NODE_KIND => Ok(NodeDispatch::LogicRetry(logic::retry::Node::new(
                &node.id,
                serde_json::from_value(node.config.clone())
                    .map_err(|e| PipelineError::new("FW_NODE_LOGIC_RETRY_CONFIG", e.to_string()))?,
                self.language.clone(),
            )?)),
            auth::token_create::NODE_KIND => {
                let Some(credentials) = &self.credentials else {
                    return Err(PipelineError::new(
                        "FW_NODE_AUTH_TOKEN_UNAVAILABLE",
                        "credential service is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::AuthTokenCreate(auth::token_create::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_AUTH_TOKEN_CONFIG", err.to_string())
                    })?,
                    credentials.clone(),
                )?))
            }
            auth::token_verify::NODE_KIND => {
                let Some(credentials) = &self.credentials else {
                    return Err(PipelineError::new(
                        "FW_NODE_AUTH_VERIFY_UNAVAILABLE",
                        "credential service is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::AuthTokenVerify(auth::token_verify::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_AUTH_VERIFY_CONFIG", err.to_string())
                    })?,
                    credentials.clone(),
                )?))
            }
            concept::NODE_KIND => Ok(NodeDispatch::Concept(concept::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_CONCEPT_CONFIG", err.to_string())
                })?,
            ))),
            mail::send::NODE_KIND => {
                let Some(credentials) = &self.credentials else {
                    return Err(PipelineError::new(
                        "FW_NODE_MAIL_UNAVAILABLE",
                        "credential service is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::MailSend(mail::send::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_MAIL_CONFIG", err.to_string())
                    })?,
                    credentials.clone(),
                    self.platform.clone(),
                )?))
            }
            weberror::NODE_KIND => Ok(NodeDispatch::WebError(weberror::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_WEBERROR_CONFIG", err.to_string())
                })?,
            ))),
            ws::trigger::NODE_KIND => Ok(NodeDispatch::WsTrigger(ws::trigger::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_WS_TRIGGER_CONFIG", err.to_string())
                })?,
            ))),
            ws::sync_state::NODE_KIND => {
                let Some(ws_hub) = &self.ws_hub else {
                    return Err(PipelineError::new(
                        "FW_NODE_WS_SYNC_STATE_UNAVAILABLE",
                        "ws hub is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::WsSyncState(ws::sync_state::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_WS_SYNC_STATE_CONFIG", err.to_string())
                    })?,
                    ws_hub.clone(),
                )?))
            }
            ws::emit::NODE_KIND => {
                let Some(ws_hub) = &self.ws_hub else {
                    return Err(PipelineError::new(
                        "FW_NODE_WS_EMIT_UNAVAILABLE",
                        "ws hub is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::WsEmit(ws::emit::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|err| {
                        PipelineError::new("FW_NODE_WS_EMIT_CONFIG", err.to_string())
                    })?,
                    ws_hub.clone(),
                )?))
            }
            crypto::NODE_KIND => Ok(NodeDispatch::Crypto(crypto::Node::new(
                serde_json::from_value(node.config.clone())
                    .map_err(|err| PipelineError::new("FW_NODE_CRYPTO_CONFIG", err.to_string()))?,
            )?)),
            trigger_function::NODE_KIND => {
                let config: trigger_function::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                Ok(NodeDispatch::TriggerFunction(trigger_function::Node::new(
                    config,
                )))
            }
            function::call::NODE_KIND => {
                let config: function::call::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                Ok(NodeDispatch::FunctionCall(function::call::Node::new(
                    config,
                    self.platform.clone(),
                )))
            }
            fs::save::NODE_KIND => {
                let config: fs::save::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FILE_SAVE",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::FileSave(fs::save::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            fs::object::LIST_NODE_KIND
            | fs::object::HEAD_NODE_KIND
            | fs::object::GET_NODE_KIND
            | fs::object::PUT_NODE_KIND
            | fs::object::DELETE_NODE_KIND
            | fs::object::COPY_NODE_KIND
            | fs::object::MOVE_NODE_KIND
            | fs::object::MKDIR_NODE_KIND => {
                let config: fs::object::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FS_OBJECT",
                        "platform service not available in this engine context",
                    ));
                };
                let operation = match node.kind.as_str() {
                    fs::object::LIST_NODE_KIND => fs::object::Operation::List,
                    fs::object::HEAD_NODE_KIND => fs::object::Operation::Head,
                    fs::object::GET_NODE_KIND => fs::object::Operation::Get,
                    fs::object::PUT_NODE_KIND => fs::object::Operation::Put,
                    fs::object::DELETE_NODE_KIND => fs::object::Operation::Delete,
                    fs::object::COPY_NODE_KIND => fs::object::Operation::Copy,
                    fs::object::MOVE_NODE_KIND => fs::object::Operation::Move,
                    fs::object::MKDIR_NODE_KIND => fs::object::Operation::Mkdir,
                    _ => unreachable!(),
                };
                Ok(NodeDispatch::FsObject(fs::object::Node::new(
                    config,
                    platform.clone(),
                    operation,
                )?))
            }
            ms::crud::PUBLISH_KIND
            | ms::crud::UNPUBLISH_KIND
            | ms::crud::GET_KIND
            | ms::crud::LIST_KIND => {
                let config: ms::crud::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_MS",
                        "platform service not available in this engine context",
                    ));
                };
                let operation = match node.kind.as_str() {
                    ms::crud::PUBLISH_KIND => ms::crud::Operation::Publish,
                    ms::crud::UNPUBLISH_KIND => ms::crud::Operation::Unpublish,
                    ms::crud::GET_KIND => ms::crud::Operation::Get,
                    ms::crud::LIST_KIND => ms::crud::Operation::List,
                    _ => unreachable!(),
                };
                Ok(NodeDispatch::MapserverCrud(ms::crud::Node::new(
                    config,
                    platform.clone(),
                    operation,
                )?))
            }
            table::convert::NODE_KIND => {
                let config: table::convert::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_TABLE_CONVERT",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::TableConvert(table::convert::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            fs::compress::NODE_KIND => {
                let config: fs::compress::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FILE_COMPRESS",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::FileCompress(fs::compress::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            fs::decompress::NODE_KIND => {
                let config: fs::decompress::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FILE_DECOMPRESS",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::FileDecompress(fs::decompress::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            geo::inspect::NODE_KIND => {
                let config: geo::inspect::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_GEO_INSPECT",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::GeoInspect(geo::inspect::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            geo::convert::NODE_KIND => {
                let config: geo::convert::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_GEO_CONVERT",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::GeoConvert(geo::convert::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            fs::pdf::convert::NODE_KIND => {
                let config: fs::pdf::convert::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_PDF_CONVERT",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::FilePdfConvert(fs::pdf::convert::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            fs::svg::convert::NODE_KIND => {
                let config: fs::svg::convert::Config = serde_json::from_value(node.config.clone())
                    .map_err(|e| PipelineError::new("FW_NODE_FS_SVG_CONVERT_CONFIG", e.to_string()))?;
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FS_SVG_CONVERT_CONFIG",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::SvgConvert(fs::svg::convert::Node::new(config, platform.clone())?))
            }
            fs::image::chromakey::NODE_KIND => {
                let config: fs::image::chromakey::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_FS_IMAGE_CHROMAKEY_CONFIG",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::ImgChromakey(fs::image::chromakey::Node::new(config, platform.clone())?))
            }
            fs::image::thumbnail::NODE_KIND => {
                let config: fs::image::thumbnail::Config =
                    serde_json::from_value(node.config.clone()).unwrap_or_default();
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "IMG_THUMBNAIL",
                        "platform service not available in this engine context",
                    ));
                };
                Ok(NodeDispatch::ImgThumbnail(fs::image::thumbnail::Node::new(
                    config,
                    platform.clone(),
                )?))
            }
            kv::set::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvSet(kv::set::Node::new(
                    serde_json::from_value(node.config.clone())
                        .map_err(|e| PipelineError::new("FW_NODE_KV_SET_CONFIG", e.to_string()))?,
                    state_bus.clone(),
                )))
            }
            // The whole `input.*` family builds through one door: the kind
            // picks the check, the config names the field.
            kind if input::is_input_kind(kind) => {
                let node = input::Node::for_kind(kind, &node.config)
                    .expect("guard says this is an input kind")?;
                Ok(NodeDispatch::Input(node))
            }
            kv::get::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvGet(kv::get::Node::new(
                    serde_json::from_value(node.config.clone())
                        .map_err(|e| PipelineError::new("FW_NODE_KV_GET_CONFIG", e.to_string()))?,
                    state_bus.clone(),
                )))
            }
            kv::del::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvDel(kv::del::Node::new(
                    serde_json::from_value(node.config.clone())
                        .map_err(|e| PipelineError::new("FW_NODE_KV_DEL_CONFIG", e.to_string()))?,
                    state_bus.clone(),
                )))
            }
            kv::incr::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvIncr(kv::incr::Node::new(
                    serde_json::from_value(node.config.clone())
                        .map_err(|e| PipelineError::new("FW_NODE_KV_INCR_CONFIG", e.to_string()))?,
                    state_bus.clone(),
                )))
            }
            kv::publish::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvPublish(kv::publish::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_KV_PUBLISH_CONFIG", e.to_string())
                    })?,
                    state_bus.clone(),
                )))
            }
            kv::exists::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvExists(kv::exists::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_KV_EXISTS_CONFIG", e.to_string())
                    })?,
                    state_bus.clone(),
                )))
            }
            kv::expire::NODE_KIND => {
                let Some(state_bus) = &self.state_bus else {
                    return Err(PipelineError::new(
                        "FW_NODE_MEM_UNAVAILABLE",
                        "state bus is not configured on this pipeline engine",
                    ));
                };
                Ok(NodeDispatch::KvExpire(kv::expire::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_KV_EXPIRE_CONFIG", e.to_string())
                    })?,
                    state_bus.clone(),
                )))
            }
            mcp_trigger::NODE_KIND => Ok(NodeDispatch::McpTrigger(mcp_trigger::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|err| {
                    PipelineError::new("FW_NODE_MCP_TRIGGER_CONFIG", err.to_string())
                })?,
            ))),
            kv_subscribe::NODE_KIND => Ok(NodeDispatch::KvSubscribe(kv_subscribe::Node::new(
                serde_json::from_value(node.config.clone()).map_err(|e| {
                    PipelineError::new("FW_NODE_KV_SUBSCRIBE_CONFIG", e.to_string())
                })?,
            ))),
            trigger_ws_client::NODE_KIND => Ok(NodeDispatch::WsClientTrigger(
                trigger_ws_client::Node::new(serde_json::from_value(node.config.clone()).map_err(
                    |e| PipelineError::new("FW_NODE_WS_CLIENT_TRIGGER_CONFIG", e.to_string()),
                )?),
            )),
            ws::client_send::NODE_KIND => {
                let Some(ws_client_manager) = &self.ws_client_manager else {
                    return Err(PipelineError::new(
                        "FW_NODE_WS_CLIENT_SEND_UNAVAILABLE",
                        "ws client manager is not configured on this framework engine",
                    ));
                };
                Ok(NodeDispatch::WsClientSend(ws::client_send::Node::new(
                    serde_json::from_value(node.config.clone()).map_err(|e| {
                        PipelineError::new("FW_NODE_WS_CLIENT_SEND_CONFIG", e.to_string())
                    })?,
                    ws_client_manager.clone(),
                )?))
            }
            // Anything the arms above did not claim is not a native node, so it
            // is provided by a bundle. Curated bundles live in `n.*` and
            // third-party ones in `n.x.*`, so the namespace cannot be used to
            // tell them apart; the manifest decides role and implementation at
            // execution time, and reports a missing package if there is none.
            other if other.starts_with("n.") => {
                let Some(platform) = &self.platform else {
                    return Err(PipelineError::new(
                        "FW_NODE_INSTALLED_NO_PLATFORM",
                        format!(
                            "installed node '{}': platform service not injected into engine",
                            other
                        ),
                    ));
                };
                Ok(NodeDispatch::InstalledNode {
                    kind: other.to_string(),
                    config: node.config.clone(),
                    platform: platform.clone(),
                    // A bundle composing another bundle's node stays governed by
                    // its own declaration, so the policy in force here descends
                    // with the dispatch rather than being replaced by it.
                    egress: self.bundle_egress.clone(),
                })
            }
            other => Err(PipelineError::new(
                "FW_NODE_KIND_UNSUPPORTED",
                format!("unsupported node kind '{}'", other),
            )),
        }
    }
}

#[async_trait]
impl PipelineEngine for BasicPipelineEngine {
    fn id(&self) -> &'static str {
        "pipeline.basic"
    }

    fn validate_graph(&self, graph: &PipelineGraph) -> Result<(), PipelineError> {
        crate::contracts::kinds::validate_pipeline_activation(graph).map_err(|error| {
            PipelineError::new(
                error.violation_code().unwrap_or("FW_PIPELINE_CONTRACT"),
                error.to_string(),
            )
        })?;
        for node in &graph.nodes {
            // Skip upfront validation for nodes whose config contains {{ expr }} placeholders —
            // those are resolved per-input at runtime, so type validation must happen there.
            if scan_exprs(&node.config).is_empty() {
                self.build_node(node)?;
            }
        }
        build_nodes_retention_plan(graph)?;
        Ok(())
    }

    async fn execute_with_options_async(
        &self,
        graph: &PipelineGraph,
        ctx: &PipelineContext,
        options: &ExecuteOptions,
    ) -> Result<PipelineOutput, PipelineError> {
        // The run is bracketed on the bus: `run_start` before the first node,
        // `run_done` after the last, whatever the outcome — the n8n-style
        // canvas badges read these, and so may any SSE client.
        let started = std::time::Instant::now();
        emit_lifecycle(
            &options.bus,
            "run_start",
            format!("run {} start", ctx.request_id),
            None,
            Some(json!({ "run_id": ctx.request_id })),
            &started,
        );
        let result = self.run_graph(graph, ctx, options).await;
        let status = if result.is_ok() { "ok" } else { "error" };
        let duration_ms = started.elapsed().as_millis() as u64;
        emit_lifecycle(
            &options.bus,
            "run_done",
            format!("run {} {} {} ms", ctx.request_id, status, duration_ms),
            None,
            Some(json!({ "run_id": ctx.request_id, "status": status, "duration_ms": duration_ms })),
            &started,
        );
        result
    }
}

/// One engine lifecycle signal on the run's bus, when the run has one.
///
/// The engine announces `run_start`, then per node `node_start` followed by
/// exactly one of `node_ok` / `node_skip` / `node_fail` / `node_retry` /
/// `node_error_routed`, then `run_done`. The last two are a failure an
/// `:error` edge consumed: `node_retry` when the edge reaches `logic.retry`
/// (`data: { attempt, max_attempts, duration_ms, error_code, message }`),
/// `node_error_routed` for any other consumer (`{ duration_ms, error_code,
/// message, to_node }`); `node_fail` is the unrouted failure only. Every
/// consumer of the bus forwards every signal, so a client that only cared
/// for node-emitted ones (an agent's steps) now sees these too — the SSE
/// help says so.
fn emit_lifecycle(
    bus: &Option<Arc<ExecutionBus>>,
    kind: &str,
    message: String,
    node: Option<(&str, &str)>,
    data: Option<Value>,
    run_started: &std::time::Instant,
) {
    let Some(bus) = bus else { return };
    let (node_id, node_kind) = node.unwrap_or(("", ""));
    bus.emit(Signal {
        kind: kind.to_string(),
        message,
        node_id: node_id.to_string(),
        node_kind: node_kind.to_string(),
        data,
        at: format!("{}ms", run_started.elapsed().as_millis()),
    });
}

/// `node_fail` for one node, whether the failure came from the node itself
/// or from resolving its config or building it.
fn emit_node_fail(
    bus: &Option<Arc<ExecutionBus>>,
    node_id: &str,
    node_kind: &str,
    error: &PipelineError,
    node_start: &std::time::Instant,
    run_started: &std::time::Instant,
) {
    let duration_ms = node_start.elapsed().as_millis() as u64;
    emit_lifecycle(
        bus,
        "node_fail",
        format!("{node_id} {} fail {duration_ms} ms", short_kind_word(node_kind)),
        Some((node_id, node_kind)),
        Some(json!({
            "duration_ms": duration_ms,
            "error_code": error.code,
            "error_class": crate::pipeline::error_class::class_of(error.code).as_status_word(),
            "error": error.message,
        })),
        run_started,
    );
}

/// `http.request` for `n.http.request` — the message is for a human.
fn short_kind_word(kind: &str) -> &str {
    kind.strip_prefix("n.").unwrap_or(kind)
}

impl BasicPipelineEngine {
    /// The run itself. `execute_with_options_async` brackets it with
    /// `run_start` / `run_done` on the bus.
    async fn run_graph(
        &self,
        graph: &PipelineGraph,
        ctx: &PipelineContext,
        options: &ExecuteOptions,
    ) -> Result<PipelineOutput, PipelineError> {
        self.validate_graph(graph)?;
        let run_started = std::time::Instant::now();

        let node_map: HashMap<&str, &PipelineNode> = graph
            .nodes
            .iter()
            .map(|node| (node.id.as_str(), node))
            .collect();
        let mut outgoing: HashMap<(&str, &str), Vec<(&str, &str)>> = HashMap::new();
        let mut incoming_counts: HashMap<&str, usize> = HashMap::new();
        for edge in &graph.edges {
            outgoing
                .entry((edge.from_node.as_str(), edge.from_pin.as_str()))
                .or_default()
                .push((edge.to_node.as_str(), edge.to_pin.as_str()));
            *incoming_counts.entry(edge.to_node.as_str()).or_default() += 1;
        }

        // Roots come from connectivity, never from source order
        // (`kinds/pipeline/README.md` Frozen Defaults). A graph with two
        // independent triggers has two roots and both run.
        let start_nodes: Vec<String> = graph
            .entry_node_ids()
            .into_iter()
            .map(ToString::to_string)
            .collect();

        let bus = options.bus.clone();
        // Project configuration is immutable for this invocation. Snapshot the
        // timeout once instead of reparsing zebflow.yaml for every node.
        let project_config = self
            .platform
            .as_ref()
            .map(|platform| {
                platform
                    .zebflow_cfg
                    .read_or_default(&ctx.owner, &ctx.project)
            })
            .transpose()
            .map_err(|err| PipelineError::new(err.code, err.message))?;
        let project_capture = project_config
            .as_ref()
            .and_then(|config| config.configs.pipelines.logging.trace_capture.as_ref())
            .cloned()
            .unwrap_or_default();
        let pipeline_capture = graph
            .metadata
            .as_ref()
            .and_then(|metadata| metadata.settings.trace_capture.as_ref());
        project_capture
            .validate()
            .map_err(|message| PipelineError::new("FW_TRACE_CONFIG", message))?;
        if let Some(settings) = pipeline_capture {
            settings
                .validate()
                .map_err(|message| PipelineError::new("FW_TRACE_CONFIG", message))?;
        }
        // Rule 3: whatever this project has taken out of the credential store is
        // masked in every payload, at every level, without any node declaring
        // it. Scoped to this owner/project — never another tenant's secrets.
        let mut trace_capture = TraceCapture::new(project_capture.resolve(pipeline_capture))
            .with_confidential(crate::platform::services::credential::confidential_values(
                &ctx.owner,
                &ctx.project,
            ));
        let project_timeout_secs = project_config
            .as_ref()
            .map(|config| config.configs.pipelines.effective_node_timeout_secs())
            .or_else(|| {
                std::env::var("PIPELINE_NODE_TIMEOUT_SECS")
                    .ok()
                    .and_then(|value| value.parse().ok())
            })
            .unwrap_or(crate::platform::model::default_pipeline_node_timeout_secs());
        // nodes_output: accumulates only completed node outputs required by the
        // graph's `$nodes`/`ctx.nodes` retention plan.
        // Declared here (before the initial queue push) so entry-node metadata can include it.
        let nodes_retention = build_nodes_retention_plan(graph)?;
        let mut nodes_output = serde_json::Map::<String, Value>::new();
        let mut queue = VecDeque::new();
        for node_id in start_nodes {
            let node = node_map
                .get(node_id.as_str())
                .ok_or_else(|| PipelineError::new("FW_ENTRY_NODE", "entry node missing"))?;
            let first_pin = node.input_pins.first().cloned().unwrap_or_default();
            queue.push_back(NodeExecutionInput {
                node_id: node.id.clone(),
                input_pin: first_pin,
                payload: ctx.input.clone(),
                metadata: execution_metadata(
                    ctx,
                    nodes_scope_for_target(&node.id, &nodes_output, &nodes_retention),
                    ctx.placeholder.clone(),
                ),
                bus: bus.clone(),
            });
        }

        let mut trace = vec![format!("engine={}", self.id())];
        let mut last_value = Value::Null;
        let mut node_trace: Vec<NodeTraceEntry> = Vec::new();
        // merge_pending: node_id -> { pin_name -> payload }
        let mut collect_pending: HashMap<String, HashMap<String, Value>> = HashMap::new();
        let mut reduce_pending: HashMap<String, ReducePendingState> = HashMap::new();

        while let Some(input) = queue.pop_front() {
            let node = node_map.get(input.node_id.as_str()).ok_or_else(|| {
                PipelineError::new("FW_EXEC_NODE", format!("node '{}' missing", input.node_id))
            })?;

            // The node's clock starts here, before its config is resolved and
            // it is built, so a failure in either is announced as this
            // node's `node_fail` and the badge never stays "running".
            let node_start = std::time::Instant::now();
            emit_lifecycle(
                &bus,
                "node_start",
                format!("{} {} start", node.id, short_kind_word(&node.kind)),
                Some((&node.id, &node.kind)),
                None,
                &run_started,
            );

            // Resolve {{ expr }} placeholders in the node's config before building.
            // Uses input.metadata["nodes"] (snapshot at queue-time) for $nodes scope,
            // so each node only sees outputs of its transitive predecessors.
            let effective_config = match resolve_config_expressions(
                node.config.clone(),
                &input.payload,
                &input.metadata,
                &self.language,
            ) {
                Ok(config) => config,
                Err(e) => {
                    emit_node_fail(&bus, &node.id, &node.kind, &e, &node_start, &run_started);
                    return Err(e);
                }
            };
            trace_capture.begin_node();
            // A declared canvas preview asks for that payload in the record —
            // see `TraceCapture::records_payload_for`. Read off the stored
            // config: `preview` is presentation, never expression-resolved.
            // An input node's widget is its input view — the Run form shows
            // what went in from the record — so the family asks for its input
            // payload the way a declared `--preview-in` does.
            let preview_in =
                node.config["preview"]["in"].is_object() || input::is_input_kind(&node.kind);
            let preview_out = node.config["preview"]["out"].is_object();
            // Captured before the node runs, because the error path needs it and
            // the outcome is not known yet. At `on-error` a successful node
            // drops it below; at `none` it is never taken at all.
            let base_trace_config = if trace_capture.records_any_payload() {
                trace_capture.config(&effective_config)
            } else {
                None
            };
            let built = if effective_config == node.config {
                // No expressions resolved — use original node directly (common fast path).
                self.build_node(node)
            } else {
                self.build_node(&PipelineNode {
                    config: effective_config.clone(),
                    ..(*node).clone()
                })
            };
            let dispatch = match built {
                Ok(dispatch) => dispatch,
                Err(e) => {
                    emit_node_fail(&bus, &node.id, &node.kind, &e, &node_start, &run_started);
                    return Err(e);
                }
            };

            // Capture context for per-node trace before consuming `input`.
            let trace_node_id = node.id.clone();
            let trace_node_kind = node.kind.clone();
            let input_snapshot = input.payload.clone();

            // Per-node timeout: prevents slow HTTP/DB nodes from hanging pipelines.
            // Priority: node config `timeout_secs` → project config → env var → default(30s).
            let node_timeout_secs: u64 = effective_config
                .get("timeout_secs")
                .and_then(|v| {
                    v.as_u64()
                        .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
                })
                .map(|v| v.clamp(5, 3600))
                .unwrap_or(project_timeout_secs);
            let mut input_for_exec = input.clone();
            if node.kind == logic::reduce::NODE_KIND
                && let Some(acc) = reduce_pending
                    .get(node.id.as_str())
                    .and_then(|state| state.acc.clone())
                && let Some(map) = input_for_exec.metadata.as_object_mut()
            {
                map.insert("reduce_acc".to_string(), acc);
            }
            let exec_fut = async {
                match dispatch {
                    NodeDispatch::Webhook(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Schedule(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Manual(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Script(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::HttpRequest(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::BrowserRun(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::SqliteQuery(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::SekejapQuery(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::SekejapInsert(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::SqliteMutate(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::Postgres(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::InlineWebResponse { node_id, config } => {
                        let markup = config.markup.as_deref().unwrap_or("").trim();
                        if markup.is_empty() {
                            Err(PipelineError::new(
                                "FW_NODE_WEB_RESPONSE_CONFIG",
                                format!("node '{node_id}' --template set but markup not loaded"),
                            ))
                        } else {
                            // Location arrives final — {{ }} resolved above.
                            let location = config.location.clone();
                            let status = config
                                .status
                                .or_else(|| if location.is_some() { Some(302) } else { None });
                            let cookie = config
                                .set_cookie
                                .as_deref()
                                .and_then(web::response::parse_cookie_spec);
                            let headers = config.headers.clone();

                            let template_id = config.template.clone().unwrap_or_default();
                            let source_path = self
                                .template_root
                                .as_ref()
                                .and_then(|r| config.template.as_deref().map(|t| r.join(t)));
                            let options = crate::rwe::ReactiveWebOptions {
                                templates: crate::rwe::TemplateOptions {
                                    template_root: self.template_root.clone(),
                                    library_roots: self.library_roots(),
                                    style_entries: Vec::new(),
                                },
                                ..Default::default()
                            };

                            let key = hash_markup(markup);
                            let cached = self.template_cache.as_ref().and_then(|c| {
                                c.read()
                                    .unwrap_or_else(|e| e.into_inner())
                                    .get(&key)
                                    .map(|e| e.page.clone())
                            });
                            let compiled_result: Result<Arc<_>, PipelineError> = if let Some(hit) =
                                cached
                            {
                                Ok(hit)
                            } else {
                                let fresh = web::response::compile_page(
                                    &node_id,
                                    &TemplateSource {
                                        id: template_id,
                                        source_path,
                                        markup: markup.to_string(),
                                    },
                                    &options,
                                    self.rwe.as_ref(),
                                    self.language.as_ref(),
                                )
                                .map(Arc::new);
                                if let Ok(ref fresh_arc) = fresh {
                                    if let Some(cache) = &self.template_cache {
                                        let deps = fresh_arc.template.dependency_paths.clone();
                                        cache.write().unwrap_or_else(|e| e.into_inner()).insert(
                                            key,
                                            CacheEntry {
                                                page: fresh_arc.clone(),
                                                dependencies: deps,
                                            },
                                        );
                                    }
                                }
                                fresh
                            };

                            compiled_result.and_then(|compiled| {
                            let enabled_libraries: Vec<String> = self.platform
                                .as_ref()
                                .and_then(|p| {
                                    p.zebflow_cfg.get_rwe_libraries(&ctx.owner, &ctx.project).ok()
                                })
                                .map(|libs| libs.into_keys().collect())
                                .unwrap_or_default();

                            let render_out = web::response::render_compiled_page(
                                &compiled,
                                input.payload,
                                input.metadata,
                                self.rwe.as_ref(),
                                self.language.as_ref(),
                                &ctx.request_id,
                                enabled_libraries,
                            )?;
                            let envelope = serde_json::json!({
                                "status": status,
                                "location": location,
                                "set_cookie": cookie,
                                "headers": headers,
                                "html": render_out.payload.get("html"),
                                "compiled_scripts": render_out.payload.get("compiled_scripts"),
                                "hydration_payload": render_out.payload.get("hydration_payload"),
                            });
                            Ok(vec![NodeExecutionOutput {
                                output_pins: render_out.output_pins,
                                payload: serde_json::json!({ "__zf_response": envelope }),
                                trace: render_out.trace,
                            }])
                        })
                        }
                    }
                    NodeDispatch::WebResponse(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::InlineWebDocsGenerate { node_id, config } => {
                        let Some(data_root) = &self.data_root else {
                            return Err(PipelineError::new(
                                "FW_NODE_WEB_DOCS_UNAVAILABLE",
                                "data_root is not configured on this pipeline engine",
                            ));
                        };
                        let Some(template_root) = &self.template_root else {
                            return Err(PipelineError::new(
                                "FW_NODE_WEB_DOCS_TEMPLATE_ROOT",
                                "template_root is not configured on this pipeline engine",
                            ));
                        };

                        let Some(docs_root) = self
                            .repo_layout
                            .as_ref()
                            .map(|layout| layout.repo_docs_dir())
                        else {
                            return Err(PipelineError::new(
                                "FW_NODE_WEB_DOCS_TEMPLATE_ROOT",
                                "project layout is not configured on this pipeline engine",
                            ));
                        };
                        let site =
                            web::docs_generate::load_site(&config, template_root, &docs_root)?;
                        let options = crate::rwe::ReactiveWebOptions {
                            templates: crate::rwe::TemplateOptions {
                                template_root: self.template_root.clone(),
                                library_roots: self.library_roots(),
                                style_entries: Vec::new(),
                            },
                            processors: vec!["tailwind".to_string(), "markdown".to_string()],
                            ..Default::default()
                        };
                        let key = hash_markup(&site.template_source.markup);
                        let cached = self.template_cache.as_ref().and_then(|c| {
                            c.read()
                                .unwrap_or_else(|e| e.into_inner())
                                .get(&key)
                                .map(|e| e.page.clone())
                        });
                        let compiled_result: Result<Arc<_>, PipelineError> =
                            if let Some(hit) = cached {
                                Ok(hit)
                            } else {
                                let fresh = web::response::compile_page(
                                    &node_id,
                                    &site.template_source,
                                    &options,
                                    self.rwe.as_ref(),
                                    self.language.as_ref(),
                                )
                                .map(Arc::new);
                                if let Ok(ref fresh_arc) = fresh
                                    && let Some(cache) = &self.template_cache
                                {
                                    let deps = fresh_arc.template.dependency_paths.clone();
                                    cache.write().unwrap_or_else(|e| e.into_inner()).insert(
                                        key,
                                        CacheEntry {
                                            page: fresh_arc.clone(),
                                            dependencies: deps,
                                        },
                                    );
                                }
                                fresh
                            };

                        compiled_result.and_then(|compiled| {
                            let mut generated_files = 0usize;
                            let mut skipped_files = 0usize;
                            let mut urls = Vec::new();
                            let asset_group = web::static_site::asset_group_id(
                                &site.template_rel_path,
                                &site.template_source.markup,
                            );
                            let mut page_records = Vec::new();
                            let mut asset_records = Vec::new();
                            let site_root_abs = data_root
                                .join("users")
                                .join(&ctx.owner)
                                .join(&ctx.project)
                                .join("files")
                                .join(&site.site_root_rel);
                            let project_asset_root = self
                                .repo_layout
                                .as_ref()
                                .map(|layout| layout.repo_static_dir());

                            let enabled_libraries: Vec<String> = self
                                .platform
                                .as_ref()
                                .and_then(|p| {
                                    p.zebflow_cfg
                                        .get_rwe_libraries(&ctx.owner, &ctx.project)
                                        .ok()
                                })
                                .map(|libs| libs.into_keys().collect())
                                .unwrap_or_default();

                            for (page_index, page) in site.pages.iter().enumerate() {
                                let mut metadata = input.metadata.clone();
                                if let Some(map) = metadata.as_object_mut() {
                                    map.insert(
                                        "route".to_string(),
                                        Value::String(web::docs_generate::default_route(page)),
                                    );
                                }
                                let payload = web::docs_generate::page_payload(
                                    &site,
                                    page_index,
                                    input.payload.clone(),
                                )?;
                                let render_out = web::response::render_compiled_page(
                                    &compiled,
                                    payload,
                                    metadata,
                                    self.rwe.as_ref(),
                                    self.language.as_ref(),
                                    &ctx.request_id,
                                    enabled_libraries.clone(),
                                )?;
                                let html = render_out
                                    .payload
                                    .get("html")
                                    .and_then(Value::as_str)
                                    .ok_or_else(|| {
                                        PipelineError::new(
                                            "FW_NODE_WEB_DOCS_RENDER",
                                            format!("node '{node_id}' did not return rendered html"),
                                        )
                                    })?
                                    .to_string();
                                let hydration_payload = render_out
                                    .payload
                                    .get("hydration_payload")
                                    .cloned()
                                    .unwrap_or(Value::Null);
                                let compiled_scripts = render_out
                                    .payload
                                    .get("compiled_scripts")
                                    .cloned()
                                    .and_then(|value| serde_json::from_value::<Vec<crate::rwe::CompiledScript>>(value).ok())
                                    .unwrap_or_default();
                                let final_html = web::static_generate::build_static_html(
                                    html,
                                    &hydration_payload,
                                    &compiled_scripts,
                                    self.template_root.as_deref(),
                                );
                                let localized = web::static_site::localize_static_html_assets(
                                    &site_root_abs,
                                    &page.output_rel_path,
                                    &final_html,
                                    web::static_site::StaticAssetSources {
                                        owner: Some(&ctx.owner),
                                        project: Some(&ctx.project),
                                        project_asset_root_abs: project_asset_root.as_deref(),
                                    },
                                    &asset_group,
                                )?;
                                asset_records.extend(localized.assets.iter().cloned());
                                let final_html = web::docs_generate::apply_page_seo(
                                    localized.html,
                                    &site,
                                    page_index,
                                );
                                let rel_path =
                                    web::docs_generate::output_rel_path(page, &site.site_root_rel)?;
                                let abs_path = data_root
                                    .join("users")
                                    .join(&ctx.owner)
                                    .join(&ctx.project)
                                    .join("files")
                                    .join(&rel_path);
                                let status = web::static_generate::write_generated_html(
                                    &abs_path,
                                    &final_html,
                                    "overwrite",
                                )?;
                                if status == "skipped" || status == "unchanged" {
                                    skipped_files += 1;
                                } else {
                                    generated_files += 1;
                                }
                                urls.push(page.route_path.clone());
                                page_records.push(web::static_site::StaticPageRecord {
                                    path: page.output_rel_path.clone(),
                                    route: page.route_path.clone(),
                                    template: site.template_rel_path.clone(),
                                    asset_group: asset_group.clone(),
                                    generator: web::docs_generate::NODE_KIND.to_string(),
                                });
                            }

                            if !site.sitemap_xml.trim().is_empty() {
                                let sitemap_rel =
                                    web::docs_generate::sitemap_rel_path(&site.site_root_rel);
                                let sitemap_abs = data_root
                                    .join("users")
                                    .join(&ctx.owner)
                                    .join(&ctx.project)
                                    .join("files")
                                    .join(&sitemap_rel);
                                let status = web::static_generate::write_generated_html(
                                    &sitemap_abs,
                                    &site.sitemap_xml,
                                    "overwrite",
                                )?;
                                if status == "skipped" || status == "unchanged" {
                                    skipped_files += 1;
                                } else {
                                    generated_files += 1;
                                }
                            }

                            let search_index_rel =
                                web::docs_generate::search_index_rel_path(&site.site_root_rel);
                            let search_index_abs = data_root
                                .join("users")
                                .join(&ctx.owner)
                                .join(&ctx.project)
                                .join("files")
                                .join(&search_index_rel);
                            let status = web::static_generate::write_generated_html(
                                &search_index_abs,
                                &site.search_index_json,
                                "overwrite",
                            )?;
                            if status == "skipped" || status == "unchanged" {
                                skipped_files += 1;
                            } else {
                                generated_files += 1;
                            }

                            let manifest_rel =
                                web::static_site::site_manifest_rel_path(&site.site_root_rel);
                            let manifest_abs = data_root
                                .join("users")
                                .join(&ctx.owner)
                                .join(&ctx.project)
                                .join("files")
                                .join(&manifest_rel);
                            let _manifest = web::static_site::update_site_manifest(
                                &manifest_abs,
                                &site.site_root_rel,
                                site.deploy_base_url.as_deref(),
                                &site.deploy_base_path,
                                web::docs_generate::NODE_KIND,
                                &site.template_rel_path,
                                &asset_group,
                                &page_records,
                                &asset_records,
                                true,
                            )?;

                            Ok(vec![NodeExecutionOutput {
                                output_pins: vec![web::docs_generate::OUTPUT_PIN_OUT.to_string()],
                                payload: json!({
                                    "docs_generated": {
                                        "status": "ok",
                                        "site_title": site.site_title,
                                        "template": site.template_rel_path,
                                        "docs_root": config.docs_root,
                                        "output_dir": config.output_dir,
                                        "site_root": site.site_root_rel,
                                        "deploy_base_url": site.deploy_base_url,
                                        "deploy_base_path": site.deploy_base_path,
                                        "manifest_path": manifest_rel,
                                        "asset_group": asset_group,
                                        "page_count": site.pages.len(),
                                        "generated_files": generated_files,
                                        "skipped_files": skipped_files,
                                        "sitemap_path": if site.sitemap_xml.trim().is_empty() { Value::Null } else { Value::String(web::docs_generate::sitemap_rel_path(&site.site_root_rel)) },
                                        "search_index_path": search_index_rel,
                                        "urls": urls,
                                    }
                                }),
                                trace: vec![
                                    format!("node={node_id}"),
                                    format!("node_kind={}", web::docs_generate::NODE_KIND),
                                    format!("pages={}", site.pages.len()),
                                ],
                            }])
                        })
                    }
                    NodeDispatch::InlineWebStaticGenerate { node_id, config } => {
                        let Some(data_root) = &self.data_root else {
                            return Err(PipelineError::new(
                                "FW_NODE_WEB_STATIC_UNAVAILABLE",
                                "data_root is not configured on this pipeline engine",
                            ));
                        };

                        let template_source = web::static_generate::resolve_template_source(
                            &node_id,
                            &config,
                            self.template_root.as_deref(),
                        )?;

                        let options = crate::rwe::ReactiveWebOptions {
                            templates: crate::rwe::TemplateOptions {
                                template_root: self.template_root.clone(),
                                library_roots: self.library_roots(),
                                style_entries: Vec::new(),
                            },
                            ..Default::default()
                        };

                        let key = hash_markup(&template_source.markup);
                        let cached = self.template_cache.as_ref().and_then(|c| {
                            c.read()
                                .unwrap_or_else(|e| e.into_inner())
                                .get(&key)
                                .map(|e| e.page.clone())
                        });
                        let compiled_result: Result<Arc<_>, PipelineError> =
                            if let Some(hit) = cached {
                                Ok(hit)
                            } else {
                                let fresh = web::response::compile_page(
                                    &node_id,
                                    &template_source,
                                    &options,
                                    self.rwe.as_ref(),
                                    self.language.as_ref(),
                                )
                                .map(Arc::new);
                                if let Ok(ref fresh_arc) = fresh
                                    && let Some(cache) = &self.template_cache
                                {
                                    let deps = fresh_arc.template.dependency_paths.clone();
                                    cache.write().unwrap_or_else(|e| e.into_inner()).insert(
                                        key,
                                        CacheEntry {
                                            page: fresh_arc.clone(),
                                            dependencies: deps,
                                        },
                                    );
                                }
                                fresh
                            };

                        compiled_result.and_then(|compiled| {
                            let rel_path = web::static_generate::effective_output_rel_path(&config)?;
                            let abs_path = data_root
                                .join("users")
                                .join(&ctx.owner)
                                .join(&ctx.project)
                                .join("files")
                                .join(&rel_path);
                            let route =
                                if let Some(explicit_route) =
                                    config.route.clone().filter(|s| !s.trim().is_empty())
                                {
                                    explicit_route
                                } else if let Some(deploy_base_path) =
                                    web::static_generate::effective_deploy_base_path(&config)?
                                {
                                    let page_output_path =
                                        web::static_generate::effective_page_output_path(&config)?;
                                    web::static_site::route_path_for_output_path(
                                        &deploy_base_path,
                                        &page_output_path,
                                    )?
                                } else {
                                    web::static_generate::default_route(
                                        &ctx.owner,
                                        &ctx.project,
                                        &rel_path,
                                    )
                                };

                            let mut metadata = input.metadata.clone();
                            if let Some(map) = metadata.as_object_mut() {
                                map.insert("route".to_string(), Value::String(route.clone()));
                            }

                            let enabled_libraries: Vec<String> = self
                                .platform
                                .as_ref()
                                .and_then(|p| {
                                    p.zebflow_cfg
                                        .get_rwe_libraries(&ctx.owner, &ctx.project)
                                        .ok()
                                })
                                .map(|libs| libs.into_keys().collect())
                                .unwrap_or_default();

                            let render_out = web::response::render_compiled_page(
                                &compiled,
                                input.payload,
                                metadata,
                                self.rwe.as_ref(),
                                self.language.as_ref(),
                                &ctx.request_id,
                                enabled_libraries,
                            )?;

                            let html = render_out
                                .payload
                                .get("html")
                                .and_then(Value::as_str)
                                .ok_or_else(|| {
                                    PipelineError::new(
                                        "FW_NODE_WEB_STATIC_RENDER",
                                        format!("node '{node_id}' did not return rendered html"),
                                    )
                                })?
                                .to_string();

                            let hydration_payload = render_out
                                .payload
                                .get("hydration_payload")
                                .cloned()
                                .unwrap_or(Value::Null);
                            let compiled_scripts = render_out
                                .payload
                                .get("compiled_scripts")
                                .cloned()
                                .and_then(|value| {
                                    serde_json::from_value::<Vec<crate::rwe::CompiledScript>>(value)
                                        .ok()
                                })
                                .unwrap_or_default();

                            let final_html = web::static_generate::build_static_html(
                                html,
                                &hydration_payload,
                                &compiled_scripts,
                                self.template_root.as_deref(),
                            );
                            let asset_group = web::static_site::asset_group_id(
                                &template_source.id,
                                &template_source.markup,
                            );
                            let project_asset_root = self
                                .repo_layout
                                .as_ref()
                                .map(|layout| layout.repo_static_dir());
                            let localized = if let Some(site_root_rel) =
                                web::static_generate::effective_site_root_rel_path(&config)?
                            {
                                let site_root_abs = data_root
                                    .join("users")
                                    .join(&ctx.owner)
                                    .join(&ctx.project)
                                    .join("files")
                                    .join(&site_root_rel);
                                web::static_site::localize_static_html_assets(
                                    &site_root_abs,
                                    &web::static_site::normalize_page_output_path(
                                        &config.output_path,
                                    )?,
                                    &final_html,
                                    web::static_site::StaticAssetSources {
                                        owner: Some(&ctx.owner),
                                        project: Some(&ctx.project),
                                        project_asset_root_abs: project_asset_root.as_deref(),
                                    },
                                    &asset_group,
                                )?
                            } else {
                                let page_dir_abs =
                                    abs_path.parent().unwrap_or(abs_path.as_path()).to_path_buf();
                                let page_file_name = abs_path
                                    .file_name()
                                    .and_then(|name| name.to_str())
                                    .ok_or_else(|| {
                                        PipelineError::new(
                                            "FW_NODE_WEB_STATIC_OUTPUT_NAME",
                                            format!(
                                                "node '{node_id}' produced an invalid output path '{}'",
                                                abs_path.display()
                                            ),
                                        )
                                    })?
                                    .to_string();
                                web::static_site::localize_static_html_assets(
                                    &page_dir_abs,
                                    &page_file_name,
                                    &final_html,
                                    web::static_site::StaticAssetSources {
                                        owner: Some(&ctx.owner),
                                        project: Some(&ctx.project),
                                        project_asset_root_abs: project_asset_root.as_deref(),
                                    },
                                    &asset_group,
                                )?
                            };
                            let status = web::static_generate::write_generated_html(
                                &abs_path,
                                &localized.html,
                                &config.on_conflict,
                            )?;
                            let bytes = localized.html.as_bytes().len() as u64;
                            let url = format!("/fs/{}/{}/{}", ctx.owner, ctx.project, rel_path);
                            let site_root_rel =
                                web::static_generate::effective_site_root_rel_path(&config)?;
                            let manifest_rel = if let Some(site_root_rel) = site_root_rel.as_deref()
                            {
                                let manifest_rel =
                                    web::static_site::site_manifest_rel_path(site_root_rel);
                                let manifest_abs = data_root
                                    .join("users")
                                    .join(&ctx.owner)
                                    .join(&ctx.project)
                                    .join("files")
                                    .join(&manifest_rel);
                                let page_path = web::static_site::normalize_page_output_path(
                                    &config.output_path,
                                )?;
                                let page_record = web::static_site::StaticPageRecord {
                                    path: page_path,
                                    route: route.clone(),
                                    template: template_source.id.clone(),
                                    asset_group: asset_group.clone(),
                                    generator: web::static_generate::NODE_KIND.to_string(),
                                };
                                let _manifest = web::static_site::update_site_manifest(
                                    &manifest_abs,
                                    site_root_rel,
                                    web::static_generate::effective_deploy_base_url(&config)
                                        .as_deref(),
                                    web::static_generate::effective_deploy_base_path(&config)?
                                        .as_deref()
                                        .unwrap_or("/"),
                                    web::static_generate::NODE_KIND,
                                    &template_source.id,
                                    &asset_group,
                                    &[page_record],
                                    &localized.assets,
                                    false,
                                )?;
                                Some(manifest_rel)
                            } else {
                                None
                            };

                            let mut trace = render_out.trace;
                            trace.push(format!("node_kind={}", web::static_generate::NODE_KIND));
                            trace.push(format!("generated_path={rel_path}"));
                            trace.push(format!("generated_status={status}"));

                            Ok(vec![NodeExecutionOutput {
                                output_pins: vec![web::static_generate::OUTPUT_PIN_OUT.to_string()],
                                payload: json!({
                                    "generated": {
                                        "status": status,
                                        "path": rel_path,
                                        "url": url,
                                        "route": route,
                                        "deploy_base_url": web::static_generate::effective_deploy_base_url(&config),
                                        "deploy_base_path": web::static_generate::effective_deploy_base_path(&config)?,
                                        "template": template_source.id,
                                        "site_root": site_root_rel,
                                        "manifest_path": manifest_rel,
                                        "asset_group": asset_group,
                                        "bytes": bytes,
                                    }
                                }),
                                trace,
                            }])
                        })
                    }
                    NodeDispatch::Agent(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::AiTts(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::LogicIf(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::LogicMatch(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::LogicCollect(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::LogicForeach(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::LogicReduce(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::LogicRetry(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::AuthTokenCreate(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::AuthTokenVerify(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::MailSend(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Concept(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::WebError(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::WsTrigger(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::WsSyncState(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::WsEmit(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Crypto(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::TriggerFunction(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::FunctionCall(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::FileSave(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::FsObject(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::MapserverCrud(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::TableConvert(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::TableQuery(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::FileCompress(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::FileDecompress(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::GeoInspect(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::GeoConvert(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::FilePdfConvert(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::ImgThumbnail(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::SvgConvert(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::ImgChromakey(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::Input(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvSet(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvGet(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvDel(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvExists(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvExpire(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvIncr(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvPublish(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::McpTrigger(node) => node.execute_many_async(input_for_exec).await,
                    NodeDispatch::KvSubscribe(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::WsClientTrigger(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::WsClientSend(node) => {
                        node.execute_many_async(input_for_exec).await
                    }
                    NodeDispatch::InstalledNode {
                        kind,
                        config,
                        platform,
                        egress,
                    } => {
                        execute_installed_node(kind, config, platform, egress, input_for_exec).await
                    }
                }
            }; // end exec_fut
            let timeout_node_id = trace_node_id.clone();
            let timeout_is_per_node = effective_config.get("timeout_secs").is_some();
            let exec_result: Result<Vec<NodeExecutionOutput>, PipelineError> =
                tokio::time::timeout(std::time::Duration::from_secs(node_timeout_secs), exec_fut)
                    .await
                    .unwrap_or_else(|_| {
                        let source = if timeout_is_per_node {
                            "per-node"
                        } else {
                            "project-level"
                        };
                        Err(PipelineError::new(
                            "FW_NODE_TIMEOUT",
                            format!(
                                "node '{}' timed out after {node_timeout_secs}s ({source} timeout)",
                                timeout_node_id
                            ),
                        ))
                    });

            let outputs = match exec_result {
                Ok(mut outs) => {
                    let mut processed_payloads: Vec<Value> = Vec::new();
                    // Skipped entirely when not recorded: not capturing is
                    // cheaper than capturing and discarding, which is the
                    // point on a device.
                    let record_input = trace_capture.records_payload_for(true, preview_in);
                    let record_output = trace_capture.records_payload_for(true, preview_out);
                    let trace_input = if record_input {
                        trace_capture.capture(&input_snapshot)
                    } else {
                        Value::Null
                    };

                    for out in &mut outs {
                        out.payload = materialize_node_output_files(
                            self.platform.as_ref(),
                            ctx,
                            &trace_node_kind,
                            out.payload.clone(),
                        )?;
                        let mut output_payload = out.payload.clone();
                        let payload_redact_tokens = take_private_redact_tokens(&mut output_payload);
                        let payload_redact_except_paths =
                            take_private_redact_except_paths(&mut output_payload);
                        let payload_output = if payload_redact_tokens.is_empty() {
                            output_payload
                        } else {
                            redact_json_value(
                                &output_payload,
                                &payload_redact_tokens,
                                &payload_redact_except_paths,
                                &[],
                            )
                        };
                        processed_payloads.push(payload_output.clone());
                        out.payload = payload_output;
                    }

                    let redacted_config = base_trace_config.clone();
                    // Rule 3 by position: the node kind declares where a
                    // secret sits in its own output, and only the record is
                    // masked. `outs` — what the next node receives — is
                    // untouched, because a level shapes the record, never the
                    // run.
                    let node_output_value = if !record_output {
                        Value::Null
                    } else {
                        match declared_secret_paths(&trace_node_kind) {
                            Some(paths) => crate::pipeline::trace_capture::mask_secret_paths(
                                &trace_capture.outputs(&outs),
                                paths,
                            ),
                            None => trace_capture.outputs(&outs),
                        }
                    };
                    let mut nodes_output_value = if processed_payloads.len() == 1 {
                        processed_payloads[0].clone()
                    } else {
                        Value::Array(processed_payloads)
                    };
                    // An input node passes the envelope through untouched, but
                    // *its* value — what `$nodes.<id>` answers — is the field it
                    // checked. The family's own reader says which, on the same
                    // payload it just passed, so nothing new rides the output.
                    if let Some(checked) =
                        input::scope_value(&trace_node_kind, &effective_config, &nodes_output_value)
                    {
                        nodes_output_value = checked;
                    }
                    // A declared image preview of a temporary file keeps a
                    // small copy in the record, since the file itself is
                    // deleted with the run. Read off the stored config, like
                    // `preview_in` / `preview_out` above.
                    let preview_snapshot = if preview_in || preview_out {
                        crate::pipeline::trace_capture::preview_snapshot::snapshot_for(
                            self.platform.as_ref(),
                            &ctx.owner,
                            &ctx.project,
                            &node.config,
                            &input_snapshot,
                            &nodes_output_value,
                        )
                    } else {
                        None
                    };
                    if should_retain_node_output(&trace_node_id, &nodes_retention) {
                        nodes_output.insert(trace_node_id.clone(), nodes_output_value);
                    }
                    // A `logic.retry` that sent a verdict round again is a
                    // wait, and the canvas should say so: `retry` in the
                    // record, `node_retry` on the bus, with the count the
                    // node stamped in `__zf_retry` (the platform's key, not
                    // the author's body). On the error road the failing node
                    // has already told this story, so the retry node stays
                    // `ok` there and the status line keeps the error text.
                    let verdict_retry = trace_node_kind == logic::retry::NODE_KIND
                        && outs.iter().any(|o| {
                            o.output_pins.iter().any(|p| p == logic::retry::OUTPUT_PIN_RETRY)
                        })
                        && input_snapshot
                            .get(RETRY_STATE_KEY)
                            .and_then(|s| s.get("failing_node_id"))
                            .is_none();
                    let status_word = if verdict_retry {
                        "retry"
                    } else if outs.is_empty() {
                        // Ran and emitted nothing — a match with no case, a
                        // filter that filtered everything. Not an error; grey
                        // in a run view where six green dots would lie.
                        "skip"
                    } else {
                        "ok"
                    };
                    node_trace.push(NodeTraceEntry {
                        node_id: trace_node_id.clone(),
                        node_kind: trace_node_kind.clone(),
                        config: if trace_capture.records_successful_payloads() {
                            redacted_config
                        } else {
                            None
                        },
                        duration_ms: node_start.elapsed().as_millis() as u64,
                        input: trace_input,
                        output: node_output_value,
                        error: None,
                        status: status_word.to_string(),
                        error_code: None,
                        preview_snapshot,
                    });
                    let duration_ms = node_start.elapsed().as_millis() as u64;
                    if verdict_retry {
                        let state = outs
                            .first()
                            .and_then(|o| o.payload.get(RETRY_STATE_KEY))
                            .cloned()
                            .unwrap_or(Value::Null);
                        let attempt = state.get("attempt").and_then(Value::as_u64).unwrap_or(1);
                        let max_attempts = state.get("max_attempts").and_then(Value::as_u64);
                        // The reason a wait is a wait: the pause the node
                        // took before the next attempt — the wait it stamped
                        // (grown by `--backoff`), or its `--delay-ms` when
                        // an older record has no stamp.
                        let message = state
                            .get("next_delay_ms")
                            .and_then(Value::as_u64)
                            .or_else(|| {
                                node.config.get("delay_ms").and_then(|d| {
                                    d.as_u64().or_else(|| d.as_str().and_then(|s| s.trim().parse().ok()))
                                })
                            })
                            .filter(|d| *d > 0)
                            .map(|d| format!("next attempt in {d} ms"));
                        emit_lifecycle(
                            &bus,
                            "node_retry",
                            match max_attempts {
                                Some(max) => format!(
                                    "{trace_node_id} {} waiting {attempt}/{max} {duration_ms} ms",
                                    short_kind_word(&trace_node_kind)
                                ),
                                None => format!(
                                    "{trace_node_id} {} waiting {attempt} {duration_ms} ms",
                                    short_kind_word(&trace_node_kind)
                                ),
                            },
                            Some((&trace_node_id, &trace_node_kind)),
                            Some(json!({
                                "attempt": attempt,
                                "max_attempts": max_attempts,
                                "duration_ms": duration_ms,
                                "message": message,
                            })),
                            &run_started,
                        );
                    } else {
                        let outcome = if outs.is_empty() { "node_skip" } else { "node_ok" };
                        emit_lifecycle(
                            &bus,
                            outcome,
                            format!(
                                "{trace_node_id} {} {duration_ms} ms",
                                short_kind_word(&trace_node_kind)
                            ),
                            Some((&trace_node_id, &trace_node_kind)),
                            Some(json!({ "duration_ms": duration_ms })),
                            &run_started,
                        );
                    }
                    outs
                }
                Err(mut e) => {
                    // Attribute error to the failing node if not already set.
                    if e.node_id.is_none() {
                        e.node_id = Some(trace_node_id.clone());
                        e.node_kind = Some(trace_node_kind.clone());
                    }
                    // A failure an `:error` edge consumes is not the run
                    // failing: it is a retry (the consumer is `logic.retry`)
                    // or an error handled by whatever the edge reaches. The
                    // record and the bus say which, so a poll loop that
                    // waited eight times and then succeeded does not show
                    // eight red crosses. `node_fail` is for the unrouted
                    // failure that ends the run.
                    let routed_edges = outgoing.get(&(node.id.as_str(), "error"));
                    let retry_consumer = routed_edges.and_then(|edges| {
                        edges
                            .iter()
                            .filter_map(|(to_node, _)| node_map.get(to_node).copied())
                            .find(|target| target.kind == logic::retry::NODE_KIND)
                    });
                    let error_payload = routed_edges.map(|_| {
                        let last_attempt = retry_consumer
                            .map(|retry| retry_last_attempt(&nodes_output, &retry.id))
                            .unwrap_or(0);
                        build_retry_error_payload(&input_snapshot, &e, last_attempt)
                    });
                    let duration_ms = node_start.elapsed().as_millis() as u64;
                    let routed_status = match (routed_edges, retry_consumer) {
                        (None, _) => None,
                        (Some(_), Some(retry)) => {
                            let attempt = error_payload
                                .as_ref()
                                .map(retry_attempt_from_payload)
                                .unwrap_or(1);
                            let max_attempts = retry_max_attempts(retry);
                            emit_lifecycle(
                                &bus,
                                "node_retry",
                                match max_attempts {
                                    Some(max) => format!(
                                        "{trace_node_id} {} retry {attempt}/{max} {duration_ms} ms",
                                        short_kind_word(&trace_node_kind)
                                    ),
                                    None => format!(
                                        "{trace_node_id} {} retry {attempt} {duration_ms} ms",
                                        short_kind_word(&trace_node_kind)
                                    ),
                                },
                                Some((&trace_node_id, &trace_node_kind)),
                                Some(json!({
                                    "attempt": attempt,
                                    "max_attempts": max_attempts,
                                    "duration_ms": duration_ms,
                                    "error_code": e.code,
                                    "message": e.message,
                                })),
                                &run_started,
                            );
                            Some("retry")
                        }
                        (Some(edges), None) => {
                            let to_node = edges.first().map(|(to, _)| *to).unwrap_or("");
                            emit_lifecycle(
                                &bus,
                                "node_error_routed",
                                format!(
                                    "{trace_node_id} {} error → {to_node} {duration_ms} ms",
                                    short_kind_word(&trace_node_kind)
                                ),
                                Some((&trace_node_id, &trace_node_kind)),
                                Some(json!({
                                    "duration_ms": duration_ms,
                                    "error_code": e.code,
                                    "message": e.message,
                                    "to_node": to_node,
                                })),
                                &run_started,
                            );
                            Some("error_routed")
                        }
                    };
                    if routed_status.is_none() {
                        emit_node_fail(
                            &bus,
                            &trace_node_id,
                            &trace_node_kind,
                            &e,
                            &node_start,
                            &run_started,
                        );
                    }
                    node_trace.push(NodeTraceEntry {
                        node_id: trace_node_id.clone(),
                        node_kind: trace_node_kind.clone(),
                        config: base_trace_config,
                        duration_ms,
                        input: if trace_capture.records_any_payload() {
                            trace_capture.capture(&input_snapshot)
                        } else {
                            Value::Null
                        },
                        output: Value::Null,
                        // The error text stays so the Logs panel says why,
                        // whichever way the failure went.
                        error: Some(e.message.clone()),
                        status: match routed_status {
                            Some(word) => word.to_string(),
                            None => crate::pipeline::error_class::class_of(e.code)
                                .as_status_word()
                                .to_string(),
                        },
                        error_code: Some(e.code.to_string()),
                        preview_snapshot: None,
                    });
                    if let (Some(next_edges), Some(error_payload)) = (routed_edges, error_payload) {
                        for (to_node, to_pin) in next_edges {
                            queue.push_back(NodeExecutionInput {
                                node_id: (*to_node).to_string(),
                                input_pin: (*to_pin).to_string(),
                                payload: error_payload.clone(),
                                metadata: execution_metadata(
                                    ctx,
                                    nodes_scope_for_target(
                                        to_node,
                                        &nodes_output,
                                        &nodes_retention,
                                    ),
                                    ctx.placeholder.clone(),
                                ),
                                bus: bus.clone(),
                            });
                        }
                        continue;
                    }
                    e.node_trace = node_trace.clone();
                    return Err(e);
                }
            };
            let outputs = if node.kind == logic::reduce::NODE_KIND {
                let mut final_outputs = Vec::new();
                if let Some(last_output) = outputs.last() {
                    let state = reduce_pending.entry(node.id.clone()).or_default();
                    state.acc = Some(last_output.payload.clone());
                    state.received += 1;
                    if state.expected.is_none() {
                        state.expected = input_snapshot
                            .get("count")
                            .and_then(Value::as_u64)
                            .map(|n| n as usize);
                    }
                    let expected = state.expected.unwrap_or(1);
                    if state.received >= expected {
                        final_outputs.push(last_output.clone());
                        reduce_pending.remove(node.id.as_str());
                    }
                }
                final_outputs
            } else {
                outputs
            };

            for mut output in outputs {
                // ── __signal extraction ──────────────────────────────────────
                // Strip `__signal` from node output and route through the bus.
                // Supports: string, object {kind,message,data}, or array of either.
                if let Some(raw_signal) = output
                    .payload
                    .as_object_mut()
                    .and_then(|m| m.remove("__signal"))
                {
                    if let Some(ref bus) = bus {
                        let emit = |s: &Value| {
                            let signal = match s {
                                Value::String(msg) => Signal {
                                    kind: "signal".to_string(),
                                    message: msg.clone(),
                                    node_id: trace_node_id.clone(),
                                    node_kind: trace_node_kind.clone(),
                                    data: None,
                                    at: format!("{}ms", node_start.elapsed().as_millis()),
                                },
                                Value::Object(obj) => Signal {
                                    kind: obj
                                        .get("kind")
                                        .and_then(Value::as_str)
                                        .unwrap_or("signal")
                                        .to_string(),
                                    message: obj
                                        .get("message")
                                        .and_then(Value::as_str)
                                        .unwrap_or("")
                                        .to_string(),
                                    node_id: trace_node_id.clone(),
                                    node_kind: trace_node_kind.clone(),
                                    data: obj.get("data").cloned(),
                                    at: format!("{}ms", node_start.elapsed().as_millis()),
                                },
                                _ => return,
                            };
                            bus.emit(signal);
                        };
                        if let Value::Array(items) = &raw_signal {
                            for item in items {
                                emit(item);
                            }
                        } else {
                            emit(&raw_signal);
                        }
                    }
                }
                trace.extend(output.trace.clone());
                last_value = output.payload.clone();

                for emitted_pin in &output.output_pins {
                    if let Some(next_edges) =
                        outgoing.get(&(node.id.as_str(), emitted_pin.as_str()))
                    {
                        for (to_node, to_pin) in next_edges {
                            let target = node_map.get(to_node).ok_or_else(|| {
                                PipelineError::new(
                                    "FW_EXEC_EDGE",
                                    format!("target node '{}' missing", to_node),
                                )
                            })?;
                            if is_logic_collect(target) {
                                let pending =
                                    collect_pending.entry((*to_node).to_string()).or_default();
                                let key = if emitted_pin == "out" {
                                    node.id.clone()
                                } else {
                                    format!("{}:{}", node.id, emitted_pin)
                                };
                                pending.insert(key, output.payload.clone());
                                let expected = incoming_counts.get(*to_node).copied().unwrap_or(1);
                                if pending.len() >= expected {
                                    let combined = Value::Object(
                                        pending
                                            .iter()
                                            .map(|(k, v)| (k.clone(), v.clone()))
                                            .collect(),
                                    );
                                    collect_pending.remove(*to_node);
                                    queue.push_back(NodeExecutionInput {
                                        node_id: (*to_node).to_string(),
                                        input_pin: (*to_pin).to_string(),
                                        payload: combined,
                                        metadata: execution_metadata(
                                            ctx,
                                            nodes_scope_for_target(
                                                to_node,
                                                &nodes_output,
                                                &nodes_retention,
                                            ),
                                            ctx.placeholder.clone(),
                                        ),
                                        bus: bus.clone(),
                                    });
                                }
                            } else {
                                queue.push_back(NodeExecutionInput {
                                    node_id: (*to_node).to_string(),
                                    input_pin: (*to_pin).to_string(),
                                    payload: output.payload.clone(),
                                    metadata: execution_metadata(
                                        ctx,
                                        nodes_scope_for_target(
                                            to_node,
                                            &nodes_output,
                                            &nodes_retention,
                                        ),
                                        ctx.placeholder.clone(),
                                    ),
                                    bus: bus.clone(),
                                });
                            }
                        }
                    }
                }
            }
        }

        Ok(PipelineOutput {
            value: strip_private_markers(last_value),
            trace,
            node_trace,
        })
    }
}

/// Removes the engine's private redaction markers from a pipeline's result.
///
/// `__zf_private_trace_redact` and its siblings are how a node tells the trace
/// sanitizer which literal values to strike out; they travel in the payload
/// because that is the only channel between nodes. They are not part of
/// anybody's API, and the trace marker's contents are the secrets themselves,
/// so none of them may ride out to a caller in the response body.
///
/// The walk is recursive because a marker does not stay at the top: an
/// `n.web.response` in the chain nests the whole upstream payload under
/// `__zf_response.body`, and a top-level-only sweep would leave it there.
fn strip_private_markers(value: Value) -> Value {
    match value {
        Value::Object(map) => Value::Object(
            map.into_iter()
                .filter(|(key, _)| {
                    !matches!(
                        key.as_str(),
                        "__zf_private_trace_redact"
                            | "__zf_private_redact"
                            | "__zf_private_redact_except_paths"
                    )
                })
                .map(|(key, item)| (key, strip_private_markers(item)))
                .collect(),
        ),
        Value::Array(items) => Value::Array(items.into_iter().map(strip_private_markers).collect()),
        other => other,
    }
}

/// Declared secret positions per node kind — `NodeDefinition::secret_paths`.
///
/// Built once. Looking these up per node execution by rebuilding the whole
/// definition list would cost more than the masking it enables.
static SECRET_PATHS_BY_KIND: std::sync::LazyLock<
    std::collections::HashMap<String, Vec<String>>,
> = std::sync::LazyLock::new(|| {
    crate::pipeline::nodes::builtin_node_definitions()
        .into_iter()
        .filter(|def| !def.secret_paths.is_empty())
        .map(|def| (def.kind, def.secret_paths))
        .collect()
});

/// Declared secret positions for a node kind, if it declares any.
fn declared_secret_paths(kind: &str) -> Option<&'static [String]> {
    SECRET_PATHS_BY_KIND.get(kind).map(Vec::as_slice)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use serde_json::json;

    use super::{
        BasicPipelineEngine, NodesAccess, NodesAccessAccumulator, PipelineContext,
        redact_json_value, sanitized_trace_value, scan_text_nodes_access,
        take_private_redact_except_paths, take_private_redact_tokens, trace_config_snapshot,
    };
    use crate::pipeline::interface::PipelineEngine;
    use crate::pipeline::model::{PipelineEdge, PipelineGraph, PipelineNode};
    use crate::pipeline::nodes::basic::{
        script,
        table::convert::{TableFormat, collect_columns, encode_rows},
    };
    use crate::platform::model::PlatformConfig;
    use crate::platform::services::PlatformService;
    use crate::platform::shell::parser::build_pipeline_graph;

    /// Compaction affects logs only, including across intermediate nodes. A
    /// zero override must preserve all array items while other caps still apply.
    #[tokio::test]
    async fn trace_capture_preserves_execution_data_and_pipeline_overrides() {
        use crate::pipeline::model::{PipelineGraphMetadata, PipelineGraphSettings};
        use crate::pipeline::trace_capture::TraceCaptureSettings;
        let mut graph = build_pipeline_graph(
            "trace-capture",
            r#"
[a] trigger.manual
[b] logic.if --expr "1 == 1"
[c] script -- "return { count: input.rows.length, last: input.rows[input.rows.length - 1] };"
[a] -> [b]
[b]:true -> [c]
"#,
        )
        .expect("graph");
        graph.metadata = Some(PipelineGraphMetadata {
            settings: PipelineGraphSettings {
                trace_capture: Some(TraceCaptureSettings {
                    // This test is about what capture *does* to a payload, so
                    // it asks for the level that records one. The default is
                    // on-error, which deliberately records nothing for a node
                    // that succeeded.
                    level: Some(crate::pipeline::trace_capture::CaptureLevel::Full),
                    array_sample_count: Some(1),
                    ..Default::default()
                }),
                ..Default::default()
            },
            ..Default::default()
        });
        let ctx = PipelineContext {
            owner: "test".into(),
            project: "test".into(),
            pipeline: "trace-capture".into(),
            request_id: "trace-test".into(),
            route: String::new(),
            input: json!({"rows": (0..100).collect::<Vec<_>>()}),
            trigger: None,
            placeholder: None,
        };
        let engine = BasicPipelineEngine::default();
        let sampled = engine
            .execute_async(&graph, &ctx)
            .await
            .expect("sampled execution");
        assert_eq!(sampled.value, json!({"count":100,"last":99}));
        for trace in &sampled.node_trace {
            assert_eq!(trace.input["rows"]["len"], 100);
            assert_eq!(trace.input["rows"]["preview"], json!([0]));
        }
        graph
            .metadata
            .as_mut()
            .unwrap()
            .settings
            .trace_capture
            .as_mut()
            .unwrap()
            .array_sample_count = Some(0);
        let full = engine
            .execute_async(&graph, &ctx)
            .await
            .expect("unsampled execution");
        assert_eq!(full.value, sampled.value);
        assert_eq!(
            full.node_trace[0].input["rows"].as_array().unwrap().len(),
            100
        );
    }

    /// Capture levels decide what a run records — `kinds/invocation-record`
    /// § Capture Level. Asserted against a canary that would be plainly visible
    /// in a payload if the level kept it.
    #[tokio::test]
    async fn capture_levels_decide_what_a_run_records() {
        use crate::pipeline::model::{PipelineGraphMetadata, PipelineGraphSettings};
        use crate::pipeline::trace_capture::{CaptureLevel, TraceCaptureSettings};

        async fn run_at(level: CaptureLevel, body: &str) -> crate::pipeline::model::PipelineOutput {
            let mut graph = build_pipeline_graph(
                "levels",
                &format!("[a] trigger.manual\n[b] script -- \"{body}\"\n[a] -> [b]\n"),
            )
            .expect("graph");
            graph.metadata = Some(PipelineGraphMetadata {
                settings: PipelineGraphSettings {
                    trace_capture: Some(TraceCaptureSettings {
                        level: Some(level),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
                ..Default::default()
            });
            let ctx = PipelineContext {
                owner: "test".into(),
                project: "test".into(),
                pipeline: "levels".into(),
                request_id: "levels-test".into(),
                route: String::new(),
                input: json!({ "canary": "CANARY-IN-INPUT" }),
                trigger: None,
                placeholder: None,
            };
            BasicPipelineEngine::default()
                .execute_async(&graph, &ctx)
                .await
                .expect("execution")
        }

        let ok_body = "return { seen: input.canary };";

        // full — everything is there.
        let out = run_at(CaptureLevel::Full, ok_body).await;
        let text = serde_json::to_string(&out.node_trace).unwrap();
        assert!(text.contains("CANARY-IN-INPUT"), "full must record payloads");

        // on-error — a node that succeeded records nothing, but the run still
        // returns its real value: a level shapes the record, never the run.
        let out = run_at(CaptureLevel::OnError, ok_body).await;
        let text = serde_json::to_string(&out.node_trace).unwrap();
        assert!(
            !text.contains("CANARY-IN-INPUT"),
            "on-error must not record a successful node's payload: {text}"
        );
        assert_eq!(out.value, json!({ "seen": "CANARY-IN-INPUT" }));
        assert!(
            out.node_trace.iter().all(|t| t.status == "ok"),
            "status must survive at on-error"
        );

        // none — nothing, even when a node fails.
        let out = run_at(CaptureLevel::None, ok_body).await;
        let text = serde_json::to_string(&out.node_trace).unwrap();
        assert!(!text.contains("CANARY-IN-INPUT"), "none must record nothing");
        assert!(
            !out.node_trace.is_empty(),
            "none still records that the nodes ran"
        );
    }

    /// The point of `on-error`: the run that fails is the one you can read.
    #[tokio::test]
    async fn on_error_records_the_node_that_failed() {
        use crate::pipeline::model::{PipelineGraphMetadata, PipelineGraphSettings};
        use crate::pipeline::trace_capture::{CaptureLevel, TraceCaptureSettings};

        let mut graph = build_pipeline_graph(
            "levels-fail",
            "[a] trigger.manual\n[b] script -- \"throw new Error('boom');\"\n[a] -> [b]\n",
        )
        .expect("graph");
        graph.metadata = Some(PipelineGraphMetadata {
            settings: PipelineGraphSettings {
                trace_capture: Some(TraceCaptureSettings {
                    level: Some(CaptureLevel::OnError),
                    ..Default::default()
                }),
                ..Default::default()
            },
            ..Default::default()
        });
        let ctx = PipelineContext {
            owner: "test".into(),
            project: "test".into(),
            pipeline: "levels-fail".into(),
            request_id: "levels-fail-test".into(),
            route: String::new(),
            input: json!({ "canary": "CANARY-IN-INPUT" }),
            trigger: None,
            placeholder: None,
        };
        let result = BasicPipelineEngine::default().execute_async(&graph, &ctx).await;
        let trace = match &result {
            Ok(out) => out.node_trace.clone(),
            Err(e) => e.node_trace.clone(),
        };
        let text = serde_json::to_string(&trace).unwrap();
        assert!(
            text.contains("CANARY-IN-INPUT"),
            "on-error must record the failing node's input: {text}"
        );
    }

    /// A declared preview records its own payload at `on-error` — and only
    /// its own: the node beside it, and its other half, stay unrecorded. At
    /// `none` the preview changes nothing.
    #[tokio::test]
    async fn a_declared_preview_records_its_payload_at_on_error() {
        use crate::pipeline::model::{PipelineGraphMetadata, PipelineGraphSettings};
        use crate::pipeline::trace_capture::{CaptureLevel, TraceCaptureSettings};

        async fn run_at(level: CaptureLevel) -> crate::pipeline::model::PipelineOutput {
            let mut graph = build_pipeline_graph(
                "preview-levels",
                "[a] trigger.manual\n[b] script --preview json -- \"return { seen: input.canary };\"\n[a] -> [b]\n",
            )
            .expect("graph");
            assert_eq!(graph.nodes[1].config["preview"]["out"], json!({ "as": "json" }));
            graph.metadata = Some(PipelineGraphMetadata {
                settings: PipelineGraphSettings {
                    trace_capture: Some(TraceCaptureSettings {
                        level: Some(level),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
                ..Default::default()
            });
            let ctx = PipelineContext {
                owner: "test".into(),
                project: "test".into(),
                pipeline: "preview-levels".into(),
                request_id: "preview-levels-test".into(),
                route: String::new(),
                input: json!({ "canary": "CANARY-IN-INPUT" }),
                trigger: None,
                placeholder: None,
            };
            BasicPipelineEngine::default()
                .execute_async(&graph, &ctx)
                .await
                .expect("execution")
        }

        let out = run_at(CaptureLevel::OnError).await;
        let a = out.node_trace.iter().find(|t| t.node_id == "a").expect("node a");
        let b = out.node_trace.iter().find(|t| t.node_id == "b").expect("node b");
        assert_eq!(a.output, serde_json::Value::Null, "no preview on a: nothing recorded");
        assert_eq!(a.input, serde_json::Value::Null);
        assert_eq!(b.input, serde_json::Value::Null, "b previews its output, not its input");
        assert_eq!(
            b.output,
            json!({ "seen": "CANARY-IN-INPUT" }),
            "b previews its output, so on-error records it"
        );
        assert!(b.config.is_none(), "config capture is unchanged by a preview");

        let out = run_at(CaptureLevel::None).await;
        let text = serde_json::to_string(&out.node_trace).unwrap();
        assert!(!text.contains("CANARY-IN-INPUT"), "none records nothing, preview or not: {text}");
    }

    /// The engine announces the run on the bus: `run_start`, then per node
    /// `node_start` and exactly one outcome, then `run_done` — in execution
    /// order, with the node named and its duration in `data`. A failing
    /// node gets `node_fail` with its code, and the run still gets its
    /// `run_done` (status `error`).
    #[tokio::test]
    async fn the_engine_announces_run_and_node_lifecycle_on_the_bus() {
        use crate::pipeline::model::{
            ExecuteOptions, ExecutionBus, PipelineError, PipelineOutput, Signal,
        };

        async fn signals_of(dsl: &str) -> (Result<PipelineOutput, PipelineError>, Vec<Signal>) {
            let graph = build_pipeline_graph("lifecycle", dsl).expect("graph");
            let ctx = PipelineContext {
                owner: "test".into(),
                project: "test".into(),
                pipeline: "lifecycle".into(),
                request_id: "lifecycle-run-1".into(),
                route: String::new(),
                input: json!({ "n": 1 }),
                trigger: None,
                placeholder: None,
            };
            let bus = Arc::new(ExecutionBus::new(64));
            let mut rx = bus.subscribe();
            let options = ExecuteOptions { bus: Some(bus) };
            let result = BasicPipelineEngine::default()
                .execute_with_options_async(&graph, &ctx, &options)
                .await;
            let mut signals = Vec::new();
            while let Ok(signal) = rx.try_recv() {
                signals.push(signal);
            }
            (result, signals)
        }

        let (result, signals) = signals_of(
            "[a] trigger.manual\n[b] script -- \"return { n: input.n + 1 };\"\n[a] -> [b]\n",
        )
        .await;
        result.expect("the run succeeds");
        let seen: Vec<(String, String)> = signals
            .iter()
            .map(|s| (s.kind.clone(), s.node_id.clone()))
            .collect();
        assert_eq!(
            seen,
            [
                ("run_start", ""),
                ("node_start", "a"),
                ("node_ok", "a"),
                ("node_start", "b"),
                ("node_ok", "b"),
                ("run_done", ""),
            ]
            .map(|(k, n)| (k.to_string(), n.to_string()))
        );
        let node_ok = &signals[4];
        assert_eq!(node_ok.node_kind, "n.script");
        assert!(node_ok.data.as_ref().unwrap()["duration_ms"].is_u64());
        assert!(node_ok.message.starts_with("b script "), "{}", node_ok.message);
        let run_done = signals.last().unwrap();
        let data = run_done.data.as_ref().unwrap();
        assert_eq!(data["run_id"], "lifecycle-run-1");
        assert_eq!(data["status"], "ok");
        assert!(data["duration_ms"].is_u64());

        let (result, signals) = signals_of(
            "[a] trigger.manual\n[b] script -- \"throw new Error('boom');\"\n[a] -> [b]\n",
        )
        .await;
        assert!(result.is_err());
        let kinds: Vec<&str> = signals.iter().map(|s| s.kind.as_str()).collect();
        assert_eq!(
            kinds,
            ["run_start", "node_start", "node_ok", "node_start", "node_fail", "run_done"]
        );
        let fail = &signals[4];
        assert_eq!(fail.node_id, "b");
        let data = fail.data.as_ref().unwrap();
        assert!(data["error_code"].as_str().is_some_and(|c| !c.is_empty()));
        assert!(data["error_class"].as_str().is_some_and(|c| c == "refused" || c == "failed"));
        assert!(data["error"].as_str().unwrap().contains("boom"), "{data}");
        assert_eq!(signals.last().unwrap().data.as_ref().unwrap()["status"], "error");
    }

    /// Rule 3 of `kinds/invocation-record` under a preview: a declared
    /// `--preview json` makes `on-error` record a successful node's payload,
    /// and that record goes through the same capture as every other, so a
    /// value the project took out of its credential store is masked in it.
    #[tokio::test]
    async fn a_previewed_payload_never_shows_a_credential_value() {
        use crate::pipeline::model::{PipelineGraphMetadata, PipelineGraphSettings};
        use crate::pipeline::trace_capture::{CaptureLevel, TraceCaptureSettings};

        const SECRET: &str = "sk-live-CREDENTIAL-VALUE-9f2c8d";
        crate::platform::services::credential::register_confidential_for_test(
            "preview-rule3",
            "preview-rule3",
            &json!({ "api_key": SECRET }),
        );
        let mut graph = build_pipeline_graph(
            "preview-rule3",
            "[a] trigger.manual\n[b] script --preview json -- \"return { token: input.secret, note: 'ok' };\"\n[a] -> [b]\n",
        )
        .expect("graph");
        graph.metadata = Some(PipelineGraphMetadata {
            settings: PipelineGraphSettings {
                trace_capture: Some(TraceCaptureSettings {
                    level: Some(CaptureLevel::OnError),
                    ..Default::default()
                }),
                ..Default::default()
            },
            ..Default::default()
        });
        let ctx = PipelineContext {
            owner: "preview-rule3".into(),
            project: "preview-rule3".into(),
            pipeline: "preview-rule3".into(),
            request_id: "preview-rule3-test".into(),
            route: String::new(),
            input: json!({ "secret": SECRET }),
            trigger: None,
            placeholder: None,
        };
        let out = BasicPipelineEngine::default()
            .execute_async(&graph, &ctx)
            .await
            .expect("execution");
        assert_eq!(out.value["token"], json!(SECRET), "the run itself is never masked");
        let b = out.node_trace.iter().find(|t| t.node_id == "b").expect("node b");
        assert_eq!(b.output["note"], json!("ok"), "the preview recorded the payload");
        let text = serde_json::to_string(&out.node_trace).unwrap();
        assert!(!text.contains(SECRET), "the record shows no credential value: {text}");
    }

    /// Reproducible capture-only comparison, without external I/O. This is an
    /// ignored diagnostic, not a timing assertion or a production throughput
    /// promise. The legacy implementation is compiled only into tests.
    #[test]
    #[ignore = "manual capture performance measurement"]
    fn trace_capture_benchmark() {
        use crate::pipeline::trace_capture::{TraceCapture, TraceCaptureSettings};
        use std::hint::black_box;
        use std::time::Instant;
        for (name, rows, repeats, nodes) in [
            ("small", 4, 1000, 1),
            ("large", 10000, 5, 1),
            ("long-chain", 10000, 1, 30),
        ] {
            let payload = json!({"rows": (0..rows).map(|i| json!({"id": i, "name": "example-row", "nested": {"values": [1,2,3,4]}, "body": "x".repeat(256)})).collect::<Vec<_>>()});
            for mode in ["disabled", "legacy", "compact"] {
                let start = Instant::now();
                let mut bytes = 0;
                for _ in 0..repeats {
                    let mut capture = TraceCapture::new(
                        TraceCaptureSettings {
                            array_sample_count: Some(1),
                            ..Default::default()
                        }
                        .resolve(None),
                    );
                    for _ in 0..nodes {
                        if mode == "disabled" {
                            black_box(&payload);
                            continue;
                        }
                        capture.begin_node();
                        let log = if mode == "legacy" {
                            super::legacy_sanitized_trace_value(black_box(&payload))
                        } else {
                            capture.capture(black_box(&payload))
                        };
                        bytes += serde_json::to_vec(black_box(&log)).unwrap().len();
                    }
                }
                eprintln!(
                    "capture-bench {name} {mode}: {:.3} ms/capture, {} bytes/capture ({} captures, debug_assertions={})",
                    start.elapsed().as_secs_f64() * 1000.0 / (repeats * nodes) as f64,
                    bytes / (repeats * nodes),
                    repeats * nodes,
                    cfg!(debug_assertions)
                );
            }
        }
    }

    /// A curated bundle ships in the binary, so a declaration it cannot run
    /// under is a startup-shaped bug rather than a user's problem.
    ///
    /// Checks both halves of the guard against every embedded bundle: no inner
    /// node is refused as uncheckable, and every host its functions name
    /// outright is one the bundle declared. A URL assembled at run time is not
    /// readable here, which is exactly why the check also exists at the moment
    /// of the request.
    ///
    /// The uncheckable half runs for bundles that declare nothing too, because
    /// that is now when it runs in production.
    #[test]
    fn embedded_bundles_can_run_under_their_own_declared_hosts() {
        use crate::contracts::decode_contract;
        use crate::contracts::kinds::{NodeBundleContract, decode_pipeline_graph};
        use crate::pipeline::security::BundleEgress;
        use crate::platform::web::embedded::{
            PLATFORM_COMPOSITE_NODE_ASSETS, platform_composite_node_asset,
        };

        fn literal_host(url: &str) -> Option<String> {
            let rest = url
                .strip_prefix("https://")
                .or_else(|| url.strip_prefix("http://"))?;
            let host = rest.split(['/', '?', '#']).next().unwrap_or_default();
            let host = host.split('@').next_back().unwrap_or_default();
            let host = host.split(':').next().unwrap_or_default();
            (!host.is_empty() && !host.contains("{{")).then(|| host.to_string())
        }

        let mut declaring_bundles = 0;
        for asset in PLATFORM_COMPOSITE_NODE_ASSETS
            .iter()
            .filter(|asset| asset.path.ends_with("/definition.json"))
        {
            let slug = asset.path.trim_end_matches("/definition.json");
            let spec = decode_contract::<NodeBundleContract>(asset.bytes)
                .unwrap_or_else(|error| panic!("embedded bundle '{slug}': {error}"))
                .spec;
            let egress = BundleEgress::extend(None, &spec.package, &spec.hosts);
            if egress.is_active() {
                declaring_bundles += 1;
            }

            for path in spec.functions.values() {
                let bytes = platform_composite_node_asset(&format!("{slug}/{path}"))
                    .unwrap_or_else(|| panic!("embedded bundle '{slug}' is missing '{path}'"));
                let graph = decode_pipeline_graph(bytes)
                    .unwrap_or_else(|error| panic!("embedded bundle '{slug}'/{path}: {error}"))
                    .spec;
                for node in &graph.nodes {
                    // The sandbox denies fetch as shipped, which is the state a
                    // curated bundle must run in; what happens when an operator
                    // widens it is asserted separately.
                    super::refuse_uncheckable_egress_node(&egress, &node.kind, false)
                        .unwrap_or_else(|error| {
                            panic!("embedded bundle '{slug}'/{path}: {}", error.message)
                        });
                    let Some(url) = node.config.get("url").and_then(|value| value.as_str()) else {
                        continue;
                    };
                    let Some(host) = literal_host(url) else {
                        continue;
                    };
                    egress
                        .check_host(&host, &node.kind)
                        .unwrap_or_else(|error| {
                            panic!("embedded bundle '{slug}'/{path}: {}", error.message)
                        });
                }
            }
        }
        assert!(
            declaring_bundles > 0,
            "no embedded bundle declares hosts, so this proved nothing"
        );
    }

    /// The incentive this guard exists to remove: if refusing an unreadable
    /// destination only happened inside a bundle that declared hosts, an author
    /// who wanted `n.pg.query` would be better off declaring nothing, and only
    /// the honest author would be constrained.
    #[test]
    fn a_bundle_that_declares_no_hosts_still_cannot_use_an_uncheckable_node() {
        use crate::pipeline::security::BundleEgress;

        let silent = BundleEgress::extend(None, "silent", &[]);
        assert!(!silent.is_active(), "the bundle declared nothing");

        for kind in [
            "n.ai.agent",
            "n.pg.query",
            "n.table.query",
            "n.ws.client.send",
            "n.trigger.ws.client",
        ] {
            let error = super::refuse_uncheckable_egress_node(&silent, kind, false)
                .expect_err("an unreadable destination is refused whatever was declared");
            assert_eq!(error.code, "FW_EGRESS_UNCHECKED_NODE");
            assert!(
                error.message.contains(kind) && error.message.contains("'silent'"),
                "{}",
                error.message
            );
        }

        // What a declaration does buy is still bought: a readable destination is
        // checked against the list rather than refused outright.
        super::refuse_uncheckable_egress_node(&silent, super::http::request::NODE_KIND, false)
            .expect("a host-checked node is not an unreadable destination");
    }

    /// `n.script` is a network node only when an operator has made it one, so
    /// the guard reads the sandbox in force rather than the sandbox as shipped.
    #[test]
    fn a_script_is_refused_inside_a_bundle_only_when_its_sandbox_reaches_the_network() {
        use crate::language::{
            DenoSandboxConfigPatch, DenoSandboxDangerZonePatch, DenoSandboxEngine, LanguageEngine,
        };
        use crate::pipeline::security::BundleEgress;

        let egress = BundleEgress::extend(None, "telegram", &["api.telegram.org".to_string()]);

        super::refuse_uncheckable_egress_node(&egress, super::script::NODE_KIND, false)
            .expect("a sandbox that denies fetch reaches nothing to check");

        let error = super::refuse_uncheckable_egress_node(&egress, super::script::NODE_KIND, true)
            .expect_err("a sandbox that may fetch is an egress path the guard cannot read");
        assert_eq!(error.code, "FW_EGRESS_UNCHECKED_NODE");
        assert!(
            error.message.contains("n.script") && error.message.contains("'telegram'"),
            "{}",
            error.message
        );

        // And the boolean is not hypothetical: it is what the engine answers.
        assert!(
            !DenoSandboxEngine::default().grants_network(),
            "the shipped sandbox denies fetch"
        );
        let widened = DenoSandboxEngine::new(
            DenoSandboxConfigPatch {
                danger_zone: Some(DenoSandboxDangerZonePatch {
                    allow_net: Some(true),
                    ..Default::default()
                }),
                ..Default::default()
            },
            DenoSandboxConfigPatch::default(),
        );
        assert!(
            widened.grants_network(),
            "an operator's platform patch is what the guard must read"
        );
    }

    #[test]
    fn private_redact_masks_overlaps_without_applying_log_limits_to_execution() {
        let tail = "retained execution text ".repeat(1024);
        let mut payload = json!({
            "__zf_private_redact": ["", "abc", "bcdef"],
            "__zf_private_redact_except_paths": ["response.body"],
            "message": format!("abcdef {tail}"),
            "nested": ["abcdef"],
            "response": {"body": "abcdef"}
        });
        let tokens = take_private_redact_tokens(&mut payload);
        let exceptions = take_private_redact_except_paths(&mut payload);
        let redacted = redact_json_value(&payload, &tokens, &exceptions, &[]);
        assert_eq!(redacted["message"], format!("•••••• {tail}"));
        assert_eq!(redacted["nested"][0], "••••••");
        assert_eq!(redacted["response"]["body"], "abcdef");
        assert!(redacted["message"].as_str().unwrap().len() > 8192);
    }

    #[test]
    fn private_redact_tokens_are_removed_and_applied_recursively() {
        let mut payload = json!({
            "__zf_private_redact": ["abc123", "secret-value"],
            "password": "abc123",
            "nested": {
                "preview": "token=secret-value",
                "array": ["abc123", "ok"]
            }
        });

        let tokens = take_private_redact_tokens(&mut payload);
        assert_eq!(tokens, vec!["abc123", "secret-value"]);
        assert!(payload.get("__zf_private_redact").is_none());

        let redacted = redact_json_value(&payload, &tokens, &[], &[]);
        assert_eq!(redacted["password"], "••••••");
        assert_eq!(redacted["nested"]["preview"], "token=••••••");
        assert_eq!(redacted["nested"]["array"][0], "••••••");
        assert_eq!(redacted["nested"]["array"][1], "ok");
    }

    #[test]
    fn private_redact_can_preserve_response_body_subtree() {
        let mut payload = json!({
            "__zf_private_redact": ["https://secret.example/api/login", "token-123"],
            "__zf_private_redact_except_paths": ["response.body"],
            "request": {
                "url": "https://secret.example/api/login",
                "summary": "token-123"
            },
            "response": {
                "body": {
                    "echoed_url": "https://secret.example/api/login",
                    "echoed_token": "token-123"
                }
            }
        });

        let tokens = take_private_redact_tokens(&mut payload);
        let except_paths = take_private_redact_except_paths(&mut payload);
        let redacted = redact_json_value(&payload, &tokens, &except_paths, &[]);

        assert_eq!(redacted["request"]["url"], "••••••");
        assert_eq!(redacted["request"]["summary"], "••••••");
        assert_eq!(
            redacted["response"]["body"]["echoed_url"],
            "https://secret.example/api/login"
        );
        assert_eq!(redacted["response"]["body"]["echoed_token"], "token-123");
    }

    #[test]
    fn private_trace_redact_tokens_are_removed_without_touching_payload_redact_keys() {
        let mut payload = json!({
            "__zf_private_trace_redact": ["abc123"],
            "token": "abc123",
            "nested": { "__zf_private_trace_redact": ["deep-secret"] }
        });

        let mut tokens = Vec::new();
        super::sweep_private_markers(&mut payload, &mut tokens);
        assert_eq!(
            tokens,
            vec!["abc123".to_string(), "deep-secret".to_string()]
        );
        assert!(payload.get("__zf_private_trace_redact").is_none());
        assert!(payload["nested"].get("__zf_private_trace_redact").is_none());
        assert_eq!(payload["token"], "abc123");
    }

    #[test]
    fn trace_config_snapshot_redacts_sensitive_keys_but_keeps_query_text() {
        let snapshot = trace_config_snapshot(&json!({
            "query": "SELECT * FROM posts LIMIT 10",
            "limit": 20,
            "ui": { "x": 100, "y": 120 },
            "password": "super-secret",
            "headers": {
                "authorization": "Bearer abc123"
            }
        }))
        .expect("config snapshot");

        assert_eq!(snapshot["query"], "SELECT * FROM posts LIMIT 10");
        assert_eq!(snapshot["limit"], 20);
        assert!(snapshot.get("ui").is_none());
        assert_eq!(snapshot["password"], "••••••");
        assert_eq!(snapshot["headers"]["authorization"], "••••••");
    }

    /// The fallback exists for the node this tree does not own. `access_token`
    /// was matched and a plainly-named `token` was not, which is the gap a
    /// third-party node would fall into.
    #[test]
    fn trace_config_snapshot_redacts_a_plainly_named_token() {
        let snapshot = trace_config_snapshot(&json!({
            "endpoint": "https://api.example.com/v1",
            "token": "t-abc123",
            "api_token": "t-def456",
            "auth_token": "t-ghi789",
            "session_token": "t-jkl012",
            "bearer_token": "t-mno345",
        }))
        .expect("config snapshot");

        assert_eq!(snapshot["endpoint"], "https://api.example.com/v1");
        for key in [
            "token",
            "api_token",
            "auth_token",
            "session_token",
            "bearer_token",
        ] {
            assert_eq!(snapshot[key], "••••••", "{key} must be redacted");
        }
    }

    #[test]
    fn trace_summary_summarizes_large_numeric_arrays_generically() {
        let vector = (0..128).map(|i| json!(i as f64 / 10.0)).collect::<Vec<_>>();
        let summary = sanitized_trace_value(&json!({
            "id": "row-1",
            "values": vector,
            "nested": {
                "items": [
                    { "payload": (0..80).map(|i| json!(i)).collect::<Vec<_>>() }
                ]
            }
        }));

        assert_eq!(summary["id"], "row-1");
        assert_eq!(summary["values"]["__zf_trace_summary"], "array");
        assert_eq!(summary["values"]["len"], 128);
        assert_eq!(
            summary["nested"]["items"][0]["payload"]["__zf_trace_summary"],
            "array"
        );
        assert_eq!(summary["nested"]["items"][0]["payload"]["len"], 80);
    }

    /// S4 regression, engine half. Every trigger's payload passes through the
    /// trace sanitizer, and only the webhook node ever publishes redaction
    /// tokens -- so a cron, WS, or script-built run carrying a secret had
    /// nothing standing between it and `trace.input` on disk. Remove the
    /// `blank_sensitive_keys` call in `sanitized_trace_value` and the password
    /// comes back in clear.
    #[test]
    fn a_secret_under_a_secret_key_never_reaches_a_trace() {
        let trace = sanitized_trace_value(&json!({
            "body": {
                "username": "wawan",
                "password": "toryoto",
                "profile": { "apiKey": "abc123", "city": "Jakarta" },
                "tokens": ["t-one", "t-two"]
            },
            "method": "POST"
        }));

        assert_eq!(
            trace["body"]["password"],
            "\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}"
        );
        assert_eq!(
            trace["body"]["profile"]["apiKey"],
            "\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}"
        );
        assert_eq!(
            trace["body"]["tokens"][0],
            "\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}"
        );
        assert_eq!(
            trace["body"]["tokens"][1],
            "\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}\u{2022}"
        );

        // What is not a secret is still legible, or the trace stops being useful.
        assert_eq!(trace["body"]["username"], "wawan");
        assert_eq!(
            sanitized_trace_value(&json!({ "authorization": true }))["authorization"],
            true,
            "a flag carries no secret and must stay readable"
        );
        assert_eq!(trace["body"]["profile"]["city"], "Jakarta");
        assert_eq!(trace["method"], "POST");

        let serialized = trace.to_string();
        for secret in ["toryoto", "abc123", "t-one", "t-two"] {
            assert!(
                !serialized.contains(secret),
                "'{secret}' survived: {serialized}"
            );
        }
    }

    /// The redaction markers are an engine-internal channel between nodes, and
    /// the trace one carries the secrets verbatim. A pipeline that returns its
    /// payload must not return them with it.
    #[test]
    fn private_markers_never_ride_out_in_a_pipeline_result() {
        let result = super::strip_private_markers(json!({
            "body": { "username": "wawan" },
            "__zf_private_trace_redact": ["toryoto"],
            "__zf_private_redact": ["abc123"],
            "__zf_private_redact_except_paths": ["response.body"]
        }));
        assert_eq!(result, json!({ "body": { "username": "wawan" } }));

        // `n.web.response` nests the whole upstream payload, so the sweep has
        // to reach down into it.
        let nested = super::strip_private_markers(json!({
            "__zf_response": {
                "status": 200,
                "body": {
                    "username": "wawan",
                    "__zf_private_trace_redact": ["toryoto"]
                }
            }
        }));
        assert!(
            !nested.to_string().contains("__zf_private"),
            "a nested marker survived: {nested}"
        );
        assert_eq!(nested["__zf_response"]["body"]["username"], "wawan");
    }

    /// A marker that has been nested by a downstream node is still read, still
    /// removed, and its contents still redacted -- the shape `n.web.response`
    /// produces, where the whole upstream payload becomes `__zf_response.body`.
    #[test]
    fn a_nested_redaction_marker_is_not_printed_into_the_run_history() {
        let trace = sanitized_trace_value(&json!({
            "__zf_response": {
                "status": 200,
                "body": {
                    "body": { "username": "wawan", "password": "toryoto" },
                    "audit": "login attempt with toryoto",
                    "__zf_private_trace_redact": ["toryoto"]
                }
            }
        }));

        let serialized = trace.to_string();
        assert!(
            !serialized.contains("__zf_private"),
            "an engine marker reached the trace: {serialized}"
        );
        assert!(
            !serialized.contains("toryoto"),
            "the secret reached the trace: {serialized}"
        );
        assert_eq!(trace["__zf_response"]["body"]["body"]["username"], "wawan");
    }

    /// The one escape hatch a pipeline already had keeps working: a field it
    /// deliberately traces is not blanked just because of what it is called.
    #[test]
    fn an_excepted_path_is_still_traced_verbatim() {
        let trace = sanitized_trace_value(&json!({
            "__zf_private_redact_except_paths": ["response.token"],
            "response": { "token": "deliberately-visible" },
            "request": { "token": "hidden" }
        }));

        assert_eq!(trace["response"]["token"], "deliberately-visible");
        assert_ne!(trace["request"]["token"], "hidden");
    }

    /// S4 regression, ingress half. The webhook collector published the tokens
    /// that let the engine strike a secret out of *every* node's trace, not
    /// just the trigger's -- and it read top-level keys only, so on the nested
    /// shape the ingress actually produces it published nothing.
    #[test]
    fn a_webhook_secret_is_struck_out_of_a_later_nodes_trace_too() {
        let ingress = json!({
            "body": { "password": "toryoto" },
            "method": "POST"
        });
        let tokens =
            crate::pipeline::nodes::basic::trigger::webhook::collect_trace_private_tokens_for_test(
                &ingress,
            );
        assert_eq!(tokens, vec!["toryoto".to_string()]);

        // A downstream node copied the secret into a differently-named field;
        // the token list still reaches it.
        let downstream = json!({
            "__zf_private_trace_redact": tokens,
            "audit_line": "login attempt for wawan with toryoto"
        });
        let trace = sanitized_trace_value(&downstream);
        assert!(
            !trace.to_string().contains("toryoto"),
            "a copy of the secret survived: {trace}"
        );
    }

    #[test]
    fn nodes_access_scanner_detects_literal_references() {
        let mut access = NodesAccessAccumulator::default();
        scan_text_nodes_access(
            &PipelineNode {
                id: "reader".to_string(),
                kind: script::NODE_KIND.to_string(),
                input_pins: vec!["in".to_string()],
                output_pins: vec!["out".to_string()],
                config: json!({}),
            },
            "return ctx.nodes.prepare.id + $nodes['geo-convert'].path + $nodes.match.value;",
            &mut access,
        )
        .expect("scan exact refs");
        let result = access.finish();
        let NodesAccess::Exact(ids) = result else {
            panic!("expected exact node references");
        };
        assert!(ids.contains("prepare"));
        assert!(ids.contains("geo-convert"));
        assert!(ids.contains("match"));
    }

    #[test]
    fn nodes_access_scanner_rejects_dynamic_references() {
        let mut dynamic = NodesAccessAccumulator::default();
        let err = scan_text_nodes_access(
            &PipelineNode {
                id: "reader".to_string(),
                kind: script::NODE_KIND.to_string(),
                input_pins: vec!["in".to_string()],
                output_pins: vec!["out".to_string()],
                config: json!({}),
            },
            "const key = input.key; return ctx.nodes[key];",
            &mut dynamic,
        )
        .expect_err("dynamic node access should be rejected");
        assert_eq!(err.code, "FW_NODES_SCOPE_DYNAMIC");
    }

    #[tokio::test]
    async fn nodes_scope_only_carries_referenced_upstream_outputs() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { big: 'huge-marker-that-should-not-leak' };"
[c] script -- "return { small: 7 };"
[d] script -- "return { small: ctx.nodes.c.small, leaked: JSON.stringify(ctx).includes('huge-marker-that-should-not-leak') };"

[a] -> [b]
[b] -> [c]
[c] -> [d]
"#;

        let graph = build_pipeline_graph("nodes-retention-exact-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "nodes-retention-exact-test".to_string(),
                    request_id: "req-nodes-retention-exact".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["small"], 7);
        assert_eq!(out.value["leaked"], false);
    }

    #[tokio::test]
    async fn nodes_scope_rejects_dynamic_script_access() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { big: 'dynamic-access-should-not-work' };"
[c] script -- "return { small: 7 };"
[d] script -- "const key = 'b'; return { big: ctx.nodes[key].big, small: ctx.nodes.c.small };"

[a] -> [b]
[b] -> [c]
[c] -> [d]
"#;

        let graph = build_pipeline_graph("nodes-retention-dynamic-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let err = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "nodes-retention-dynamic-test".to_string(),
                    request_id: "req-nodes-retention-dynamic".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect_err("dynamic nodes access should fail validation");

        assert_eq!(err.code, "FW_NODES_SCOPE_DYNAMIC");
    }

    /// `kinds/pipeline/README.md` Frozen Defaults: an omitted
    /// `spec.entry_nodes` means "Runtime derives roots from graph
    /// connectivity". Two independent roots are two roots — taking the first
    /// node in source order runs half the pipeline and silently drops the
    /// rest.
    #[tokio::test]
    async fn every_root_of_a_multi_root_graph_runs() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { left: true };"
[c] trigger.manual
[d] script -- "return { right: true };"

[a] -> [b]
[c] -> [d]
"#;

        // The DSL parser fills `entry_nodes` in; a graph registered as raw
        // contract JSON does not have to, and the frozen default is what is
        // under test.
        let mut graph = build_pipeline_graph("multi-root-test", dsl).expect("graph");
        graph.entry_nodes.clear();
        assert_eq!(graph.entry_node_ids(), vec!["a", "c"]);

        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "multi-root-test".to_string(),
                    request_id: "req-multi-root".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        let executed: Vec<&str> = out
            .node_trace
            .iter()
            .map(|entry| entry.node_id.as_str())
            .collect();
        for node in ["a", "b", "c", "d"] {
            assert!(
                executed.contains(&node),
                "node '{node}' never ran; executed {executed:?}"
            );
        }
    }

    /// A graph whose every node has an incoming edge has no root by
    /// connectivity. Cycles are valid pipelines, so it still starts — source
    /// order picks the entry in that one case, and only in that case.
    #[test]
    fn a_fully_cyclic_graph_still_has_a_starting_node() {
        let dsl = r#"
[a] script -- "return input;"
[b] script -- "return input;"

[a] -> [b]
[b] -> [a]
"#;
        let graph = build_pipeline_graph("cyclic-root-test", dsl).expect("graph");
        assert_eq!(graph.entry_node_ids(), vec!["a"]);
    }

    #[tokio::test]
    async fn logic_collect_groups_multiple_upstreams_before_continuing() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { user: { id: 'u_42' } };"
[c] script -- "return { orders: [{ id: 'o_1' }] };"
[d] logic.collect
[e] script -- "return input;"

[a] -> [b]
[a] -> [c]
[b] -> [d]
[c] -> [d]
[d] -> [e]
"#;

        let graph = build_pipeline_graph("logic-collect-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-collect-test".to_string(),
                    request_id: "req-1".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["b"]["user"]["id"], "u_42");
        assert_eq!(out.value["c"]["orders"][0]["id"], "o_1");
    }

    #[tokio::test]
    async fn fs_object_nodes_cover_s3_like_crud_flow() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "fs_object_e2e";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );

        let dsl = r#"
[a] trigger.manual
[b] fs.put --path qa/fs/hello.txt --text "hello fs"
[c] fs.get --path qa/fs/hello.txt
[d] fs.copy --from qa/fs/hello.txt --to qa/fs/copy.txt
[e] fs.move --from qa/fs/copy.txt --to qa/fs/moved.txt
[f] fs.list --path qa/fs
[g] fs.delete --path qa/fs/hello.txt
[h] fs.mkdir --path qa/fs/prefix

[a] -> [b]
[b] -> [c]
[c] -> [d]
[d] -> [e]
[e] -> [f]
[f] -> [g]
[g] -> [h]
"#;

        let graph = build_pipeline_graph("fs-object-e2e", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform.clone());
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "fs-object-e2e".to_string(),
                    request_id: "req-fs-object".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["fs"]["operation"], "mkdir");
        assert_eq!(out.value["fs"]["object"]["kind"], "prefix");

        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        let zebfs = layout.open_files();
        let moved = zebfs.get("qa/fs/moved.txt").expect("moved object");
        assert_eq!(String::from_utf8(moved.bytes).expect("utf8"), "hello fs");
        assert!(zebfs.head("qa/fs/prefix").is_ok());
        assert!(zebfs.head("qa/fs/hello.txt").is_err());
        assert!(zebfs.head("qa/fs/copy.txt").is_err());
    }

    #[tokio::test]
    async fn node_output_files_materialize_for_any_node_output() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "node_output_files_e2e";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );

        let dsl = r#"
[a] trigger.manual
[b] script -- "return { ok: true, state_sequence: '{{file:state-sequence.json}}', __zf_files: [{ name: 'state-sequence.json', content_type: 'application/json', encoding: 'json', data: { frames: [1, 2, 3] } }] };"
[c] script -- "return input;"

[a] -> [b]
[b] -> [c]
"#;

        let graph = build_pipeline_graph("node-output-files-e2e", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform.clone());
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "node-output-files-e2e".to_string(),
                    request_id: "req-node-output-files".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        let file_ref = &out.value["state_sequence"];
        assert_eq!(file_ref["__zf_type"], "file_ref");
        // The eleven contract fields, and nothing that was dropped
        // (`kinds/file-ref/README.md`).
        crate::pipeline::nodes::shared::file_ref::validate_file_ref(file_ref)
            .expect("valid FileRef");
        assert_eq!(file_ref["mime"], "application/json");
        assert_eq!(file_ref["kind"], "json");
        for dropped in ["path", "name", "content_type", "url"] {
            assert!(
                file_ref.get(dropped).is_none(),
                "FileRef must not carry '{dropped}'"
            );
        }
        assert!(out.value.get("__zf_files").is_none());
        assert_eq!(
            out.value["file_refs"]["state-sequence.json"]["ref"],
            file_ref["ref"]
        );

        let rel_path = file_ref["ref"].as_str().expect("file ref path");
        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        let zebfs = layout.open_files();
        let object = zebfs.get(rel_path).expect("materialized object");
        let stored: serde_json::Value = serde_json::from_slice(&object.bytes).expect("json file");
        assert_eq!(stored, json!({ "frames": [1, 2, 3] }));
    }

    #[tokio::test]
    async fn table_convert_parquet_roundtrips_through_pipeline_and_zebfs() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "table_convert_e2e";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );

        let dsl = r#"
[a] trigger.manual
[b] table.convert --from "{{ input.rows }}" --to datasets/posts.parquet
[c] table.convert --from datasets/posts.parquet --to-json --preview-rows 2

[a] -> [b]
[b] -> [c]
"#;

        let graph = build_pipeline_graph("table-convert-parquet-e2e", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform.clone());
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "table-convert-parquet-e2e".to_string(),
                    request_id: "req-table-convert".to_string(),
                    route: String::new(),
                    input: json!({
                        "rows": [
                            { "id": 1, "title": "First", "score": 1.5, "active": true },
                            { "id": 2, "title": "Second", "score": 2.0, "active": false }
                        ]
                    }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["table"]["from_format"], "parquet");
        assert_eq!(out.value["table"]["data"][0]["id"], 1);
        assert_eq!(out.value["table"]["data"][0]["title"], "First");
        assert_eq!(out.value["table"]["data"][0]["score"], 1.5);
        assert_eq!(out.value["table"]["data"][0]["active"], true);
        assert_eq!(out.value["table"]["rows"], 2);

        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        let object = layout
            .open_files()
            .get("datasets/posts.parquet")
            .expect("parquet object");
        assert!(!object.bytes.is_empty());
    }

    #[tokio::test]
    async fn table_query_joins_multiple_zebfs_sources_with_params() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "table_query_e2e";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );
        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        let zebfs = layout.open_files();
        zebfs
            .put(
                "datasets/posts.csv",
                b"id,author_id,title\n1,10,First\n2,20,Second\n",
            )
            .expect("posts csv");
        zebfs
            .put("datasets/authors.csv", b"id,name\n10,Ada\n20,Bob\n")
            .expect("authors csv");

        let dsl = r#"
[a] trigger.manual
[b] table.query --from "datasets/posts.csv as posts" --from "datasets/authors.csv as authors" --params "{{ [input.post_id] }}" --to-json --preview-rows 1 --query "select p.id, p.title, a.name from posts p join authors a on p.author_id = a.id where p.id = $1"

[a] -> [b]
"#;

        let graph = build_pipeline_graph("table-query-e2e", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform.clone());
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "table-query-e2e".to_string(),
                    request_id: "req-table-query".to_string(),
                    route: String::new(),
                    input: json!({ "post_id": 1 }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["table"]["engine"], "geodatafusion");
        assert_eq!(out.value["table"]["rows"], 1);
        assert_eq!(out.value["table"]["data"][0]["id"], 1);
        assert_eq!(out.value["table"]["data"][0]["title"], "First");
        assert_eq!(out.value["table"]["data"][0]["name"], "Ada");
        assert_eq!(out.value["table"]["preview"][0]["title"], "First");
    }

    #[tokio::test]
    async fn table_query_joins_multiple_parquet_sources() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "table_query_parquet_join";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );
        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        let zebfs = layout.open_files();

        let posts = vec![
            json!({ "id": 1, "author_id": 10, "title": "First", "score": 7.5 }),
            json!({ "id": 2, "author_id": 20, "title": "Second", "score": 3.0 }),
            json!({ "id": 3, "author_id": 30, "title": "Third", "score": 9.0 }),
        ];
        let posts_bytes = encode_rows(&posts, &collect_columns(&posts), TableFormat::Parquet)
            .expect("posts parquet");
        zebfs
            .put("datasets/posts.parquet", &posts_bytes)
            .expect("posts parquet put");

        let authors = vec![
            json!({ "id": 10, "name": "Ada", "active": true }),
            json!({ "id": 20, "name": "Bob", "active": false }),
            json!({ "id": 30, "name": "Cora", "active": true }),
        ];
        let authors_bytes = encode_rows(&authors, &collect_columns(&authors), TableFormat::Parquet)
            .expect("authors parquet");
        zebfs
            .put("datasets/authors.parquet", &authors_bytes)
            .expect("authors parquet put");

        let dsl = r#"
[a] trigger.manual
[b] table.query --from "datasets/posts.parquet as posts" --from "datasets/authors.parquet as authors" --to-json --preview-rows 2 --query "select p.id, p.title, a.name from posts p join authors a on p.author_id = a.id where a.active = true order by p.id"

[a] -> [b]
"#;

        let graph = build_pipeline_graph("table-query-parquet-join", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform);
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "table-query-parquet-join".to_string(),
                    request_id: "req-table-query-parquet-join".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["table"]["engine"], "geodatafusion");
        assert_eq!(out.value["table"]["rows"], 2);
        assert_eq!(out.value["table"]["data"][0]["id"], 1);
        assert_eq!(out.value["table"]["data"][0]["title"], "First");
        assert_eq!(out.value["table"]["data"][0]["name"], "Ada");
        assert_eq!(out.value["table"]["data"][1]["id"], 3);
        assert_eq!(out.value["table"]["data"][1]["title"], "Third");
        assert_eq!(out.value["table"]["data"][1]["name"], "Cora");
    }

    #[tokio::test]
    async fn table_query_accepts_ui_source_binding_rows() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "table_query_ui_rows";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );
        let layout = platform
            .file
            .ensure_project_layout(owner, project)
            .expect("project layout");
        layout
            .open_files()
            .put("datasets/posts.csv", b"id,title\n1,First\n2,Second\n")
            .expect("posts csv");

        let graph = PipelineGraph {
            id: "table-query-ui-rows".to_string(),
            description: None,
            metadata: None,
            notes: Vec::new(),
            entry_nodes: Vec::new(),
            nodes: vec![
                PipelineNode {
                    id: "a".to_string(),
                    kind: "n.trigger.manual".to_string(),
                    input_pins: vec![],
                    output_pins: vec!["out".to_string()],
                    config: json!({}),
                },
                PipelineNode {
                    id: "b".to_string(),
                    kind: "n.table.query".to_string(),
                    input_pins: vec!["in".to_string()],
                    output_pins: vec!["out".to_string()],
                    config: json!({
                        "sources": [
                            { "source": "datasets/posts.csv", "alias": "posts" }
                        ],
                        "query": "select * from posts where id = $1",
                        "params": "{{ [input.post_id] }}",
                        "to_json": true,
                        "preview_rows": 1
                    }),
                },
            ],
            edges: vec![PipelineEdge {
                from_node: "a".to_string(),
                from_pin: "out".to_string(),
                to_node: "b".to_string(),
                to_pin: "in".to_string(),
            }],
        };

        let engine = BasicPipelineEngine::default().with_platform(platform);
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "table-query-ui-rows".to_string(),
                    request_id: "req-table-query-ui".to_string(),
                    route: String::new(),
                    input: json!({ "post_id": 2 }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["table"]["rows"], 1);
        assert_eq!(out.value["table"]["data"][0]["title"], "Second");
        assert_eq!(out.value["table"]["sources"][0]["alias"], "posts");
        assert_eq!(
            out.value["table"]["preview"].as_array().map(Vec::len),
            Some(1),
            "preview_rows samples rows: {}",
            out.value["table"]
        );
    }

    #[tokio::test]
    async fn table_query_geodatafusion_runs_geospatial_sql() {
        let data_root = tempfile::tempdir().expect("temp data root");
        let owner = "superadmin";
        let project = "table_query_geodatafusion";
        let platform = Arc::new(
            PlatformService::from_config(PlatformConfig {
                data_root: data_root.path().to_path_buf(),
                default_password: "secret".to_string(),
                default_project: project.to_string(),
                ..Default::default()
            })
            .expect("platform"),
        );

        let dsl = r#"
[a] trigger.manual
[b] table.query --engine geodatafusion --from "$input.rows as points" --to-json --preview-rows 1 --query "select id, ST_AsText(ST_Point(x, y)) as geom from points where id = 1"

[a] -> [b]
"#;

        let graph = build_pipeline_graph("table-query-geodatafusion", dsl).expect("graph");
        let engine = BasicPipelineEngine::default().with_platform(platform);
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: owner.to_string(),
                    project: project.to_string(),
                    pipeline: "table-query-geodatafusion".to_string(),
                    request_id: "req-table-query-geodatafusion".to_string(),
                    route: String::new(),
                    input: json!({
                        "rows": [
                            { "id": 1, "x": 30.0, "y": 10.0 },
                            { "id": 2, "x": 40.0, "y": 20.0 }
                        ],
                    }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["table"]["engine"], "geodatafusion");
        assert_eq!(out.value["table"]["rows"], 1);
        assert_eq!(out.value["table"]["data"][0]["id"], 1);
        assert_eq!(out.value["table"]["data"][0]["geom"], "POINT(30 10)");
    }

    #[tokio::test]
    async fn logic_if_supports_dsl_input_scope() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { type: 'billing' };"
[c] logic.if --expr "$input.type == 'billing'"
[d] script -- "return { branch: 'true' };"
[e] script -- "return { branch: 'false' };"

[a] -> [b]
[b] -> [c]
[c]:true -> [d]
[c]:false -> [e]
"#;

        let graph = build_pipeline_graph("logic-if-scope-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-if-scope-test".to_string(),
                    request_id: "req-if".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["branch"], "true");
    }

    #[tokio::test]
    async fn logic_match_supports_dsl_nodes_scope() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { kind: 'billing' };"
[c] logic.match --expr "$nodes.b.kind" --cases billing,technical --default default
[d] script -- "return { lane: 'billing' };"
[e] script -- "return { lane: 'technical' };"
[f] script -- "return { lane: 'default' };"

[a] -> [b]
[b] -> [c]
[c]:billing -> [d]
[c]:technical -> [e]
[c]:default -> [f]
"#;

        let graph = build_pipeline_graph("logic-match-scope-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-match-scope-test".to_string(),
                    request_id: "req-match".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["lane"], "billing");
    }

    /// An input node leaves the envelope alone for the next node and answers
    /// `$nodes.<id>` with the value it checked; a required field that was not
    /// sent refuses the run naming the field.
    #[tokio::test]
    async fn input_nodes_pass_the_envelope_through_and_expose_the_checked_value() {
        let dsl = r#"
[a] trigger.manual
[b] input.text prompt --label "Caption"
[c] input.number count --default 3
[d] script -- "return { echoed: input.body.prompt, prompt: $nodes.b, count: $nodes.c, keys: Object.keys(input).sort() };"

[a] -> [b]
[b] -> [c]
[c] -> [d]
"#;
        let graph = build_pipeline_graph("input-nodes-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let ctx = |input: serde_json::Value| PipelineContext {
            owner: "test".to_string(),
            project: "test".to_string(),
            pipeline: "input-nodes-test".to_string(),
            request_id: "req-input".to_string(),
            route: String::new(),
            input,
            trigger: None,
            placeholder: None,
        };
        let out = engine
            .execute_async(&graph, &ctx(json!({ "body": { "prompt": "hello", "extra": true } })))
            .await
            .expect("execute");
        assert_eq!(out.value["echoed"], "hello");
        assert_eq!(out.value["prompt"], "hello");
        assert_eq!(out.value["count"], 3.0);
        assert_eq!(out.value["keys"], json!(["body"]));

        let err = engine
            .execute_async(&graph, &ctx(json!({ "body": {} })))
            .await
            .expect_err("a required input that was not sent refuses the run");
        assert_eq!(err.code, "FW_NODE_INPUT_MISSING");
        assert!(err.message.contains("'prompt'"), "{}", err.message);
        assert_eq!(err.node_id.as_deref(), Some("b"));
    }

    #[tokio::test]
    async fn logic_foreach_emits_one_run_per_item() {
        let dsl = r#"
[a] trigger.manual
[b] logic.foreach --items-expr "$input.rows"

[a] -> [b]
"#;

        let mut graph = build_pipeline_graph("logic-foreach-test", dsl).expect("graph");
        // Asserts on the foreach node's recorded output, so it needs the level
        // that records a successful node. Default is on-error.
        graph.metadata = Some(crate::pipeline::model::PipelineGraphMetadata {
            settings: crate::pipeline::model::PipelineGraphSettings {
                trace_capture: Some(crate::pipeline::trace_capture::TraceCaptureSettings {
                    level: Some(crate::pipeline::trace_capture::CaptureLevel::Full),
                    ..Default::default()
                }),
                ..Default::default()
            },
            ..Default::default()
        });
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-foreach-test".to_string(),
                    request_id: "req-2".to_string(),
                    route: String::new(),
                    input: json!({
                        "rows": [
                            { "id": "r1" },
                            { "id": "r2" }
                        ]
                    }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        let foreach_trace = out
            .node_trace
            .iter()
            .find(|entry| entry.node_kind == "n.logic.foreach")
            .expect("foreach trace");
        assert_eq!(foreach_trace.output["count"], 2);
        assert_eq!(foreach_trace.output["emissions"][0]["item"]["id"], "r1");
        assert_eq!(foreach_trace.output["emissions"][1]["item"]["id"], "r2");
        assert_eq!(out.value["item"]["id"], "r2");
        assert_eq!(out.value["index"], 1);
        assert_eq!(out.value["count"], 2);
        assert!(out.value.get("rows").is_none());
    }

    #[tokio::test]
    async fn logic_foreach_keep_input_preserves_parent_payload_when_requested() {
        let dsl = r#"
[a] trigger.manual
[b] logic.foreach --items-expr "$input.rows" --keep-input

[a] -> [b]
"#;

        let graph = build_pipeline_graph("logic-foreach-keep-input-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-foreach-keep-input-test".to_string(),
                    request_id: "req-foreach-keep-input".to_string(),
                    route: String::new(),
                    input: json!({
                        "rows": [
                            { "id": "r1" },
                            { "id": "r2" }
                        ],
                        "batch_marker": "kept-only-when-requested"
                    }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["item"]["id"], "r2");
        assert_eq!(out.value["batch_marker"], "kept-only-when-requested");
        assert_eq!(out.value["rows"][0]["id"], "r1");
    }

    #[tokio::test]
    async fn logic_reduce_accumulates_foreach_series() {
        let dsl = r#"
[a] trigger.manual
[b] logic.foreach --items-expr "$input.rows"
[c] logic.reduce --init-expr "{ total: 0 }" --step-expr "{ total: $acc.total + $input.item.amount }"

[a] -> [b]
[b]:item -> [c]
"#;

        let graph = build_pipeline_graph("logic-reduce-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-reduce-test".to_string(),
                    request_id: "req-3".to_string(),
                    route: String::new(),
                    input: json!({
                        "rows": [
                            { "amount": 10 },
                            { "amount": 15 },
                            { "amount": 7 }
                        ]
                    }),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["total"], 32);
    }


    /// The one resolution mechanism must reach the response node like any
    /// other: a whole {{ }} in --message or --location is the typed value.
    #[tokio::test]
    async fn a_web_response_config_is_resolved_like_any_other_node() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "return { u: 'hello-from-expr' };"
[c] web.response --message "{{ input.u }}"

[a] -> [b]
[b] -> [c]
"#;
        let graph = build_pipeline_graph("web-response-expr-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "web-response-expr-test".to_string(),
                    request_id: "req-wr".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");
        assert_eq!(
            out.value["__zf_response"]["message"], "hello-from-expr",
            "resolved message did not reach the envelope: {}",
            out.value
        );
    }

    #[tokio::test]
    async fn logic_retry_retries_until_success() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "const attempt = input.__zf_retry?.attempt ?? 0; if (attempt < 2) { throw new Error('retry me'); } return { ok: true, attempt };"
[r] logic.retry --max-attempts 3
[c] script -- "return input;"
[d] script -- "return input;"

[a] -> [b]
[b]:error -> [r]
[r]:retry -> [b]
[b] -> [c]
[r]:failed -> [d]
"#;

        let graph = build_pipeline_graph("logic-retry-success-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-retry-success-test".to_string(),
                    request_id: "req-4".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert_eq!(out.value["ok"], true);
        assert_eq!(out.value["attempt"], 2);
    }

    /// `input.* --default` reaches the envelope on every path: an execute that
    /// omits the field finds the default at `input.body.<name>` downstream,
    /// the same as `$nodes.<id>`; a sent value wins.
    #[tokio::test]
    async fn an_input_default_is_written_into_the_envelope_when_the_field_is_omitted() {
        let dsl = r#"
[t] trigger.manual
[who] input.text who --optional --default world
[s] script -- "return { hi: input.body.who, own: ctx.nodes.who };"
[t] -> [who]
[who] -> [s]
"#;
        let graph = build_pipeline_graph("input-default-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let run = |input: serde_json::Value| {
            let graph = &graph;
            let engine = &engine;
            async move {
                engine
                    .execute_async(
                        graph,
                        &PipelineContext {
                            owner: "test".to_string(),
                            project: "test".to_string(),
                            pipeline: "input-default-test".to_string(),
                            request_id: "req-default".to_string(),
                            route: String::new(),
                            input,
                            trigger: None,
                            placeholder: None,
                        },
                    )
                    .await
                    .expect("execute")
            }
        };
        let omitted = run(json!({})).await;
        assert_eq!(omitted.value["hi"], "world");
        assert_eq!(omitted.value["own"], "world");
        let empty_body = run(json!({ "body": {} })).await;
        assert_eq!(empty_body.value["hi"], "world");
        let sent = run(json!({ "body": { "who": "x" } })).await;
        assert_eq!(sent.value["hi"], "x");
        assert_eq!(sent.value["own"], "x");
    }

    #[tokio::test]
    async fn logic_retry_routes_to_failed_after_budget() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "throw new Error('always fail');"
[r] logic.retry --max-attempts 2
[c] script -- "return input;"

[a] -> [b]
[b]:error -> [r]
[r]:retry -> [b]
[r]:failed -> [c]
"#;

        let graph = build_pipeline_graph("logic-retry-failed-test", dsl).expect("graph");
        let engine = BasicPipelineEngine::default();
        let out = engine
            .execute_async(
                &graph,
                &PipelineContext {
                    owner: "test".to_string(),
                    project: "test".to_string(),
                    pipeline: "logic-retry-failed-test".to_string(),
                    request_id: "req-5".to_string(),
                    route: String::new(),
                    input: json!({}),
                    trigger: None,
                    placeholder: None,
                },
            )
            .await
            .expect("execute");

        assert!(
            out.value["error"]["message"]
                .as_str()
                .expect("error message")
                .contains("always fail")
        );
        assert_eq!(out.value["__zf_retry"]["attempt"], 2);
    }

    async fn run_with_bus(
        id: &str,
        dsl: &str,
    ) -> (
        Result<crate::pipeline::model::PipelineOutput, crate::pipeline::model::PipelineError>,
        Vec<crate::pipeline::model::Signal>,
    ) {
        use crate::pipeline::model::{ExecuteOptions, ExecutionBus};
        let graph = build_pipeline_graph(id, dsl).expect("graph");
        let ctx = PipelineContext {
            owner: "test".into(),
            project: "test".into(),
            pipeline: id.into(),
            request_id: format!("{id}-run"),
            route: String::new(),
            input: json!({}),
            trigger: None,
            placeholder: None,
        };
        let bus = Arc::new(ExecutionBus::new(128));
        let mut rx = bus.subscribe();
        let options = ExecuteOptions { bus: Some(bus) };
        let result = BasicPipelineEngine::default()
            .execute_with_options_async(&graph, &ctx, &options)
            .await;
        let mut signals = Vec::new();
        while let Ok(signal) = rx.try_recv() {
            signals.push(signal);
        }
        (result, signals)
    }

    /// A failure an `:error` edge hands to `logic.retry` is a retry, not a
    /// failure: the record says `retry`, the bus says `node_retry` with the
    /// attempt counted, `node_fail` is never announced, and the run is `ok`.
    /// The error text stays in the entry so the log still says why.
    #[tokio::test]
    async fn a_routed_failure_is_a_retry_in_the_record_and_on_the_bus() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "const attempt = input.__zf_retry?.attempt ?? 0; if (attempt < 2) { throw new Error('not yet'); } return { ok: true, attempt };"
[r] logic.retry --max-attempts 5
[c] script -- "return input;"
[d] script -- "return { gaveup: true };"
[a] -> [b]
[b]:error -> [r]
[r]:retry -> [b]
[b] -> [c]
[r]:failed -> [d]
"#;
        let (result, signals) = run_with_bus("routed-retry", dsl).await;
        let out = result.expect("the run succeeds");
        assert_eq!(out.value["ok"], true);

        let b_entries: Vec<&crate::pipeline::model::NodeTraceEntry> =
            out.node_trace.iter().filter(|t| t.node_id == "b").collect();
        let statuses: Vec<&str> = b_entries.iter().map(|t| t.status.as_str()).collect();
        assert_eq!(statuses, ["retry", "retry", "ok"]);
        assert!(b_entries[0].error.as_deref().is_some_and(|e| e.contains("not yet")));
        assert!(b_entries[0].error_code.is_some());
        assert!(b_entries[2].error.is_none());

        let retries: Vec<&crate::pipeline::model::Signal> =
            signals.iter().filter(|s| s.kind == "node_retry").collect();
        assert_eq!(retries.len(), 2, "{:?}", signals.iter().map(|s| &s.kind).collect::<Vec<_>>());
        for (i, signal) in retries.iter().enumerate() {
            assert_eq!(signal.node_id, "b");
            let data = signal.data.as_ref().unwrap();
            assert_eq!(data["attempt"], (i + 1) as u64, "{data}");
            assert_eq!(data["max_attempts"], 5);
            assert!(data["duration_ms"].is_u64());
            assert!(data["error_code"].as_str().is_some_and(|c| !c.is_empty()));
            assert!(data["message"].as_str().unwrap().contains("not yet"));
        }
        assert!(retries[0].message.contains("retry 1/5"), "{}", retries[0].message);
        assert!(signals.iter().all(|s| s.kind != "node_fail"), "a routed failure is never node_fail");
        assert_eq!(signals.last().unwrap().kind, "run_done");
        assert_eq!(signals.last().unwrap().data.as_ref().unwrap()["status"], "ok");

        // The record's key for an error group skips routed entries: this run
        // has no failing entry at all, so it could never name a group.
        let entry = crate::platform::model::PipelineInvocationEntry {
            run_id: "routed-retry-run".into(),
            at: 0,
            duration_ms: 1,
            status: "ok".into(),
            trigger: "manual".into(),
            error: None,
            trace: out.node_trace.clone(),
        };
        assert!(crate::platform::model::failing_trace_entry(&entry.trace).is_none());
    }

    /// A failure routed to something other than `logic.retry` is
    /// `error_routed`, naming where it went; an unrouted one is still
    /// `node_fail` and fails the run.
    #[tokio::test]
    async fn a_failure_routed_elsewhere_is_error_routed_and_an_unrouted_one_still_fails() {
        let dsl = r#"
[a] trigger.manual
[b] script -- "throw new Error('handled here');"
[h] script -- "return { handled: input.error.message };"
[a] -> [b]
[b]:error -> [h]
"#;
        let (result, signals) = run_with_bus("routed-elsewhere", dsl).await;
        let out = result.expect("the run succeeds");
        assert!(out.value["handled"].as_str().unwrap().contains("handled here"), "{}", out.value);
        let b = out.node_trace.iter().find(|t| t.node_id == "b").unwrap();
        assert_eq!(b.status, "error_routed");
        let routed = signals.iter().find(|s| s.kind == "node_error_routed").expect("node_error_routed");
        assert_eq!(routed.node_id, "b");
        assert_eq!(routed.data.as_ref().unwrap()["to_node"], "h");
        assert!(signals.iter().all(|s| s.kind != "node_fail"));

        let (result, signals) = run_with_bus(
            "unrouted",
            "[a] trigger.manual\n[b] script -- \"throw new Error('boom');\"\n[a] -> [b]\n",
        )
        .await;
        let err = result.expect_err("the run fails");
        assert_eq!(err.node_trace.last().unwrap().status, "failed");
        assert!(signals.iter().any(|s| s.kind == "node_fail"));
        assert!(
            crate::platform::model::failing_trace_entry(&err.node_trace)
                .is_some_and(|t| t.node_id == "b")
        );
    }

    /// The verdict road: a check that never throws answers `retry: true`
    /// while waiting; `logic.retry` counts its own attempts from `$nodes`
    /// even though `poll` replaces the payload every round, then `done`
    /// carries the payload on. No failure anywhere, so every badge is green.
    #[tokio::test]
    async fn logic_retry_takes_a_verdict_and_counts_for_itself() {
        let dsl = r#"
[a] trigger.manual
[poll] script -- "return { polled: true };"
[check] script -- "const seen = (ctx.nodes.wait && ctx.nodes.wait.__zf_retry && ctx.nodes.wait.__zf_retry.attempt) || 0; return { retry: seen < 2, seen };"
[wait] logic.retry --max-attempts 5
[done] script -- "return { done: true, attempts: input.__zf_retry.attempt, seen: input.seen };"
[gaveup] script -- "return { gaveup: true };"
[a] -> [poll]
[poll] -> [check]
[check] -> [wait]
[wait]:retry -> [poll]
[wait]:done -> [done]
[wait]:failed -> [gaveup]
"#;
        let (result, signals) = run_with_bus("verdict-done", dsl).await;
        let out = result.expect("the run succeeds");
        assert_eq!(out.value["done"], true);
        assert_eq!(out.value["attempts"], 2, "{}", out.value);
        assert_eq!(out.value["seen"], 2);
        assert_eq!(out.node_trace.iter().filter(|t| t.node_id == "poll").count(), 3);
        // The wait is told on the retry node — `retry` twice, then `ok` when
        // `done` fires — and nowhere else: every other entry is `ok`.
        let wait_statuses: Vec<&str> = out
            .node_trace
            .iter()
            .filter(|t| t.node_id == "wait")
            .map(|t| t.status.as_str())
            .collect();
        assert_eq!(wait_statuses, ["retry", "retry", "ok"]);
        assert!(
            out.node_trace.iter().filter(|t| t.node_id != "wait").all(|t| t.status == "ok"),
            "{:?}",
            out.node_trace.iter().map(|t| (&t.node_id, &t.status)).collect::<Vec<_>>()
        );
        assert!(signals.iter().all(|s| s.kind != "node_fail"));
        let waits: Vec<&crate::pipeline::model::Signal> =
            signals.iter().filter(|s| s.kind == "node_retry").collect();
        assert_eq!(waits.len(), 2);
        assert!(waits.iter().all(|s| s.node_id == "wait"));
        assert_eq!(waits[0].data.as_ref().unwrap()["attempt"], 1);
        assert_eq!(waits[1].data.as_ref().unwrap()["attempt"], 2);
        assert_eq!(waits[1].data.as_ref().unwrap()["max_attempts"], 5);

        // The budget spent on a verdict that never turns false is `failed`.
        let dsl = r#"
[a] trigger.manual
[poll] script -- "return { polled: true };"
[check] script -- "return { retry: true };"
[wait] logic.retry --max-attempts 2
[done] script -- "return { done: true };"
[gaveup] script -- "return { gaveup: true, attempts: input.__zf_retry.attempt };"
[a] -> [poll]
[poll] -> [check]
[check] -> [wait]
[wait]:retry -> [poll]
[wait]:done -> [done]
[wait]:failed -> [gaveup]
"#;
        let (result, _) = run_with_bus("verdict-failed", dsl).await;
        let out = result.expect("the run succeeds through the failed pin");
        assert_eq!(out.value["gaveup"], true);
        assert_eq!(out.value["attempts"], 2);
    }
}

enum NodeDispatch {
    Webhook(webhook::Node),
    Schedule(schedule::Node),
    Manual(manual::Node),
    Script(script::Node),
    HttpRequest(http::request::Node),
    BrowserRun(browser::run::Node),
    SqliteQuery(sqlite::query::Node),
    SekejapQuery(sekejap::query::Node),
    SekejapInsert(sekejap::insert::Node),
    SqliteMutate(sqlite::mutate::Node),
    Postgres(pg::query::Node),
    InlineWebResponse {
        node_id: String,
        config: web::response::Config,
    },
    InlineWebStaticGenerate {
        node_id: String,
        config: web::static_generate::Config,
    },
    InlineWebDocsGenerate {
        node_id: String,
        config: web::docs_generate::Config,
    },
    WebResponse(web::response::Node),
    Agent(ai::agent::Node),
    AiTts(ai::tts::Node),
    LogicIf(logic::if_::Node),
    LogicMatch(logic::match_::Node),
    LogicCollect(logic::collect::Node),
    LogicForeach(logic::foreach_::Node),
    LogicReduce(logic::reduce::Node),
    LogicRetry(logic::retry::Node),
    AuthTokenCreate(auth::token_create::Node),
    AuthTokenVerify(auth::token_verify::Node),
    MailSend(mail::send::Node),
    Concept(concept::Node),
    WebError(weberror::Node),
    WsTrigger(ws::trigger::Node),
    WsSyncState(ws::sync_state::Node),
    WsEmit(ws::emit::Node),
    Crypto(crypto::Node),
    TriggerFunction(trigger_function::Node),
    FunctionCall(function::call::Node),
    FileSave(fs::save::Node),
    FsObject(fs::object::Node),
    MapserverCrud(ms::crud::Node),
    TableConvert(table::convert::Node),
    TableQuery(table::query::Node),
    FileCompress(fs::compress::Node),
    FileDecompress(fs::decompress::Node),
    GeoInspect(geo::inspect::Node),
    GeoConvert(geo::convert::Node),
    FilePdfConvert(fs::pdf::convert::Node),
    ImgThumbnail(fs::image::thumbnail::Node),
    SvgConvert(fs::svg::convert::Node),
    ImgChromakey(fs::image::chromakey::Node),
    /// Any `n.input.*` kind — a pass-through validator of one envelope field.
    Input(input::Node),
    KvSet(kv::set::Node),
    KvGet(kv::get::Node),
    KvDel(kv::del::Node),
    KvExists(kv::exists::Node),
    KvExpire(kv::expire::Node),
    KvIncr(kv::incr::Node),
    KvPublish(kv::publish::Node),
    KvSubscribe(kv_subscribe::Node),
    WsClientTrigger(trigger_ws_client::Node),
    WsClientSend(ws::client_send::Node),
    McpTrigger(mcp_trigger::Node),
    /// A node provided by an installed bundle (`n.x.*`).
    ///
    /// Role and implementation come from the package manifest at execution
    /// time, not from the node kind, so a node may move between composite and
    /// WASM without changing its kind.
    InstalledNode {
        kind: String,
        config: serde_json::Value,
        platform: std::sync::Arc<crate::platform::services::PlatformService>,
        /// Declared-host policy already in force, when this dispatch is itself
        /// inside another bundle's function pipeline.
        egress: Option<Arc<crate::pipeline::security::BundleEgress>>,
    },
}

/// Node kinds whose outbound destination reaches an egress guard as a URL, so a
/// bundle's declared hosts can be checked against what they actually contact.
const HOST_CHECKED_NETWORK_NODES: &[&str] = &[http::request::NODE_KIND, browser::run::NODE_KIND];

/// Refuses a network-capable node whose destination cannot be host-checked.
///
/// Capability comes from the same compiled-in table the package review reads, so
/// a network node added later is refused until it is given a guard rather than
/// silently becoming a way out of a bundle's declaration. Kinds absent from that
/// table are bundle-provided, and are governed by the dispatch they recurse into.
///
/// This runs for every bundle-provided node, whether or not that bundle declared
/// a host. Asking only inside a declaring bundle would have made silence the
/// cheapest way to reach an unreadable destination, so an author who declared
/// truthfully would be the only one constrained.
///
/// `n.script` is not a network node in the capability table, and with the
/// sandbox denying `fetch` it is not one in fact either. When an operator has
/// granted the sandbox network access, it becomes an egress path this guard
/// cannot read, and is refused on the same grounds as the rest.
fn refuse_uncheckable_egress_node(
    egress: &crate::pipeline::security::BundleEgress,
    kind: &str,
    sandbox_reaches_network: bool,
) -> Result<(), PipelineError> {
    if kind == script::NODE_KIND {
        if sandbox_reaches_network {
            return Err(egress.refuse_uncheckable(kind));
        }
        return Ok(());
    }
    if HOST_CHECKED_NETWORK_NODES.contains(&kind) {
        return Ok(());
    }
    let reaches_network = crate::pipeline::nodes::native_node_capabilities()
        .get(kind)
        .is_some_and(|capabilities| {
            capabilities.contains(&crate::pipeline::model::NodeCapability::Network)
        });
    if reaches_network {
        return Err(egress.refuse_uncheckable(kind));
    }
    Ok(())
}

fn is_logic_collect(node: &PipelineNode) -> bool {
    node.kind == logic::collect::NODE_KIND
}