ergo_runtime/runtime/

execute.rs

//! runtime/execute.rs — Kernel graph execution engine
//!
//! Purpose:
//! - Executes a validated expanded graph against registered primitive
//!   implementations (sources, computes, triggers, actions) within a
//!   supplied `ExecutionContext`.
//!
//! Owns:
//! - Topological execution ordering of nodes
//! - Value propagation along wired edges
//! - Type conversion between runtime value domains (trigger ↔ action)
//! - Context key injection for source nodes
//! - Parameter binding for compute and trigger nodes
//! - Action skip/invoke gating (R.7)
//! - Non-finite numeric output rejection (NUM-FINITE-1)
//!
//! Does not own:
//! - Graph validation (see `validate.rs`)
//! - Primitive registration or catalog ownership (see `catalog.rs`)
//! - Episode lifecycle or replay (see `ergo-supervisor`)
//!
//! Connects to:
//! - `validate.rs` — receives `ValidatedGraph` after validation
//! - `types.rs` — uses `ExecError`, `ExecutionReport`, `Registries`
//! - `catalog.rs` — queries primitive metadata for parameter binding
//!
//! Safety notes:
//! - All execution paths must remain deterministic for replay integrity
//! - R.7: NotEmitted triggers must be caught by should_skip_action
//!   before reaching action value conversion; violation returns
//!   `ExecError::ActionSkipViolation` instead of panicking

use std::collections::HashMap;

use crate::action::{ActionOutcome, ActionValue};
use crate::cluster::{PrimitiveKind, ValueType};
use crate::common::{derive_intent_id, ActionEffect, EffectWrite, IntentField, IntentRecord};
use crate::trigger::{TriggerEvent, TriggerValue};

use super::types::{
    Endpoint, ExecError, ExecutionContext, ExecutionReport, Registries, RuntimeEvent, RuntimeValue,
    ValidatedEdge, ValidatedGraph, ValidatedNode,
};

pub fn execute(
    graph: &ValidatedGraph,
    registries: &Registries,
    ctx: &ExecutionContext,
) -> Result<ExecutionReport, ExecError> {
    if let Some(node) = first_intent_emitting_action(graph, registries) {
        return Err(ExecError::IntentMetadataRequired { node });
    }
    execute_with_metadata(graph, registries, ctx, "graph", "event")
}

pub fn execute_with_metadata(
    graph: &ValidatedGraph,
    registries: &Registries,
    ctx: &ExecutionContext,
    graph_id: &str,
    event_id: &str,
) -> Result<ExecutionReport, ExecError> {
    let mut node_outputs: HashMap<String, HashMap<String, RuntimeValue>> = HashMap::new();
    let mut effects: Vec<ActionEffect> = Vec::new();

    for node_id in &graph.topo_order {
        let node = graph
            .nodes
            .get(node_id)
            .ok_or_else(|| ExecError::MissingNode {
                node: node_id.clone(),
            })?;

        let inputs = collect_inputs(node_id, &node.inputs, &graph.edges, &node_outputs)?;

        let outputs = match node.kind {
            PrimitiveKind::Source => execute_source(node, inputs, registries, ctx)?,
            PrimitiveKind::Compute => execute_compute(node, inputs, registries)?,
            PrimitiveKind::Trigger => execute_trigger(node, inputs, registries)?,
            PrimitiveKind::Action => {
                // R.7: Actions execute only when all trigger event inputs are Emitted.
                // If any Event input is TriggerEvent::NotEmitted, skip execution.
                if should_skip_action(&inputs) {
                    produce_skipped_outputs(node)
                } else {
                    let (action_outputs, action_effects) =
                        execute_action(node, inputs, registries, graph_id, event_id)?;
                    effects.extend(action_effects);
                    action_outputs
                }
            }
        };

        node_outputs.insert(node_id.clone(), outputs);
    }

    let mut outputs: HashMap<String, RuntimeValue> = HashMap::new();
    for out in &graph.boundary_outputs {
        if let Some(node_outs) = node_outputs.get(&out.maps_to.node_id) {
            if let Some(val) = node_outs.get(&out.maps_to.port_name) {
                outputs.insert(out.name.clone(), val.clone());
            } else {
                return Err(ExecError::MissingOutput {
                    node: out.maps_to.node_id.clone(),
                    output: out.maps_to.port_name.clone(),
                });
            }
        } else {
            return Err(ExecError::MissingOutput {
                node: out.maps_to.node_id.clone(),
                output: out.maps_to.port_name.clone(),
            });
        }
    }

    Ok(ExecutionReport { outputs, effects })
}

fn collect_inputs(
    target: &str,
    input_specs: &[crate::cluster::InputMetadata],
    edges: &[ValidatedEdge],
    node_outputs: &HashMap<String, HashMap<String, RuntimeValue>>,
) -> Result<HashMap<String, RuntimeValue>, ExecError> {
    let mut inputs: HashMap<String, RuntimeValue> = HashMap::new();

    for edge in edges {
        let Endpoint::NodePort {
            node_id: to_node,
            port_name: to_port,
        } = &edge.to;
        if to_node == target {
            let Endpoint::NodePort {
                node_id: from,
                port_name: from_port,
            } = &edge.from;
            let outs = node_outputs
                .get(from)
                .ok_or_else(|| ExecError::MissingOutput {
                    node: from.clone(),
                    output: from_port.clone(),
                })?;
            let val = outs
                .get(from_port)
                .ok_or_else(|| ExecError::MissingOutput {
                    node: from.clone(),
                    output: from_port.clone(),
                })?;
            inputs.insert(to_port.clone(), val.clone());
        }
    }

    // fill missing required
    for spec in input_specs {
        if spec.required && !inputs.contains_key(&spec.name) {
            return Err(ExecError::MissingOutput {
                node: target.to_string(),
                output: spec.name.clone(),
            });
        }
    }

    Ok(inputs)
}

fn execute_source(
    node: &ValidatedNode,
    _inputs: HashMap<String, RuntimeValue>,
    registries: &Registries,
    ctx: &ExecutionContext,
) -> Result<HashMap<String, RuntimeValue>, ExecError> {
    let primitive =
        registries
            .sources
            .get(&node.impl_id)
            .ok_or_else(|| ExecError::UnknownPrimitive {
                id: node.impl_id.clone(),
                version: node.version.clone(),
            })?;

    let manifest = primitive.manifest();
    let manifest_name_parameters = source_manifest_name_parameters(node, manifest);
    for req in &manifest.requires.context {
        // Resolve $key bindings before the required check so optional
        // parameter-bound keys are still resolved.
        let resolved_name =
            crate::common::resolve_manifest_name(&req.name, &manifest_name_parameters).map_err(
                |_| ExecError::MissingRequiredContextKey {
                    node: node.runtime_id.clone(),
                    key: req.name.clone(),
                },
            )?;
        if !req.required {
            continue;
        }
        match ctx.value(&resolved_name) {
            None => {
                return Err(ExecError::MissingRequiredContextKey {
                    node: node.runtime_id.clone(),
                    key: resolved_name,
                });
            }
            Some(val) => {
                if val.value_type() != req.ty {
                    return Err(ExecError::ContextKeyTypeMismatch {
                        node: node.runtime_id.clone(),
                        key: resolved_name,
                        expected: req.ty.clone(),
                        got: val.value_type(),
                    });
                }
            }
        }
    }

    let mut mapped_parameters: HashMap<String, crate::source::ParameterValue> = HashMap::new();
    for (name, val) in &node.parameters {
        let mapped = map_to_source_parameter_value(val).ok_or_else(|| {
            ExecError::ParameterTypeConversionFailed {
                node: node.runtime_id.clone(),
                parameter: name.clone(),
            }
        })?;
        mapped_parameters.insert(name.clone(), mapped);
    }

    let outputs = primitive.produce(&mapped_parameters, ctx);
    ensure_finite(&node.runtime_id, &outputs)?;
    Ok(outputs
        .into_iter()
        .map(|(k, v)| (k, map_common_value(v)))
        .collect())
}

fn execute_compute(
    node: &ValidatedNode,
    inputs: HashMap<String, RuntimeValue>,
    registries: &Registries,
) -> Result<HashMap<String, RuntimeValue>, ExecError> {
    let primitive =
        registries
            .computes
            .get(&node.impl_id)
            .ok_or_else(|| ExecError::UnknownPrimitive {
                id: node.impl_id.clone(),
                version: node.version.clone(),
            })?;

    let mut mapped_inputs: HashMap<String, crate::common::Value> = HashMap::new();
    for (name, val) in inputs {
        let mapped = map_to_compute_value(&val).ok_or_else(|| ExecError::TypeConversionFailed {
            node: node.runtime_id.clone(),
            port: name.clone(),
        })?;
        mapped_inputs.insert(name, mapped);
    }

    let mut mapped_parameters: HashMap<String, crate::common::Value> = HashMap::new();
    for (name, val) in &node.parameters {
        let mapped = map_to_compute_parameter_value(val).ok_or_else(|| {
            // X.11: If Int conversion failed, it's out of range; otherwise type mismatch
            if let crate::cluster::ParameterValue::Int(i) = val {
                ExecError::ParameterOutOfRange {
                    node: node.runtime_id.clone(),
                    parameter: name.clone(),
                    value: *i,
                }
            } else {
                ExecError::ParameterTypeConversionFailed {
                    node: node.runtime_id.clone(),
                    parameter: name.clone(),
                }
            }
        })?;
        mapped_parameters.insert(name.clone(), mapped);
    }

    let outputs = primitive
        .compute(&mapped_inputs, &mapped_parameters, None)
        .map_err(|error| ExecError::ComputeFailed {
            node: node.runtime_id.clone(),
            id: node.impl_id.clone(),
            version: node.version.clone(),
            error,
        })?;
    for output_name in node.outputs.keys() {
        if !outputs.contains_key(output_name) {
            return Err(ExecError::MissingOutput {
                node: node.runtime_id.clone(),
                output: output_name.clone(),
            });
        }
    }
    ensure_finite(&node.runtime_id, &outputs)?;
    Ok(outputs
        .into_iter()
        .map(|(k, v)| (k, map_common_value(v)))
        .collect())
}

fn execute_trigger(
    node: &ValidatedNode,
    inputs: HashMap<String, RuntimeValue>,
    registries: &Registries,
) -> Result<HashMap<String, RuntimeValue>, ExecError> {
    let primitive =
        registries
            .triggers
            .get(&node.impl_id)
            .ok_or_else(|| ExecError::UnknownPrimitive {
                id: node.impl_id.clone(),
                version: node.version.clone(),
            })?;

    let mut mapped_inputs: HashMap<String, TriggerValue> = HashMap::new();
    for (name, val) in inputs {
        let mapped = map_to_trigger_value(&val).ok_or_else(|| ExecError::TypeConversionFailed {
            node: node.runtime_id.clone(),
            port: name.clone(),
        })?;
        mapped_inputs.insert(name, mapped);
    }

    let mut mapped_parameters: HashMap<String, crate::trigger::ParameterValue> = HashMap::new();
    for (name, val) in &node.parameters {
        let mapped = map_to_trigger_parameter_value(val).ok_or_else(|| {
            ExecError::ParameterTypeConversionFailed {
                node: node.runtime_id.clone(),
                parameter: name.clone(),
            }
        })?;
        mapped_parameters.insert(name.clone(), mapped);
    }

    let outputs = primitive.evaluate(&mapped_inputs, &mapped_parameters);
    Ok(outputs
        .into_iter()
        .map(|(k, v)| (k, map_trigger_value(v)))
        .collect())
}

fn execute_action(
    node: &ValidatedNode,
    inputs: HashMap<String, RuntimeValue>,
    registries: &Registries,
    graph_id: &str,
    event_id: &str,
) -> Result<(HashMap<String, RuntimeValue>, Vec<ActionEffect>), ExecError> {
    let primitive =
        registries
            .actions
            .get(&node.impl_id)
            .ok_or_else(|| ExecError::UnknownPrimitive {
                id: node.impl_id.clone(),
                version: node.version.clone(),
            })?;

    let mut mapped_inputs: HashMap<String, ActionValue> = HashMap::new();
    for (name, val) in &inputs {
        let mapped = map_to_action_value(val, &node.runtime_id, name)?;
        mapped_inputs.insert(name.clone(), mapped);
    }

    let mut mapped_parameters: HashMap<String, crate::action::ParameterValue> = HashMap::new();
    for (name, val) in &node.parameters {
        let mapped = map_to_action_parameter_value(val).ok_or_else(|| {
            ExecError::ParameterTypeConversionFailed {
                node: node.runtime_id.clone(),
                parameter: name.clone(),
            }
        })?;
        mapped_parameters.insert(name.clone(), mapped);
    }

    let outputs = primitive.execute(&mapped_inputs, &mapped_parameters);

    // Build effects from manifest write declarations and v1 intent declarations.
    let manifest = primitive.manifest();
    let mut writes = Vec::new();
    for spec in &manifest.effects.writes {
        // Resolve $key via Decision 2 infrastructure
        let resolved_name = crate::common::resolve_manifest_name(&spec.name, &node.parameters)
            .map_err(|_| ExecError::ParameterTypeConversionFailed {
                node: node.runtime_id.clone(),
                parameter: spec.name.clone(),
            })?;

        // Read the write value from the action input snapshot
        let input_val = inputs
            .get(&spec.from_input)
            .ok_or_else(|| ExecError::MissingOutput {
                node: node.runtime_id.clone(),
                output: spec.from_input.clone(),
            })?;

        let value = map_runtime_value_to_common(input_val).ok_or_else(|| {
            ExecError::TypeConversionFailed {
                node: node.runtime_id.clone(),
                port: spec.from_input.clone(),
            }
        })?;

        writes.push(EffectWrite {
            key: resolved_name,
            value,
        });
    }

    let mut intents_by_kind: Vec<(String, Vec<IntentRecord>)> = Vec::new();
    for (intent_ordinal, intent_spec) in manifest.effects.intents.iter().enumerate() {
        let mut fields = Vec::new();
        let mut field_values_by_name = HashMap::new();

        for field_spec in &intent_spec.fields {
            let value = match (
                field_spec.from_input.as_ref(),
                field_spec.from_param.as_ref(),
            ) {
                (Some(from_input), None) => {
                    let input_value =
                        mapped_inputs
                            .get(from_input)
                            .ok_or_else(|| ExecError::MissingOutput {
                                node: node.runtime_id.clone(),
                                output: from_input.clone(),
                            })?;
                    map_action_value_to_common(input_value).ok_or_else(|| {
                        ExecError::TypeConversionFailed {
                            node: node.runtime_id.clone(),
                            port: from_input.clone(),
                        }
                    })?
                }
                (None, Some(from_param)) => {
                    let parameter_value = mapped_parameters.get(from_param).ok_or_else(|| {
                        ExecError::ParameterTypeConversionFailed {
                            node: node.runtime_id.clone(),
                            parameter: from_param.clone(),
                        }
                    })?;
                    map_action_parameter_value_to_common(parameter_value).ok_or_else(|| {
                        ExecError::ParameterTypeConversionFailed {
                            node: node.runtime_id.clone(),
                            parameter: from_param.clone(),
                        }
                    })?
                }
                _ => {
                    return Err(ExecError::ParameterTypeConversionFailed {
                        node: node.runtime_id.clone(),
                        parameter: field_spec.name.clone(),
                    });
                }
            };

            field_values_by_name.insert(field_spec.name.clone(), value.clone());
            fields.push(IntentField {
                name: field_spec.name.clone(),
                value,
            });
        }

        for mirror_write in &intent_spec.mirror_writes {
            let resolved_name =
                crate::common::resolve_manifest_name(&mirror_write.name, &node.parameters)
                    .map_err(|_| ExecError::ParameterTypeConversionFailed {
                        node: node.runtime_id.clone(),
                        parameter: mirror_write.name.clone(),
                    })?;
            let mirrored_value = field_values_by_name
                .get(&mirror_write.from_field)
                .ok_or_else(|| ExecError::MissingOutput {
                    node: node.runtime_id.clone(),
                    output: mirror_write.from_field.clone(),
                })?
                .clone();
            writes.push(EffectWrite {
                key: resolved_name,
                value: mirrored_value,
            });
        }

        let intent = IntentRecord {
            kind: intent_spec.name.clone(),
            intent_id: derive_intent_id(
                graph_id,
                event_id,
                &node.runtime_id,
                &intent_spec.name,
                intent_ordinal,
            ),
            fields,
        };

        if let Some((_, records)) = intents_by_kind
            .iter_mut()
            .find(|(kind, _)| kind == &intent_spec.name)
        {
            records.push(intent);
        } else {
            intents_by_kind.push((intent_spec.name.clone(), vec![intent]));
        }
    }

    let mut effects = Vec::new();
    if !writes.is_empty() {
        effects.push(ActionEffect {
            kind: "set_context".to_string(),
            writes,
            intents: vec![],
        });
    }

    for (intent_kind, intents) in intents_by_kind {
        effects.push(ActionEffect {
            kind: intent_kind,
            writes: vec![],
            intents,
        });
    }

    let runtime_outputs = outputs
        .into_iter()
        .map(|(k, v)| (k, map_action_value(v)))
        .collect();

    Ok((runtime_outputs, effects))
}

/// NUM-FINITE-1: Reject non-finite numeric outputs before propagation.
///
/// This guard ensures that NaN, inf, and -inf cannot reach triggers or actions,
/// preventing counterintuitive behavior (e.g., NaN comparisons always return false).
///
/// Called after compute and source outputs are produced, before values enter
/// the node_outputs map.
///
/// See: NUM-FINITE-1 in PHASE_INVARIANTS.md
fn ensure_finite(
    node: &str,
    outputs: &HashMap<String, crate::common::Value>,
) -> Result<(), ExecError> {
    for (port, value) in outputs {
        match value {
            crate::common::Value::Number(n) if !n.is_finite() => {
                return Err(ExecError::NonFiniteOutput {
                    node: node.to_string(),
                    port: port.to_string(),
                });
            }
            crate::common::Value::Series(values)
                if values.iter().any(|value| !value.is_finite()) =>
            {
                return Err(ExecError::NonFiniteOutput {
                    node: node.to_string(),
                    port: port.to_string(),
                });
            }
            _ => {}
        }
    }

    Ok(())
}

fn first_intent_emitting_action(graph: &ValidatedGraph, registries: &Registries) -> Option<String> {
    for node_id in &graph.topo_order {
        let Some(node) = graph.nodes.get(node_id) else {
            continue;
        };
        if node.kind != PrimitiveKind::Action {
            continue;
        }

        let Some(action) = registries.actions.get(&node.impl_id) else {
            continue;
        };
        if !action.manifest().effects.intents.is_empty() {
            return Some(node.runtime_id.clone());
        }
    }
    None
}

fn map_common_value(v: crate::common::Value) -> RuntimeValue {
    match v {
        crate::common::Value::Number(n) => RuntimeValue::Number(n),
        crate::common::Value::Series(s) => RuntimeValue::Series(s),
        crate::common::Value::Bool(b) => RuntimeValue::Bool(b),
        crate::common::Value::String(s) => RuntimeValue::String(s),
    }
}

fn map_to_compute_value(v: &RuntimeValue) -> Option<crate::common::Value> {
    match v {
        RuntimeValue::Number(n) => Some(crate::common::Value::Number(*n)),
        RuntimeValue::Series(s) => Some(crate::common::Value::Series(s.clone())),
        RuntimeValue::Bool(b) => Some(crate::common::Value::Bool(*b)),
        _ => None,
    }
}

/// X.11: Maximum safe integer for exact f64 representation (2^53).
const MAX_SAFE_INT: i64 = 9_007_199_254_740_992;

fn map_to_compute_parameter_value(
    v: &crate::cluster::ParameterValue,
) -> Option<crate::common::Value> {
    match v {
        crate::cluster::ParameterValue::Int(i) => {
            // X.11: Guard against precision loss for |i| > 2^53
            // Note: Use explicit bounds instead of i.abs() to avoid overflow panic on i64::MIN
            if *i >= -MAX_SAFE_INT && *i <= MAX_SAFE_INT {
                Some(crate::common::Value::Number(*i as f64))
            } else {
                None // Caller will produce ParameterOutOfRange with full context
            }
        }
        crate::cluster::ParameterValue::Number(n) => Some(crate::common::Value::Number(*n)),
        crate::cluster::ParameterValue::Bool(b) => Some(crate::common::Value::Bool(*b)),
        _ => None,
    }
}

fn map_trigger_value(v: TriggerValue) -> RuntimeValue {
    match v {
        TriggerValue::Number(n) => RuntimeValue::Number(n),
        TriggerValue::Series(s) => RuntimeValue::Series(s),
        TriggerValue::Bool(b) => RuntimeValue::Bool(b),
        TriggerValue::Event(e) => RuntimeValue::Event(RuntimeEvent::Trigger(e)),
    }
}

fn map_to_trigger_value(v: &RuntimeValue) -> Option<TriggerValue> {
    match v {
        RuntimeValue::Number(n) => Some(TriggerValue::Number(*n)),
        RuntimeValue::Series(s) => Some(TriggerValue::Series(s.clone())),
        RuntimeValue::Bool(b) => Some(TriggerValue::Bool(*b)),
        RuntimeValue::Event(RuntimeEvent::Trigger(e)) => Some(TriggerValue::Event(e.clone())),
        _ => None,
    }
}

fn map_to_trigger_parameter_value(
    v: &crate::cluster::ParameterValue,
) -> Option<crate::trigger::ParameterValue> {
    match v {
        crate::cluster::ParameterValue::Int(i) => Some(crate::trigger::ParameterValue::Int(*i)),
        crate::cluster::ParameterValue::Number(n) => {
            Some(crate::trigger::ParameterValue::Number(*n))
        }
        crate::cluster::ParameterValue::Bool(b) => Some(crate::trigger::ParameterValue::Bool(*b)),
        crate::cluster::ParameterValue::String(s) => {
            Some(crate::trigger::ParameterValue::String(s.clone()))
        }
        crate::cluster::ParameterValue::Enum(e) => {
            Some(crate::trigger::ParameterValue::Enum(e.clone()))
        }
    }
}

fn map_action_value(v: ActionValue) -> RuntimeValue {
    match v {
        ActionValue::Event(e) => RuntimeValue::Event(RuntimeEvent::Action(e)),
        ActionValue::Number(n) => RuntimeValue::Number(n),
        ActionValue::Series(s) => RuntimeValue::Series(s),
        ActionValue::Bool(b) => RuntimeValue::Bool(b),
        ActionValue::String(s) => RuntimeValue::String(s),
    }
}

fn map_to_action_value(v: &RuntimeValue, node: &str, port: &str) -> Result<ActionValue, ExecError> {
    Ok(match v {
        RuntimeValue::Event(RuntimeEvent::Action(e)) => ActionValue::Event(e.clone()),
        RuntimeValue::Event(RuntimeEvent::Trigger(TriggerEvent::Emitted)) => {
            ActionValue::Event(crate::action::ActionOutcome::Attempted)
        }
        RuntimeValue::Event(RuntimeEvent::Trigger(TriggerEvent::NotEmitted)) => {
            // R.7: should_skip_action() must catch NotEmitted before this point.
            // Return an error instead of panicking to maintain kernel determinism.
            return Err(ExecError::ActionSkipViolation {
                node: node.to_string(),
                port: port.to_string(),
            });
        }
        RuntimeValue::Number(n) => ActionValue::Number(*n),
        RuntimeValue::Series(s) => ActionValue::Series(s.clone()),
        RuntimeValue::Bool(b) => ActionValue::Bool(*b),
        RuntimeValue::String(s) => ActionValue::String(s.clone()),
    })
}

fn map_runtime_value_to_common(v: &RuntimeValue) -> Option<crate::common::Value> {
    match v {
        RuntimeValue::Number(n) => Some(crate::common::Value::Number(*n)),
        RuntimeValue::Bool(b) => Some(crate::common::Value::Bool(*b)),
        RuntimeValue::String(s) => Some(crate::common::Value::String(s.clone())),
        RuntimeValue::Series(s) => Some(crate::common::Value::Series(s.clone())),
        RuntimeValue::Event(_) => None,
    }
}

fn map_action_value_to_common(v: &ActionValue) -> Option<crate::common::Value> {
    match v {
        ActionValue::Number(n) => Some(crate::common::Value::Number(*n)),
        ActionValue::Series(s) => Some(crate::common::Value::Series(s.clone())),
        ActionValue::Bool(b) => Some(crate::common::Value::Bool(*b)),
        ActionValue::String(s) => Some(crate::common::Value::String(s.clone())),
        ActionValue::Event(_) => None,
    }
}

fn map_action_parameter_value_to_common(
    v: &crate::action::ParameterValue,
) -> Option<crate::common::Value> {
    match v {
        crate::action::ParameterValue::Number(n) => Some(crate::common::Value::Number(*n)),
        crate::action::ParameterValue::Bool(b) => Some(crate::common::Value::Bool(*b)),
        crate::action::ParameterValue::String(s) => Some(crate::common::Value::String(s.clone())),
        crate::action::ParameterValue::Int(_) | crate::action::ParameterValue::Enum(_) => None,
    }
}

fn map_to_action_parameter_value(
    v: &crate::cluster::ParameterValue,
) -> Option<crate::action::ParameterValue> {
    match v {
        crate::cluster::ParameterValue::Int(i) => Some(crate::action::ParameterValue::Int(*i)),
        crate::cluster::ParameterValue::Number(n) => {
            Some(crate::action::ParameterValue::Number(*n))
        }
        crate::cluster::ParameterValue::Bool(b) => Some(crate::action::ParameterValue::Bool(*b)),
        crate::cluster::ParameterValue::String(s) => {
            Some(crate::action::ParameterValue::String(s.clone()))
        }
        crate::cluster::ParameterValue::Enum(e) => {
            Some(crate::action::ParameterValue::Enum(e.clone()))
        }
    }
}

fn map_to_source_parameter_value(
    v: &crate::cluster::ParameterValue,
) -> Option<crate::source::ParameterValue> {
    match v {
        crate::cluster::ParameterValue::Int(i) => Some(crate::source::ParameterValue::Int(*i)),
        crate::cluster::ParameterValue::Number(n) => {
            Some(crate::source::ParameterValue::Number(*n))
        }
        crate::cluster::ParameterValue::Bool(b) => Some(crate::source::ParameterValue::Bool(*b)),
        crate::cluster::ParameterValue::String(s) => {
            Some(crate::source::ParameterValue::String(s.clone()))
        }
        crate::cluster::ParameterValue::Enum(e) => {
            Some(crate::source::ParameterValue::Enum(e.clone()))
        }
    }
}

fn source_manifest_name_parameters(
    node: &ValidatedNode,
    manifest: &crate::source::SourcePrimitiveManifest,
) -> HashMap<String, crate::cluster::ParameterValue> {
    let mut resolved = node.parameters.clone();

    for spec in &manifest.parameters {
        if resolved.contains_key(&spec.name) {
            continue;
        }
        let Some(default) = &spec.default else {
            continue;
        };
        if let Some(mapped) = map_source_default_to_cluster_parameter_value(default) {
            resolved.insert(spec.name.clone(), mapped);
        }
    }

    resolved
}

fn map_source_default_to_cluster_parameter_value(
    v: &crate::source::ParameterValue,
) -> Option<crate::cluster::ParameterValue> {
    match v {
        crate::source::ParameterValue::Int(i) => Some(crate::cluster::ParameterValue::Int(*i)),
        crate::source::ParameterValue::Number(n) => {
            Some(crate::cluster::ParameterValue::Number(*n))
        }
        crate::source::ParameterValue::Bool(b) => Some(crate::cluster::ParameterValue::Bool(*b)),
        crate::source::ParameterValue::String(s) => {
            Some(crate::cluster::ParameterValue::String(s.clone()))
        }
        crate::source::ParameterValue::Enum(e) => {
            Some(crate::cluster::ParameterValue::Enum(e.clone()))
        }
    }
}

/// R.7 gating: Returns true if any Event input is TriggerEvent::NotEmitted.
/// Uses AND semantics: all trigger events must be Emitted for action to execute.
fn should_skip_action(inputs: &HashMap<String, RuntimeValue>) -> bool {
    inputs.values().any(|v| {
        matches!(
            v,
            RuntimeValue::Event(RuntimeEvent::Trigger(TriggerEvent::NotEmitted))
        )
    })
}

/// Produce outputs for a skipped action. Event outputs get ActionOutcome::Skipped.
fn produce_skipped_outputs(node: &ValidatedNode) -> HashMap<String, RuntimeValue> {
    node.outputs
        .iter()
        .map(|(name, meta)| {
            let value = match meta.value_type {
                ValueType::Event => {
                    RuntimeValue::Event(RuntimeEvent::Action(ActionOutcome::Skipped))
                }
                // Non-event outputs use sensible defaults (actions are terminal per F.2).
                ValueType::Number => RuntimeValue::Number(0.0),
                ValueType::Bool => RuntimeValue::Bool(false),
                ValueType::String => RuntimeValue::String(String::new()),
                ValueType::Series => RuntimeValue::Series(vec![]),
            };
            (name.clone(), value)
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::ensure_finite;
    use crate::common::Value;

    #[test]
    fn num_finite_guard_rejects_nan() {
        let outputs =
            std::collections::HashMap::from([("result".to_string(), Value::Number(f64::NAN))]);
        let result = ensure_finite("test_node", &outputs);
        assert!(matches!(
            result,
            Err(super::ExecError::NonFiniteOutput { .. })
        ));
    }

    #[test]
    fn num_finite_guard_rejects_infinity() {
        let outputs =
            std::collections::HashMap::from([("result".to_string(), Value::Number(f64::INFINITY))]);
        let result = ensure_finite("test_node", &outputs);
        assert!(matches!(
            result,
            Err(super::ExecError::NonFiniteOutput { .. })
        ));
    }

    #[test]
    fn num_finite_guard_allows_finite() {
        let outputs =
            std::collections::HashMap::from([("result".to_string(), Value::Number(42.0))]);
        let result = ensure_finite("test_node", &outputs);
        assert!(result.is_ok());
    }
}