fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workflows): M1 — schema + definition validation + declarative reconcile

Browse Source 
First milestone of the v1.2 Workflows track (blueprint §9.1/§9.2): the durable
DAG engine's foundation — the definition model, its validation, and the
declarative `apply` reconcile path. No execution yet (the orchestrator is M2).

- Migration 0071_workflows: `workflows` (owner-polymorphic like scripts, app-
  owned in M1), `workflow_runs`, and `workflow_run_steps` (the per-step
  competing-consumer lease table the M2 orchestrator will claim). Schema golden
  reblessed.
- `shared::workflow`: the `WorkflowDefinition` / `WorkflowStepDef` DTOs (shared
  by the CLI, apply, and the future orchestrator) + run/step status enums.
- Pure, DB-free `workflow_template` (input `{{ input.x }}` / `{{ steps.a.output.y }}`
  resolution, type-preserving) and `workflow_expr` (a small safe JSON-predicate
  evaluator for `when` — keeps manager-core free of a scripting engine).
- `workflow_repo`: read trait + reconcile tx free-fns (insert/update/delete).
- apply_service: `BundleWorkflow` + `Plan.workflows` + `CurrentState.workflows`
  + `ApplyReport.workflows_*`; server-side `validate_workflow_definition`
  (unique/acyclic steps via topological sort, deps exist, function XOR workflow,
  `when`/template parse) rejecting on a group node; `diff_workflows` by
  lower(name) (Update on a definition change) + in-tx reconcile.
- CLI: `[[workflows]]` + `[[workflows.steps]]` manifest structs → wire bundle;
  plan/apply render + report counts.
- Tests: 16 manager-core lib tests (template, expr, definition validation) +
  the `workflows` CLI journey (apply → NoOp → prune; cyclic DAG rejected).

Deferred to later milestones: the orchestrator worker (M2), conditional/template
runtime wiring (M3), nested sub-workflows (M4), the SDK/API/CLI run surface
(M5), the dashboard (M6). The `group_id` column + run/step tables ship now so
those slot in without a migration churn.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-12 16:46:32 +02:00
parent c8c4f012ff
commit 44f992cbb0
18 changed files with 2059 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

345
crates/manager-core/src/apply_service.rs
View File

@@ -103,6 +103,20 @@ pub struct Bundle {
/// declines (the binding 404s instead of serving). App-only.
#[serde(default)]
pub suppress_routes: Vec<String>,
/// v1.2 Workflows: declared DAG workflow definitions. App-owned (M1); a
/// `[group]` carrying workflows is rejected in `validate_bundle_for`.
#[serde(default)]
pub workflows: Vec<BundleWorkflow>,
}
/// One declared workflow on the wire: a name + its (already-parsed) DAG.
#[derive(Debug, Clone, Deserialize)]
pub struct BundleWorkflow {
pub name: String,
pub definition: picloud_shared::workflow::WorkflowDefinition,
/// Three-state lifecycle (§4.3); omitted ⇒ active.
#[serde(default = "picloud_shared::default_true")]
pub enabled: bool,
}
/// One declared shared-collection marker on the wire: a name + its store kind.
@@ -450,6 +464,9 @@ pub struct Plan {
/// §11 tail per-app suppression markers, keyed `"{target_kind}:{reference}"`.
#[serde(default)]
pub suppressions: Vec<ResourceChange>,
/// v1.2 Workflows: workflow definitions, keyed by name.
#[serde(default)]
pub workflows: Vec<ResourceChange>,
}
impl Plan {
@@ -464,6 +481,8 @@ impl Plan {
.chain(&self.vars)
.chain(&self.extension_points)
.chain(&self.collections)
.chain(&self.suppressions)
.chain(&self.workflows)
.all(|c| c.op == Op::NoOp)
}
}
@@ -744,6 +763,8 @@ pub struct CurrentState {
/// §11 tail per-app suppression markers at this node (app-only), as
/// `(target_kind, reference)` pairs.
pub suppressions: Vec<(String, String)>,
/// v1.2 Workflows owned directly by this node (app-owned; empty for a group).
pub workflows: Vec<crate::workflow_repo::Workflow>,
}
/// One row of the read-only extension-point report (§5.5).
@@ -1125,6 +1146,20 @@ impl ApplyService {
// subtree. Both are inheritance-only (the dispatch filters gate to
// group-owned templates on the suppressing owner's chain), so no
// owner-kind guard here.
//
// v1.2 Workflows: app-owned only (M1). Reject on a group node (the
// `group_id` column ships unused). Then structurally validate each
// definition (unique/acyclic steps, deps exist, when/template parse).
if is_group && !bundle.workflows.is_empty() {
return Err(ApplyError::Invalid(
"a group cannot declare workflows — they are app-owned (group-owned \
workflow templates are a future addition)"
.into(),
));
}
for w in &bundle.workflows {
validate_workflow_definition(&w.name, &w.definition).map_err(ApplyError::Invalid)?;
}
self.validate_bundle(bundle, inherited_endpoints)
}
@@ -1568,6 +1603,62 @@ impl ApplyService {
}
}
}
// v1.2 Workflows (app-owned in M1). Create/Update always; Delete only
// under --prune (mirrors scripts). Function / sub-workflow references
// are resolved at RUN time, so there is no in-tx ordering dependency on
// the scripts applied above.
if let ApplyOwner::App(app_id) = owner {
let cur_by_name: HashMap<String, &crate::workflow_repo::Workflow> = current
.workflows
.iter()
.map(|w| (w.name.to_lowercase(), w))
.collect();
let desired_by_name: HashMap<String, &BundleWorkflow> = bundle
.workflows
.iter()
.map(|w| (w.name.to_lowercase(), w))
.collect();
for ch in &plan.workflows {
let key = ch.key.to_lowercase();
match ch.op {
Op::Create => {
let w = desired_by_name[&key];
crate::workflow_repo::insert_workflow_tx(
&mut *tx,
app_id,
&w.name,
&w.definition,
w.enabled,
)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
report.workflows_created += 1;
}
Op::Update => {
let w = desired_by_name[&key];
let cur = cur_by_name[&key];
crate::workflow_repo::update_workflow_tx(
&mut *tx,
cur.id,
&w.definition,
w.enabled,
)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
report.workflows_updated += 1;
}
Op::Delete if prune => {
let cur = cur_by_name[&key];
crate::workflow_repo::delete_workflow_tx(&mut *tx, cur.id)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
report.workflows_deleted += 1;
}
Op::NoOp | Op::Delete => {}
}
}
}
Ok(name_to_id)
}
@@ -3934,6 +4025,15 @@ impl ApplyService {
crate::suppression_repo::list_for_owner(&self.pool, owner.as_script_owner())
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
// v1.2 Workflows are app-owned (M1); a group node owns none.
let workflows = match owner {
ApplyOwner::App(app_id) => {
crate::workflow_repo::list_workflows_for_app(&self.pool, app_id)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?
}
ApplyOwner::Group(_) => Vec::new(),
};
Ok(CurrentState {
scripts,
routes,
@@ -3943,6 +4043,7 @@ impl ApplyService {
extension_point_names,
collections,
suppressions,
workflows,
})
}
@@ -4008,9 +4109,160 @@ fn compute_diff_with_names(
extension_points: diff_extension_points(current, bundle),
collections: diff_collections(current, bundle),
suppressions: diff_suppressions(current, bundle),
workflows: diff_workflows(current, bundle),
}
}
/// Diff workflows by `lower(name)`. Like scripts, a workflow has a stable name
/// identity with a mutable body, so a changed definition (or `enabled`) is an
/// `Update`, not a delete+create. Live workflows absent from the manifest are
/// `Delete` (applied only under `--prune`).
fn diff_workflows(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
let by_name: HashMap<String, &crate::workflow_repo::Workflow> = current
.workflows
.iter()
.map(|w| (w.name.to_lowercase(), w))
.collect();
let desired: HashSet<String> = bundle
.workflows
.iter()
.map(|w| w.name.to_lowercase())
.collect();
let mut out = Vec::new();
for w in &bundle.workflows {
match by_name.get(&w.name.to_lowercase()) {
None => out.push(ResourceChange {
op: Op::Create,
key: w.name.clone(),
detail: None,
}),
Some(cur) => {
let reason = if cur.definition != w.definition {
Some("definition changed".to_string())
} else if cur.enabled != w.enabled {
Some(if w.enabled { "enabled" } else { "disabled" }.to_string())
} else {
None
};
out.push(ResourceChange {
op: if reason.is_some() {
Op::Update
} else {
Op::NoOp
},
key: w.name.clone(),
detail: reason,
});
}
}
}
for w in &current.workflows {
if !desired.contains(&w.name.to_lowercase()) {
out.push(ResourceChange {
op: Op::Delete,
key: w.name.clone(),
detail: None,
});
}
}
out
}
/// Pure structural validation of a workflow definition (mirrors
/// `validate_trigger_shape`): non-empty; unique step names; each step targets
/// exactly one of function/workflow; `depends_on` references exist and aren't
/// self-loops; the DAG is acyclic; every `when` + input template parses and
/// only references declared steps. Returns a human error string on failure.
fn validate_workflow_definition(
wf_name: &str,
def: &picloud_shared::workflow::WorkflowDefinition,
) -> Result<(), String> {
use std::collections::{BTreeSet, HashMap as Map};
if def.steps.is_empty() {
return Err(format!("workflow `{wf_name}` has no steps"));
}
let mut names: BTreeSet<String> = BTreeSet::new();
for s in &def.steps {
if s.name.trim().is_empty() {
return Err(format!(
"workflow `{wf_name}` has a step with an empty name"
));
}
if !names.insert(s.name.clone()) {
return Err(format!(
"workflow `{wf_name}` has a duplicate step name `{}`",
s.name
));
}
}
for s in &def.steps {
if s.target().is_none() {
return Err(format!(
"workflow `{wf_name}` step `{}` must set exactly one of `function` or `workflow`",
s.name
));
}
for dep in &s.depends_on {
if dep == &s.name {
return Err(format!(
"workflow `{wf_name}` step `{}` depends on itself",
s.name
));
}
if !names.contains(dep) {
return Err(format!(
"workflow `{wf_name}` step `{}` depends on undeclared step `{dep}`",
s.name
));
}
}
if let Some(w) = &s.when {
crate::workflow_expr::validate(w, &names).map_err(|e| {
format!(
"workflow `{wf_name}` step `{}` has a bad `when`: {e}",
s.name
)
})?;
}
crate::workflow_template::validate(&s.input, &names).map_err(|e| {
format!(
"workflow `{wf_name}` step `{}` has a bad input template: {e}",
s.name
)
})?;
}
// DAG acyclicity via Kahn topological sort.
let mut indeg: Map<&str, usize> = def.steps.iter().map(|s| (s.name.as_str(), 0)).collect();
for s in &def.steps {
for _dep in &s.depends_on {
*indeg.get_mut(s.name.as_str()).unwrap() += 1;
}
}
let mut queue: Vec<&str> = indeg
.iter()
.filter(|(_, d)| **d == 0)
.map(|(n, _)| *n)
.collect();
let mut seen = 0usize;
while let Some(n) = queue.pop() {
seen += 1;
for s in &def.steps {
if s.depends_on.iter().any(|d| d == n) {
let e = indeg.get_mut(s.name.as_str()).unwrap();
*e -= 1;
if *e == 0 {
queue.push(s.name.as_str());
}
}
}
}
if seen != def.steps.len() {
return Err(format!("workflow `{wf_name}` has a dependency cycle"));
}
Ok(())
}
/// Diff app-owned vars by key. Unlike secrets (whose values live out-of-band),
/// a var's value IS in the manifest, so equality is value-sensitive: a changed
/// value is an `Update`. Live vars absent from the manifest are `Delete`
@@ -5140,6 +5392,12 @@ pub struct ApplyReport {
pub suppressions_created: u32,
#[serde(default)]
pub suppressions_deleted: u32,
#[serde(default)]
pub workflows_created: u32,
#[serde(default)]
pub workflows_updated: u32,
#[serde(default)]
pub workflows_deleted: u32,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub warnings: Vec<String>,
}
@@ -5483,6 +5741,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
};
let plan = compute_diff(&current, &bundle);
assert_eq!(plan.scripts.len(), 1);
@@ -5509,6 +5768,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
};
let plan = compute_diff(&current, &bundle);
assert!(plan.is_noop(), "expected all no-op, got {plan:?}");
@@ -5530,6 +5790,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
};
let plan = compute_diff(&current, &bundle);
assert_eq!(plan.scripts[0].op, Op::Update);
@@ -5552,6 +5813,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
};
let plan = compute_diff(&current, &bundle);
assert_eq!(plan.scripts[0].op, Op::Delete);
@@ -5605,6 +5867,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
};
let plan = compute_diff(&current, &bundle);
assert_eq!(plan.routes[0].op, Op::Update);
@@ -5621,6 +5884,7 @@ mod tests {
collections: Vec::new(),
suppress_triggers: Vec::new(),
suppress_routes: Vec::new(),
workflows: Vec::new(),
}
}
@@ -6350,4 +6614,85 @@ mod tests {
assert!(!bare.confirm_required("production"));
assert!(bare.gated_envs().is_empty());
}
// ---- v1.2 Workflows: definition validation ----------------------------
fn wf_def(steps: serde_json::Value) -> picloud_shared::workflow::WorkflowDefinition {
serde_json::from_value(serde_json::json!({ "steps": steps })).expect("valid def json")
}
#[test]
fn workflow_validation_accepts_a_valid_dag() {
let def = wf_def(serde_json::json!([
{ "name": "a", "function": "fa" },
{ "name": "b", "function": "fb", "depends_on": ["a"],
"when": "steps.a.output.ok == true",
"input": { "x": "{{ steps.a.output.value }}" } },
{ "name": "c", "workflow": "sub", "depends_on": ["a", "b"] },
]));
assert!(validate_workflow_definition("w", &def).is_ok());
}
#[test]
fn workflow_validation_rejects_empty_and_dupes() {
assert!(validate_workflow_definition("w", &wf_def(serde_json::json!([]))).is_err());
let dupe = wf_def(serde_json::json!([
{ "name": "a", "function": "fa" },
{ "name": "a", "function": "fb" },
]));
assert!(validate_workflow_definition("w", &dupe)
.unwrap_err()
.contains("duplicate step name"));
}
#[test]
fn workflow_validation_rejects_bad_target_and_deps() {
// neither function nor workflow
let no_target = wf_def(serde_json::json!([{ "name": "a" }]));
assert!(validate_workflow_definition("w", &no_target)
.unwrap_err()
.contains("exactly one"));
// both function and workflow
let both = wf_def(serde_json::json!([{ "name": "a", "function": "f", "workflow": "w" }]));
assert!(validate_workflow_definition("w", &both).is_err());
// dependency on an undeclared step
let bad_dep = wf_def(serde_json::json!([
{ "name": "a", "function": "fa", "depends_on": ["ghost"] },
]));
assert!(validate_workflow_definition("w", &bad_dep)
.unwrap_err()
.contains("undeclared step"));
// self-dependency
let self_dep = wf_def(serde_json::json!([
{ "name": "a", "function": "fa", "depends_on": ["a"] },
]));
assert!(validate_workflow_definition("w", &self_dep)
.unwrap_err()
.contains("depends on itself"));
}
#[test]
fn workflow_validation_detects_cycles() {
let cyclic = wf_def(serde_json::json!([
{ "name": "a", "function": "fa", "depends_on": ["c"] },
{ "name": "b", "function": "fb", "depends_on": ["a"] },
{ "name": "c", "function": "fc", "depends_on": ["b"] },
]));
assert!(validate_workflow_definition("w", &cyclic)
.unwrap_err()
.contains("cycle"));
}
#[test]
fn workflow_validation_rejects_bad_when_and_template() {
let bad_when = wf_def(serde_json::json!([
{ "name": "a", "function": "fa" },
{ "name": "b", "function": "fb", "depends_on": ["a"], "when": "steps.ghost.output.x" },
]));
assert!(validate_workflow_definition("w", &bad_when).is_err());
let bad_tmpl = wf_def(serde_json::json!([
{ "name": "a", "function": "fa", "input": { "x": "{{ steps.ghost.output }}" } },
]));
assert!(validate_workflow_definition("w", &bad_tmpl).is_err());
}
}