diff --git a/crates/ontoref-reflection/Cargo.toml b/crates/ontoref-reflection/Cargo.toml deleted file mode 100644 index 54f7ca9..0000000 --- a/crates/ontoref-reflection/Cargo.toml +++ /dev/null @@ -1,32 +0,0 @@ -[package] -name = "ontoref-reflection" -version.workspace = true -edition.workspace = true -description = "Load, validate, and execute Reflection modes (NCL DAG contracts) against project state" -license.workspace = true - -[features] -default = [] -nats = ["dep:platform-nats", "dep:bytes"] - -[dependencies] -stratum-graph = { path = "../../../../stratumiops/code/crates/stratum-graph" } -stratum-state = { path = "../../../../stratumiops/code/crates/stratum-state" } - -serde = { version = "1", features = ["derive"] } -serde_json = { version = "1" } -anyhow = { version = "1" } -thiserror = { version = "2" } -async-trait = { version = "0.1" } -tokio = { version = "1", features = ["full"] } -uuid = { version = "1", features = ["v4"] } -chrono = { version = "0.4", features = ["serde"] } -tracing = { version = "0.1" } -regex = { version = "1" } - -platform-nats = { path = "../../../../stratumiops/code/crates/platform-nats", optional = true } -bytes = { version = "1", optional = true } - -[dev-dependencies] -tokio-test = { version = "0.4" } -tempfile = { version = "3" } diff --git a/crates/ontoref-reflection/src/dag.rs b/crates/ontoref-reflection/src/dag.rs deleted file mode 100644 index e359b38..0000000 --- a/crates/ontoref-reflection/src/dag.rs +++ /dev/null @@ -1,200 +0,0 @@ -use std::collections::{HashMap, HashSet, VecDeque}; - -use crate::{error::ReflectionError, mode::ActionStep}; - -/// Full DAG validation: uniqueness + referential integrity + cycle detection. -pub fn validate(steps: &[ActionStep]) -> Result<(), ReflectionError> { - check_uniqueness(steps)?; - check_referential_integrity(steps)?; - detect_cycles(steps)?; - Ok(()) -} - -/// Kahn's algorithm: returns steps grouped into parallel execution layers. -/// Layer 0 has no dependencies; layer N depends only on layers < N. -/// Returns `Err(CycleDetected)` if the graph is not a DAG. -pub fn topological_layers(steps: &[ActionStep]) -> Result>, ReflectionError> { - // in_degree[id] = number of steps id depends on - let mut in_degree: HashMap<&str, usize> = steps - .iter() - .map(|s| (s.id.as_str(), s.depends_on.len())) - .collect(); - - // dependents[id] = steps that depend on id (reverse edges) - let mut dependents: HashMap<&str, Vec<&str>> = - steps.iter().map(|s| (s.id.as_str(), vec![])).collect(); - - for step in steps { - for dep in &step.depends_on { - dependents - .entry(dep.step.as_str()) - .or_default() - .push(step.id.as_str()); - } - } - - let mut queue: VecDeque<&str> = in_degree - .iter() - .filter_map(|(&id, &d)| (d == 0).then_some(id)) - .collect(); - - let mut layers: Vec> = Vec::new(); - let mut visited = 0usize; - - while !queue.is_empty() { - let layer: Vec<&str> = queue.drain(..).collect(); - visited += layer.len(); - - let mut next: Vec<&str> = Vec::new(); - for &node in &layer { - for &dep in dependents.get(node).map(Vec::as_slice).unwrap_or(&[]) { - let d = in_degree - .get_mut(dep) - .expect("all step ids must be in in_degree — check uniqueness first"); - *d -= 1; - if *d == 0 { - next.push(dep); - } - } - } - - layers.push(layer.iter().map(|&s| s.to_string()).collect()); - queue.extend(next); - } - - if visited != steps.len() { - Err(ReflectionError::CycleDetected) - } else { - Ok(layers) - } -} - -fn check_uniqueness(steps: &[ActionStep]) -> Result<(), ReflectionError> { - let mut seen: HashSet<&str> = HashSet::with_capacity(steps.len()); - for step in steps { - if !seen.insert(step.id.as_str()) { - return Err(ReflectionError::DuplicateStepId(step.id.clone())); - } - } - Ok(()) -} - -fn check_referential_integrity(steps: &[ActionStep]) -> Result<(), ReflectionError> { - let ids: HashSet<&str> = steps.iter().map(|s| s.id.as_str()).collect(); - let bad: Vec = steps - .iter() - .flat_map(|step| { - step.depends_on.iter().filter_map(|dep| { - if ids.contains(dep.step.as_str()) { - None - } else { - Some(format!( - "step '{}' depends_on unknown '{}'", - step.id, dep.step - )) - } - }) - }) - .collect(); - - if bad.is_empty() { - Ok(()) - } else { - Err(ReflectionError::BadDependencyRefs(bad)) - } -} - -fn detect_cycles(steps: &[ActionStep]) -> Result<(), ReflectionError> { - topological_layers(steps).map(|_| ()) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::mode::{Dependency, DependencyKind, OnError}; - - fn step(id: &str, deps: &[&str]) -> ActionStep { - ActionStep { - id: id.to_string(), - action: id.to_string(), - actor: crate::mode::Actor::Both, - cmd: None, - depends_on: deps - .iter() - .map(|d| Dependency { - step: d.to_string(), - kind: DependencyKind::Always, - condition: None, - }) - .collect(), - on_error: OnError::default(), - verify: None, - note: None, - } - } - - #[test] - fn linear_chain_produces_single_layers() { - let steps = vec![step("a", &[]), step("b", &["a"]), step("c", &["b"])]; - let layers = topological_layers(&steps).unwrap(); - assert_eq!(layers.len(), 3); - assert_eq!(layers[0], vec!["a"]); - assert_eq!(layers[1], vec!["b"]); - assert_eq!(layers[2], vec!["c"]); - } - - #[test] - fn parallel_deps_form_single_layer() { - // a → {b, c} → d - let steps = vec![ - step("a", &[]), - step("b", &["a"]), - step("c", &["a"]), - step("d", &["b", "c"]), - ]; - let layers = topological_layers(&steps).unwrap(); - assert_eq!(layers.len(), 3); - assert_eq!(layers[0], vec!["a"]); - assert!(layers[1].contains(&"b".to_string())); - assert!(layers[1].contains(&"c".to_string())); - assert_eq!(layers[2], vec!["d"]); - } - - #[test] - fn cycle_detected() { - let steps = vec![step("a", &["b"]), step("b", &["a"])]; - assert!(matches!( - topological_layers(&steps), - Err(ReflectionError::CycleDetected) - )); - } - - #[test] - fn duplicate_id_rejected() { - let steps = vec![step("a", &[]), step("a", &[])]; - assert!(matches!( - check_uniqueness(&steps), - Err(ReflectionError::DuplicateStepId(_)) - )); - } - - #[test] - fn bad_ref_rejected() { - let steps = vec![step("a", &["nonexistent"])]; - assert!(matches!( - check_referential_integrity(&steps), - Err(ReflectionError::BadDependencyRefs(_)) - )); - } - - #[test] - fn validate_passes_for_valid_dag() { - let steps = vec![ - step("init_repo", &[]), - step("copy_ontology", &["init_repo"]), - step("init_kogral", &["init_repo"]), - step("publish", &["copy_ontology", "init_kogral"]), - ]; - assert!(validate(&steps).is_ok()); - } -} diff --git a/crates/ontoref-reflection/src/error.rs b/crates/ontoref-reflection/src/error.rs deleted file mode 100644 index 44f194c..0000000 --- a/crates/ontoref-reflection/src/error.rs +++ /dev/null @@ -1,25 +0,0 @@ -use thiserror::Error; - -#[derive(Debug, Error)] -pub enum ReflectionError { - #[error("duplicate step id: '{0}'")] - DuplicateStepId(String), - - #[error("invalid depends_on references: {}", .0.join(", "))] - BadDependencyRefs(Vec), - - #[error("cycle detected in step dependency graph")] - CycleDetected, - - #[error("unknown parameter placeholders in cmd (not in RunContext.params): {}", .0.join(", "))] - UnknownParams(Vec), - - #[error("nickel export failed on '{path}': {stderr}")] - NickelExport { path: String, stderr: String }, - - #[error("failed to parse ReflectionMode from nickel output: {0}")] - ParseMode(#[from] serde_json::Error), - - #[error("step task panicked: {0}")] - TaskPanic(String), -} diff --git a/crates/ontoref-reflection/src/executor.rs b/crates/ontoref-reflection/src/executor.rs deleted file mode 100644 index 41024d8..0000000 --- a/crates/ontoref-reflection/src/executor.rs +++ /dev/null @@ -1,554 +0,0 @@ -use std::{collections::HashMap, sync::Arc, time::Duration}; - -use anyhow::{anyhow, Context, Result}; -use regex::Regex; -use stratum_graph::types::NodeId; -use stratum_state::{PipelineRun, PipelineRunId, PipelineStatus, StateTracker, StepRecord}; -use tokio::task::JoinSet; -use tracing::{info, warn}; - -use crate::{ - dag, - error::ReflectionError, - mode::{ActionStep, Actor, ErrorStrategy, GuardSeverity, ReflectionMode}, -}; - -/// Context provided to a mode execution run. -pub struct RunContext { - /// Project identifier — used in NATS trigger subject. - pub project: String, - /// Parameter values substituted into step `cmd` fields via `{key}` - /// placeholders. - pub params: HashMap, - /// State tracker for recording pipeline run and step records. - pub state: Arc, - /// Optional NATS stream for publishing step completion events. - #[cfg(feature = "nats")] - pub nats: Option>, -} - -/// Result of executing a reflection mode. -#[derive(Debug)] -pub struct ModeRun { - pub mode_id: String, - pub run_id: PipelineRunId, - pub final_status: PipelineStatus, -} - -impl ReflectionMode { - /// Validate and execute this mode against the provided context. - /// Steps within each topological layer run concurrently via `JoinSet`. - /// Step failure behaviour is governed by `on_error.strategy`. - pub async fn execute(&self, ctx: &RunContext) -> Result { - self.validate() - .with_context(|| format!("mode '{}' failed pre-execution DAG validation", self.id))?; - - // ── Guards (Active Partner) ── - for guard in &self.guards { - let result = run_cmd(&guard.cmd).await; - match (&guard.severity, result) { - (_, Ok(())) => { - info!(guard = %guard.id, mode = %self.id, "guard passed"); - } - (GuardSeverity::Block, Err(e)) => { - return Err(anyhow!( - "mode '{}' blocked by guard '{}': {} — {}", - self.id, - guard.id, - guard.reason, - e - )); - } - (GuardSeverity::Warn, Err(_)) => { - warn!( - guard = %guard.id, - mode = %self.id, - reason = %guard.reason, - "guard warning — continuing execution" - ); - } - } - } - - let trigger_subject = format!("ecosystem.reflection.{}.{}", self.id, ctx.project); - let trigger_payload = serde_json::to_value(&ctx.params) - .context("serializing RunContext.params as trigger payload")?; - - let run = PipelineRun::new(trigger_subject, trigger_payload); - let run_id = run.id.clone(); - ctx.state - .create_run(&run) - .await - .context("creating PipelineRun in state tracker")?; - - // ── Step execution with convergence loop ── - let max_iterations = self - .converge - .as_ref() - .map_or(1, |c| (c.max_iterations.max(1) + 1) as usize); - - for iteration in 0..max_iterations { - let had_failure = run_all_layers( - &self.steps, - ctx, - &run_id, - &self.id, - &self.converge, - iteration, - max_iterations, - ) - .await?; - - if self.converge.is_none() { - break; - } - - let converge = self.converge.as_ref().expect("checked above"); - if iteration != 0 && !had_failure { - break; - } - - match check_convergence(converge, &self.id, iteration, max_iterations).await { - ConvergeOutcome::Converged => break, - ConvergeOutcome::Iterate => continue, - ConvergeOutcome::Exhausted => break, - } - } - - ctx.state - .update_status(&run_id, PipelineStatus::Success) - .await - .context("updating pipeline status to Success")?; - - info!(mode = %self.id, run = %run_id, "mode completed successfully"); - - Ok(ModeRun { - mode_id: self.id.clone(), - run_id, - final_status: PipelineStatus::Success, - }) - } -} - -enum ConvergeOutcome { - Converged, - Iterate, - Exhausted, -} - -async fn check_convergence( - converge: &crate::mode::Converge, - mode_id: &str, - iteration: usize, - max_iterations: usize, -) -> ConvergeOutcome { - match run_cmd(&converge.condition).await { - Ok(()) => { - info!(mode = %mode_id, iteration = iteration + 1, "convergence condition met"); - ConvergeOutcome::Converged - } - Err(_) if iteration + 1 < max_iterations => { - info!( - mode = %mode_id, - iteration = iteration + 1, - strategy = ?converge.strategy, - "convergence condition not met — iterating" - ); - ConvergeOutcome::Iterate - } - Err(e) => { - warn!( - mode = %mode_id, - "convergence condition not met after {} iterations: {e}", - iteration + 1 - ); - ConvergeOutcome::Exhausted - } - } -} - -async fn run_all_layers( - steps: &[ActionStep], - ctx: &RunContext, - run_id: &PipelineRunId, - mode_id: &str, - converge: &Option, - iteration: usize, - max_iterations: usize, -) -> Result { - let layers = dag::topological_layers(steps) - .with_context(|| format!("computing execution layers for mode '{mode_id}'"))?; - - let step_index: HashMap<&str, &ActionStep> = steps.iter().map(|s| (s.id.as_str(), s)).collect(); - - for layer in &layers { - let failure = run_layer(layer, &step_index, ctx, run_id, mode_id).await?; - - if let Some(e) = failure { - if converge.is_none() || iteration + 1 >= max_iterations { - ctx.state - .update_status(run_id, PipelineStatus::Failed) - .await - .context("updating pipeline status to Failed")?; - return Err(e); - } - warn!(mode = %mode_id, iteration = iteration + 1, "step failure — will retry"); - return Ok(true); - } - } - - Ok(false) -} - -/// Execute all steps in a layer concurrently. Returns the first fatal error, if -/// any. -async fn run_layer( - layer: &[String], - step_index: &HashMap<&str, &ActionStep>, - ctx: &RunContext, - run_id: &PipelineRunId, - mode_id: &str, -) -> Result> { - let mut set: JoinSet<(String, Result<()>)> = JoinSet::new(); - - for step_id in layer { - let step = (*step_index - .get(step_id.as_str()) - .expect("layer ids are derived from the validated step list")) - .clone(); - let owned_step_id = step.id.clone(); - let params = ctx.params.clone(); - let state = Arc::clone(&ctx.state); - let owned_run_id = run_id.clone(); - let owned_mode_id = mode_id.to_owned(); - let owned_project = ctx.project.clone(); - #[cfg(feature = "nats")] - let nats = ctx.nats.clone(); - - set.spawn(async move { - #[cfg(feature = "nats")] - let result = execute_step( - step, - params, - state, - owned_run_id, - owned_mode_id, - owned_project, - nats, - ) - .await; - #[cfg(not(feature = "nats"))] - let result = execute_step( - step, - params, - state, - owned_run_id, - owned_mode_id, - owned_project, - ) - .await; - (owned_step_id, result) - }); - } - - let mut fatal: Option = None; - - while let Some(join_result) = set.join_next().await { - let (step_id, result) = - join_result.map_err(|e| anyhow!(ReflectionError::TaskPanic(e.to_string())))?; - - if let Err(e) = result { - let step = step_index[step_id.as_str()]; - match step.on_error.strategy { - ErrorStrategy::Stop | ErrorStrategy::Retry => { - // Retry is exhausted inside execute_step; treat the final failure as Stop. - set.abort_all(); - fatal = Some(e); - break; - } - ErrorStrategy::Continue | ErrorStrategy::Fallback | ErrorStrategy::Branch => { - warn!( - step = %step_id, - strategy = ?step.on_error.strategy, - "step failed, continuing per on_error strategy: {e}" - ); - } - } - } - } - - Ok(fatal) -} - -/// Execute a single step: record start → run cmd (with retry) → record outcome. -/// All parameters are owned so this future is `'static` and can be spawned. -#[cfg_attr(not(feature = "nats"), allow(unused_variables))] -async fn execute_step( - step: ActionStep, - params: HashMap, - state: Arc, - run_id: PipelineRunId, - mode_id: String, - project: String, - #[cfg(feature = "nats")] nats: Option>, -) -> Result<()> { - let start_record = StepRecord::start(NodeId(step.id.clone())); - state - .record_step(&run_id, &start_record) - .await - .with_context(|| format!("recording step '{}' start", step.id))?; - - info!(step = %step.id, action = %step.action, actor = ?step.actor, "executing step"); - - let outcome = match step.actor { - Actor::Human => { - info!( - step = %step.id, - "step requires human action — skipping automated execution: {}", - step.note.as_deref().unwrap_or(&step.action) - ); - Ok(()) - } - Actor::Agent | Actor::Both => match &step.cmd { - None => { - info!(step = %step.id, "no cmd — step is documentation-only"); - Ok(()) - } - Some(cmd) => { - let resolved = substitute_params(cmd, ¶ms) - .with_context(|| format!("substituting params in step '{}' cmd", step.id))?; - run_with_retry(&resolved, &step.on_error).await - } - }, - }; - - match outcome { - Ok(()) => { - state - .record_step(&run_id, &start_record.succeed(vec![])) - .await - .with_context(|| format!("recording step '{}' success", step.id))?; - - #[cfg(feature = "nats")] - if let Some(ref nats) = nats { - publish_step_event(nats, &run_id, &step.id, true).await; - publish_kogral_capture(nats, &run_id, &mode_id, &project, &step.id, "success") - .await; - } - - Ok(()) - } - Err(e) => { - let err_str = e.to_string(); - state - .record_step(&run_id, &start_record.fail(err_str.clone())) - .await - .with_context(|| format!("recording step '{}' failure", step.id))?; - - #[cfg(feature = "nats")] - if let Some(ref nats) = nats { - publish_step_event(nats, &run_id, &step.id, false).await; - publish_kogral_capture(nats, &run_id, &mode_id, &project, &step.id, "failed").await; - } - - Err(anyhow!("step '{}' failed: {}", step.id, err_str)) - } - } -} - -/// Replace `{key}` placeholders in `cmd` with values from `params`. -/// Returns `Err(UnknownParams)` if any placeholder key is absent from `params`. -fn substitute_params( - cmd: &str, - params: &HashMap, -) -> Result { - let re = Regex::new(r"\{([^}]+)\}").expect("static regex is valid"); - - let unknown: Vec = re - .captures_iter(cmd) - .filter_map(|cap| { - let key = cap[1].to_string(); - if params.contains_key(&key) { - None - } else { - Some(key) - } - }) - .collect(); - - if !unknown.is_empty() { - return Err(ReflectionError::UnknownParams(unknown)); - } - - let mut result = cmd.to_owned(); - for (key, value) in params { - result = result.replace(&format!("{{{key}}}"), value); - } - - Ok(result) -} - -/// Execute `cmd` via `sh -c`, retrying on failure when `on_error.strategy == -/// Retry`. -async fn run_with_retry(cmd: &str, on_error: &crate::mode::OnError) -> Result<()> { - let max_attempts = if on_error.strategy == ErrorStrategy::Retry { - on_error.max.max(1) - } else { - 1 - }; - - for attempt in 0..max_attempts { - match run_cmd(cmd).await { - Ok(()) => return Ok(()), - Err(e) => { - if attempt + 1 < max_attempts { - let delay = Duration::from_secs(on_error.backoff_s); - warn!( - attempt = attempt + 1, - max = max_attempts, - "cmd failed, retrying in {delay:?}: {e}" - ); - tokio::time::sleep(delay).await; - } else { - return Err(e); - } - } - } - } - - // Defensive: reached only when max_attempts == 0, which is prevented by .max(1) - // above. - Err(anyhow!("retry loop exited without result")) -} - -pub(crate) async fn run_cmd(cmd: &str) -> Result<()> { - let output = tokio::process::Command::new("sh") - .arg("-c") - .arg(cmd) - .output() - .await - .with_context(|| format!("spawning sh -c: {cmd}"))?; - - if output.status.success() { - Ok(()) - } else { - let stderr = String::from_utf8_lossy(&output.stderr); - let stdout = String::from_utf8_lossy(&output.stdout); - Err(anyhow!( - "command exited {}: stderr='{}' stdout='{}'", - output.status, - stderr.trim(), - stdout.trim() - )) - } -} - -#[cfg(feature = "nats")] -async fn publish_step_event( - nats: &platform_nats::EventStream, - run_id: &PipelineRunId, - step_id: &str, - success: bool, -) { - let subject = if success { - "ecosystem.reflection.step.completed" - } else { - "ecosystem.reflection.step.failed" - }; - - let payload = serde_json::json!({ - "run_id": run_id.0.to_string(), - "step_id": step_id, - "success": success, - }); - - match serde_json::to_vec(&payload) { - Ok(bytes) => { - if let Err(e) = nats.publish(subject, bytes::Bytes::from(bytes)).await { - warn!(step = %step_id, "failed to publish step event to '{subject}': {e}"); - } - } - Err(e) => { - warn!(step = %step_id, "failed to serialize step event: {e}"); - } - } -} - -/// Publish to `ecosystem.kogral.capture` so Kogral can record this step as an -/// Execution node in the shared knowledge graph. Matches the KogralCapture -/// payload contract defined in `nats/subjects.ncl`. -#[cfg(feature = "nats")] -async fn publish_kogral_capture( - nats: &platform_nats::EventStream, - run_id: &PipelineRunId, - mode_id: &str, - project: &str, - step_id: &str, - status: &str, -) { - const SUBJECT: &str = "ecosystem.kogral.capture"; - - let payload = serde_json::json!({ - "project": project, - "mode_id": mode_id, - "step_id": step_id, - "run_id": run_id.0.to_string(), - "action": step_id, - "status": status, - "context": {}, - }); - - match serde_json::to_vec(&payload) { - Ok(bytes) => { - if let Err(e) = nats.publish(SUBJECT, bytes::Bytes::from(bytes)).await { - warn!(step = %step_id, "failed to publish kogral capture to '{SUBJECT}': {e}"); - } - } - Err(e) => { - warn!(step = %step_id, "failed to serialize kogral capture payload: {e}"); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn substitute_replaces_known_params() { - let mut params = HashMap::new(); - params.insert("project_name".to_string(), "my-service".to_string()); - params.insert("project_dir".to_string(), "/tmp/my-service".to_string()); - - let result = - substitute_params("git -C {project_dir} init && echo {project_name}", ¶ms).unwrap(); - assert_eq!(result, "git -C /tmp/my-service init && echo my-service"); - } - - #[test] - fn substitute_rejects_unknown_params() { - let params = HashMap::new(); - let err = substitute_params("echo {unknown_key}", ¶ms).unwrap_err(); - assert!(matches!( - err, - ReflectionError::UnknownParams(keys) if keys.contains(&"unknown_key".to_string()) - )); - } - - #[test] - fn substitute_no_placeholders_is_identity() { - let result = substitute_params("ls -la /tmp", &HashMap::new()).unwrap(); - assert_eq!(result, "ls -la /tmp"); - } - - #[tokio::test] - async fn run_cmd_success() { - run_cmd("true").await.unwrap(); - } - - #[tokio::test] - async fn run_cmd_failure_returns_err() { - let err = run_cmd("false").await.unwrap_err(); - assert!(err.to_string().contains("command exited")); - } -} diff --git a/crates/ontoref-reflection/src/lib.rs b/crates/ontoref-reflection/src/lib.rs deleted file mode 100644 index 7581b5d..0000000 --- a/crates/ontoref-reflection/src/lib.rs +++ /dev/null @@ -1,22 +0,0 @@ -//! Load, validate, and execute Reflection modes as NCL DAG contracts. -//! -//! A [`mode::ReflectionMode`] is a typed DAG of [`mode::ActionStep`]s with -//! explicit actor assignments, dependency edges, and error strategies. The -//! [`executor`] validates the DAG contract before executing any step, ensuring -//! that declared `depends_on` ordering and actor policies are respected. -//! -//! Depends on `stratum-graph` and `stratum-state` for DAG traversal and FSM -//! state tracking. The `nats` feature gates the `nats` module and event -//! publishing. - -pub mod dag; -pub mod error; -pub mod executor; -pub mod mode; - -pub use error::ReflectionError; -pub use executor::{ModeRun, RunContext}; -pub use mode::{ - ActionStep, Actor, Converge, ConvergeStrategy, Dependency, DependencyKind, ErrorStrategy, - Guard, GuardSeverity, OnError, ReflectionMode, -}; diff --git a/crates/ontoref-reflection/src/mode.rs b/crates/ontoref-reflection/src/mode.rs deleted file mode 100644 index c7378bd..0000000 --- a/crates/ontoref-reflection/src/mode.rs +++ /dev/null @@ -1,213 +0,0 @@ -use std::path::Path; - -use anyhow::{Context, Result}; -use serde::{Deserialize, Serialize}; - -use crate::dag; -use crate::error::ReflectionError; - -/// A reflection mode loaded from a Nickel `.ncl` file. -/// Corresponds to the `Mode String` contract in `reflection/schema.ncl`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ReflectionMode { - pub id: String, - pub trigger: String, - #[serde(default)] - pub preconditions: Vec, - #[serde(default)] - pub guards: Vec, - pub steps: Vec, - #[serde(default)] - pub postconditions: Vec, - pub converge: Option, -} - -/// Pre-flight check executed before any step. -/// Block severity aborts mode execution; Warn prints a message and continues. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Guard { - pub id: String, - pub cmd: String, - pub reason: String, - #[serde(default)] - pub severity: GuardSeverity, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] -pub enum GuardSeverity { - #[default] - Block, - Warn, -} - -/// Post-execution convergence loop. -/// After all steps complete, evaluates `condition` — if non-zero exit, -/// re-executes steps up to `max_iterations` times. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Converge { - pub condition: String, - #[serde(default = "Converge::default_max_iterations")] - pub max_iterations: u32, - #[serde(default)] - pub strategy: ConvergeStrategy, -} - -impl Converge { - fn default_max_iterations() -> u32 { - 3 - } -} - -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] -pub enum ConvergeStrategy { - #[default] - RetryFailed, - RetryAll, -} - -impl ReflectionMode { - /// Load a `ReflectionMode` from a Nickel file by invoking `nickel export - /// --format json`. Mirrors the pattern in - /// `stratum-orchestrator::graph::loader::load_node_from_ncl`. - pub fn load(mode_file: &Path) -> Result { - let output = std::process::Command::new("nickel") - .arg("export") - .arg("--format") - .arg("json") - .arg(mode_file) - .output() - .with_context(|| format!("running nickel export on '{}'", mode_file.display()))?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr).trim().to_owned(); - return Err(ReflectionError::NickelExport { - path: mode_file.display().to_string(), - stderr, - } - .into()); - } - - serde_json::from_slice::(&output.stdout) - .map_err(ReflectionError::ParseMode) - .with_context(|| { - format!( - "parsing ReflectionMode from '{}' — confirm the .ncl exports `| (s.Mode \ - String)`", - mode_file.display() - ) - }) - } - - /// Validate the step DAG in Rust (uniqueness + referential integrity + - /// cycle detection). Complements the Nickel-side structural + - /// referential checks; adds cycle detection that Nickel cannot express. - pub fn validate(&self) -> Result<(), ReflectionError> { - dag::validate(&self.steps) - } - - /// Return steps in a flat topological order (one valid sequencing, no - /// parallelism). Useful for dry-run display. For parallel execution, - /// use `dag::topological_layers`. - pub fn execution_order(&self) -> Result, ReflectionError> { - let layers = dag::topological_layers(&self.steps)?; - Ok(layers - .into_iter() - .flat_map(|layer| { - layer - .into_iter() - .filter_map(|id| self.steps.iter().find(|s| s.id == id)) - }) - .collect()) - } -} - -/// A single executable step within a mode. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ActionStep { - pub id: String, - /// Semantic action name (string identifier in cross-project modes). - pub action: String, - #[serde(default)] - pub actor: Actor, - /// Shell command with optional `{param}` placeholders substituted from - /// `RunContext.params`. - pub cmd: Option, - #[serde(default)] - pub depends_on: Vec, - #[serde(default)] - pub on_error: OnError, - pub verify: Option, - pub note: Option, -} - -/// Who executes this step. -/// Nickel enum tags serialize with their exact case: `'Human` → `"Human"`. -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] -pub enum Actor { - Human, - Agent, - #[default] - Both, -} - -/// Dependency edge: this step waits for another step based on `kind`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Dependency { - /// ID of the step this step depends on. - pub step: String, - #[serde(default)] - pub kind: DependencyKind, - pub condition: Option, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] -pub enum DependencyKind { - #[default] - Always, - OnSuccess, - OnFailure, -} - -/// Error handling strategy for a step. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct OnError { - pub strategy: ErrorStrategy, - pub target: Option, - pub on_success: Option, - #[serde(default = "OnError::default_max")] - pub max: u32, - /// Backoff in seconds between retry attempts. - #[serde(default = "OnError::default_backoff_s")] - pub backoff_s: u64, -} - -impl OnError { - fn default_max() -> u32 { - 3 - } - - fn default_backoff_s() -> u64 { - 5 - } -} - -impl Default for OnError { - fn default() -> Self { - Self { - strategy: ErrorStrategy::Stop, - target: None, - on_success: None, - max: Self::default_max(), - backoff_s: Self::default_backoff_s(), - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub enum ErrorStrategy { - Stop, - Continue, - Retry, - Fallback, - Branch, -} diff --git a/ontology/schemas/manifest.ncl b/ontology/schemas/manifest.ncl index 3da9185..26d1ecd 100644 --- a/ontology/schemas/manifest.ncl +++ b/ontology/schemas/manifest.ncl @@ -53,10 +53,35 @@ let audit_level_type = [| # A layer is a named region of the repo with visibility rules per mode. # The `committed` flag distinguishes product (true) from process (false). +# What a region is FOR, in the ADR-020 three-layer sense. This is the export +# axis — the `pub` of an ontoref instance: +# +# 'SelfManagement Layer 1. The project describing itself: ontology, ADRs, +# reflection, its own positioning. Never travels to a +# consumer. Mandatory — every instance has one. +# 'IntegrationSurface Layer 2. What OTHER projects bind to: domain artifacts, +# protocol schemas, adoption templates. Optional, and +# biconditional with manifest.registry_provides. +# 'Internal Neither. Implementation the consumer reaches only +# through a released binary or artifact (crates, install +# tooling, session process files). +# +# Layer 3 (caller-side cabling) has no region here BY CONSTRUCTION — it lives in +# the caller's repo, which is the whole point of the layer. +# +# Default is 'Internal so the field is additive: an existing manifest keeps +# validating, and a project opts into the export axis by naming its layers. +let layer_kind_type = [| + 'SelfManagement, + 'IntegrationSurface, + 'Internal, +|] in + let layer_type = { id | String, paths | Array String, committed | Bool, + kind | layer_kind_type | default = 'Internal, description | String | default = "", } in