feat(workflows): M2 — durable orchestrator worker (claim/execute/advance)

The v1.2 Workflows durable DAG engine gains its runtime. A dedicated
background worker (`workflow_orchestrator.rs`, mirroring `cron_scheduler`,
NOT folded into the dispatcher) advances every in-flight run step-by-step,
durably, surviving restarts.

Per tick, two phases:
  A. Claim + execute — up to a small batch of `ready`, due steps are claimed
     with the same `FOR UPDATE SKIP LOCKED` competing-consumer lease the queue
     uses (`claim_ready_step`), one execution-gate permit per step acquired
     BEFORE the claim so the shared gate bounds real parallelism. Each step
     resolves its function by name in the run's app scope (never a
     script-passed arg — the isolation boundary), builds an `ExecRequest`, and
     runs through the injected `ExecutorClient`. Claimed steps run concurrently.
  B. Advance — the outcome is written and the DAG advanced in one token-gated
     transaction (`complete_step_and_advance`): pending steps whose deps are
     satisfied flip to `ready`, and the run's terminal status is recomputed.
     Fan-in falls out (a join waits until its last dep flips it); a stale worker
     matches zero rows and writes nothing.

The graph-advance decision is a pure, DB-free function (`compute_advance`) —
promotions + terminal run status, folding `on_error` fail-vs-continue and
skipped/failed dependency satisfaction — so it is unit-tested in isolation.

Retry uses the step's own policy via `compute_backoff`; a second, slower
cadence reclaims steps leased by a crashed worker (`reclaim_stale_steps`) — the
durability safety net. Steps run with no principal (like invoke_async), and log
under the new `ExecutionSource::Workflow` so `pic logs` surfaces them.

M2 executes function steps only; `when` + input templating land in M3, nested
sub-workflows in M4. Seams present (`StepTarget`, `run_input`, `workflow_depth`).

- migration 0072: widen the `execution_logs.source` CHECK with `workflow`
- shared: `ExecutionSource::Workflow`, `StepStatus::is_terminal`
- workflow_repo: run/step state — `start_run`, `claim_ready_step`,
  `complete_step_and_advance`, `reclaim_stale_steps`, `get_run`,
  `list_run_steps`, `compute_advance` (+ 7 pure unit tests)
- workflow_orchestrator: the worker + config (`PICLOUD_WORKFLOW_*` knobs),
  spawned in `picloud/src/lib.rs` beside the dispatcher/cron scheduler
- tests: 8 DB-gated integration tests (linear, parallel fan-out/fan-in, retry,
  on_error fail/continue, double-complete idempotency, stale-lease reclaim,
  end-to-end tick with a fake executor)

Verified: cargo fmt, clippy -D warnings clean, 448 manager-core lib tests,
8 workflow_orchestrator DB tests, schema snapshot reblessed, M1 workflow CLI
journeys still pass (binary boots with the orchestrator wired in).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

crates/manager-core/migrations/0072_execution_source_workflow.sql

@@ -0,0 +1,18 @@
+-- v1.2 Workflows M2: the orchestrator executes each workflow step as a normal
+-- function invocation and logs it, so `pic logs` can surface (and filter by)
+-- workflow-driven runs — exactly as it does for cron/queue/invoke triggers.
+--
+-- Add 'workflow' to the execution_logs.source CHECK (extends 0043). The column
+-- default stays 'http'; this only widens the allowed set. Keep this list in
+-- sync with `shared::ExecutionSource` (which gains the matching `Workflow`
+-- variant) and `manager-core::OutboxSourceKind`.
+--
+-- Postgres has no ALTER … ALTER CONSTRAINT for a CHECK, so drop + re-add.
+
+ALTER TABLE execution_logs DROP CONSTRAINT execution_logs_source_check;
+ALTER TABLE execution_logs
+    ADD CONSTRAINT execution_logs_source_check
+    CHECK (source IN (
+        'http', 'kv', 'docs', 'dead_letter', 'cron',
+        'files', 'pubsub', 'email', 'invoke', 'queue', 'workflow'
+    ));

crates/manager-core/src/lib.rs

@@ -109,6 +109,7 @@ pub mod vars_api;
 pub mod vars_repo;
 pub mod vars_service;
 pub mod workflow_expr;
+pub mod workflow_orchestrator;
 pub mod workflow_repo;
 pub mod workflow_template;
 
@@ -267,3 +268,4 @@ pub use users_service::{UsersServiceConfig, UsersServiceImpl};
 pub use vars_api::{vars_router, VarsApiError, VarsApiState};
 pub use vars_repo::{PostgresVarsRepo, VarOwner, VarRow, VarsRepo, VarsRepoError};
 pub use vars_service::VarsServiceImpl;
+pub use workflow_orchestrator::{spawn_workflow_orchestrator, WorkflowOrchestrator};

crates/manager-core/src/workflow_orchestrator.rs

@@ -0,0 +1,395 @@
+//! The durable workflow orchestrator (v1.2 Workflows, M2).
+//!
+//! A single dedicated background worker — mirroring [`cron_scheduler`] — that
+//! advances every in-flight workflow run step-by-step, durably, surviving
+//! restarts. It is deliberately NOT folded into `dispatcher.rs`: a per-*run*
+//! graph advance is its own transaction with its own lease semantics.
+//!
+//! Each tick has two phases:
+//!
+//!   **A. Claim + execute.** Up to a small batch of `ready`, due steps are
+//!   claimed with the same `FOR UPDATE SKIP LOCKED` competing-consumer lease
+//!   the queue uses ([`workflow_repo::claim_ready_step`]) — one execution-gate
+//!   permit acquired per step *before* the claim, so the shared gate bounds
+//!   real parallelism and no step is ever claimed without a slot to run it.
+//!   Each claimed step resolves its function by name (in the run's app scope),
+//!   builds an `ExecRequest`, and runs through the injected [`ExecutorClient`]
+//!   — the same executor path every other trigger uses. Claimed steps run
+//!   concurrently within the tick.
+//!
+//!   **B. Advance.** The step outcome is written and the DAG advanced in one
+//!   token-gated transaction ([`workflow_repo::complete_step_and_advance`]):
+//!   pending steps whose dependencies are now satisfied flip to `ready`, and
+//!   the run's terminal status is recomputed. Fan-in falls out for free (a
+//!   join step stays `pending` until its last dependency flips it). A stale
+//!   worker (its lease reclaimed) matches zero rows and writes nothing.
+//!
+//! 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`).
+//!
+//! [`cron_scheduler`]: crate::cron_scheduler
+
+use std::sync::Arc;
+use std::time::Duration as StdDuration;
+
+use chrono::{Duration as ChronoDuration, Utc};
+use sqlx::PgPool;
+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 crate::dispatcher::compute_backoff;
+use crate::repo::ScriptRepository;
+use crate::trigger_config::BackoffShape;
+use crate::workflow_repo::{
+    claim_ready_step, complete_step_and_advance, reclaim_stale_steps, ClaimedStep, StepOutcome,
+};
+
+/// Max steps claimed (and thus gate permits held) per orchestrator tick. Keeps
+/// the workflow worker from monopolizing the gate it shares with HTTP + the
+/// dispatcher; the tick cadence + reclaim pick up the rest.
+const STEP_CLAIM_BATCH: usize = 16;
+
+/// Tunables, all env-overridable (`PICLOUD_WORKFLOW_*`), matching the
+/// parse-warn-fallback pattern the rest of the crate uses.
+#[derive(Debug, Clone, Copy)]
+pub struct WorkflowOrchestratorConfig {
+    pub tick_interval_ms: u64,
+    pub reclaim_interval_ms: u64,
+    pub visibility_timeout_secs: u32,
+    pub retry_jitter_pct: u32,
+    pub step_timeout: StdDuration,
+    /// Nesting ceiling for sub-workflows (M4). Read now so the knob is stable.
+    pub max_depth: u32,
+}
+
+impl Default for WorkflowOrchestratorConfig {
+    fn default() -> Self {
+        Self {
+            tick_interval_ms: 250,
+            reclaim_interval_ms: 30_000,
+            visibility_timeout_secs: 300,
+            retry_jitter_pct: 20,
+            step_timeout: StdDuration::from_secs(300),
+            max_depth: 8,
+        }
+    }
+}
+
+impl WorkflowOrchestratorConfig {
+    #[must_use]
+    pub fn from_env() -> Self {
+        let d = Self::default();
+        Self {
+            tick_interval_ms: env_u64("PICLOUD_WORKFLOW_TICK_INTERVAL_MS", d.tick_interval_ms),
+            reclaim_interval_ms: env_u64(
+                "PICLOUD_WORKFLOW_RECLAIM_INTERVAL_MS",
+                d.reclaim_interval_ms,
+            ),
+            visibility_timeout_secs: env_u32(
+                "PICLOUD_WORKFLOW_STEP_VISIBILITY_TIMEOUT_SECS",
+                d.visibility_timeout_secs,
+            ),
+            retry_jitter_pct: env_u32("PICLOUD_WORKFLOW_RETRY_JITTER_PCT", d.retry_jitter_pct),
+            step_timeout: StdDuration::from_secs(env_u64(
+                "PICLOUD_WORKFLOW_STEP_TIMEOUT_SEC",
+                d.step_timeout.as_secs(),
+            )),
+            max_depth: env_u32("PICLOUD_WORKFLOW_MAX_DEPTH", d.max_depth),
+        }
+    }
+}
+
+fn env_u64(key: &str, default: u64) -> u64 {
+    match std::env::var(key) {
+        Ok(v) => v.trim().parse::<u64>().map_or_else(
+            |_| {
+                tracing::warn!(%key, value = %v, "ignoring invalid value; want a positive integer");
+                default
+            },
+            |n| if n > 0 { n } else { default },
+        ),
+        Err(_) => default,
+    }
+}
+
+fn env_u32(key: &str, default: u32) -> u32 {
+    u32::try_from(env_u64(key, u64::from(default))).unwrap_or(default)
+}
+
+/// The orchestrator's collaborators. All the durable state lives behind the
+/// pool free-fns in [`workflow_repo`](crate::workflow_repo); execution goes
+/// through the injected [`ExecutorClient`], never `executor-core` directly.
+pub struct WorkflowOrchestrator {
+    pub pool: PgPool,
+    pub scripts: Arc<dyn ScriptRepository>,
+    pub executor: Arc<dyn ExecutorClient>,
+    pub gate: Arc<ExecutionGate>,
+    pub log_sink: Arc<dyn ExecutionLogSink>,
+    pub config: WorkflowOrchestratorConfig,
+}
+
+impl WorkflowOrchestrator {
+    /// Spawn the advance loop + the reclaim task. Both run for the process
+    /// lifetime; returned handles are dropped on purpose (as with the
+    /// dispatcher + cron scheduler).
+    pub fn spawn(self) {
+        // Reclaim task — independent, slower cadence so it doesn't contend
+        // with the per-tick advance loop.
+        let reclaim_pool = self.pool.clone();
+        let reclaim_interval = StdDuration::from_millis(self.config.reclaim_interval_ms.max(1_000));
+        let visibility = self.config.visibility_timeout_secs;
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(reclaim_interval);
+            ticker.tick().await;
+            loop {
+                ticker.tick().await;
+                match reclaim_stale_steps(&reclaim_pool, visibility).await {
+                    Ok(0) => {}
+                    Ok(n) => {
+                        tracing::info!(reclaimed = n, "workflow step visibility-timeout reclaim");
+                    }
+                    Err(e) => tracing::warn!(?e, "workflow reclaim task errored"),
+                }
+            }
+        });
+
+        // Advance loop.
+        let interval = StdDuration::from_millis(self.config.tick_interval_ms.max(50));
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(interval);
+            ticker.tick().await;
+            loop {
+                ticker.tick().await;
+                self.tick_once().await;
+            }
+        });
+    }
+
+    /// One advance tick: claim a permit-bounded batch of ready steps, then run
+    /// them (execute + advance) concurrently. Public so integration tests can
+    /// drive the loop deterministically (one wave per call).
+    pub async fn tick_once(&self) {
+        use futures::stream::{self, StreamExt};
+
+        let mut batch: Vec<(ClaimedStep, OwnedSemaphorePermit)> = Vec::new();
+        for _ in 0..STEP_CLAIM_BATCH {
+            // Acquire the gate slot BEFORE claiming so a claimed step always
+            // has somewhere to run; if the gate is saturated, stop this tick.
+            let Ok(permit) = self.gate.try_acquire() else {
+                break;
+            };
+            match claim_ready_step(&self.pool).await {
+                Ok(Some(claimed)) => batch.push((claimed, permit)),
+                Ok(None) => {
+                    drop(permit);
+                    break;
+                }
+                Err(e) => {
+                    drop(permit);
+                    tracing::warn!(?e, "workflow claim errored");
+                    break;
+                }
+            }
+        }
+        if batch.is_empty() {
+            return;
+        }
+        // Each entry already holds a permit, so run them all concurrently.
+        stream::iter(batch)
+            .for_each_concurrent(None, |(claimed, permit)| async move {
+                self.run_step(claimed, permit).await;
+            })
+            .await;
+    }
+
+    /// Execute one claimed step, then write its outcome + advance the run.
+    async fn run_step(&self, claimed: ClaimedStep, permit: OwnedSemaphorePermit) {
+        let outcome = self.execute_step(&claimed).await;
+        // Release the gate slot before the (DB-only, fast) advance write.
+        drop(permit);
+        if let Err(e) = complete_step_and_advance(&self.pool, &claimed, outcome).await {
+            tracing::error!(
+                ?e,
+                run_id = %claimed.run_id,
+                step = %claimed.step_name,
+                "workflow step advance write failed"
+            );
+        }
+    }
+
+    /// Resolve + run a step's function, returning the outcome. Failures are
+    /// converted to [`StepOutcome::Failed`] carrying the retry delay derived
+    /// from the step's own retry policy (the repo decides retry-vs-terminal).
+    #[allow(clippy::too_many_lines)]
+    async fn execute_step(&self, claimed: &ClaimedStep) -> StepOutcome {
+        let Some(step) = claimed.step_def() else {
+            return self.fail(
+                None,
+                claimed,
+                "step not found in workflow definition".into(),
+            );
+        };
+
+        let fn_name = match step.target() {
+            Some(StepTarget::Function(f)) => f.to_string(),
+            Some(StepTarget::Workflow(_)) => {
+                // Nested sub-workflows land in M4.
+                return self.fail(
+                    Some(step),
+                    claimed,
+                    "sub-workflow steps not yet supported".into(),
+                );
+            }
+            None => {
+                return self.fail(
+                    Some(step),
+                    claimed,
+                    "step declares neither `function` nor `workflow`".into(),
+                );
+            }
+        };
+
+        // Resolve the function by name in the RUN's app scope (an inherited
+        // group script is reachable, nearest-owner-wins) — never off a
+        // script-passed arg. cx isolation boundary preserved.
+        let script = match self
+            .scripts
+            .get_by_name_inherited(claimed.app_id, &fn_name)
+            .await
+        {
+            Ok(Some(s)) if s.enabled => s,
+            Ok(Some(_)) => {
+                return self.fail(
+                    Some(step),
+                    claimed,
+                    format!("function {fn_name:?} is disabled"),
+                );
+            }
+            Ok(None) => {
+                return self.fail(
+                    Some(step),
+                    claimed,
+                    format!("function {fn_name:?} not found"),
+                );
+            }
+            Err(e) => {
+                return self.fail(Some(step), claimed, format!("resolving {fn_name:?}: {e}"));
+            }
+        };
+
+        // M2 input: pass the step's static `input` if present, else the run
+        // input. M3 replaces this with `{{ … }}` template resolution against
+        // prior step outputs.
+        let body = if step.input.is_null() {
+            claimed.run_input.clone()
+        } else {
+            step.input.clone()
+        };
+
+        let execution_id = ExecutionId::new();
+        let request_id = RequestId::new();
+        let req = ExecRequest {
+            execution_id,
+            request_id,
+            script_id: script.id,
+            script_name: script.name.clone(),
+            invocation_type: InvocationType::Function,
+            path: format!("/workflow/{}", claimed.step_name),
+            method: String::new(),
+            headers: std::collections::BTreeMap::new(),
+            body: body.clone(),
+            params: std::collections::BTreeMap::new(),
+            query: std::collections::BTreeMap::new(),
+            rest: String::new(),
+            sandbox_overrides: script.sandbox,
+            app_id: claimed.app_id,
+            script_owner: script.owner(),
+            // Like invoke_async / HTTP outbox rows: steps run with no
+            // principal (the run's origin is forensic, not re-authenticated).
+            principal: None,
+            trigger_depth: 0,
+            root_execution_id: ExecutionId::from(claimed.root_execution_id),
+            is_dead_letter_handler: false,
+            event: None,
+        };
+        let identity = ScriptIdentity {
+            script_id: script.id,
+            updated_at: script.updated_at,
+        };
+
+        let started = Utc::now();
+        let outcome = self
+            .executor
+            .execute_with_identity(identity, &script.source, req, self.config.step_timeout)
+            .await;
+        let finished = Utc::now();
+
+        // Log the run so `pic logs` surfaces workflow steps (source=workflow).
+        let log = build_execution_log(
+            claimed.app_id,
+            script.id,
+            request_id,
+            format!("/workflow/{}", claimed.step_name),
+            std::collections::BTreeMap::new(),
+            body,
+            ExecutionSource::Workflow,
+            &outcome,
+            started,
+            finished,
+        );
+        if let Err(e) = self.log_sink.record(log).await {
+            tracing::warn!(?e, step = %claimed.step_name, "workflow step log write failed");
+        }
+
+        match outcome {
+            Ok(resp) => StepOutcome::Succeeded(resp.body),
+            Err(err) => self.fail(Some(step), claimed, err.to_string()),
+        }
+    }
+
+    /// Build a `Failed` outcome, computing the retry backoff from the step's
+    /// own retry policy (zero if it has none — the repo then treats the first
+    /// failure as terminal).
+    fn fail(
+        &self,
+        step: Option<&WorkflowStepDef>,
+        claimed: &ClaimedStep,
+        error: String,
+    ) -> StepOutcome {
+        let retry_delay = step
+            .and_then(|s| s.retry.as_ref())
+            .map_or(ChronoDuration::zero(), |r| {
+                let ms = compute_backoff(
+                    u32::try_from(claimed.attempt.max(1)).unwrap_or(1),
+                    map_backoff(r.backoff),
+                    r.base_ms,
+                    self.config.retry_jitter_pct,
+                );
+                ChronoDuration::milliseconds(i64::from(ms))
+            });
+        StepOutcome::Failed { error, retry_delay }
+    }
+}
+
+/// Map the self-contained workflow backoff DTO onto the shared trigger backoff.
+fn map_backoff(b: WorkflowBackoff) -> BackoffShape {
+    match b {
+        WorkflowBackoff::Constant => BackoffShape::Constant,
+        WorkflowBackoff::Linear => BackoffShape::Linear,
+        WorkflowBackoff::Exponential => BackoffShape::Exponential,
+    }
+}
+
+/// Consume the orchestrator and spawn its background tasks (mirrors
+/// [`spawn_cron_scheduler`](crate::spawn_cron_scheduler)).
+pub fn spawn_workflow_orchestrator(orchestrator: WorkflowOrchestrator) {
+    orchestrator.spawn();
+}

crates/manager-core/src/workflow_repo.rs

@@ -6,8 +6,14 @@
 //! declarative `apply` reconcile engine (Create/Update/Delete by `lower(name)`,
 //! all in one transaction) — mirroring `trigger_repo::insert_trigger_tx`.
 //!
-//! Run/step state (`workflow_runs`, `workflow_run_steps`) is added in M2 with
-//! the orchestrator.
+//! Run/step state (`workflow_runs`, `workflow_run_steps`) lands in M2: the
+//! durable orchestrator seeds a run (`start_run`), claims a `ready` step with
+//! the same `FOR UPDATE SKIP LOCKED` lease `queue_repo` uses, and — in one
+//! token-gated transaction — writes the step outcome and advances the DAG
+//! (`complete_step_and_advance`). The graph-advance decision is a pure,
+//! DB-free function ([`compute_advance`]) so it is unit-testable.
+
+use std::collections::BTreeMap;
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
@@ -15,8 +21,10 @@ use sqlx::PgPool;
 use thiserror::Error;
 use uuid::Uuid;
 
-use picloud_shared::workflow::WorkflowDefinition;
-use picloud_shared::{AppId, WorkflowId};
+use picloud_shared::workflow::{
+    OnError, RunStatus, StepStatus, WorkflowDefinition, WorkflowStepDef,
+};
+use picloud_shared::{AppId, WorkflowId, WorkflowRunId, WorkflowRunStepId};
 
 #[derive(Debug, Error)]
 pub enum WorkflowRepoError {
@@ -227,3 +235,785 @@ pub async fn delete_workflow_tx(
         .await?;
     Ok(())
 }
+
+// ===========================================================================
+// M2 — run/step state + the durable orchestrator's persistence surface.
+// ===========================================================================
+
+/// Everything needed to seed a fresh run (mirrors `NewQueueMessage`). The
+/// orchestrator's M5 `workflow::start` and the M2 tests both build this.
+#[derive(Debug, Clone)]
+pub struct NewWorkflowRun {
+    pub workflow_id: WorkflowId,
+    pub app_id: AppId,
+    pub input: serde_json::Value,
+    /// Correlates every step execution in `execution_logs` under one id.
+    pub root_execution_id: Uuid,
+    /// 0 for a top-level run; a nested sub-workflow (M4) is parent + 1.
+    pub workflow_depth: i32,
+    pub parent_run_id: Option<WorkflowRunId>,
+    pub parent_step_id: Option<WorkflowRunStepId>,
+}
+
+/// A run row, read for the admin/CLI surface + test assertions.
+#[derive(Debug, Clone)]
+pub struct WorkflowRun {
+    pub id: WorkflowRunId,
+    pub workflow_id: WorkflowId,
+    pub app_id: AppId,
+    pub status: RunStatus,
+    pub input: serde_json::Value,
+    pub output: Option<serde_json::Value>,
+    pub error: Option<String>,
+    pub root_execution_id: Uuid,
+    pub workflow_depth: i32,
+    pub parent_run_id: Option<WorkflowRunId>,
+    pub parent_step_id: Option<WorkflowRunStepId>,
+    pub started_at: Option<DateTime<Utc>>,
+    pub finished_at: Option<DateTime<Utc>>,
+    pub created_at: DateTime<Utc>,
+}
+
+/// A step row within a run (read for the admin/CLI surface + assertions).
+#[derive(Debug, Clone)]
+pub struct WorkflowRunStep {
+    pub id: WorkflowRunStepId,
+    pub run_id: WorkflowRunId,
+    pub step_name: String,
+    pub status: StepStatus,
+    pub attempt: i32,
+    pub max_attempts: i32,
+    pub output: Option<serde_json::Value>,
+    pub error: Option<String>,
+    pub child_run_id: Option<WorkflowRunId>,
+}
+
+/// A claimed `ready` step, joined with its run + workflow definition so the
+/// orchestrator has everything it needs to resolve, execute, and advance
+/// without a second round trip.
+#[derive(Debug, Clone)]
+pub struct ClaimedStep {
+    pub step_id: WorkflowRunStepId,
+    pub run_id: WorkflowRunId,
+    pub step_name: String,
+    /// 1-indexed current attempt (the claim increments it).
+    pub attempt: i32,
+    pub max_attempts: i32,
+    pub claim_token: Uuid,
+    // ---- run context ----
+    pub app_id: AppId,
+    pub workflow_id: WorkflowId,
+    pub root_execution_id: Uuid,
+    pub workflow_depth: i32,
+    pub run_input: serde_json::Value,
+    pub definition: WorkflowDefinition,
+}
+
+impl ClaimedStep {
+    /// The definition of the step being executed.
+    #[must_use]
+    pub fn step_def(&self) -> Option<&WorkflowStepDef> {
+        self.definition
+            .steps
+            .iter()
+            .find(|s| s.name == self.step_name)
+    }
+}
+
+/// The outcome the orchestrator hands back after executing a step. On
+/// `Failed`, the repo decides retry-vs-terminal from `attempt < max_attempts`;
+/// the orchestrator supplies the delay to use if a retry remains.
+#[derive(Debug, Clone)]
+pub enum StepOutcome {
+    Succeeded(serde_json::Value),
+    Failed {
+        error: String,
+        retry_delay: chrono::Duration,
+    },
+}
+
+/// What `complete_step_and_advance` did — surfaced mostly for tests/logging.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AdvanceResult {
+    /// The claim token no longer matched (a stale worker); nothing written.
+    Stale,
+    /// The step failed but had attempts left; it was re-armed as `ready`.
+    Retried,
+    /// The step reached a terminal status and the run graph was advanced.
+    Advanced,
+}
+
+// ---- the pure graph-advance decision (DB-free, unit-tested) --------------
+
+/// The write plan `complete_step_and_advance` derives after a step reaches a
+/// terminal status: which `pending` steps become `ready`, and the run's new
+/// status/output/error.
+#[derive(Debug, Clone, PartialEq)]
+pub struct AdvancePlan {
+    pub promote: Vec<String>,
+    pub run_status: RunStatus,
+    pub run_error: Option<String>,
+    pub run_output: Option<serde_json::Value>,
+}
+
+/// Is `dep` satisfied for the purpose of unblocking a dependent? Succeeded and
+/// skipped always satisfy; a `failed` step satisfies **only** when its
+/// definition says `on_error = continue` (the run limps on).
+fn dep_satisfied(status: StepStatus, on_error: OnError) -> bool {
+    match status {
+        StepStatus::Succeeded | StepStatus::Skipped => true,
+        StepStatus::Failed => matches!(on_error, OnError::Continue),
+        _ => false,
+    }
+}
+
+/// Pure DAG advance. Given the definition and the current per-step
+/// status/output snapshot, decide the promotions + terminal run state.
+///
+/// - A `pending` step whose every `depends_on` is satisfied becomes `ready`.
+/// - If any step `failed` with `on_error = fail`, the run fails immediately.
+/// - Else if every step is terminal, the run succeeds; its output is the
+///   object of `{ step_name: output }` over the succeeded steps.
+/// - Otherwise the run is still `running`.
+#[must_use]
+pub fn compute_advance(
+    def: &WorkflowDefinition,
+    statuses: &BTreeMap<String, StepStatus>,
+    outputs: &BTreeMap<String, serde_json::Value>,
+) -> AdvancePlan {
+    let by_name: BTreeMap<&str, &WorkflowStepDef> =
+        def.steps.iter().map(|s| (s.name.as_str(), s)).collect();
+    let status_of = |name: &str| statuses.get(name).copied().unwrap_or(StepStatus::Pending);
+
+    // Promote every pending step whose deps are all satisfied.
+    let mut promote = Vec::new();
+    for step in &def.steps {
+        if status_of(&step.name) != StepStatus::Pending {
+            continue;
+        }
+        let ready = step.depends_on.iter().all(|dep| {
+            let d_on_err = by_name
+                .get(dep.as_str())
+                .map_or(OnError::Fail, |s| s.on_error);
+            dep_satisfied(status_of(dep), d_on_err)
+        });
+        if ready {
+            promote.push(step.name.clone());
+        }
+    }
+
+    // A hard failure (on_error = fail that exhausted) fails the whole run.
+    if let Some(failed) = def
+        .steps
+        .iter()
+        .find(|s| status_of(&s.name) == StepStatus::Failed && matches!(s.on_error, OnError::Fail))
+    {
+        return AdvancePlan {
+            promote,
+            run_status: RunStatus::Failed,
+            run_error: Some(format!("step {:?} failed", failed.name)),
+            run_output: None,
+        };
+    }
+
+    // All steps terminal (accounting for the promotions we just queued — a
+    // promoted step is pending → ready, i.e. not terminal, so the run stays
+    // running whenever `promote` is non-empty).
+    let all_terminal =
+        promote.is_empty() && def.steps.iter().all(|s| status_of(&s.name).is_terminal());
+    if all_terminal {
+        let mut out = serde_json::Map::new();
+        for step in &def.steps {
+            if status_of(&step.name) == StepStatus::Succeeded {
+                if let Some(v) = outputs.get(&step.name) {
+                    out.insert(step.name.clone(), v.clone());
+                }
+            }
+        }
+        return AdvancePlan {
+            promote,
+            run_status: RunStatus::Succeeded,
+            run_error: None,
+            run_output: Some(serde_json::Value::Object(out)),
+        };
+    }
+
+    AdvancePlan {
+        promote,
+        run_status: RunStatus::Running,
+        run_error: None,
+        run_output: None,
+    }
+}
+
+// ---- run seeding ----------------------------------------------------------
+
+/// Seed a run: insert the `workflow_runs` row + one `workflow_run_steps` row
+/// per definition step (root steps — no `depends_on` — start `ready`, the rest
+/// `pending`). All in one transaction so a partially-seeded run is never
+/// visible to the orchestrator's claim scan.
+pub async fn start_run(
+    pool: &PgPool,
+    new: NewWorkflowRun,
+    definition: &WorkflowDefinition,
+) -> Result<WorkflowRunId, WorkflowRepoError> {
+    let mut tx = pool.begin().await?;
+    let (run_id,): (Uuid,) = sqlx::query_as(
+        "INSERT INTO workflow_runs \
+            (workflow_id, app_id, status, input, root_execution_id, \
+             workflow_depth, parent_run_id, parent_step_id) \
+         VALUES ($1, $2, 'pending', $3, $4, $5, $6, $7) RETURNING id",
+    )
+    .bind(new.workflow_id.into_inner())
+    .bind(new.app_id.into_inner())
+    .bind(&new.input)
+    .bind(new.root_execution_id)
+    .bind(new.workflow_depth)
+    .bind(new.parent_run_id.map(WorkflowRunId::into_inner))
+    .bind(new.parent_step_id.map(WorkflowRunStepId::into_inner))
+    .fetch_one(&mut *tx)
+    .await?;
+
+    for step in &definition.steps {
+        let status = if step.depends_on.is_empty() {
+            "ready"
+        } else {
+            "pending"
+        };
+        let max_attempts = i32::try_from(step.max_attempts()).unwrap_or(1);
+        sqlx::query(
+            "INSERT INTO workflow_run_steps \
+                (run_id, step_name, status, max_attempts) \
+             VALUES ($1, $2, $3, $4)",
+        )
+        .bind(run_id)
+        .bind(&step.name)
+        .bind(status)
+        .bind(max_attempts)
+        .execute(&mut *tx)
+        .await?;
+    }
+
+    tx.commit().await?;
+    Ok(run_id.into())
+}
+
+// ---- claim / complete / reclaim ------------------------------------------
+
+#[derive(sqlx::FromRow)]
+struct ClaimedRow {
+    id: Uuid,
+    run_id: Uuid,
+    step_name: String,
+    attempt: i32,
+    max_attempts: i32,
+    app_id: Uuid,
+    workflow_id: Uuid,
+    root_execution_id: Uuid,
+    workflow_depth: i32,
+    input: serde_json::Value,
+    definition: serde_json::Value,
+}
+
+/// Atomically claim one `ready`, due step across all active runs — the queue
+/// competing-consumer lease (`FOR UPDATE SKIP LOCKED`), so parallel workers
+/// (and, cluster mode later, parallel nodes) never grab the same step. The
+/// claimed step flips to `running` with a fresh `claim_token`; its run flips
+/// `pending → running`. Returns `None` when nothing is claimable.
+pub async fn claim_ready_step(pool: &PgPool) -> Result<Option<ClaimedStep>, WorkflowRepoError> {
+    let token = Uuid::new_v4();
+    let row: Option<ClaimedRow> = sqlx::query_as(
+        "WITH claimed AS ( \
+            UPDATE workflow_run_steps \
+            SET status = 'running', claim_token = $1, claimed_at = NOW(), \
+                attempt = attempt + 1, updated_at = NOW() \
+            WHERE id = ( \
+                SELECT s.id FROM workflow_run_steps s \
+                JOIN workflow_runs r ON r.id = s.run_id \
+                WHERE s.status = 'ready' \
+                  AND (s.next_attempt_at IS NULL OR s.next_attempt_at <= NOW()) \
+                  AND r.status IN ('pending', 'running') \
+                ORDER BY s.created_at \
+                FOR UPDATE OF s SKIP LOCKED \
+                LIMIT 1 \
+            ) \
+            RETURNING id, run_id, step_name, attempt, max_attempts \
+        ) \
+        SELECT c.id, c.run_id, c.step_name, c.attempt, c.max_attempts, \
+               r.app_id, r.workflow_id, r.root_execution_id, r.workflow_depth, \
+               r.input, w.definition \
+        FROM claimed c \
+        JOIN workflow_runs r ON r.id = c.run_id \
+        JOIN workflows w ON w.id = r.workflow_id",
+    )
+    .bind(token)
+    .fetch_optional(pool)
+    .await?;
+
+    let Some(r) = row else { return Ok(None) };
+
+    // Reflect the run as running promptly (advance also does this, but a
+    // long-running first step would otherwise leave the run 'pending').
+    sqlx::query(
+        "UPDATE workflow_runs SET status = 'running', started_at = COALESCE(started_at, NOW()) \
+         WHERE id = $1 AND status = 'pending'",
+    )
+    .bind(r.run_id)
+    .execute(pool)
+    .await?;
+
+    let definition = serde_json::from_value(r.definition)
+        .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?;
+    Ok(Some(ClaimedStep {
+        step_id: r.id.into(),
+        run_id: r.run_id.into(),
+        step_name: r.step_name,
+        attempt: r.attempt,
+        max_attempts: r.max_attempts,
+        claim_token: token,
+        app_id: r.app_id.into(),
+        workflow_id: r.workflow_id.into(),
+        root_execution_id: r.root_execution_id,
+        workflow_depth: r.workflow_depth,
+        run_input: r.input,
+        definition,
+    }))
+}
+
+#[derive(sqlx::FromRow)]
+struct StepStateRow {
+    step_name: String,
+    status: String,
+    output: Option<serde_json::Value>,
+}
+
+/// Write a claimed step's outcome and advance the run graph — all in one
+/// token-gated transaction. A stale worker (claim reclaimed out from under it)
+/// matches zero rows and writes nothing (`AdvanceResult::Stale`).
+///
+/// - `Succeeded` → the step is marked `succeeded` with its output, then the
+///   graph advances (promotions + terminal run status).
+/// - `Failed` with attempts left → the step is re-armed `ready` with a backoff
+///   gate; the run keeps running (`Retried`, no advance).
+/// - `Failed` exhausted → the step is marked `failed`, then the graph advances
+///   (its `on_error` decides whether the run fails or limps on).
+#[allow(clippy::too_many_lines)]
+pub async fn complete_step_and_advance(
+    pool: &PgPool,
+    claimed: &ClaimedStep,
+    outcome: StepOutcome,
+) -> Result<AdvanceResult, WorkflowRepoError> {
+    let mut tx = pool.begin().await?;
+
+    // 1. Token-gated step write.
+    match &outcome {
+        StepOutcome::Succeeded(out) => {
+            let res = sqlx::query(
+                "UPDATE workflow_run_steps \
+                 SET status = 'succeeded', output = $3, 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)
+            .bind(out)
+            .execute(&mut *tx)
+            .await?;
+            if res.rows_affected() == 0 {
+                tx.rollback().await?;
+                return Ok(AdvanceResult::Stale);
+            }
+        }
+        StepOutcome::Failed { error, retry_delay } => {
+            if claimed.attempt < claimed.max_attempts {
+                let next = Utc::now() + *retry_delay;
+                let res = sqlx::query(
+                    "UPDATE workflow_run_steps \
+                     SET status = 'ready', error = $3, claim_token = NULL, \
+                         claimed_at = NULL, next_attempt_at = $4, updated_at = NOW() \
+                     WHERE id = $1 AND claim_token = $2",
+                )
+                .bind(claimed.step_id.into_inner())
+                .bind(claimed.claim_token)
+                .bind(error)
+                .bind(next)
+                .execute(&mut *tx)
+                .await?;
+                if res.rows_affected() == 0 {
+                    tx.rollback().await?;
+                    return Ok(AdvanceResult::Stale);
+                }
+                tx.commit().await?;
+                return Ok(AdvanceResult::Retried);
+            }
+            let res = sqlx::query(
+                "UPDATE workflow_run_steps \
+                 SET status = 'failed', error = $3, 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)
+            .bind(error)
+            .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
+    // serialize (each does a full recompute, so this is idempotent).
+    let run: Option<(String,)> =
+        sqlx::query_as("SELECT status FROM workflow_runs WHERE id = $1 FOR UPDATE")
+            .bind(claimed.run_id.into_inner())
+            .fetch_optional(&mut *tx)
+            .await?;
+    let Some((run_status,)) = run else {
+        tx.commit().await?;
+        return Ok(AdvanceResult::Advanced);
+    };
+    if RunStatus::from_str(&run_status).is_some_and(RunStatus::is_terminal) {
+        // Another worker already finished the run; nothing to do.
+        tx.commit().await?;
+        return Ok(AdvanceResult::Advanced);
+    }
+
+    let step_rows: Vec<StepStateRow> = sqlx::query_as(
+        "SELECT step_name, status, output FROM workflow_run_steps WHERE run_id = $1",
+    )
+    .bind(claimed.run_id.into_inner())
+    .fetch_all(&mut *tx)
+    .await?;
+    let mut statuses = BTreeMap::new();
+    let mut outputs = BTreeMap::new();
+    for r in step_rows {
+        if let Some(st) = StepStatus::from_str(&r.status) {
+            statuses.insert(r.step_name.clone(), st);
+        }
+        if let Some(o) = r.output {
+            outputs.insert(r.step_name, o);
+        }
+    }
+
+    let plan = compute_advance(&claimed.definition, &statuses, &outputs);
+
+    for name in &plan.promote {
+        sqlx::query(
+            "UPDATE workflow_run_steps \
+             SET status = 'ready', next_attempt_at = NULL, updated_at = NOW() \
+             WHERE run_id = $1 AND step_name = $2 AND status = 'pending'",
+        )
+        .bind(claimed.run_id.into_inner())
+        .bind(name)
+        .execute(&mut *tx)
+        .await?;
+    }
+
+    match plan.run_status {
+        RunStatus::Succeeded => {
+            sqlx::query(
+                "UPDATE workflow_runs \
+                 SET status = 'succeeded', output = $2, finished_at = NOW() WHERE id = $1",
+            )
+            .bind(claimed.run_id.into_inner())
+            .bind(plan.run_output)
+            .execute(&mut *tx)
+            .await?;
+        }
+        RunStatus::Failed => {
+            sqlx::query(
+                "UPDATE workflow_runs \
+                 SET status = 'failed', error = $2, finished_at = NOW() WHERE id = $1",
+            )
+            .bind(claimed.run_id.into_inner())
+            .bind(plan.run_error)
+            .execute(&mut *tx)
+            .await?;
+        }
+        _ => {
+            sqlx::query(
+                "UPDATE workflow_runs \
+                 SET status = 'running', started_at = COALESCE(started_at, NOW()) WHERE id = $1",
+            )
+            .bind(claimed.run_id.into_inner())
+            .execute(&mut *tx)
+            .await?;
+        }
+    }
+
+    tx.commit().await?;
+    Ok(AdvanceResult::Advanced)
+}
+
+/// Periodic safety net: a step leased by a crashed worker (still `running`
+/// with a `claim_token` past the visibility timeout) is re-armed `ready` so
+/// another worker retries it. Only touches steps of still-active runs.
+/// Returns the number of steps reclaimed.
+pub async fn reclaim_stale_steps(
+    pool: &PgPool,
+    visibility_timeout_secs: u32,
+) -> Result<u64, WorkflowRepoError> {
+    let res = sqlx::query(
+        "UPDATE workflow_run_steps s \
+         SET status = 'ready', claim_token = NULL, claimed_at = NULL, updated_at = NOW() \
+         FROM workflow_runs r \
+         WHERE s.run_id = r.id \
+           AND s.claim_token IS NOT NULL \
+           AND s.status = 'running' \
+           AND s.claimed_at < NOW() - ($1 || ' seconds')::INTERVAL \
+           AND r.status IN ('pending', 'running')",
+    )
+    .bind(i64::from(visibility_timeout_secs))
+    .execute(pool)
+    .await?;
+    Ok(res.rows_affected())
+}
+
+// ---- run/step reads (admin surface + test assertions) --------------------
+
+#[derive(sqlx::FromRow)]
+struct RunRow {
+    id: Uuid,
+    workflow_id: Uuid,
+    app_id: Uuid,
+    status: String,
+    input: serde_json::Value,
+    output: Option<serde_json::Value>,
+    error: Option<String>,
+    root_execution_id: Uuid,
+    workflow_depth: i32,
+    parent_run_id: Option<Uuid>,
+    parent_step_id: Option<Uuid>,
+    started_at: Option<DateTime<Utc>>,
+    finished_at: Option<DateTime<Utc>>,
+    created_at: DateTime<Utc>,
+}
+
+impl TryFrom<RunRow> for WorkflowRun {
+    type Error = WorkflowRepoError;
+    fn try_from(r: RunRow) -> Result<Self, Self::Error> {
+        Ok(Self {
+            id: r.id.into(),
+            workflow_id: r.workflow_id.into(),
+            app_id: r.app_id.into(),
+            status: RunStatus::from_str(&r.status).ok_or_else(|| {
+                WorkflowRepoError::Serde(format!("bad run status {:?}", r.status))
+            })?,
+            input: r.input,
+            output: r.output,
+            error: r.error,
+            root_execution_id: r.root_execution_id,
+            workflow_depth: r.workflow_depth,
+            parent_run_id: r.parent_run_id.map(Into::into),
+            parent_step_id: r.parent_step_id.map(Into::into),
+            started_at: r.started_at,
+            finished_at: r.finished_at,
+            created_at: r.created_at,
+        })
+    }
+}
+
+const RUN_COLS: &str = "id, workflow_id, app_id, status, input, output, error, \
+    root_execution_id, workflow_depth, parent_run_id, parent_step_id, \
+    started_at, finished_at, created_at";
+
+/// Read a single run by id (app-scoped for isolation).
+pub async fn get_run(
+    pool: &PgPool,
+    app_id: AppId,
+    run_id: WorkflowRunId,
+) -> Result<Option<WorkflowRun>, WorkflowRepoError> {
+    let row: Option<RunRow> = sqlx::query_as(&format!(
+        "SELECT {RUN_COLS} FROM workflow_runs WHERE id = $1 AND app_id = $2"
+    ))
+    .bind(run_id.into_inner())
+    .bind(app_id.into_inner())
+    .fetch_optional(pool)
+    .await?;
+    row.map(TryInto::try_into).transpose()
+}
+
+#[derive(sqlx::FromRow)]
+struct StepRow {
+    id: Uuid,
+    run_id: Uuid,
+    step_name: String,
+    status: String,
+    attempt: i32,
+    max_attempts: i32,
+    output: Option<serde_json::Value>,
+    error: Option<String>,
+    child_run_id: Option<Uuid>,
+}
+
+impl TryFrom<StepRow> for WorkflowRunStep {
+    type Error = WorkflowRepoError;
+    fn try_from(r: StepRow) -> Result<Self, Self::Error> {
+        Ok(Self {
+            id: r.id.into(),
+            run_id: r.run_id.into(),
+            step_name: r.step_name,
+            status: StepStatus::from_str(&r.status).ok_or_else(|| {
+                WorkflowRepoError::Serde(format!("bad step status {:?}", r.status))
+            })?,
+            attempt: r.attempt,
+            max_attempts: r.max_attempts,
+            output: r.output,
+            error: r.error,
+            child_run_id: r.child_run_id.map(Into::into),
+        })
+    }
+}
+
+/// All steps of a run, ordered by name (for the admin/CLI surface + tests).
+pub async fn list_run_steps(
+    pool: &PgPool,
+    run_id: WorkflowRunId,
+) -> Result<Vec<WorkflowRunStep>, WorkflowRepoError> {
+    let rows: Vec<StepRow> = sqlx::query_as(
+        "SELECT id, run_id, step_name, status, attempt, max_attempts, output, error, \
+                child_run_id \
+         FROM workflow_run_steps WHERE run_id = $1 ORDER BY step_name",
+    )
+    .bind(run_id.into_inner())
+    .fetch_all(pool)
+    .await?;
+    rows.into_iter().map(TryInto::try_into).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use picloud_shared::workflow::WorkflowStepDef;
+    use serde_json::json;
+
+    fn step(name: &str, deps: &[&str]) -> WorkflowStepDef {
+        WorkflowStepDef {
+            name: name.into(),
+            function: Some(format!("fn_{name}")),
+            workflow: None,
+            input: serde_json::Value::Null,
+            depends_on: deps.iter().map(|s| (*s).to_string()).collect(),
+            when: None,
+            retry: None,
+            on_error: OnError::Fail,
+        }
+    }
+
+    fn def(steps: Vec<WorkflowStepDef>) -> WorkflowDefinition {
+        WorkflowDefinition { steps }
+    }
+
+    fn statuses(pairs: &[(&str, StepStatus)]) -> BTreeMap<String, StepStatus> {
+        pairs.iter().map(|(n, s)| ((*n).to_string(), *s)).collect()
+    }
+
+    #[test]
+    fn promotes_dependent_when_dep_succeeds() {
+        // a -> b. a succeeded, b still pending → b becomes ready.
+        let d = def(vec![step("a", &[]), step("b", &["a"])]);
+        let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Pending)]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert_eq!(plan.promote, vec!["b".to_string()]);
+        assert_eq!(plan.run_status, RunStatus::Running);
+    }
+
+    #[test]
+    fn fan_in_waits_for_all_deps() {
+        // c depends on [a, b]. Only a done → c must NOT be promoted.
+        let d = def(vec![step("a", &[]), step("b", &[]), step("c", &["a", "b"])]);
+        let st = statuses(&[
+            ("a", StepStatus::Succeeded),
+            ("b", StepStatus::Running),
+            ("c", StepStatus::Pending),
+        ]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert!(plan.promote.is_empty());
+        assert_eq!(plan.run_status, RunStatus::Running);
+
+        // Now both a and b done → c promoted.
+        let st = statuses(&[
+            ("a", StepStatus::Succeeded),
+            ("b", StepStatus::Succeeded),
+            ("c", StepStatus::Pending),
+        ]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert_eq!(plan.promote, vec!["c".to_string()]);
+    }
+
+    #[test]
+    fn all_terminal_succeeds_with_output_object() {
+        let d = def(vec![step("a", &[]), step("b", &["a"])]);
+        let st = statuses(&[("a", StepStatus::Succeeded), ("b", StepStatus::Succeeded)]);
+        let mut out = BTreeMap::new();
+        out.insert("a".to_string(), json!({ "x": 1 }));
+        out.insert("b".to_string(), json!("done"));
+        let plan = compute_advance(&d, &st, &out);
+        assert!(plan.promote.is_empty());
+        assert_eq!(plan.run_status, RunStatus::Succeeded);
+        assert_eq!(
+            plan.run_output,
+            Some(json!({ "a": { "x": 1 }, "b": "done" }))
+        );
+    }
+
+    #[test]
+    fn on_error_fail_exhausted_fails_the_run() {
+        let d = def(vec![step("a", &[]), step("b", &["a"])]);
+        // a failed (on_error = fail) → run fails, b never promoted.
+        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert!(plan.promote.is_empty());
+        assert_eq!(plan.run_status, RunStatus::Failed);
+        assert!(plan.run_error.is_some());
+    }
+
+    #[test]
+    fn on_error_continue_lets_dependents_proceed_and_run_succeed() {
+        let mut a = step("a", &[]);
+        a.on_error = OnError::Continue;
+        let d = def(vec![a, step("b", &["a"])]);
+        // a failed but continue → b promoted.
+        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Pending)]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert_eq!(plan.promote, vec!["b".to_string()]);
+        assert_eq!(plan.run_status, RunStatus::Running);
+
+        // a failed-continue + b succeeded → run succeeds (a contributes no output).
+        let st = statuses(&[("a", StepStatus::Failed), ("b", StepStatus::Succeeded)]);
+        let mut out = BTreeMap::new();
+        out.insert("b".to_string(), json!(42));
+        let plan = compute_advance(&d, &st, &out);
+        assert_eq!(plan.run_status, RunStatus::Succeeded);
+        assert_eq!(plan.run_output, Some(json!({ "b": 42 })));
+    }
+
+    #[test]
+    fn skipped_dep_satisfies_dependents() {
+        let d = def(vec![step("a", &[]), step("b", &["a"])]);
+        let st = statuses(&[("a", StepStatus::Skipped), ("b", StepStatus::Pending)]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert_eq!(plan.promote, vec!["b".to_string()]);
+    }
+
+    #[test]
+    fn parallel_fan_out_promotes_all_ready_children() {
+        // root -> b, root -> c (diamond top). After root succeeds, both promote.
+        let d = def(vec![
+            step("root", &[]),
+            step("b", &["root"]),
+            step("c", &["root"]),
+        ]);
+        let st = statuses(&[
+            ("root", StepStatus::Succeeded),
+            ("b", StepStatus::Pending),
+            ("c", StepStatus::Pending),
+        ]);
+        let plan = compute_advance(&d, &st, &BTreeMap::new());
+        assert_eq!(plan.promote, vec!["b".to_string(), "c".to_string()]);
+        assert_eq!(plan.run_status, RunStatus::Running);
+    }
+}

crates/manager-core/tests/expected_schema.txt

@@ -873,7 +873,7 @@ constraints on email_trigger_details:
   [PRIMARY KEY] email_trigger_details_pkey: PRIMARY KEY (trigger_id)
 
 constraints on execution_logs:
-  [CHECK] execution_logs_source_check: CHECK ((source = ANY (ARRAY['http'::text, 'kv'::text, 'docs'::text, 'dead_letter'::text, 'cron'::text, 'files'::text, 'pubsub'::text, 'email'::text, 'invoke'::text, 'queue'::text])))
+  [CHECK] execution_logs_source_check: CHECK ((source = ANY (ARRAY['http'::text, 'kv'::text, 'docs'::text, 'dead_letter'::text, 'cron'::text, 'files'::text, 'pubsub'::text, 'email'::text, 'invoke'::text, 'queue'::text, 'workflow'::text])))
   [CHECK] execution_logs_status_check: CHECK ((status = ANY (ARRAY['success'::text, 'error'::text, 'timeout'::text, 'budget_exceeded'::text])))
   [FOREIGN KEY] execution_logs_app_id_fk: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
   [FOREIGN KEY] execution_logs_script_id_fkey: FOREIGN KEY (script_id) REFERENCES scripts(id) ON DELETE SET NULL
@@ -1120,3 +1120,4 @@ constraints on workflows:
   0069: email secret version
   0070: admin session absolute expiry
   0071: workflows
+  0072: execution source workflow

crates/manager-core/tests/workflow_orchestrator.rs

@@ -0,0 +1,537 @@
+//! v1.2 Workflows M2 — durable orchestrator integration tests.
+//!
+//! Two layers, both DB-gated (skip cleanly when `DATABASE_URL` is unset):
+//!
+//!   * **State machine** — drive `claim_ready_step` → `complete_step_and_advance`
+//!     in a deterministic loop (no executor, feeding canned outcomes) to pin the
+//!     durable DAG advance: linear chains, parallel fan-out/fan-in, retry,
+//!     on_error fail vs continue, stale-token idempotency, and stale-lease
+//!     reclaim.
+//!   * **End-to-end** — a real orchestrator `tick()` with a fake `ExecutorClient`
+//!     and real `scripts` rows, asserting steps resolve + execute + advance to a
+//!     terminal run.
+
+#![allow(clippy::too_many_lines)]
+
+use std::sync::{Arc, LazyLock};
+
+use async_trait::async_trait;
+use picloud_executor_core::{ExecError, ExecRequest, ExecResponse, ExecStats};
+use picloud_manager_core::repo::PostgresScriptRepository;
+use picloud_manager_core::workflow_orchestrator::{
+    WorkflowOrchestrator, WorkflowOrchestratorConfig,
+};
+use picloud_manager_core::workflow_repo::{
+    claim_ready_step, complete_step_and_advance, get_run, list_run_steps, reclaim_stale_steps,
+    start_run, AdvanceResult, ClaimedStep, NewWorkflowRun, StepOutcome,
+};
+use picloud_orchestrator_core::{ExecutionGate, ExecutorClient, ScriptIdentity};
+use picloud_shared::workflow::{
+    OnError, RunStatus, StepStatus, WorkflowBackoff, WorkflowDefinition, WorkflowRetry,
+    WorkflowStepDef,
+};
+use picloud_shared::{AppId, ExecutionLogSink, LogSinkError, WorkflowId};
+use serde_json::{json, Value};
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use std::time::Duration as StdDuration;
+use uuid::Uuid;
+
+/// `claim_ready_step` is deliberately **global** (one production orchestrator
+/// claims every app's ready steps). These tests share one DB, so a sibling
+/// test's concurrent claim would steal steps from the run under assertion. This
+/// process-wide lock serializes the claim-driving tests — each still creates
+/// its own app/workflow/run, and each finishes with no `ready` steps, so once
+/// serialized they don't interfere. (See the shared-DB test-harness note.)
+static CLAIM_LOCK: LazyLock<tokio::sync::Mutex<()>> = LazyLock::new(|| tokio::sync::Mutex::new(()));
+
+/// Acquire the serialization lock AND clear any leftover runs/steps so the
+/// global `claim_ready_step`/`reclaim_stale_steps` see only this test's rows.
+/// (The shared dev DB accumulates rows across runs; `workflow_runs` CASCADEs to
+/// its steps.) Returns the held guard.
+async fn lock_and_reset(pool: &PgPool) -> tokio::sync::MutexGuard<'static, ()> {
+    let guard = CLAIM_LOCK.lock().await;
+    sqlx::query("DELETE FROM workflow_runs")
+        .execute(pool)
+        .await
+        .expect("reset workflow_runs");
+    guard
+}
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("workflow_orchestrator: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(4)
+        .connect(&url)
+        .await
+        .expect("connect to DATABASE_URL");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("apply migrations");
+    Some(pool)
+}
+
+fn step(name: &str, function: &str, deps: &[&str]) -> WorkflowStepDef {
+    WorkflowStepDef {
+        name: name.into(),
+        function: Some(function.into()),
+        workflow: None,
+        input: Value::Null,
+        depends_on: deps.iter().map(|s| (*s).to_string()).collect(),
+        when: None,
+        retry: None,
+        on_error: OnError::Fail,
+    }
+}
+
+/// Insert an app + a workflow definition; return `(app_id, workflow_id)`.
+async fn seed_workflow(pool: &PgPool, def: &WorkflowDefinition) -> (AppId, WorkflowId) {
+    let sfx = Uuid::new_v4().simple().to_string();
+    let grp: (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("wf-grp-{sfx}"))
+            .fetch_one(pool)
+            .await
+            .expect("insert group");
+    let app: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("wf-app-{sfx}"))
+            .bind(grp.0)
+            .fetch_one(pool)
+            .await
+            .expect("insert app");
+    let def_json = serde_json::to_value(def).unwrap();
+    let wf: (Uuid,) = sqlx::query_as(
+        "INSERT INTO workflows (app_id, name, definition) VALUES ($1, $2, $3) RETURNING id",
+    )
+    .bind(app.0)
+    .bind(format!("wf-{sfx}"))
+    .bind(def_json)
+    .fetch_one(pool)
+    .await
+    .expect("insert workflow");
+    (app.0.into(), wf.0.into())
+}
+
+fn new_run(app_id: AppId, wf_id: WorkflowId) -> NewWorkflowRun {
+    NewWorkflowRun {
+        workflow_id: wf_id,
+        app_id,
+        input: json!({ "seed": true }),
+        root_execution_id: Uuid::new_v4(),
+        workflow_depth: 0,
+        parent_run_id: None,
+        parent_step_id: None,
+    }
+}
+
+/// Drive the run to a terminal state by claiming + completing steps, deciding
+/// each outcome via `decide` (keyed on step name + current attempt).
+async fn run_to_terminal<F>(pool: &PgPool, mut decide: F)
+where
+    F: FnMut(&ClaimedStep) -> StepOutcome,
+{
+    // Bounded so a bug can't spin forever.
+    for _ in 0..200 {
+        let Some(claimed) = claim_ready_step(pool).await.expect("claim") else {
+            break;
+        };
+        let outcome = decide(&claimed);
+        complete_step_and_advance(pool, &claimed, outcome)
+            .await
+            .expect("advance");
+    }
+}
+
+fn ok(body: Value) -> StepOutcome {
+    StepOutcome::Succeeded(body)
+}
+
+fn fail(msg: &str) -> StepOutcome {
+    StepOutcome::Failed {
+        error: msg.into(),
+        retry_delay: chrono::Duration::zero(),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn seeds_roots_ready_and_completes_linear_chain() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    // a -> b -> c
+    let def = WorkflowDefinition {
+        steps: vec![
+            step("a", "fn_a", &[]),
+            step("b", "fn_b", &["a"]),
+            step("c", "fn_c", &["b"]),
+        ],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    // Root `a` starts ready; b, c pending.
+    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::Ready);
+    assert_eq!(by("b"), StepStatus::Pending);
+    assert_eq!(by("c"), StepStatus::Pending);
+
+    run_to_terminal(&pool, |c| ok(json!({ "step": c.step_name }))).await;
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+    assert_eq!(
+        r.output,
+        Some(json!({
+            "a": { "step": "a" },
+            "b": { "step": "b" },
+            "c": { "step": "c" },
+        }))
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn parallel_fan_out_and_fan_in() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    // a -> {b, c} -> d (diamond).
+    let def = WorkflowDefinition {
+        steps: vec![
+            step("a", "fn_a", &[]),
+            step("b", "fn_b", &["a"]),
+            step("c", "fn_c", &["a"]),
+            step("d", "fn_d", &["b", "c"]),
+        ],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    // After `a` completes, BOTH b and c must be claimable before d.
+    let a = claim_ready_step(&pool).await.unwrap().expect("claim a");
+    assert_eq!(a.step_name, "a");
+    complete_step_and_advance(&pool, &a, ok(json!("a")))
+        .await
+        .unwrap();
+
+    let s1 = claim_ready_step(&pool)
+        .await
+        .unwrap()
+        .expect("first parallel");
+    let s2 = claim_ready_step(&pool)
+        .await
+        .unwrap()
+        .expect("second parallel");
+    let mut names = [s1.step_name.clone(), s2.step_name.clone()];
+    names.sort();
+    assert_eq!(names, ["b".to_string(), "c".to_string()]);
+    // d must NOT be claimable yet (fan-in waits).
+    assert!(claim_ready_step(&pool).await.unwrap().is_none());
+
+    complete_step_and_advance(&pool, &s1, ok(json!("x")))
+        .await
+        .unwrap();
+    // Still waiting on the other parallel branch.
+    assert!(claim_ready_step(&pool).await.unwrap().is_none());
+    complete_step_and_advance(&pool, &s2, ok(json!("y")))
+        .await
+        .unwrap();
+
+    let d = claim_ready_step(&pool)
+        .await
+        .unwrap()
+        .expect("d after fan-in");
+    assert_eq!(d.step_name, "d");
+    complete_step_and_advance(&pool, &d, ok(json!("d")))
+        .await
+        .unwrap();
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn retry_then_succeed() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let mut a = step("a", "fn_a", &[]);
+    a.retry = Some(WorkflowRetry {
+        max_attempts: 3,
+        backoff: WorkflowBackoff::Constant,
+        base_ms: 0, // no delay → immediately re-claimable
+    });
+    let def = WorkflowDefinition { steps: vec![a] };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    // First claim fails (attempt 1) → retried.
+    let c1 = claim_ready_step(&pool).await.unwrap().expect("attempt 1");
+    assert_eq!(c1.attempt, 1);
+    let res = complete_step_and_advance(&pool, &c1, fail("boom"))
+        .await
+        .unwrap();
+    assert_eq!(res, AdvanceResult::Retried);
+
+    // Second claim (attempt 2) succeeds → run done.
+    let c2 = claim_ready_step(&pool).await.unwrap().expect("attempt 2");
+    assert_eq!(c2.attempt, 2);
+    complete_step_and_advance(&pool, &c2, ok(json!("recovered")))
+        .await
+        .unwrap();
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn on_error_fail_fails_the_run() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    // a (fail) -> b; a has no retry and on_error=fail → run fails, b never runs.
+    let def = WorkflowDefinition {
+        steps: vec![step("a", "fn_a", &[]), step("b", "fn_b", &["a"])],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    run_to_terminal(&pool, |c| {
+        if c.step_name == "a" {
+            fail("kaboom")
+        } else {
+            ok(json!("b"))
+        }
+    })
+    .await;
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Failed);
+    assert!(r.error.unwrap().contains('a'));
+    let steps = list_run_steps(&pool, run).await.unwrap();
+    let b = steps.iter().find(|s| s.step_name == "b").unwrap();
+    assert_eq!(b.status, StepStatus::Pending); // never promoted
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn on_error_continue_run_succeeds() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let mut a = step("a", "fn_a", &[]);
+    a.on_error = OnError::Continue;
+    let def = WorkflowDefinition {
+        steps: vec![a, step("b", "fn_b", &["a"])],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    run_to_terminal(&pool, |c| {
+        if c.step_name == "a" {
+            fail("soft")
+        } else {
+            ok(json!("b-ran"))
+        }
+    })
+    .await;
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(
+        r.status,
+        RunStatus::Succeeded,
+        "continue lets the run finish"
+    );
+    assert_eq!(r.output, Some(json!({ "b": "b-ran" })));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn double_complete_is_idempotent() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let def = WorkflowDefinition {
+        steps: vec![step("a", "fn_a", &[])],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    let c = claim_ready_step(&pool).await.unwrap().expect("claim");
+    let first = complete_step_and_advance(&pool, &c, ok(json!("first")))
+        .await
+        .unwrap();
+    assert_eq!(first, AdvanceResult::Advanced);
+
+    // Replaying the SAME claim (stale token, lease long gone) writes nothing.
+    let second = complete_step_and_advance(&pool, &c, ok(json!("SECOND")))
+        .await
+        .unwrap();
+    assert_eq!(second, AdvanceResult::Stale);
+
+    let steps = list_run_steps(&pool, run).await.unwrap();
+    assert_eq!(steps[0].output, Some(json!("first")), "first write wins");
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn stale_lease_is_reclaimed() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    let def = WorkflowDefinition {
+        steps: vec![step("a", "fn_a", &[])],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    let c = claim_ready_step(&pool).await.unwrap().expect("claim");
+    // Nothing else claimable while `a` is leased.
+    assert!(claim_ready_step(&pool).await.unwrap().is_none());
+
+    // Backdate the lease so it's past the (0s) visibility window.
+    sqlx::query(
+        "UPDATE workflow_run_steps SET claimed_at = NOW() - INTERVAL '1 hour' WHERE id = $1",
+    )
+    .bind(c.step_id.into_inner())
+    .execute(&pool)
+    .await
+    .unwrap();
+    let reclaimed = reclaim_stale_steps(&pool, 0).await.unwrap();
+    assert_eq!(reclaimed, 1);
+
+    // Now claimable again (a fresh attempt).
+    let c2 = claim_ready_step(&pool)
+        .await
+        .unwrap()
+        .expect("re-claim after reclaim");
+    assert_eq!(c2.step_name, "a");
+    complete_step_and_advance(&pool, &c2, ok(json!("done")))
+        .await
+        .unwrap();
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+}
+
+// ---- end-to-end: a real orchestrator tick with a fake executor ------------
+
+struct FakeExecutor {
+    calls: std::sync::Mutex<Vec<String>>,
+}
+
+#[async_trait]
+impl ExecutorClient for FakeExecutor {
+    async fn execute(
+        &self,
+        _source: &str,
+        req: ExecRequest,
+        _timeout: StdDuration,
+    ) -> Result<ExecResponse, ExecError> {
+        self.calls.lock().unwrap().push(req.script_name.clone());
+        Ok(ExecResponse {
+            status_code: 200,
+            headers: std::collections::BTreeMap::new(),
+            body: json!({ "ran": req.script_name }),
+            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
+    }
+}
+
+struct NoopLogSink;
+#[async_trait]
+impl ExecutionLogSink for NoopLogSink {
+    async fn record(&self, _log: picloud_shared::ExecutionLog) -> Result<(), LogSinkError> {
+        Ok(())
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn end_to_end_tick_executes_functions_and_completes() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let _claim_guard = lock_and_reset(&pool).await;
+    // a -> b; both bound to real scripts (fn_a, fn_b) so name resolution works.
+    let def = WorkflowDefinition {
+        steps: vec![step("a", "fn_a", &[]), step("b", "fn_b", &["a"])],
+    };
+    let (app, wf) = seed_workflow(&pool, &def).await;
+    for fn_name in ["fn_a", "fn_b"] {
+        sqlx::query("INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2)")
+            .bind(fn_name)
+            .bind(app.into_inner())
+            .execute(&pool)
+            .await
+            .expect("insert script");
+    }
+    let run = start_run(&pool, new_run(app, wf), &def)
+        .await
+        .expect("start");
+
+    let executor = Arc::new(FakeExecutor {
+        calls: std::sync::Mutex::new(vec![]),
+    });
+    let orch = WorkflowOrchestrator {
+        pool: pool.clone(),
+        scripts: Arc::new(PostgresScriptRepository::new(pool.clone())),
+        executor: executor.clone(),
+        gate: Arc::new(ExecutionGate::new(8)),
+        log_sink: Arc::new(NoopLogSink),
+        config: WorkflowOrchestratorConfig::default(),
+    };
+
+    // Drive ticks until the run is terminal (bounded).
+    for _ in 0..50 {
+        orch.tick_once().await;
+        let r = get_run(&pool, app, run).await.unwrap().unwrap();
+        if r.status.is_terminal() {
+            break;
+        }
+    }
+
+    let r = get_run(&pool, app, run).await.unwrap().unwrap();
+    assert_eq!(r.status, RunStatus::Succeeded);
+    let calls = executor.calls.lock().unwrap().clone();
+    assert!(calls.contains(&"fn_a".to_string()));
+    assert!(calls.contains(&"fn_b".to_string()));
+    assert_eq!(
+        r.output,
+        Some(json!({ "a": { "ran": "fn_a" }, "b": { "ran": "fn_b" } }))
+    );
+}