diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index 8bea2a5..a8562c0 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -3616,6 +3616,7 @@ impl ApplyService { warnings.extend(unreachable_endpoint_warnings(bundle)); } warnings.extend(self.dangling_suppress_warnings(owner, bundle).await?); + warnings.extend(workflow_nesting_warnings(bundle)); Ok(warnings) } @@ -5280,6 +5281,64 @@ fn disabled_target_warnings(bundle: &Bundle) -> Vec { out } +/// M4 soft-warn: a `workflow`-kind step that (directly or transitively, within +/// this bundle) nests back into its own workflow forms a cycle. This is not an +/// error — the runtime depth ceiling (`PICLOUD_WORKFLOW_MAX_DEPTH`) bounds it — +/// but it's almost always a mistake, so we surface it at plan time. Only edges +/// to workflows present in the same bundle are considered (a reference to an +/// external workflow can't be resolved here). +fn workflow_nesting_warnings(bundle: &Bundle) -> Vec { + use std::collections::{BTreeSet, HashMap, HashSet}; + if bundle.workflows.is_empty() { + return Vec::new(); + } + let names: HashSet = bundle + .workflows + .iter() + .map(|w| w.name.to_lowercase()) + .collect(); + // from → the set of in-bundle workflows it nests into. + let mut edges: HashMap> = HashMap::new(); + for w in &bundle.workflows { + let from = w.name.to_lowercase(); + for s in &w.definition.steps { + if let Some(t) = s.workflow.as_deref() { + let t = t.to_lowercase(); + if names.contains(&t) { + edges.entry(from.clone()).or_default().insert(t); + } + } + } + } + let mut out = Vec::new(); + for w in &bundle.workflows { + let start = w.name.to_lowercase(); + // Reachability from `start`: if we can get back to `start`, it's cyclic. + let mut seen: HashSet = HashSet::new(); + let mut stack: Vec = edges.get(&start).into_iter().flatten().cloned().collect(); + let mut cyclic = false; + while let Some(n) = stack.pop() { + if n == start { + cyclic = true; + break; + } + if seen.insert(n.clone()) { + if let Some(next) = edges.get(&n) { + stack.extend(next.iter().cloned()); + } + } + } + if cyclic { + out.push(format!( + "workflow `{}` can nest into itself (a sub-workflow cycle) — runs are \ + bounded by PICLOUD_WORKFLOW_MAX_DEPTH, not infinite, but this is likely a bug", + w.name + )); + } + } + out +} + /// Per-kind structural validation for one desired trigger — kept at parity /// with the interactive trigger API so `apply` can't write a trigger the /// dashboard would have rejected. Pure (no live state), so it's unit-tested @@ -6695,4 +6754,47 @@ mod tests { ])); assert!(validate_workflow_definition("w", &bad_tmpl).is_err()); } + + fn bundle_workflow(name: &str, steps: serde_json::Value) -> BundleWorkflow { + BundleWorkflow { + name: name.into(), + definition: wf_def(steps), + enabled: true, + } + } + + #[test] + fn workflow_nesting_warning_flags_self_and_mutual_cycles() { + // Direct self-nest. + let mut b = empty_bundle(); + b.workflows = vec![bundle_workflow( + "loop", + serde_json::json!([{ "name": "s", "workflow": "loop" }]), + )]; + let w = workflow_nesting_warnings(&b); + assert_eq!(w.len(), 1); + assert!(w[0].contains("loop") && w[0].contains("cycle")); + + // Mutual cycle w1 -> w2 -> w1. + let mut b = empty_bundle(); + b.workflows = vec![ + bundle_workflow("w1", serde_json::json!([{ "name": "s", "workflow": "w2" }])), + bundle_workflow("w2", serde_json::json!([{ "name": "s", "workflow": "w1" }])), + ]; + assert_eq!(workflow_nesting_warnings(&b).len(), 2); + + // Acyclic nesting (parent -> child, child is a leaf) → no warning. + let mut b = empty_bundle(); + b.workflows = vec![ + bundle_workflow( + "parent", + serde_json::json!([{ "name": "s", "workflow": "child" }]), + ), + bundle_workflow( + "child", + serde_json::json!([{ "name": "s", "function": "f" }]), + ), + ]; + assert!(workflow_nesting_warnings(&b).is_empty()); + } } diff --git a/crates/manager-core/src/workflow_orchestrator.rs b/crates/manager-core/src/workflow_orchestrator.rs index 41e80e9..db760a9 100644 --- a/crates/manager-core/src/workflow_orchestrator.rs +++ b/crates/manager-core/src/workflow_orchestrator.rs @@ -27,9 +27,13 @@ //! A second, slower cadence reclaims steps leased by a crashed worker //! ([`workflow_repo::reclaim_stale_steps`]) — the durability safety net. //! -//! M2 executes **function** steps only; `workflow`-kind (nested sub-workflow) -//! steps land in M4, and `when` conditions + input templating in M3. The seams -//! are present (the step target enum, `run_input`, `workflow_depth`). +//! Each claimed step is *prepared* ([`Prep`]) — the run context is built, its +//! `when` condition evaluated (false → `skipped`, M3), and its `input` template +//! resolved (M3) — then dispatched by target: a **function** runs through the +//! executor; a **`workflow`** starts a nested child run and parks the parent +//! step on it ([`workflow_repo::start_child_and_park`], M4), resumed by +//! [`workflow_repo::resume_finished_children`] once the child terminates. Nested +//! runs are bounded by [`WorkflowOrchestratorConfig::max_depth`]. //! //! [`cron_scheduler`]: crate::cron_scheduler @@ -42,19 +46,41 @@ use tokio::sync::OwnedSemaphorePermit; use picloud_executor_core::{build_execution_log, ExecRequest, InvocationType}; use picloud_orchestrator_core::{ExecutionGate, ExecutorClient, ScriptIdentity}; -use picloud_shared::workflow::{StepTarget, WorkflowBackoff, WorkflowStepDef}; -use picloud_shared::{ExecutionId, ExecutionLogSink, ExecutionSource, RequestId}; +use picloud_shared::workflow::{StepTarget, WorkflowBackoff, WorkflowDefinition, WorkflowStepDef}; +use picloud_shared::{ + ExecutionId, ExecutionLogSink, ExecutionSource, RequestId, Script, WorkflowId, +}; use crate::dispatcher::compute_backoff; use crate::repo::ScriptRepository; use crate::trigger_config::BackoffShape; use crate::workflow_expr; use crate::workflow_repo::{ - claim_ready_step, complete_step_and_advance, load_run_step_outputs, reclaim_stale_steps, - ClaimedStep, StepOutcome, + claim_ready_step, complete_step_and_advance, get_workflow_by_name, load_run_step_outputs, + reclaim_stale_steps, resume_finished_children, start_child_and_park, ClaimedStep, StepOutcome, }; use crate::workflow_template::{self, RunContext}; +/// The prepared action for a claimed step, after context build + `when` + input +/// resolution + target lookup. The orchestrator dispatches on this. +enum Prep { + /// `when` evaluated false — skip the step (never executed). + Skip, + /// A resolution error, already turned into a `Failed` outcome. + Fail(StepOutcome), + /// A function step: run this script with this resolved body. + Function { + script: Box