ontoref-code/crates/ontoref-ops/src/runtime.rs

//! Pipeline executor — `load → pre_validate → exec → apply → render → post_validate → emit_witness`.
//!
//! Phase 1 wires the skeleton: every step is instrumented with
//! microsecond-resolution metrics (ADR-026 SLA observability). Concrete
//! precondition loaders, validator dispatch, effect application, and
//! NCL render paths plug in alongside their consuming ops (O3 onward).

use std::time::Instant;

use serde::{Deserialize, Serialize};

use crate::context::OpContext;
use crate::error::OpError;
use crate::output::OpOutput;
use crate::registry::OperationEntry;
use crate::validation::{dispatch as validation_dispatch, ValidationCtx, Verdict};
use crate::witness::{sign_witness, PipelineWitness};

/// Microsecond-resolution metrics emitted alongside every pipeline
/// invocation. ADR-026 SLA tracking consumes these once the runtime
/// learns to publish them.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PipelineMetrics {
    /// Time spent loading the precondition slice.
    pub precondition_load_us: u64,
    /// Time spent running pre-validators.
    pub pre_validate_us: u64,
    /// Time spent in the op body.
    pub exec_us: u64,
    /// Time spent applying effects to the substrate.
    pub apply_us: u64,
    /// Time spent rendering NCL paths.
    pub render_us: u64,
    /// Time spent running post-validators.
    pub post_validate_us: u64,
    /// Time spent computing the state root + signing the witness.
    pub witness_us: u64,
}

/// Aggregate result of one pipeline invocation.
#[derive(Debug, Clone)]
pub struct PipelineResult {
    /// Signed witness — the only object that crosses the trust boundary.
    pub witness: PipelineWitness,
    /// Effects the op declared (already folded into the substrate).
    pub effects: Vec<crate::output::EffectRecord>,
    /// Per-step latencies.
    pub metrics: PipelineMetrics,
    /// Oplog id the witness was persisted under (`substrate` feature).
    /// `None` when the context carried no oplog — the witness was signed
    /// and returned but not persisted. This is the witness's DAG position
    /// (bl-022): the id covers the body's `parents` + `timestamp`.
    #[cfg(feature = "substrate")]
    pub oplog_op_id: Option<ontoref_types::OpId>,
}

/// Execute one operation through the full pipeline.
///
/// # Errors
///
/// Any pipeline step may abort with [`OpError`]; in particular,
/// [`OpError::ValidatorRejected`] indicates a Reject verdict and
/// [`OpError::ActorPolicyDenied`] indicates the calling actor is not
/// permitted for the op.
pub fn execute_op(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &mut OpContext,
) -> Result<PipelineResult, OpError> {
    let mut metrics = PipelineMetrics::default();

    enforce_actor_policy(entry, &ctx.actor_id)?;
    // Stamp the op id onto the context so validators and the body can
    // reference it without an extra parameter.
    ctx.op_id = entry.id.to_owned();

    // 1. Precondition load — dispatched by entry.precondition tag.
    //    Bundles the loaded slice with the op's inputs so validators
    //    see both without an extra parameter.
    let t0 = Instant::now();
    let slice = load_precondition(entry, inputs, ctx)?;
    metrics.precondition_load_us = elapsed_us(t0);

    // 2. Pre-validators — block the op when any non-Accept verdict.
    let t0 = Instant::now();
    let validation_ctx = ValidationCtx::synthetic(&ctx.actor_id, entry.id);
    run_validators(entry.id, entry.pre_validators, &slice, &validation_ctx)?;
    metrics.pre_validate_us = elapsed_us(t0);

    // 3. Op body.
    let t0 = Instant::now();
    let output = (entry.execute)(inputs, ctx)?;
    metrics.exec_us = elapsed_us(t0);

    // 4. Apply effects (declarative hook — currently a no-op; ops mutate
    //    ctx.commit_layer directly inside the body).
    let t0 = Instant::now();
    apply_effects(entry, &output, ctx);
    metrics.apply_us = elapsed_us(t0);

    // 5. Render NCL paths — Phase 1 supports state.ncl when
    //    ctx.state + ctx.project_root are set.
    let t0 = Instant::now();
    render_paths(entry, ctx)?;
    metrics.render_us = elapsed_us(t0);

    // 6. Post-validators — re-load the slice (state has mutated) and run.
    let t0 = Instant::now();
    let post_slice = load_precondition(entry, inputs, ctx)?;
    run_validators(entry.id, entry.post_validators, &post_slice, &validation_ctx)?;
    metrics.post_validate_us = elapsed_us(t0);

    // 7. Emit witness.
    let t0 = Instant::now();
    let state_root = ctx.commit_layer_root();
    let witness = sign_witness(
        &ctx.signing_key,
        entry.id,
        state_root,
        &output,
        entry.witness_shape,
    );
    metrics.witness_us = elapsed_us(t0);

    // 8. Persist the witness to the oplog as the final stage (substrate
    //    feature). Append-first semantics: the oplog is the source of
    //    truth, the rendered NCL is a projection. Done here, after the
    //    in-memory render, the only divergence window is render-ok /
    //    append-fail, surfaced as OpError::WitnessPersist for the caller
    //    to reconcile.
    #[cfg(feature = "substrate")]
    let oplog_op_id = persist_witness(&witness, ctx)?;

    Ok(PipelineResult {
        witness,
        effects: output.effects,
        metrics,
        #[cfg(feature = "substrate")]
        oplog_op_id,
    })
}

/// Persist the signed pipeline witness to the oplog as an
/// [`OpPayload::Witness`] operation. Returns the resulting [`OpId`]
/// (the witness's DAG position) or `None` when the context carries no
/// oplog. The op's parents are the current DAG heads, so each witness
/// extends every head; its HLC is ticked strictly past every parent's,
/// satisfying the oplog's monotonicity invariant.
///
/// # Errors
///
/// [`OpError::WitnessPersist`] when reading heads, loading a parent, or
/// the append itself fails.
#[cfg(feature = "substrate")]
fn persist_witness(
    witness: &PipelineWitness,
    ctx: &OpContext,
) -> Result<Option<ontoref_types::OpId>, OpError> {
    use std::time::{SystemTime, UNIX_EPOCH};

    use ontoref_types::operation::{OpBody, OpPayload, Operation};
    use ontoref_types::Hlc;

    let Some(oplog) = ctx.oplog.as_ref() else {
        return Ok(None);
    };

    let persist_err = |reason: String| OpError::WitnessPersist {
        op_id: witness.op_id.clone(),
        reason,
    };

    // Parents = current DAG heads; the witness extends every head.
    let parents = oplog
        .heads()
        .map_err(|e| persist_err(format!("reading oplog heads: {e}")))?;

    // HLC must be strictly greater than every parent's timestamp.
    let mut max_parent_ts = Hlc::genesis();
    for parent in &parents {
        let parent_op = oplog
            .get(parent)
            .map_err(|e| persist_err(format!("loading parent {parent:?}: {e}")))?
            .ok_or_else(|| persist_err(format!("head {parent:?} absent from oplog")))?;
        if parent_op.body.timestamp > max_parent_ts {
            max_parent_ts = parent_op.body.timestamp;
        }
    }
    let now_physical = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
        .unwrap_or(0);
    let timestamp = Hlc::tick(max_parent_ts, now_physical);

    let body = OpBody {
        actor: witness.actor,
        parents,
        payload: OpPayload::Witness {
            effects_hash: witness.effects_hash,
            extra_payload: witness.extra_payload.clone(),
            op_id: witness.op_id.clone(),
            state_root: witness.state_root,
            witness_shape: witness.witness_shape.clone(),
            witness_sig: witness.signature,
        },
        timestamp,
    };
    let op = Operation::sign(body, &ctx.signing_key);
    let id = oplog
        .append(&op)
        .map_err(|e| persist_err(format!("oplog append: {e}")))?;
    Ok(Some(id))
}

fn enforce_actor_policy(entry: &OperationEntry, actor_id: &str) -> Result<(), OpError> {
    if entry.actor_policy.is_empty() {
        return Ok(());
    }
    if entry.actor_policy.contains(&actor_id) {
        return Ok(());
    }
    Err(OpError::ActorPolicyDenied {
        actor_id: actor_id.to_owned(),
        op_id: entry.id.to_owned(),
        allowed: entry.actor_policy.iter().map(|s| s.to_string()).collect(),
    })
}

fn load_precondition(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    match entry.precondition {
        "" => Ok(serde_json::Value::Null),
        "fsm_dimension_by_id" => load_fsm_dimension_slice(entry, inputs, ctx),
        "ondaod" => load_ondaod_slice(entry, inputs, ctx),
        "ontology_node_by_id" => load_ontology_node_slice(entry, inputs, ctx),
        "ontology_edge_dag_check" => load_ontology_edge_dag_check_slice(entry, inputs, ctx),
        "backlog_open_items" => load_backlog_slice(entry, inputs, ctx),
        other => Err(OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: format!("no loader registered for precondition tag '{other}'"),
        }),
    }
}

fn load_ondaod_slice(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    // Phase 1: the ondaod evaluation comes in the inputs alongside the
    // op-specific fields. The precondition load surface confirms its
    // shape so pre-validators can rely on it; full integration with
    // crate::precondition::ondaod::load arrives when the runtime takes
    // ownership of the qa.ncl + core.ncl JSON snapshots.
    let mut slice = serde_json::Map::new();
    let ondaod = inputs.get("ondaod_evaluation").cloned().unwrap_or_else(|| {
        serde_json::json!({ "engaged_tensions": [], "synthesis_state": "" })
    });
    slice.insert("ondaod".to_owned(), ondaod);
    slice.insert("inputs".to_owned(), inputs.clone());
    slice.insert(
        "actor".to_owned(),
        serde_json::Value::String(ctx.actor_id.clone()),
    );
    let _ = entry;
    Ok(serde_json::Value::Object(slice))
}

fn load_ontology_node_slice(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    let node_id = inputs
        .get("node_id")
        .and_then(|v| v.as_str())
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: "missing required input 'node_id'".to_owned(),
        })?;
    let core = ctx.core.as_ref().ok_or_else(|| OpError::PreconditionLoad {
        op_id: entry.id.to_owned(),
        reason: "ctx.core is None — ontology_node_by_id requires a loaded core snapshot".to_owned(),
    })?;
    let existing_node = core
        .get("nodes")
        .and_then(|n| n.as_array())
        .and_then(|arr| {
            arr.iter()
                .find(|n| n.get("id").and_then(|v| v.as_str()) == Some(node_id))
        })
        .cloned();
    let mut slice = serde_json::Map::new();
    slice.insert(
        "existing_node".to_owned(),
        existing_node.unwrap_or(serde_json::Value::Null),
    );
    slice.insert("inputs".to_owned(), inputs.clone());
    Ok(serde_json::Value::Object(slice))
}

fn load_backlog_slice(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    _ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    let _ = entry;
    let mut slice = serde_json::Map::new();
    slice.insert("inputs".to_owned(), inputs.clone());
    Ok(serde_json::Value::Object(slice))
}

fn load_ontology_edge_dag_check_slice(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    let from = inputs
        .get("from")
        .and_then(|v| v.as_str())
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: "missing 'from' input".to_owned(),
        })?;
    let to = inputs
        .get("to")
        .and_then(|v| v.as_str())
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: "missing 'to' input".to_owned(),
        })?;
    // The reverse edge (to → from) is what would close a 1-step cycle.
    let existing_reverse = ctx
        .core
        .as_ref()
        .and_then(|c| c.get("edges"))
        .and_then(|e| e.as_array())
        .and_then(|arr| {
            arr.iter()
                .find(|e| {
                    e.get("from").and_then(|v| v.as_str()) == Some(to)
                        && e.get("to").and_then(|v| v.as_str()) == Some(from)
                })
                .cloned()
        })
        .unwrap_or(serde_json::Value::Null);
    let mut slice = serde_json::Map::new();
    slice.insert("existing_reverse".to_owned(), existing_reverse);
    slice.insert("inputs".to_owned(), inputs.clone());
    Ok(serde_json::Value::Object(slice))
}

fn load_fsm_dimension_slice(
    entry: &OperationEntry,
    inputs: &serde_json::Value,
    ctx: &OpContext,
) -> Result<serde_json::Value, OpError> {
    let dim_id = inputs
        .get("dimension")
        .and_then(|v| v.as_str())
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: "missing required input 'dimension'".to_owned(),
        })?;
    let state = ctx
        .state
        .as_ref()
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: "ctx.state is None — fsm_dimension_by_id requires a loaded StateConfig".to_owned(),
        })?;
    let dimension = state
        .dimensions
        .iter()
        .find(|d| d.id == dim_id)
        .ok_or_else(|| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: format!("dimension '{dim_id}' not present in ctx.state"),
        })?;
    let mut slice = serde_json::Map::new();
    slice.insert(
        "dimension".to_owned(),
        serde_json::to_value(dimension).map_err(|e| OpError::PreconditionLoad {
            op_id: entry.id.to_owned(),
            reason: format!("serialising dimension: {e}"),
        })?,
    );
    slice.insert("inputs".to_owned(), inputs.clone());
    Ok(serde_json::Value::Object(slice))
}

fn run_validators(
    op_id: &str,
    validator_ids: &[&str],
    slice: &serde_json::Value,
    ctx: &ValidationCtx,
) -> Result<(), OpError> {
    for vid in validator_ids {
        if vid.is_empty() {
            continue;
        }
        match validation_dispatch::dispatch(vid, slice, ctx) {
            Some(Verdict::Accept) => {}
            Some(other) => {
                return Err(OpError::ValidatorRejected {
                    op_id: op_id.to_owned(),
                    verdict: other,
                });
            }
            None => {
                // Unknown validator id — fail loud rather than silently
                // accept. Catalog/Rust coherence checks (O6) will catch
                // this earlier in CI.
                return Err(OpError::ValidatorRejected {
                    op_id: op_id.to_owned(),
                    verdict: Verdict::Unknown,
                });
            }
        }
    }
    Ok(())
}

fn apply_effects(_entry: &OperationEntry, _output: &OpOutput, _ctx: &mut OpContext) {
    // O2 stub: the op body is currently the only path that mutates the
    // substrate. Declarative effect application lands when concrete
    // `effects` strings are parsed in a later gate.
}

fn render_paths(entry: &OperationEntry, ctx: &mut OpContext) -> Result<(), OpError> {
    let Some(root) = ctx.project_root.as_ref() else {
        return Ok(());
    };
    let Some(state) = ctx.state.as_ref() else {
        return Ok(());
    };
    for rel in entry.render_paths {
        if rel.is_empty() {
            continue;
        }
        let target = root.join(rel);
        // Phase 1: only state.ncl renders are supported. Other render
        // paths surface as OpError::Render with the NotImplemented
        // reason — visible to consumers, no silent skip.
        match ontoref_ontology::render_pipeline::render_and_write(state, &target) {
            Ok(()) => {}
            Err(e) => {
                return Err(OpError::Render {
                    path: target.display().to_string(),
                    reason: e.to_string(),
                });
            }
        }
    }
    Ok(())
}

trait CommitLayerRoot {
    fn commit_layer_root(&self) -> ontoref_commit::StateRoot;
}

impl CommitLayerRoot for OpContext {
    fn commit_layer_root(&self) -> ontoref_commit::StateRoot {
        self.commit_layer.root()
    }
}

fn elapsed_us(t0: Instant) -> u64 {
    let elapsed = t0.elapsed();
    elapsed.as_micros().min(u64::MAX as u128) as u64
}

#[cfg(test)]
#[test]
fn pipeline() {
    use ed25519_dalek::SigningKey;
    use ontoref_derive::onto_operation;

    // Noop op — empty inputs, no validators, no effects, no render
    // paths. Exercises every pipeline step in skeleton form.
    #[onto_operation(
        id = "noop_pipeline_smoke",
        description = "Noop op used by the O2 runtime smoke test — no precondition, no validators, no effects, no render paths.",
        validation_sla = "Synchronous",
        witness_shape = "noop"
    )]
    fn noop_pipeline_smoke(
        _inputs: &serde_json::Value,
        _ctx: &mut OpContext,
    ) -> Result<OpOutput, OpError> {
        Ok(OpOutput::default())
    }

    let entry = inventory::iter::<OperationEntry>()
        .find(|e| e.id == "noop_pipeline_smoke")
        .expect("noop op must be registered");

    let signing_key = SigningKey::from_bytes(&[3u8; 32]);
    let mut ctx = OpContext::new_in_memory("test-actor", "", signing_key);

    let inputs = serde_json::json!({});
    let result = execute_op(entry, &inputs, &mut ctx).expect("pipeline succeeds");

    // Witness signature verifies under the actor's public key.
    assert_eq!(result.witness.op_id, "noop_pipeline_smoke");
    assert_eq!(result.witness.witness_shape, "noop");
    result
        .witness
        .verify()
        .expect("witness must verify under signer's public key");

    // No effects, no render paths — but metrics are populated.
    assert!(result.effects.is_empty());
    // The witness step always runs and records a non-zero (or at least
    // observed) measurement.
    let total_us = result.metrics.precondition_load_us
        + result.metrics.pre_validate_us
        + result.metrics.exec_us
        + result.metrics.apply_us
        + result.metrics.render_us
        + result.metrics.post_validate_us
        + result.metrics.witness_us;
    let _ = total_us; // metrics emitted; the value itself is wall-clock dependent.

    // Negative: tampering with the state root must fail verification.
    let mut tampered = result.witness.clone();
    tampered.state_root = [0xFFu8; 32];
    assert!(
        tampered.verify().is_err(),
        "tampered state_root must not verify"
    );
}

#[cfg(test)]
#[test]
fn pipeline_rejects_disallowed_actor() {
    use ed25519_dalek::SigningKey;
    use ontoref_derive::onto_operation;

    /// Restricted op for actor-policy negative coverage.
    #[onto_operation(
        id = "restricted_op_smoke",
        description = "Restricted to the admin role — used by O2 to verify actor policy enforcement.",
        validation_sla = "Synchronous",
        actor_policy = "admin",
        witness_shape = "noop"
    )]
    fn restricted_op_smoke(
        _inputs: &serde_json::Value,
        _ctx: &mut OpContext,
    ) -> Result<OpOutput, OpError> {
        Ok(OpOutput::default())
    }

    let entry = inventory::iter::<OperationEntry>()
        .find(|e| e.id == "restricted_op_smoke")
        .expect("op must be registered");

    let signing_key = SigningKey::from_bytes(&[5u8; 32]);
    let mut ctx = OpContext::new_in_memory("viewer", "", signing_key);
    let err = execute_op(entry, &serde_json::json!({}), &mut ctx)
        .expect_err("viewer must be rejected when admin is the only allowed role");
    assert!(matches!(err, OpError::ActorPolicyDenied { .. }));
}

#[cfg(all(test, feature = "substrate"))]
#[test]
fn witness_persists_to_oplog_and_verifies_in_fresh_process() {
    use std::sync::Arc;

    use ed25519_dalek::SigningKey;
    use ontoref_derive::onto_operation;
    use ontoref_oplog::OpLog;
    use ontoref_types::operation::OpPayload;

    /// Noop op exercising the full pipeline including witness persistence.
    #[onto_operation(
        id = "substrate_persist_smoke",
        description = "Noop op used by the substrate witness-persistence test — dispatches, signs, and appends to the oplog.",
        validation_sla = "Synchronous",
        witness_shape = "noop"
    )]
    fn substrate_persist_smoke(
        _inputs: &serde_json::Value,
        _ctx: &mut OpContext,
    ) -> Result<OpOutput, OpError> {
        Ok(OpOutput::default())
    }

    let entry = inventory::iter::<OperationEntry>()
        .find(|e| e.id == "substrate_persist_smoke")
        .expect("op must be registered");

    let dir = tempfile::tempdir().expect("tempdir");
    let log_path = dir.path().join("oplog.redb");
    let signing_key = SigningKey::from_bytes(&[7u8; 32]);

    // Two dispatches share one oplog; the second witness must chain onto
    // the first. Scoped so the redb handle closes before we reopen it,
    // simulating a fresh process.
    let (id1, id2) = {
        let oplog = Arc::new(OpLog::open(&log_path).expect("open log"));
        let mut ctx =
            OpContext::new_in_memory("ci", "", signing_key).with_oplog(Arc::clone(&oplog));
        let r1 = execute_op(entry, &serde_json::json!({}), &mut ctx).expect("dispatch 1");
        let r2 = execute_op(entry, &serde_json::json!({}), &mut ctx).expect("dispatch 2");
        let id1 = r1.oplog_op_id.expect("witness 1 persisted");
        let id2 = r2.oplog_op_id.expect("witness 2 persisted");
        assert_ne!(id1, id2, "distinct dispatches yield distinct oplog ids");
        (id1, id2)
    };

    // Fresh process: reopen the durable log and verify every witness
    // against the actor's public key — no trust in the producer.
    let reopened = OpLog::open(&log_path).expect("reopen log");
    assert_eq!(reopened.len().expect("len"), 2, "both witnesses durable");

    let op2 = reopened.get(&id2).expect("get op2").expect("op2 present");
    op2.verify_signature()
        .expect("oplog op must verify in a fresh process");
    assert_eq!(
        op2.body.parents,
        vec![id1],
        "second witness chains onto the first — DAG position is bound"
    );

    // The persisted payload reconstructs the PipelineWitness, whose own
    // Ed25519 signature also verifies.
    assert!(
        matches!(op2.body.payload, OpPayload::Witness { .. }),
        "witness op must carry a Witness payload"
    );
    if let OpPayload::Witness {
        effects_hash,
        extra_payload,
        op_id,
        state_root,
        witness_shape,
        witness_sig,
    } = op2.body.payload.clone()
    {
        let reconstructed = PipelineWitness {
            op_id,
            state_root,
            effects_hash,
            extra_payload,
            witness_shape,
            actor: op2.body.actor,
            signature: witness_sig,
        };
        reconstructed
            .verify()
            .expect("reconstructed pipeline witness must verify");
    }

    // bl-022: tampering with the DAG position invalidates the operation
    // signature — reorder/replay of effects is detectable from the
    // witness alone.
    let mut tampered = op2.clone();
    tampered.body.parents.clear();
    assert!(
        tampered.verify_signature().is_err(),
        "clearing parents must invalidate the witness signature"
    );
}