feat(workflows): M3 — conditional when + input templating at run time

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 <noreply@anthropic.com>

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();

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<BTreeMap<String, serde_json::Value>, WorkflowRepoError> {
+    let rows: Vec<(String, Option<serde_json::Value>)> = 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::*;

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<String, Value>,
+    captured: std::sync::Mutex<Vec<(String, Value)>>,
+}
+
+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<Value> {
+        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<ExecResponse, ExecError> {
+        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<ExecResponse, ExecError> {
+        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<dyn ExecutorClient>,
+    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"));
+}