schema: layer_type gains kind — the export axis of an ontoref instance (ADR-020)
'SelfManagement | 'IntegrationSurface | 'Internal, defaulting to 'Internal so a manifest that never mentions it keeps validating. Additive but ORDERED: Nickel record contracts are closed, so a consumer declaring kind against a data-dir schema predating this field fails with `extra field kind`. Reinstall precedes tagging. ontoref did not catch this on its own — its manifest imports the schema source by relative path while every consumer imports the installed copy, so the self-hosting project is the one project that structurally cannot feel this class of break.
This commit is contained in:
Jesús Pérez
parent
c2111b9189
commit
2949efb8fd
GPG key ID:
9F243E355E0BC939
7 changed files with 25 additions and 1046 deletions
|
|
@ -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" }
|
||||
|
|
@ -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<Vec<Vec<String>>, 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<String>> = 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<String> = 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());
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String>),
|
||||
|
||||
#[error("cycle detected in step dependency graph")]
|
||||
CycleDetected,
|
||||
|
||||
#[error("unknown parameter placeholders in cmd (not in RunContext.params): {}", .0.join(", "))]
|
||||
UnknownParams(Vec<String>),
|
||||
|
||||
#[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),
|
||||
}
|
||||
|
|
@ -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<String, String>,
|
||||
/// State tracker for recording pipeline run and step records.
|
||||
pub state: Arc<dyn StateTracker>,
|
||||
/// Optional NATS stream for publishing step completion events.
|
||||
#[cfg(feature = "nats")]
|
||||
pub nats: Option<Arc<platform_nats::EventStream>>,
|
||||
}
|
||||
|
||||
/// 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<ModeRun> {
|
||||
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<crate::mode::Converge>,
|
||||
iteration: usize,
|
||||
max_iterations: usize,
|
||||
) -> Result<bool> {
|
||||
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<Option<anyhow::Error>> {
|
||||
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<anyhow::Error> = 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<String, String>,
|
||||
state: Arc<dyn StateTracker>,
|
||||
run_id: PipelineRunId,
|
||||
mode_id: String,
|
||||
project: String,
|
||||
#[cfg(feature = "nats")] nats: Option<Arc<platform_nats::EventStream>>,
|
||||
) -> 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<String, String>,
|
||||
) -> Result<String, ReflectionError> {
|
||||
let re = Regex::new(r"\{([^}]+)\}").expect("static regex is valid");
|
||||
|
||||
let unknown: Vec<String> = 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"));
|
||||
}
|
||||
}
|
||||
|
|
@ -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,
|
||||
};
|
||||
|
|
@ -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<String>,
|
||||
#[serde(default)]
|
||||
pub guards: Vec<Guard>,
|
||||
pub steps: Vec<ActionStep>,
|
||||
#[serde(default)]
|
||||
pub postconditions: Vec<String>,
|
||||
pub converge: Option<Converge>,
|
||||
}
|
||||
|
||||
/// 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<Self> {
|
||||
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::<Self>(&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<Vec<&ActionStep>, 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<String>,
|
||||
#[serde(default)]
|
||||
pub depends_on: Vec<Dependency>,
|
||||
#[serde(default)]
|
||||
pub on_error: OnError,
|
||||
pub verify: Option<String>,
|
||||
pub note: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
#[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<String>,
|
||||
pub on_success: Option<String>,
|
||||
#[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,
|
||||
}
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue