From 41a21c0551620ebcf2ca44d3f88487fb1a6dcb9a Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sun, 12 Jul 2026 17:18:01 +0200 Subject: [PATCH] =?UTF-8?q?feat(workflows):=20M3=20=E2=80=94=20conditional?= =?UTF-8?q?=20`when`=20+=20input=20templating=20at=20run=20time?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the two pure M1 modules into the orchestrator's execute path: - Before running a step, build a `RunContext` from the run input + every prior succeeded step's output (`load_run_step_outputs`, read fresh at execution time so a step sees all upstream results). - `workflow_expr::eval` the step's `when` condition (if any): false → the step is `skipped` — never executed — and counts as satisfied-but-empty for its dependents (a new `StepOutcome::Skipped`, written in `complete_step_and_advance` then advanced; `compute_advance` already treats skipped as satisfying deps). - `workflow_template::resolve` the step's `input` against the context: an exact single `{{ ref }}` preserves the referenced JSON type, an embedded ref interpolates as text; a missing reference is a hard step failure (a definition bug surfaced, not hidden), never a silent null. `when` and templates were already parse-validated at apply time (M1); this is the runtime half. Tests (DB-gated, end-to-end via a scripted fake executor): - `when_false_skips_step` — b skipped, never executed, omitted from run output - `step_output_flows_into_downstream_input` — a's `{n:7}` feeds b's input, type-preserved for a bare ref and interpolated in a larger string - `missing_input_ref_fails_the_step` — an unresolved ref fails b → run fails Verified: cargo fmt, clippy -D warnings clean, 448 manager-core lib tests, 11 workflow_orchestrator DB tests (3 new). Co-Authored-By: Claude Opus 4.8 --- .../manager-core/src/workflow_orchestrator.rs | 44 +++- crates/manager-core/src/workflow_repo.rs | 40 +++ .../tests/workflow_orchestrator.rs | 227 ++++++++++++++++++ 3 files changed, 306 insertions(+), 5 deletions(-) diff --git a/crates/manager-core/src/workflow_orchestrator.rs b/crates/manager-core/src/workflow_orchestrator.rs index 4ed5b0a..41e80e9 100644 --- a/crates/manager-core/src/workflow_orchestrator.rs +++ b/crates/manager-core/src/workflow_orchestrator.rs @@ -48,9 +48,12 @@ use picloud_shared::{ExecutionId, ExecutionLogSink, ExecutionSource, RequestId}; 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, reclaim_stale_steps, ClaimedStep, StepOutcome, + claim_ready_step, complete_step_and_advance, load_run_step_outputs, reclaim_stale_steps, + ClaimedStep, StepOutcome, }; +use crate::workflow_template::{self, RunContext}; /// Max steps claimed (and thus gate permits held) per orchestrator tick. Keeps /// the workflow worker from monopolizing the gate it shares with HTTP + the @@ -238,6 +241,31 @@ impl WorkflowOrchestrator { ); }; + // M3: build the run context (run input + prior succeeded outputs) once; + // it backs both the `when` condition and the input template. + let outputs = match load_run_step_outputs(&self.pool, claimed.run_id).await { + Ok(o) => o, + Err(e) => { + return self.fail(Some(step), claimed, format!("loading step context: {e}")); + } + }; + let ctx = RunContext { + input: claimed.run_input.clone(), + steps: outputs, + }; + + // M3: a false `when` condition skips the step — never executed, and + // satisfied-but-empty for its dependents (`compute_advance`). + if let Some(cond) = &step.when { + match workflow_expr::eval(cond, &ctx) { + Ok(true) => {} + Ok(false) => return StepOutcome::Skipped, + Err(e) => { + return self.fail(Some(step), claimed, format!("evaluating `when`: {e}")); + } + } + } + let fn_name = match step.target() { Some(StepTarget::Function(f)) => f.to_string(), Some(StepTarget::Workflow(_)) => { @@ -285,13 +313,19 @@ impl WorkflowOrchestrator { } }; - // M2 input: pass the step's static `input` if present, else the run - // input. M3 replaces this with `{{ … }}` template resolution against - // prior step outputs. + // M3: resolve `{{ … }}` references in the step input against the + // context. A null input passes the run input through unchanged; a + // missing reference is a hard step failure (a definition bug we + // surface, not hide). let body = if step.input.is_null() { claimed.run_input.clone() } else { - step.input.clone() + match workflow_template::resolve(&step.input, &ctx) { + Ok(v) => v, + Err(e) => { + return self.fail(Some(step), claimed, format!("resolving input: {e}")); + } + } }; let execution_id = ExecutionId::new(); diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs index 3cc2153..bb51878 100644 --- a/crates/manager-core/src/workflow_repo.rs +++ b/crates/manager-core/src/workflow_repo.rs @@ -330,6 +330,9 @@ pub enum StepOutcome { error: String, retry_delay: chrono::Duration, }, + /// The step's `when` condition evaluated false — it is `skipped` (never + /// executed) and counts as satisfied-but-empty for its dependents (M3). + Skipped, } /// What `complete_step_and_advance` did — surfaced mostly for tests/logging. @@ -662,6 +665,22 @@ pub async fn complete_step_and_advance( return Ok(AdvanceResult::Stale); } } + StepOutcome::Skipped => { + let res = sqlx::query( + "UPDATE workflow_run_steps \ + SET status = 'skipped', output = NULL, error = NULL, \ + claim_token = NULL, claimed_at = NULL, updated_at = NOW() \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(claimed.step_id.into_inner()) + .bind(claimed.claim_token) + .execute(&mut *tx) + .await?; + if res.rows_affected() == 0 { + tx.rollback().await?; + return Ok(AdvanceResult::Stale); + } + } } // 2. Advance — lock the run row so concurrent advances of the same run @@ -884,6 +903,27 @@ pub async fn list_run_steps( rows.into_iter().map(TryInto::try_into).collect() } +/// The `{ step_name: output }` map of a run's **succeeded** steps — the +/// accumulated context M3 resolves a step's `when` + `input` templates against +/// (see `workflow_template::RunContext`). Read fresh at execution time so a +/// step sees every prior step's output. +pub async fn load_run_step_outputs( + pool: &PgPool, + run_id: WorkflowRunId, +) -> Result, WorkflowRepoError> { + let rows: Vec<(String, Option)> = sqlx::query_as( + "SELECT step_name, output FROM workflow_run_steps \ + WHERE run_id = $1 AND status = 'succeeded'", + ) + .bind(run_id.into_inner()) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .filter_map(|(name, out)| out.map(|v| (name, v))) + .collect()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/manager-core/tests/workflow_orchestrator.rs b/crates/manager-core/tests/workflow_orchestrator.rs index cee4ea1..b16e041 100644 --- a/crates/manager-core/tests/workflow_orchestrator.rs +++ b/crates/manager-core/tests/workflow_orchestrator.rs @@ -535,3 +535,230 @@ async fn end_to_end_tick_executes_functions_and_completes() { Some(json!({ "a": { "ran": "fn_a" }, "b": { "ran": "fn_b" } })) ); } + +// =========================================================================== +// M3 — conditional `when` + input templating (end-to-end via the orchestrator). +// =========================================================================== + +/// An executor that returns a canned body per script name (default +/// `{ "ran": name }`) and captures every `(script_name, request_body)` — so a +/// test can assert what input a downstream step actually received. +struct ScriptedExecutor { + outputs: std::collections::HashMap, + captured: std::sync::Mutex>, +} + +impl ScriptedExecutor { + fn new(outputs: &[(&str, Value)]) -> Self { + Self { + outputs: outputs + .iter() + .map(|(k, v)| ((*k).to_string(), v.clone())) + .collect(), + captured: std::sync::Mutex::new(vec![]), + } + } + fn body_for(&self, name: &str) -> Option { + self.captured + .lock() + .unwrap() + .iter() + .find(|(n, _)| n == name) + .map(|(_, b)| b.clone()) + } +} + +#[async_trait] +impl ExecutorClient for ScriptedExecutor { + async fn execute( + &self, + _source: &str, + req: ExecRequest, + _timeout: StdDuration, + ) -> Result { + self.captured + .lock() + .unwrap() + .push((req.script_name.clone(), req.body.clone())); + let body = self + .outputs + .get(&req.script_name) + .cloned() + .unwrap_or_else(|| json!({ "ran": req.script_name })); + Ok(ExecResponse { + status_code: 200, + headers: std::collections::BTreeMap::new(), + body, + logs: vec![], + stats: ExecStats::default(), + }) + } + + async fn execute_with_identity( + &self, + _identity: ScriptIdentity, + source: &str, + req: ExecRequest, + timeout: StdDuration, + ) -> Result { + self.execute(source, req, timeout).await + } +} + +async fn seed_scripts(pool: &PgPool, app: AppId, names: &[&str]) { + for name in names { + sqlx::query("INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2)") + .bind(name) + .bind(app.into_inner()) + .execute(pool) + .await + .expect("insert script"); + } +} + +/// Drive a real orchestrator to terminal using `executor`. +async fn drive_to_terminal( + pool: &PgPool, + executor: Arc, + app: AppId, + run: picloud_shared::WorkflowRunId, +) { + let orch = WorkflowOrchestrator { + pool: pool.clone(), + scripts: Arc::new(PostgresScriptRepository::new(pool.clone())), + executor, + gate: Arc::new(ExecutionGate::new(8)), + log_sink: Arc::new(NoopLogSink), + config: WorkflowOrchestratorConfig::default(), + }; + for _ in 0..50 { + orch.tick_once().await; + if get_run(pool, app, run) + .await + .unwrap() + .unwrap() + .status + .is_terminal() + { + break; + } + } +} + +fn run_with_input(app: AppId, wf: WorkflowId, input: Value) -> NewWorkflowRun { + NewWorkflowRun { + workflow_id: wf, + app_id: app, + input, + root_execution_id: Uuid::new_v4(), + workflow_depth: 0, + parent_run_id: None, + parent_step_id: None, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn when_false_skips_step() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // a -> b, but b runs only `when input.flag` (false here) → b is skipped. + let mut b = step("b", "fn_b", &["a"]); + b.when = Some("input.flag".into()); + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[]), b], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + seed_scripts(&pool, app, &["fn_a", "fn_b"]).await; + let run = start_run( + &pool, + run_with_input(app, wf, json!({ "flag": false })), + &def, + ) + .await + .expect("start"); + + let executor = Arc::new(ScriptedExecutor::new(&[])); + drive_to_terminal(&pool, executor.clone(), app, run).await; + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); + let steps = list_run_steps(&pool, run).await.unwrap(); + let by = |n: &str| steps.iter().find(|s| s.step_name == n).unwrap().status; + assert_eq!(by("a"), StepStatus::Succeeded); + assert_eq!(by("b"), StepStatus::Skipped); + // A skipped step is never executed. + assert!( + executor.body_for("fn_b").is_none(), + "skipped step must not run" + ); + // Its output is omitted from the run output. + assert_eq!(r.output, Some(json!({ "a": { "ran": "fn_a" } }))); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn step_output_flows_into_downstream_input() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + // b's input pulls a field out of a's output, type-preserved. + let mut b = step("b", "fn_b", &["a"]); + b.input = json!({ "v": "{{ steps.a.output.n }}", "label": "n={{ steps.a.output.n }}" }); + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[]), b], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + seed_scripts(&pool, app, &["fn_a", "fn_b"]).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + // fn_a outputs { n: 7 }. + let executor = Arc::new(ScriptedExecutor::new(&[("fn_a", json!({ "n": 7 }))])); + drive_to_terminal(&pool, executor.clone(), app, run).await; + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!(r.status, RunStatus::Succeeded); + // b received the resolved input: exact-single-ref preserves the number 7, + // the embedded ref interpolates as text. + assert_eq!( + executor.body_for("fn_b"), + Some(json!({ "v": 7, "label": "n=7" })) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn missing_input_ref_fails_the_step() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let mut b = step("b", "fn_b", &["a"]); + b.input = json!({ "v": "{{ steps.a.output.does_not_exist }}" }); + let def = WorkflowDefinition { + steps: vec![step("a", "fn_a", &[]), b], + }; + let (app, wf) = seed_workflow(&pool, &def).await; + seed_scripts(&pool, app, &["fn_a", "fn_b"]).await; + let run = start_run(&pool, new_run(app, wf), &def) + .await + .expect("start"); + + // fn_a outputs { n: 7 } — no `does_not_exist`, so b's input resolve fails. + let executor = Arc::new(ScriptedExecutor::new(&[("fn_a", json!({ "n": 7 }))])); + drive_to_terminal(&pool, executor.clone(), app, run).await; + + let r = get_run(&pool, app, run).await.unwrap().unwrap(); + assert_eq!( + r.status, + RunStatus::Failed, + "missing input ref fails the run" + ); + assert!(executor.body_for("fn_b").is_none(), "b never executes"); + let steps = list_run_steps(&pool, run).await.unwrap(); + let bstep = steps.iter().find(|s| s.step_name == "b").unwrap(); + assert_eq!(bstep.status, StepStatus::Failed); + assert!(bstep.error.as_deref().unwrap_or("").contains("input")); +}