From 44f992cbb084e752bf63b2d2aa0ba6ba7de3a9f1 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sun, 12 Jul 2026 16:46:32 +0200 Subject: [PATCH] =?UTF-8?q?feat(workflows):=20M1=20=E2=80=94=20schema=20+?= =?UTF-8?q?=20definition=20validation=20+=20declarative=20reconcile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First milestone of the v1.2 Workflows track (blueprint §9.1/§9.2): the durable DAG engine's foundation — the definition model, its validation, and the declarative `apply` reconcile path. No execution yet (the orchestrator is M2). - Migration 0071_workflows: `workflows` (owner-polymorphic like scripts, app- owned in M1), `workflow_runs`, and `workflow_run_steps` (the per-step competing-consumer lease table the M2 orchestrator will claim). Schema golden reblessed. - `shared::workflow`: the `WorkflowDefinition` / `WorkflowStepDef` DTOs (shared by the CLI, apply, and the future orchestrator) + run/step status enums. - Pure, DB-free `workflow_template` (input `{{ input.x }}` / `{{ steps.a.output.y }}` resolution, type-preserving) and `workflow_expr` (a small safe JSON-predicate evaluator for `when` — keeps manager-core free of a scripting engine). - `workflow_repo`: read trait + reconcile tx free-fns (insert/update/delete). - apply_service: `BundleWorkflow` + `Plan.workflows` + `CurrentState.workflows` + `ApplyReport.workflows_*`; server-side `validate_workflow_definition` (unique/acyclic steps via topological sort, deps exist, function XOR workflow, `when`/template parse) rejecting on a group node; `diff_workflows` by lower(name) (Update on a definition change) + in-tx reconcile. - CLI: `[[workflows]]` + `[[workflows.steps]]` manifest structs → wire bundle; plan/apply render + report counts. - Tests: 16 manager-core lib tests (template, expr, definition validation) + the `workflows` CLI journey (apply → NoOp → prune; cyclic DAG rejected). Deferred to later milestones: the orchestrator worker (M2), conditional/template runtime wiring (M3), nested sub-workflows (M4), the SDK/API/CLI run surface (M5), the dashboard (M6). The `group_id` column + run/step tables ship now so those slot in without a migration churn. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../migrations/0071_workflows.sql | 120 +++++ crates/manager-core/src/apply_service.rs | 345 ++++++++++++ crates/manager-core/src/lib.rs | 3 + crates/manager-core/src/workflow_expr.rs | 490 ++++++++++++++++++ crates/manager-core/src/workflow_repo.rs | 229 ++++++++ crates/manager-core/src/workflow_template.rs | 315 +++++++++++ crates/manager-core/tests/expected_schema.txt | 82 +++ crates/picloud-cli/src/client.rs | 10 + crates/picloud-cli/src/cmds/apply.rs | 14 + crates/picloud-cli/src/cmds/init.rs | 1 + crates/picloud-cli/src/cmds/plan.rs | 24 +- crates/picloud-cli/src/cmds/pull.rs | 1 + crates/picloud-cli/src/manifest.rs | 64 +++ crates/picloud-cli/tests/cli.rs | 1 + crates/picloud-cli/tests/workflows.rs | 155 ++++++ crates/shared/src/ids.rs | 3 + crates/shared/src/lib.rs | 3 +- crates/shared/src/workflow.rs | 202 ++++++++ 18 files changed, 2059 insertions(+), 3 deletions(-) create mode 100644 crates/manager-core/migrations/0071_workflows.sql create mode 100644 crates/manager-core/src/workflow_expr.rs create mode 100644 crates/manager-core/src/workflow_repo.rs create mode 100644 crates/manager-core/src/workflow_template.rs create mode 100644 crates/picloud-cli/tests/workflows.rs create mode 100644 crates/shared/src/workflow.rs diff --git a/crates/manager-core/migrations/0071_workflows.sql b/crates/manager-core/migrations/0071_workflows.sql new file mode 100644 index 0000000..0e0ac3e --- /dev/null +++ b/crates/manager-core/migrations/0071_workflows.sql @@ -0,0 +1,120 @@ +-- v1.2 Workflows (blueprint §9.1/§9.2): a durable DAG orchestration engine. +-- +-- A `workflow` is a declarative graph of steps; each step invokes a function +-- (a script, by name, resolved in the app's scope) or a nested sub-workflow, +-- with `depends_on` edges, a per-step `when` condition, input mapping, and a +-- per-step retry + `on_error` policy. A `workflow::start(name, input)` (or the +-- admin run API) creates a RUN; a dedicated background orchestrator advances it +-- step-by-step, durably, surviving restarts. +-- +-- Three tables: +-- * workflows — the definition (owner-polymorphic like scripts/0050: +-- app XOR group; M1 authors app-owned only, the group_id +-- column ships unused so group-owned templates stay a +-- door-open follow-up). +-- * workflow_runs — one row per start(); the durable run record. +-- * workflow_run_steps — one row per step per run; carries the competing- +-- consumer lease (claim_token/claimed_at) exactly like +-- queue_messages (0034), so parallel steps complete +-- lock-free and a crashed worker's step is reclaimable. + +-- --------------------------------------------------------------------------- +-- workflows: the definition (owner-polymorphic, mirrors scripts/0050). +-- --------------------------------------------------------------------------- +CREATE TABLE workflows ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + -- exactly one of app_id / group_id (the exactly-one CHECK below). Code is + -- not data, so RESTRICT (a group can't be deleted out from under a workflow + -- it owns), mirroring scripts. + app_id UUID NULL REFERENCES apps(id) ON DELETE CASCADE, + group_id UUID NULL REFERENCES groups(id) ON DELETE RESTRICT, + name TEXT NOT NULL, + -- the parsed DAG: { "steps": [ { name, function|workflow, input, depends_on, + -- when, retry, on_error } ] }. Validated server-side (acyclic, unique step + -- names, deps exist) before it is ever written. + definition JSONB NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT workflows_owner_exactly_one CHECK ((group_id IS NULL) <> (app_id IS NULL)) +); + +-- Per-owner, case-insensitive name uniqueness (partial, one per owner column). +CREATE UNIQUE INDEX workflows_app_name_uidx + ON workflows (app_id, LOWER(name)) WHERE app_id IS NOT NULL; +CREATE UNIQUE INDEX workflows_group_name_uidx + ON workflows (group_id, LOWER(name)) WHERE group_id IS NOT NULL; +CREATE INDEX workflows_group_id_idx ON workflows (group_id) WHERE group_id IS NOT NULL; + +-- --------------------------------------------------------------------------- +-- workflow_runs: one row per start(). Carries app_id NOT NULL (the data-plane +-- invariant — every run-listing query filters by app without a join) even +-- though it is derivable from the workflow. +-- --------------------------------------------------------------------------- +CREATE TABLE workflow_runs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + workflow_id UUID NOT NULL REFERENCES workflows(id) ON DELETE RESTRICT, + app_id UUID NOT NULL REFERENCES apps(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'running', 'succeeded', 'failed', 'canceled')), + input JSONB NOT NULL DEFAULT '{}'::jsonb, + output JSONB NULL, + error TEXT NULL, + -- correlates every step execution in execution_logs under one id. + root_execution_id UUID NOT NULL, + -- nesting: a sub-workflow run links back to the parent step that spawned it. + workflow_depth INT NOT NULL DEFAULT 0, + parent_run_id UUID NULL REFERENCES workflow_runs(id) ON DELETE SET NULL, + parent_step_id UUID NULL, + started_at TIMESTAMPTZ NULL, + finished_at TIMESTAMPTZ NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX workflow_runs_app_status_idx ON workflow_runs (app_id, status); +CREATE INDEX workflow_runs_workflow_idx ON workflow_runs (workflow_id, created_at DESC); +CREATE INDEX workflow_runs_parent_idx ON workflow_runs (parent_run_id) WHERE parent_run_id IS NOT NULL; + +-- --------------------------------------------------------------------------- +-- workflow_run_steps: one row per step per run. The claim_token/claimed_at +-- lease is the queue_messages (0034) competing-consumer pattern — the +-- orchestrator claims a `ready` step FOR UPDATE SKIP LOCKED, and the reclaim +-- task frees a stale lease so a crashed worker's step is retried. +-- --------------------------------------------------------------------------- +CREATE TABLE workflow_run_steps ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL REFERENCES workflow_runs(id) ON DELETE CASCADE, + step_name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'ready', 'running', 'succeeded', 'failed', 'skipped')), + attempt INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 1, + output JSONB NULL, + error TEXT NULL, + -- competing-consumer lease (mirrors queue_messages). + claim_token UUID NULL, + claimed_at TIMESTAMPTZ NULL, + -- retry backoff / initial dispatch gate (NULL = due immediately once ready). + next_attempt_at TIMESTAMPTZ NULL, + -- set when this step is a nested sub-workflow: the child run it is waiting on. + child_run_id UUID NULL REFERENCES workflow_runs(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (run_id, step_name) +); + +-- The orchestrator's claim scan: ready steps whose backoff gate has elapsed. +-- (deliver-gate NOW() comparison is applied as a filter on the small partial +-- set, as in queue_messages.) +CREATE INDEX workflow_run_steps_claimable_idx + ON workflow_run_steps (next_attempt_at) + WHERE status = 'ready'; + +-- Advance re-eval + reclaim scans by run. +CREATE INDEX workflow_run_steps_run_idx ON workflow_run_steps (run_id); + +-- Reclaim-task scan: leased steps whose claim is older than the visibility +-- timeout (bounded by in-flight step count). +CREATE INDEX workflow_run_steps_claimed_idx + ON workflow_run_steps (claimed_at) + WHERE claim_token IS NOT NULL; diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index e9690bc..8bea2a5 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -103,6 +103,20 @@ pub struct Bundle { /// declines (the binding 404s instead of serving). App-only. #[serde(default)] pub suppress_routes: Vec, + /// v1.2 Workflows: declared DAG workflow definitions. App-owned (M1); a + /// `[group]` carrying workflows is rejected in `validate_bundle_for`. + #[serde(default)] + pub workflows: Vec, +} + +/// One declared workflow on the wire: a name + its (already-parsed) DAG. +#[derive(Debug, Clone, Deserialize)] +pub struct BundleWorkflow { + pub name: String, + pub definition: picloud_shared::workflow::WorkflowDefinition, + /// Three-state lifecycle (§4.3); omitted ⇒ active. + #[serde(default = "picloud_shared::default_true")] + pub enabled: bool, } /// One declared shared-collection marker on the wire: a name + its store kind. @@ -450,6 +464,9 @@ pub struct Plan { /// §11 tail per-app suppression markers, keyed `"{target_kind}:{reference}"`. #[serde(default)] pub suppressions: Vec, + /// v1.2 Workflows: workflow definitions, keyed by name. + #[serde(default)] + pub workflows: Vec, } impl Plan { @@ -464,6 +481,8 @@ impl Plan { .chain(&self.vars) .chain(&self.extension_points) .chain(&self.collections) + .chain(&self.suppressions) + .chain(&self.workflows) .all(|c| c.op == Op::NoOp) } } @@ -744,6 +763,8 @@ pub struct CurrentState { /// §11 tail per-app suppression markers at this node (app-only), as /// `(target_kind, reference)` pairs. pub suppressions: Vec<(String, String)>, + /// v1.2 Workflows owned directly by this node (app-owned; empty for a group). + pub workflows: Vec, } /// One row of the read-only extension-point report (§5.5). @@ -1125,6 +1146,20 @@ impl ApplyService { // subtree. Both are inheritance-only (the dispatch filters gate to // group-owned templates on the suppressing owner's chain), so no // owner-kind guard here. + // + // v1.2 Workflows: app-owned only (M1). Reject on a group node (the + // `group_id` column ships unused). Then structurally validate each + // definition (unique/acyclic steps, deps exist, when/template parse). + if is_group && !bundle.workflows.is_empty() { + return Err(ApplyError::Invalid( + "a group cannot declare workflows — they are app-owned (group-owned \ + workflow templates are a future addition)" + .into(), + )); + } + for w in &bundle.workflows { + validate_workflow_definition(&w.name, &w.definition).map_err(ApplyError::Invalid)?; + } self.validate_bundle(bundle, inherited_endpoints) } @@ -1568,6 +1603,62 @@ impl ApplyService { } } } + + // v1.2 Workflows (app-owned in M1). Create/Update always; Delete only + // under --prune (mirrors scripts). Function / sub-workflow references + // are resolved at RUN time, so there is no in-tx ordering dependency on + // the scripts applied above. + if let ApplyOwner::App(app_id) = owner { + let cur_by_name: HashMap = current + .workflows + .iter() + .map(|w| (w.name.to_lowercase(), w)) + .collect(); + let desired_by_name: HashMap = bundle + .workflows + .iter() + .map(|w| (w.name.to_lowercase(), w)) + .collect(); + for ch in &plan.workflows { + let key = ch.key.to_lowercase(); + match ch.op { + Op::Create => { + let w = desired_by_name[&key]; + crate::workflow_repo::insert_workflow_tx( + &mut *tx, + app_id, + &w.name, + &w.definition, + w.enabled, + ) + .await + .map_err(|e| ApplyError::Backend(e.to_string()))?; + report.workflows_created += 1; + } + Op::Update => { + let w = desired_by_name[&key]; + let cur = cur_by_name[&key]; + crate::workflow_repo::update_workflow_tx( + &mut *tx, + cur.id, + &w.definition, + w.enabled, + ) + .await + .map_err(|e| ApplyError::Backend(e.to_string()))?; + report.workflows_updated += 1; + } + Op::Delete if prune => { + let cur = cur_by_name[&key]; + crate::workflow_repo::delete_workflow_tx(&mut *tx, cur.id) + .await + .map_err(|e| ApplyError::Backend(e.to_string()))?; + report.workflows_deleted += 1; + } + Op::NoOp | Op::Delete => {} + } + } + } Ok(name_to_id) } @@ -3934,6 +4025,15 @@ impl ApplyService { crate::suppression_repo::list_for_owner(&self.pool, owner.as_script_owner()) .await .map_err(|e| ApplyError::Backend(e.to_string()))?; + // v1.2 Workflows are app-owned (M1); a group node owns none. + let workflows = match owner { + ApplyOwner::App(app_id) => { + crate::workflow_repo::list_workflows_for_app(&self.pool, app_id) + .await + .map_err(|e| ApplyError::Backend(e.to_string()))? + } + ApplyOwner::Group(_) => Vec::new(), + }; Ok(CurrentState { scripts, routes, @@ -3943,6 +4043,7 @@ impl ApplyService { extension_point_names, collections, suppressions, + workflows, }) } @@ -4008,9 +4109,160 @@ fn compute_diff_with_names( extension_points: diff_extension_points(current, bundle), collections: diff_collections(current, bundle), suppressions: diff_suppressions(current, bundle), + workflows: diff_workflows(current, bundle), } } +/// Diff workflows by `lower(name)`. Like scripts, a workflow has a stable name +/// identity with a mutable body, so a changed definition (or `enabled`) is an +/// `Update`, not a delete+create. Live workflows absent from the manifest are +/// `Delete` (applied only under `--prune`). +fn diff_workflows(current: &CurrentState, bundle: &Bundle) -> Vec { + let by_name: HashMap = current + .workflows + .iter() + .map(|w| (w.name.to_lowercase(), w)) + .collect(); + let desired: HashSet = bundle + .workflows + .iter() + .map(|w| w.name.to_lowercase()) + .collect(); + + let mut out = Vec::new(); + for w in &bundle.workflows { + match by_name.get(&w.name.to_lowercase()) { + None => out.push(ResourceChange { + op: Op::Create, + key: w.name.clone(), + detail: None, + }), + Some(cur) => { + let reason = if cur.definition != w.definition { + Some("definition changed".to_string()) + } else if cur.enabled != w.enabled { + Some(if w.enabled { "enabled" } else { "disabled" }.to_string()) + } else { + None + }; + out.push(ResourceChange { + op: if reason.is_some() { + Op::Update + } else { + Op::NoOp + }, + key: w.name.clone(), + detail: reason, + }); + } + } + } + for w in ¤t.workflows { + if !desired.contains(&w.name.to_lowercase()) { + out.push(ResourceChange { + op: Op::Delete, + key: w.name.clone(), + detail: None, + }); + } + } + out +} + +/// Pure structural validation of a workflow definition (mirrors +/// `validate_trigger_shape`): non-empty; unique step names; each step targets +/// exactly one of function/workflow; `depends_on` references exist and aren't +/// self-loops; the DAG is acyclic; every `when` + input template parses and +/// only references declared steps. Returns a human error string on failure. +fn validate_workflow_definition( + wf_name: &str, + def: &picloud_shared::workflow::WorkflowDefinition, +) -> Result<(), String> { + use std::collections::{BTreeSet, HashMap as Map}; + if def.steps.is_empty() { + return Err(format!("workflow `{wf_name}` has no steps")); + } + let mut names: BTreeSet = BTreeSet::new(); + for s in &def.steps { + if s.name.trim().is_empty() { + return Err(format!( + "workflow `{wf_name}` has a step with an empty name" + )); + } + if !names.insert(s.name.clone()) { + return Err(format!( + "workflow `{wf_name}` has a duplicate step name `{}`", + s.name + )); + } + } + for s in &def.steps { + if s.target().is_none() { + return Err(format!( + "workflow `{wf_name}` step `{}` must set exactly one of `function` or `workflow`", + s.name + )); + } + for dep in &s.depends_on { + if dep == &s.name { + return Err(format!( + "workflow `{wf_name}` step `{}` depends on itself", + s.name + )); + } + if !names.contains(dep) { + return Err(format!( + "workflow `{wf_name}` step `{}` depends on undeclared step `{dep}`", + s.name + )); + } + } + if let Some(w) = &s.when { + crate::workflow_expr::validate(w, &names).map_err(|e| { + format!( + "workflow `{wf_name}` step `{}` has a bad `when`: {e}", + s.name + ) + })?; + } + crate::workflow_template::validate(&s.input, &names).map_err(|e| { + format!( + "workflow `{wf_name}` step `{}` has a bad input template: {e}", + s.name + ) + })?; + } + // DAG acyclicity via Kahn topological sort. + let mut indeg: Map<&str, usize> = def.steps.iter().map(|s| (s.name.as_str(), 0)).collect(); + for s in &def.steps { + for _dep in &s.depends_on { + *indeg.get_mut(s.name.as_str()).unwrap() += 1; + } + } + let mut queue: Vec<&str> = indeg + .iter() + .filter(|(_, d)| **d == 0) + .map(|(n, _)| *n) + .collect(); + let mut seen = 0usize; + while let Some(n) = queue.pop() { + seen += 1; + for s in &def.steps { + if s.depends_on.iter().any(|d| d == n) { + let e = indeg.get_mut(s.name.as_str()).unwrap(); + *e -= 1; + if *e == 0 { + queue.push(s.name.as_str()); + } + } + } + } + if seen != def.steps.len() { + return Err(format!("workflow `{wf_name}` has a dependency cycle")); + } + Ok(()) +} + /// Diff app-owned vars by key. Unlike secrets (whose values live out-of-band), /// a var's value IS in the manifest, so equality is value-sensitive: a changed /// value is an `Update`. Live vars absent from the manifest are `Delete` @@ -5140,6 +5392,12 @@ pub struct ApplyReport { pub suppressions_created: u32, #[serde(default)] pub suppressions_deleted: u32, + #[serde(default)] + pub workflows_created: u32, + #[serde(default)] + pub workflows_updated: u32, + #[serde(default)] + pub workflows_deleted: u32, #[serde(skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, } @@ -5483,6 +5741,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), }; let plan = compute_diff(¤t, &bundle); assert_eq!(plan.scripts.len(), 1); @@ -5509,6 +5768,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), }; let plan = compute_diff(¤t, &bundle); assert!(plan.is_noop(), "expected all no-op, got {plan:?}"); @@ -5530,6 +5790,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), }; let plan = compute_diff(¤t, &bundle); assert_eq!(plan.scripts[0].op, Op::Update); @@ -5552,6 +5813,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), }; let plan = compute_diff(¤t, &bundle); assert_eq!(plan.scripts[0].op, Op::Delete); @@ -5605,6 +5867,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), }; let plan = compute_diff(¤t, &bundle); assert_eq!(plan.routes[0].op, Op::Update); @@ -5621,6 +5884,7 @@ mod tests { collections: Vec::new(), suppress_triggers: Vec::new(), suppress_routes: Vec::new(), + workflows: Vec::new(), } } @@ -6350,4 +6614,85 @@ mod tests { assert!(!bare.confirm_required("production")); assert!(bare.gated_envs().is_empty()); } + + // ---- v1.2 Workflows: definition validation ---------------------------- + + fn wf_def(steps: serde_json::Value) -> picloud_shared::workflow::WorkflowDefinition { + serde_json::from_value(serde_json::json!({ "steps": steps })).expect("valid def json") + } + + #[test] + fn workflow_validation_accepts_a_valid_dag() { + let def = wf_def(serde_json::json!([ + { "name": "a", "function": "fa" }, + { "name": "b", "function": "fb", "depends_on": ["a"], + "when": "steps.a.output.ok == true", + "input": { "x": "{{ steps.a.output.value }}" } }, + { "name": "c", "workflow": "sub", "depends_on": ["a", "b"] }, + ])); + assert!(validate_workflow_definition("w", &def).is_ok()); + } + + #[test] + fn workflow_validation_rejects_empty_and_dupes() { + assert!(validate_workflow_definition("w", &wf_def(serde_json::json!([]))).is_err()); + let dupe = wf_def(serde_json::json!([ + { "name": "a", "function": "fa" }, + { "name": "a", "function": "fb" }, + ])); + assert!(validate_workflow_definition("w", &dupe) + .unwrap_err() + .contains("duplicate step name")); + } + + #[test] + fn workflow_validation_rejects_bad_target_and_deps() { + // neither function nor workflow + let no_target = wf_def(serde_json::json!([{ "name": "a" }])); + assert!(validate_workflow_definition("w", &no_target) + .unwrap_err() + .contains("exactly one")); + // both function and workflow + let both = wf_def(serde_json::json!([{ "name": "a", "function": "f", "workflow": "w" }])); + assert!(validate_workflow_definition("w", &both).is_err()); + // dependency on an undeclared step + let bad_dep = wf_def(serde_json::json!([ + { "name": "a", "function": "fa", "depends_on": ["ghost"] }, + ])); + assert!(validate_workflow_definition("w", &bad_dep) + .unwrap_err() + .contains("undeclared step")); + // self-dependency + let self_dep = wf_def(serde_json::json!([ + { "name": "a", "function": "fa", "depends_on": ["a"] }, + ])); + assert!(validate_workflow_definition("w", &self_dep) + .unwrap_err() + .contains("depends on itself")); + } + + #[test] + fn workflow_validation_detects_cycles() { + let cyclic = wf_def(serde_json::json!([ + { "name": "a", "function": "fa", "depends_on": ["c"] }, + { "name": "b", "function": "fb", "depends_on": ["a"] }, + { "name": "c", "function": "fc", "depends_on": ["b"] }, + ])); + assert!(validate_workflow_definition("w", &cyclic) + .unwrap_err() + .contains("cycle")); + } + + #[test] + fn workflow_validation_rejects_bad_when_and_template() { + let bad_when = wf_def(serde_json::json!([ + { "name": "a", "function": "fa" }, + { "name": "b", "function": "fb", "depends_on": ["a"], "when": "steps.ghost.output.x" }, + ])); + assert!(validate_workflow_definition("w", &bad_when).is_err()); + let bad_tmpl = wf_def(serde_json::json!([ + { "name": "a", "function": "fa", "input": { "x": "{{ steps.ghost.output }}" } }, + ])); + assert!(validate_workflow_definition("w", &bad_tmpl).is_err()); + } } diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index a3a31f7..62cff63 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -108,6 +108,9 @@ pub mod users_service; pub mod vars_api; pub mod vars_repo; pub mod vars_service; +pub mod workflow_expr; +pub mod workflow_repo; +pub mod workflow_template; pub use abandoned_repo::{ AbandonedRepo, AbandonedRepoError, NewAbandonedExecution, PostgresAbandonedRepo, diff --git a/crates/manager-core/src/workflow_expr.rs b/crates/manager-core/src/workflow_expr.rs new file mode 100644 index 0000000..e7830aa --- /dev/null +++ b/crates/manager-core/src/workflow_expr.rs @@ -0,0 +1,490 @@ +//! Workflow `when` condition evaluation (v1.2 Workflows) — pure, DB-free. +//! +//! A step's optional `when` is a small, safe JSON predicate over the run +//! context (`input` + prior step `output`s, resolved via [`workflow_template`]). +//! When it evaluates false the step is `skipped`. This keeps `manager-core` +//! free of a scripting engine (the boundary rule) — it is NOT Rhai and never +//! grows into one; a `when` that ever needs real scripting is a deliberate +//! future ExecutorClient round-trip, not an engine dependency here. +//! +//! Grammar (precedence low→high): `||`, `&&`, `!`, comparison +//! (`== != < > <= >=`), primary (`( … )`, `exists `, literal, ``). +//! Literals: numbers, `'…'`/`"…"` strings, `true`, `false`, `null`. +//! A bare reference is truthy per [`is_truthy`]; a reference that doesn't +//! resolve is `null` (falsy) at run time — apply-time [`validate`] guards typos. +//! +//! [`workflow_template`]: crate::workflow_template + +use std::collections::BTreeSet; + +use serde_json::Value; +use thiserror::Error; + +use crate::workflow_template::{RunContext, TemplateError}; + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ExprError { + #[error("workflow `when` parse error: {0}")] + Parse(String), + #[error("workflow `when` reference error: {0}")] + Ref(#[from] TemplateError), +} + +/// Parse a `when` expression (syntactic validation). Returns the AST. +/// +/// # Errors +/// [`ExprError::Parse`] on malformed syntax. +pub fn parse(src: &str) -> Result { + let toks = lex(src)?; + let mut p = Parser { + toks: &toks, + pos: 0, + }; + let e = p.parse_or()?; + if p.pos != p.toks.len() { + return Err(ExprError::Parse(format!( + "trailing tokens near {:?}", + p.toks.get(p.pos) + ))); + } + Ok(e) +} + +/// Parse + evaluate `src` against `ctx`. +/// +/// # Errors +/// [`ExprError::Parse`] on malformed syntax. +pub fn eval(src: &str, ctx: &RunContext) -> Result { + Ok(eval_bool(&parse(src)?, ctx)) +} + +/// Apply-time validation: parse, then check every referenced `steps.` +/// names a declared step. +/// +/// # Errors +/// [`ExprError`] on a parse error or a reference to an undeclared step. +pub fn validate(src: &str, step_names: &BTreeSet) -> Result<(), ExprError> { + let ast = parse(src)?; + let mut refs = Vec::new(); + ast.collect_refs(&mut refs); + for r in refs { + let parts: Vec<&str> = r.split('.').collect(); + match parts.first().copied() { + Some("input") => {} + Some("steps") => { + let name = parts + .get(1) + .filter(|s| !s.is_empty()) + .ok_or_else(|| ExprError::Ref(TemplateError::MalformedRef(r.clone())))?; + if parts.get(2).copied() != Some("output") { + return Err(ExprError::Ref(TemplateError::MalformedRef(r.clone()))); + } + if !step_names.contains(*name) { + return Err(ExprError::Ref(TemplateError::UnknownStep( + r.clone(), + (*name).to_string(), + ))); + } + } + _ => return Err(ExprError::Ref(TemplateError::MalformedRef(r.clone()))), + } + } + Ok(()) +} + +// ---- AST ------------------------------------------------------------------ + +#[derive(Debug, Clone, PartialEq)] +pub enum Expr { + Or(Box, Box), + And(Box, Box), + Not(Box), + Cmp(Operand, CmpOp, Operand), + Truthy(Operand), + Exists(String), +} + +#[derive(Debug, Clone, PartialEq)] +pub enum Operand { + Ref(String), + Lit(Value), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CmpOp { + Eq, + Ne, + Lt, + Gt, + Le, + Ge, +} + +impl Expr { + fn collect_refs(&self, acc: &mut Vec) { + match self { + Self::Or(a, b) | Self::And(a, b) => { + a.collect_refs(acc); + b.collect_refs(acc); + } + Self::Not(a) => a.collect_refs(acc), + Self::Cmp(l, _, r) => { + l.collect_ref(acc); + r.collect_ref(acc); + } + Self::Truthy(o) => o.collect_ref(acc), + Self::Exists(r) => acc.push(r.clone()), + } + } +} + +impl Operand { + fn collect_ref(&self, acc: &mut Vec) { + if let Self::Ref(r) = self { + acc.push(r.clone()); + } + } +} + +// ---- Evaluation ----------------------------------------------------------- + +fn eval_bool(e: &Expr, ctx: &RunContext) -> bool { + match e { + Expr::Or(a, b) => eval_bool(a, ctx) || eval_bool(b, ctx), + Expr::And(a, b) => eval_bool(a, ctx) && eval_bool(b, ctx), + Expr::Not(a) => !eval_bool(a, ctx), + Expr::Truthy(o) => is_truthy(&operand_value(o, ctx)), + Expr::Exists(r) => ctx.resolve_ref(r).is_some_and(|v| !v.is_null()), + Expr::Cmp(l, op, r) => { + let lv = operand_value(l, ctx); + let rv = operand_value(r, ctx); + eval_cmp(&lv, *op, &rv) + } + } +} + +fn operand_value(o: &Operand, ctx: &RunContext) -> Value { + match o { + Operand::Lit(v) => v.clone(), + Operand::Ref(r) => ctx.resolve_ref(r).cloned().unwrap_or(Value::Null), + } +} + +/// JSON truthiness: null/false/0/""/[]/{}` are falsy; everything else truthy. +#[must_use] +pub fn is_truthy(v: &Value) -> bool { + match v { + Value::Null => false, + Value::Bool(b) => *b, + Value::Number(n) => n.as_f64().is_some_and(|f| f != 0.0), + Value::String(s) => !s.is_empty(), + Value::Array(a) => !a.is_empty(), + Value::Object(m) => !m.is_empty(), + } +} + +fn eval_cmp(l: &Value, op: CmpOp, r: &Value) -> bool { + match op { + CmpOp::Eq => json_eq(l, r), + CmpOp::Ne => !json_eq(l, r), + CmpOp::Lt | CmpOp::Gt | CmpOp::Le | CmpOp::Ge => match order(l, r) { + Some(ord) => match op { + CmpOp::Lt => ord.is_lt(), + CmpOp::Gt => ord.is_gt(), + CmpOp::Le => ord.is_le(), + CmpOp::Ge => ord.is_ge(), + _ => unreachable!(), + }, + None => false, + }, + } +} + +fn json_eq(l: &Value, r: &Value) -> bool { + match (l.as_f64(), r.as_f64()) { + (Some(a), Some(b)) => (a - b).abs() < f64::EPSILON, + _ => l == r, + } +} + +fn order(l: &Value, r: &Value) -> Option { + if let (Some(a), Some(b)) = (l.as_f64(), r.as_f64()) { + return a.partial_cmp(&b); + } + if let (Value::String(a), Value::String(b)) = (l, r) { + return Some(a.cmp(b)); + } + None +} + +// ---- Lexer ---------------------------------------------------------------- + +#[derive(Debug, Clone, PartialEq)] +enum Tok { + Word(String), + Num(f64), + Str(String), + Op(&'static str), + LParen, + RParen, +} + +#[allow(clippy::too_many_lines)] +fn lex(src: &str) -> Result, ExprError> { + let b = src.as_bytes(); + let mut i = 0; + let mut out = Vec::new(); + while i < b.len() { + let c = b[i]; + match c { + _ if c.is_ascii_whitespace() => i += 1, + b'(' => { + out.push(Tok::LParen); + i += 1; + } + b')' => { + out.push(Tok::RParen); + i += 1; + } + b'\'' | b'"' => { + let quote = c; + let start = i + 1; + let mut j = start; + while j < b.len() && b[j] != quote { + j += 1; + } + if j >= b.len() { + return Err(ExprError::Parse("unterminated string".to_string())); + } + out.push(Tok::Str(src[start..j].to_string())); + i = j + 1; + } + b'0'..=b'9' | b'-' if c != b'-' || next_is_digit(b, i) => { + let start = i; + i += 1; + while i < b.len() && (b[i].is_ascii_digit() || b[i] == b'.') { + i += 1; + } + let f: f64 = src[start..i] + .parse() + .map_err(|_| ExprError::Parse(format!("bad number `{}`", &src[start..i])))?; + out.push(Tok::Num(f)); + } + b'=' | b'!' | b'<' | b'>' | b'&' | b'|' => { + let two = if i + 1 < b.len() { &src[i..=i + 1] } else { "" }; + match two { + "==" => two_op(&mut out, &mut i, "=="), + "!=" => two_op(&mut out, &mut i, "!="), + "<=" => two_op(&mut out, &mut i, "<="), + ">=" => two_op(&mut out, &mut i, ">="), + "&&" => two_op(&mut out, &mut i, "&&"), + "||" => two_op(&mut out, &mut i, "||"), + _ => match c { + b'<' => one_op(&mut out, &mut i, "<"), + b'>' => one_op(&mut out, &mut i, ">"), + b'!' => one_op(&mut out, &mut i, "!"), + _ => return Err(ExprError::Parse(format!("unexpected `{}`", c as char))), + }, + } + } + _ if c.is_ascii_alphabetic() || c == b'_' => { + let start = i; + while i < b.len() && (b[i].is_ascii_alphanumeric() || b[i] == b'_' || b[i] == b'.') + { + i += 1; + } + out.push(Tok::Word(src[start..i].to_string())); + } + _ => return Err(ExprError::Parse(format!("unexpected `{}`", c as char))), + } + } + Ok(out) +} + +fn next_is_digit(b: &[u8], i: usize) -> bool { + b.get(i + 1).is_some_and(u8::is_ascii_digit) +} + +fn two_op(out: &mut Vec, i: &mut usize, op: &'static str) { + out.push(Tok::Op(op)); + *i += 2; +} + +fn one_op(out: &mut Vec, i: &mut usize, op: &'static str) { + out.push(Tok::Op(op)); + *i += 1; +} + +// ---- Parser --------------------------------------------------------------- + +struct Parser<'a> { + toks: &'a [Tok], + pos: usize, +} + +impl Parser<'_> { + fn peek(&self) -> Option<&Tok> { + self.toks.get(self.pos) + } + + fn parse_or(&mut self) -> Result { + let mut left = self.parse_and()?; + while matches!(self.peek(), Some(Tok::Op("||"))) { + self.pos += 1; + let right = self.parse_and()?; + left = Expr::Or(Box::new(left), Box::new(right)); + } + Ok(left) + } + + fn parse_and(&mut self) -> Result { + let mut left = self.parse_not()?; + while matches!(self.peek(), Some(Tok::Op("&&"))) { + self.pos += 1; + let right = self.parse_not()?; + left = Expr::And(Box::new(left), Box::new(right)); + } + Ok(left) + } + + fn parse_not(&mut self) -> Result { + if matches!(self.peek(), Some(Tok::Op("!"))) { + self.pos += 1; + return Ok(Expr::Not(Box::new(self.parse_not()?))); + } + self.parse_cmp() + } + + fn parse_cmp(&mut self) -> Result { + // `( … )` group is a full boolean sub-expression. + if matches!(self.peek(), Some(Tok::LParen)) { + self.pos += 1; + let inner = self.parse_or()?; + if !matches!(self.peek(), Some(Tok::RParen)) { + return Err(ExprError::Parse("expected `)`".to_string())); + } + self.pos += 1; + return Ok(inner); + } + // `exists ` + if matches!(self.peek(), Some(Tok::Word(w)) if w == "exists") { + self.pos += 1; + let Some(Tok::Word(r)) = self.peek().cloned() else { + return Err(ExprError::Parse( + "expected reference after `exists`".to_string(), + )); + }; + self.pos += 1; + return Ok(Expr::Exists(r)); + } + let left = self.parse_operand()?; + if let Some(op) = self.peek().and_then(cmp_op) { + self.pos += 1; + let right = self.parse_operand()?; + return Ok(Expr::Cmp(left, op, right)); + } + Ok(Expr::Truthy(left)) + } + + fn parse_operand(&mut self) -> Result { + match self.peek().cloned() { + Some(Tok::Num(n)) => { + self.pos += 1; + Ok(Operand::Lit(serde_json::json!(n))) + } + Some(Tok::Str(s)) => { + self.pos += 1; + Ok(Operand::Lit(Value::String(s))) + } + Some(Tok::Word(w)) => { + self.pos += 1; + Ok(match w.as_str() { + "true" => Operand::Lit(Value::Bool(true)), + "false" => Operand::Lit(Value::Bool(false)), + "null" => Operand::Lit(Value::Null), + _ => Operand::Ref(w), + }) + } + other => Err(ExprError::Parse(format!( + "expected operand, found {other:?}" + ))), + } + } +} + +fn cmp_op(t: &Tok) -> Option { + match t { + Tok::Op("==") => Some(CmpOp::Eq), + Tok::Op("!=") => Some(CmpOp::Ne), + Tok::Op("<") => Some(CmpOp::Lt), + Tok::Op(">") => Some(CmpOp::Gt), + Tok::Op("<=") => Some(CmpOp::Le), + Tok::Op(">=") => Some(CmpOp::Ge), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use std::collections::BTreeMap; + + fn ctx() -> RunContext { + let mut steps = BTreeMap::new(); + steps.insert("check".to_string(), json!({ "ok": true, "score": 7 })); + RunContext { + input: json!({ "amount": 100, "mode": "prod", "flag": false }), + steps, + } + } + + #[test] + fn comparisons() { + let c = ctx(); + assert!(eval("input.amount > 50", &c).unwrap()); + assert!(!eval("input.amount > 500", &c).unwrap()); + assert!(eval("input.amount == 100", &c).unwrap()); + assert!(eval("input.mode == 'prod'", &c).unwrap()); + assert!(eval("steps.check.output.score >= 7", &c).unwrap()); + assert!(eval("steps.check.output.ok == true", &c).unwrap()); + } + + #[test] + fn boolean_ops_and_precedence() { + let c = ctx(); + assert!(eval("input.amount > 50 && input.mode == 'prod'", &c).unwrap()); + assert!(!eval("input.amount > 50 && input.mode == 'dev'", &c).unwrap()); + assert!(eval("input.amount > 500 || input.mode == 'prod'", &c).unwrap()); + assert!(eval("!(input.mode == 'dev')", &c).unwrap()); + // && binds tighter than || + assert!(eval("false && false || input.amount == 100", &c).unwrap()); + } + + #[test] + fn truthiness_and_exists() { + let c = ctx(); + assert!(eval("steps.check.output.ok", &c).unwrap()); + assert!(!eval("input.flag", &c).unwrap()); + assert!(eval("exists input.amount", &c).unwrap()); + assert!(!eval("exists input.nope", &c).unwrap()); + // a missing ref is falsy, not an error, in a condition + assert!(!eval("input.nope == 1", &c).unwrap()); + } + + #[test] + fn parse_errors() { + assert!(parse("input.a >").is_err()); + assert!(parse("&& input.a").is_err()); + assert!(parse("(input.a == 1").is_err()); + assert!(parse("'unterminated").is_err()); + } + + #[test] + fn validate_checks_declared_steps() { + let steps: BTreeSet = ["check".to_string()].into_iter().collect(); + assert!(validate("steps.check.output.ok && input.x > 1", &steps).is_ok()); + assert!(validate("steps.ghost.output.ok", &steps).is_err()); + assert!(validate("steps.check.result", &steps).is_err()); + } +} diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs new file mode 100644 index 0000000..7d1ed8f --- /dev/null +++ b/crates/manager-core/src/workflow_repo.rs @@ -0,0 +1,229 @@ +//! Persistence for workflow definitions (v1.2 Workflows). +//! +//! The `workflows` table (0071) is owner-polymorphic like `scripts`; M1 authors +//! **app-owned** workflows only (the `group_id` column ships unused). The read +//! trait backs the admin/CLI surface; the `*_tx` free-fns are used by the +//! 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. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use sqlx::PgPool; +use thiserror::Error; +use uuid::Uuid; + +use picloud_shared::workflow::WorkflowDefinition; +use picloud_shared::{AppId, WorkflowId}; + +#[derive(Debug, Error)] +pub enum WorkflowRepoError { + #[error(transparent)] + Db(#[from] sqlx::Error), + #[error("workflow definition (de)serialization failed: {0}")] + Serde(String), +} + +/// A stored workflow definition (app-owned in M1). +#[derive(Debug, Clone)] +pub struct Workflow { + pub id: WorkflowId, + pub app_id: AppId, + pub name: String, + pub definition: WorkflowDefinition, + pub enabled: bool, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +/// Raw row shape (definition decoded separately). +#[derive(sqlx::FromRow)] +struct WorkflowRow { + id: Uuid, + app_id: Uuid, + name: String, + definition: serde_json::Value, + enabled: bool, + created_at: DateTime, + updated_at: DateTime, +} + +impl TryFrom for Workflow { + type Error = WorkflowRepoError; + fn try_from(r: WorkflowRow) -> Result { + Ok(Self { + id: r.id.into(), + app_id: r.app_id.into(), + name: r.name, + definition: serde_json::from_value(r.definition) + .map_err(|e| WorkflowRepoError::Serde(e.to_string()))?, + enabled: r.enabled, + created_at: r.created_at, + updated_at: r.updated_at, + }) + } +} + +const SELECT_COLS: &str = "id, app_id, name, definition, enabled, created_at, updated_at"; + +#[async_trait] +pub trait WorkflowRepo: Send + Sync { + /// All workflows owned by `app_id`, ordered by name. + async fn list_for_app(&self, app_id: AppId) -> Result, WorkflowRepoError>; + /// A single app-owned workflow by case-insensitive name. + async fn get_by_name( + &self, + app_id: AppId, + name: &str, + ) -> Result, WorkflowRepoError>; + /// A single workflow by id (app-scoped for isolation). + async fn get_by_id( + &self, + app_id: AppId, + id: WorkflowId, + ) -> Result, WorkflowRepoError>; +} + +pub struct PostgresWorkflowRepo { + pool: PgPool, +} + +impl PostgresWorkflowRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl WorkflowRepo for PostgresWorkflowRepo { + async fn list_for_app(&self, app_id: AppId) -> Result, WorkflowRepoError> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)" + )) + .bind(app_id.into_inner()) + .fetch_all(&self.pool) + .await?; + rows.into_iter().map(TryInto::try_into).collect() + } + + async fn get_by_name( + &self, + app_id: AppId, + name: &str, + ) -> Result, WorkflowRepoError> { + let row: Option = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND LOWER(name) = LOWER($2)" + )) + .bind(app_id.into_inner()) + .bind(name) + .fetch_optional(&self.pool) + .await?; + row.map(TryInto::try_into).transpose() + } + + async fn get_by_id( + &self, + app_id: AppId, + id: WorkflowId, + ) -> Result, WorkflowRepoError> { + let row: Option = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND id = $2" + )) + .bind(app_id.into_inner()) + .bind(id.into_inner()) + .fetch_optional(&self.pool) + .await?; + row.map(TryInto::try_into).transpose() + } +} + +/// Pool-based read of an app's workflows (used by `apply`'s `load_current`, +/// which reads several markers directly off the pool). +pub async fn list_workflows_for_app( + pool: &PgPool, + app_id: AppId, +) -> Result, WorkflowRepoError> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)" + )) + .bind(app_id.into_inner()) + .fetch_all(pool) + .await?; + rows.into_iter().map(TryInto::try_into).collect() +} + +// ---- reconcile tx free-fns (used by apply_service) ------------------------ + +/// Load an app's workflows inside the apply transaction (so the diff sees a +/// snapshot consistent with the writes). +pub async fn list_workflows_for_app_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + app_id: AppId, +) -> Result, WorkflowRepoError> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 ORDER BY LOWER(name)" + )) + .bind(app_id.into_inner()) + .fetch_all(&mut **tx) + .await?; + rows.into_iter().map(TryInto::try_into).collect() +} + +/// Insert an app-owned workflow. +pub async fn insert_workflow_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + app_id: AppId, + name: &str, + definition: &WorkflowDefinition, + enabled: bool, +) -> Result { + let def = + serde_json::to_value(definition).map_err(|e| WorkflowRepoError::Serde(e.to_string()))?; + let (id,): (Uuid,) = sqlx::query_as( + "INSERT INTO workflows (app_id, name, definition, enabled) \ + VALUES ($1, $2, $3, $4) RETURNING id", + ) + .bind(app_id.into_inner()) + .bind(name) + .bind(def) + .bind(enabled) + .fetch_one(&mut **tx) + .await?; + Ok(id.into()) +} + +/// Update a workflow's definition + enabled flag (name/identity unchanged). +pub async fn update_workflow_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + id: WorkflowId, + definition: &WorkflowDefinition, + enabled: bool, +) -> Result<(), WorkflowRepoError> { + let def = + serde_json::to_value(definition).map_err(|e| WorkflowRepoError::Serde(e.to_string()))?; + sqlx::query( + "UPDATE workflows SET definition = $2, enabled = $3, updated_at = NOW() WHERE id = $1", + ) + .bind(id.into_inner()) + .bind(def) + .bind(enabled) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Delete a workflow (prune). RESTRICTs if runs reference it — a workflow with +/// history is kept; this is surfaced as an error to the reconcile caller. +pub async fn delete_workflow_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + id: WorkflowId, +) -> Result<(), WorkflowRepoError> { + sqlx::query("DELETE FROM workflows WHERE id = $1") + .bind(id.into_inner()) + .execute(&mut **tx) + .await?; + Ok(()) +} diff --git a/crates/manager-core/src/workflow_template.rs b/crates/manager-core/src/workflow_template.rs new file mode 100644 index 0000000..95c12ed --- /dev/null +++ b/crates/manager-core/src/workflow_template.rs @@ -0,0 +1,315 @@ +//! Workflow input reference resolution (v1.2 Workflows) — pure, DB-free. +//! +//! A step's declared `input` is a JSON value whose string leaves may carry +//! `{{ … }}` references, resolved at run time against the run's accumulated +//! context: the run `input` and each prior step's `output`. +//! +//! Reference grammar: +//! `{{ input }}` · `{{ input.a.b }}` +//! `{{ steps..output }}` · `{{ steps..output.a.b }}` +//! +//! A string that is *exactly* a single `{{ … }}` reference is replaced by the +//! referenced JSON value (type-preserving). A reference embedded in a larger +//! string interpolates as text. A missing reference is a hard error (a workflow +//! referencing a non-existent output is a definition bug we surface, not hide). +//! +//! The same reference walker backs the `when` condition evaluator +//! (`workflow_expr`). + +use std::collections::{BTreeMap, BTreeSet}; + +use serde_json::{Map, Value}; +use thiserror::Error; + +/// The run context a template/condition resolves against. +#[derive(Debug, Default, Clone)] +pub struct RunContext { + /// The run's top-level input. + pub input: Value, + /// Prior step outputs, keyed by step name. + pub steps: BTreeMap, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum TemplateError { + #[error("unresolved reference `{0}` (no such input/step output)")] + MissingRef(String), + #[error("malformed reference `{0}`")] + MalformedRef(String), + #[error("reference `{0}` targets undeclared step `{1}`")] + UnknownStep(String, String), +} + +impl RunContext { + /// Resolve a dotted reference path (`input.a.b` / `steps.name.output.a.b`) + /// to the JSON value it points at, or `None` if absent. + #[must_use] + pub fn resolve_ref(&self, path: &str) -> Option<&Value> { + let parts: Vec<&str> = path.split('.').collect(); + let (root_val, rest): (&Value, &[&str]) = match parts.first()?.trim() { + "input" => (&self.input, &parts[1..]), + "steps" => { + // steps..output[.deep...] + let name = parts.get(1)?; + if parts.get(2).map(|s| s.trim()) != Some("output") { + return None; + } + (self.steps.get(*name)?, &parts[3..]) + } + _ => return None, + }; + let mut cur = root_val; + for seg in rest { + let seg = seg.trim(); + cur = match cur { + Value::Object(m) => m.get(seg)?, + Value::Array(a) => a.get(seg.parse::().ok()?)?, + _ => return None, + }; + } + Some(cur) + } +} + +/// Resolve every `{{ … }}` reference in `value` against `ctx`. +/// +/// # Errors +/// [`TemplateError::MissingRef`] if a reference does not resolve; +/// [`TemplateError::MalformedRef`] on a syntactically-bad reference. +pub fn resolve(value: &Value, ctx: &RunContext) -> Result { + match value { + Value::String(s) => resolve_string(s, ctx), + Value::Array(a) => Ok(Value::Array( + a.iter() + .map(|v| resolve(v, ctx)) + .collect::>()?, + )), + Value::Object(m) => { + let mut out = Map::with_capacity(m.len()); + for (k, v) in m { + out.insert(k.clone(), resolve(v, ctx)?); + } + Ok(Value::Object(out)) + } + other => Ok(other.clone()), + } +} + +fn resolve_string(s: &str, ctx: &RunContext) -> Result { + // Exact-single-reference: preserve the referenced value's JSON type. + if let Some(inner) = exact_single_ref(s) { + return ctx + .resolve_ref(inner) + .cloned() + .ok_or_else(|| TemplateError::MissingRef(inner.to_string())); + } + // Otherwise interpolate references into the surrounding text. + let mut out = String::with_capacity(s.len()); + let mut rest = s; + while let Some(open) = rest.find("{{") { + out.push_str(&rest[..open]); + let after = &rest[open + 2..]; + let close = after + .find("}}") + .ok_or_else(|| TemplateError::MalformedRef(s.to_string()))?; + let refpath = after[..close].trim(); + let val = ctx + .resolve_ref(refpath) + .ok_or_else(|| TemplateError::MissingRef(refpath.to_string()))?; + out.push_str(&render_scalar(val)); + rest = &after[close + 2..]; + } + out.push_str(rest); + Ok(Value::String(out)) +} + +/// If `s` is exactly one `{{ ref }}` (ignoring surrounding whitespace and with +/// no other braces), return the trimmed inner reference. +fn exact_single_ref(s: &str) -> Option<&str> { + let t = s.trim(); + let inner = t.strip_prefix("{{")?.strip_suffix("}}")?; + if inner.contains("{{") || inner.contains("}}") { + return None; + } + Some(inner.trim()) +} + +/// Render a JSON scalar for string interpolation (strings raw, others via JSON). +fn render_scalar(v: &Value) -> String { + match v { + Value::String(s) => s.clone(), + Value::Null => String::new(), + other => other.to_string(), + } +} + +/// Apply-time validation: every reference in `value` must be syntactically valid +/// and target either `input` or a declared step's `output`. Step outputs don't +/// exist yet at apply time, so this checks *shape*, not resolvability. +/// +/// # Errors +/// [`TemplateError`] on a malformed reference or one naming an undeclared step. +pub fn validate(value: &Value, step_names: &BTreeSet) -> Result<(), TemplateError> { + for r in collect_refs(value)? { + let parts: Vec<&str> = r.split('.').map(str::trim).collect(); + match parts.first().copied() { + Some("input") => {} + Some("steps") => { + let name = parts + .get(1) + .filter(|s| !s.is_empty()) + .ok_or_else(|| TemplateError::MalformedRef(r.clone()))?; + if parts.get(2).copied() != Some("output") { + return Err(TemplateError::MalformedRef(r.clone())); + } + if !step_names.contains(*name) { + return Err(TemplateError::UnknownStep(r.clone(), (*name).to_string())); + } + } + _ => return Err(TemplateError::MalformedRef(r.clone())), + } + } + Ok(()) +} + +/// Collect every `{{ … }}` reference path found anywhere in `value`. +fn collect_refs(value: &Value) -> Result, TemplateError> { + let mut acc = Vec::new(); + collect_into(value, &mut acc)?; + Ok(acc) +} + +fn collect_into(value: &Value, acc: &mut Vec) -> Result<(), TemplateError> { + match value { + Value::String(s) => { + let mut rest = s.as_str(); + while let Some(open) = rest.find("{{") { + let after = &rest[open + 2..]; + let close = after + .find("}}") + .ok_or_else(|| TemplateError::MalformedRef(s.clone()))?; + acc.push(after[..close].trim().to_string()); + rest = &after[close + 2..]; + } + } + Value::Array(a) => { + for v in a { + collect_into(v, acc)?; + } + } + Value::Object(m) => { + for v in m.values() { + collect_into(v, acc)?; + } + } + _ => {} + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn ctx() -> RunContext { + let mut steps = BTreeMap::new(); + steps.insert("validate".to_string(), json!({ "ok": true, "score": 7 })); + steps.insert("fetch".to_string(), json!([10, 20, 30])); + RunContext { + input: json!({ "order": { "id": "abc", "total": 42 } }), + steps, + } + } + + #[test] + fn exact_single_ref_preserves_type() { + let c = ctx(); + assert_eq!( + resolve(&json!("{{ input.order.total }}"), &c).unwrap(), + json!(42) + ); + assert_eq!( + resolve(&json!("{{ steps.validate.output.ok }}"), &c).unwrap(), + json!(true) + ); + // whole-object ref + assert_eq!( + resolve(&json!("{{ steps.validate.output }}"), &c).unwrap(), + json!({ "ok": true, "score": 7 }) + ); + } + + #[test] + fn array_index_and_nested_object() { + let c = ctx(); + assert_eq!( + resolve(&json!("{{ steps.fetch.output.1 }}"), &c).unwrap(), + json!(20) + ); + assert_eq!( + resolve(&json!({ "id": "{{ input.order.id }}" }), &c).unwrap(), + json!({ "id": "abc" }) + ); + } + + #[test] + fn interpolation_renders_text() { + let c = ctx(); + assert_eq!( + resolve( + &json!("order {{ input.order.id }} scored {{ steps.validate.output.score }}"), + &c + ) + .unwrap(), + json!("order abc scored 7") + ); + } + + #[test] + fn missing_ref_is_error() { + let c = ctx(); + assert_eq!( + resolve(&json!("{{ steps.nope.output }}"), &c).unwrap_err(), + TemplateError::MissingRef("steps.nope.output".to_string()) + ); + assert!(matches!( + resolve(&json!("{{ input.order.missing }}"), &c).unwrap_err(), + TemplateError::MissingRef(_) + )); + } + + #[test] + fn validate_shape_against_declared_steps() { + let steps: BTreeSet = ["validate".to_string()].into_iter().collect(); + // ok: input + declared step + assert!(validate( + &json!({ "a": "{{ input.x }}", "b": "{{ steps.validate.output.ok }}" }), + &steps + ) + .is_ok()); + // unknown step + assert!(matches!( + validate(&json!("{{ steps.ghost.output }}"), &steps).unwrap_err(), + TemplateError::UnknownStep(_, _) + )); + // malformed root + assert!(matches!( + validate(&json!("{{ bogus.x }}"), &steps).unwrap_err(), + TemplateError::MalformedRef(_) + )); + // steps without .output + assert!(matches!( + validate(&json!("{{ steps.validate.result }}"), &steps).unwrap_err(), + TemplateError::MalformedRef(_) + )); + } + + #[test] + fn non_template_values_pass_through() { + let c = ctx(); + assert_eq!(resolve(&json!(123), &c).unwrap(), json!(123)); + assert_eq!(resolve(&json!(null), &c).unwrap(), json!(null)); + assert_eq!(resolve(&json!("plain"), &c).unwrap(), json!("plain")); + } +} diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index 162e69c..be69dab 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/crates/manager-core/tests/expected_schema.txt @@ -468,6 +468,48 @@ table: vars created_at: timestamp with time zone NOT NULL default=now() updated_at: timestamp with time zone NOT NULL default=now() +table: workflow_run_steps + id: uuid NOT NULL default=gen_random_uuid() + run_id: uuid NOT NULL + step_name: text NOT NULL + status: text NOT NULL default='pending'::text + attempt: integer NOT NULL default=0 + max_attempts: integer NOT NULL default=1 + output: jsonb NULL + error: text NULL + claim_token: uuid NULL + claimed_at: timestamp with time zone NULL + next_attempt_at: timestamp with time zone NULL + child_run_id: uuid NULL + created_at: timestamp with time zone NOT NULL default=now() + updated_at: timestamp with time zone NOT NULL default=now() + +table: workflow_runs + id: uuid NOT NULL default=gen_random_uuid() + workflow_id: uuid NOT NULL + app_id: uuid NOT NULL + status: text NOT NULL default='pending'::text + input: jsonb NOT NULL default='{}'::jsonb + output: jsonb NULL + error: text NULL + root_execution_id: uuid NOT NULL + workflow_depth: integer NOT NULL default=0 + parent_run_id: uuid NULL + parent_step_id: uuid NULL + started_at: timestamp with time zone NULL + finished_at: timestamp with time zone NULL + created_at: timestamp with time zone NOT NULL default=now() + +table: workflows + id: uuid NOT NULL default=gen_random_uuid() + app_id: uuid NULL + group_id: uuid NULL + name: text NOT NULL + definition: jsonb NOT NULL + enabled: boolean NOT NULL default=true + created_at: timestamp with time zone NOT NULL default=now() + updated_at: timestamp with time zone NOT NULL default=now() + ## indexes indexes on abandoned_executions: @@ -709,6 +751,25 @@ indexes on vars: vars_group_uidx: public.vars USING btree (group_id, environment_scope, key) WHERE (group_id IS NOT NULL) vars_pkey: public.vars USING btree (id) +indexes on workflow_run_steps: + workflow_run_steps_claimable_idx: public.workflow_run_steps USING btree (next_attempt_at) WHERE (status = 'ready'::text) + workflow_run_steps_claimed_idx: public.workflow_run_steps USING btree (claimed_at) WHERE (claim_token IS NOT NULL) + workflow_run_steps_pkey: public.workflow_run_steps USING btree (id) + workflow_run_steps_run_id_step_name_key: public.workflow_run_steps USING btree (run_id, step_name) + workflow_run_steps_run_idx: public.workflow_run_steps USING btree (run_id) + +indexes on workflow_runs: + workflow_runs_app_status_idx: public.workflow_runs USING btree (app_id, status) + workflow_runs_parent_idx: public.workflow_runs USING btree (parent_run_id) WHERE (parent_run_id IS NOT NULL) + workflow_runs_pkey: public.workflow_runs USING btree (id) + workflow_runs_workflow_idx: public.workflow_runs USING btree (workflow_id, created_at DESC) + +indexes on workflows: + workflows_app_name_uidx: public.workflows USING btree (app_id, lower(name)) WHERE (app_id IS NOT NULL) + workflows_group_id_idx: public.workflows USING btree (group_id) WHERE (group_id IS NOT NULL) + workflows_group_name_uidx: public.workflows USING btree (group_id, lower(name)) WHERE (group_id IS NOT NULL) + workflows_pkey: public.workflows USING btree (id) + ## constraints constraints on abandoned_executions: @@ -967,6 +1028,26 @@ constraints on vars: [FOREIGN KEY] vars_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE CASCADE [PRIMARY KEY] vars_pkey: PRIMARY KEY (id) +constraints on workflow_run_steps: + [CHECK] workflow_run_steps_status_check: CHECK ((status = ANY (ARRAY['pending'::text, 'ready'::text, 'running'::text, 'succeeded'::text, 'failed'::text, 'skipped'::text]))) + [FOREIGN KEY] workflow_run_steps_child_run_id_fkey: FOREIGN KEY (child_run_id) REFERENCES workflow_runs(id) ON DELETE SET NULL + [FOREIGN KEY] workflow_run_steps_run_id_fkey: FOREIGN KEY (run_id) REFERENCES workflow_runs(id) ON DELETE CASCADE + [PRIMARY KEY] workflow_run_steps_pkey: PRIMARY KEY (id) + [UNIQUE] workflow_run_steps_run_id_step_name_key: UNIQUE (run_id, step_name) + +constraints on workflow_runs: + [CHECK] workflow_runs_status_check: CHECK ((status = ANY (ARRAY['pending'::text, 'running'::text, 'succeeded'::text, 'failed'::text, 'canceled'::text]))) + [FOREIGN KEY] workflow_runs_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE + [FOREIGN KEY] workflow_runs_parent_run_id_fkey: FOREIGN KEY (parent_run_id) REFERENCES workflow_runs(id) ON DELETE SET NULL + [FOREIGN KEY] workflow_runs_workflow_id_fkey: FOREIGN KEY (workflow_id) REFERENCES workflows(id) ON DELETE RESTRICT + [PRIMARY KEY] workflow_runs_pkey: PRIMARY KEY (id) + +constraints on workflows: + [CHECK] workflows_owner_exactly_one: CHECK (((group_id IS NULL) <> (app_id IS NULL))) + [FOREIGN KEY] workflows_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE + [FOREIGN KEY] workflows_group_id_fkey: FOREIGN KEY (group_id) REFERENCES groups(id) ON DELETE RESTRICT + [PRIMARY KEY] workflows_pkey: PRIMARY KEY (id) + ## applied migrations 0001: init 0002: sandbox @@ -1038,3 +1119,4 @@ constraints on vars: 0068: group dead letters 0069: email secret version 0070: admin session absolute expiry + 0071: workflows diff --git a/crates/picloud-cli/src/client.rs b/crates/picloud-cli/src/client.rs index a8daa11..1cef39c 100644 --- a/crates/picloud-cli/src/client.rs +++ b/crates/picloud-cli/src/client.rs @@ -1619,6 +1619,8 @@ pub struct PlanDto { pub collections: Vec, #[serde(default)] pub suppressions: Vec, + #[serde(default)] + pub workflows: Vec, /// Fingerprint of the live state this plan was computed against; carried /// in `.picloud/` and replayed to `apply` for the bound-plan check. #[serde(default)] @@ -1694,6 +1696,8 @@ pub struct NodePlanDto { pub collections: Vec, #[serde(default)] pub suppressions: Vec, + #[serde(default)] + pub workflows: Vec, /// §7 M3: this node's ownership outcome under the tree's `[project]`. #[serde(default)] pub ownership: Option, @@ -1753,6 +1757,12 @@ pub struct ApplyReportDto { #[serde(default)] pub suppressions_deleted: u32, #[serde(default)] + pub workflows_created: u32, + #[serde(default)] + pub workflows_updated: u32, + #[serde(default)] + pub workflows_deleted: u32, + #[serde(default)] pub warnings: Vec, } diff --git a/crates/picloud-cli/src/cmds/apply.rs b/crates/picloud-cli/src/cmds/apply.rs index 2aaf37a..d684891 100644 --- a/crates/picloud-cli/src/cmds/apply.rs +++ b/crates/picloud-cli/src/cmds/apply.rs @@ -148,6 +148,13 @@ pub async fn run( "+{} -{}", report.suppressions_created, report.suppressions_deleted ), + ) + .field( + "workflows", + format!( + "+{} ~{} -{}", + report.workflows_created, report.workflows_updated, report.workflows_deleted + ), ); for w in &report.warnings { block.field("warning", w.clone()); @@ -269,6 +276,13 @@ pub async fn run_tree( "+{} -{}", report.suppressions_created, report.suppressions_deleted ), + ) + .field( + "workflows", + format!( + "+{} ~{} -{}", + report.workflows_created, report.workflows_updated, report.workflows_deleted + ), ); for w in &report.warnings { block.field("warning", w.clone()); diff --git a/crates/picloud-cli/src/cmds/init.rs b/crates/picloud-cli/src/cmds/init.rs index 588afe5..d3f0b24 100644 --- a/crates/picloud-cli/src/cmds/init.rs +++ b/crates/picloud-cli/src/cmds/init.rs @@ -143,6 +143,7 @@ fn scaffold_manifest(slug: &str, name: &str) -> Manifest { secrets: crate::manifest::ManifestSecrets::default(), vars: std::collections::BTreeMap::new(), suppress: crate::manifest::ManifestSuppress::default(), + workflows: Vec::new(), } } diff --git a/crates/picloud-cli/src/cmds/plan.rs b/crates/picloud-cli/src/cmds/plan.rs index 291c7cd..44a8738 100644 --- a/crates/picloud-cli/src/cmds/plan.rs +++ b/crates/picloud-cli/src/cmds/plan.rs @@ -106,7 +106,7 @@ fn render_tree(plan: &TreePlanDto, mode: OutputMode) { ]); } } - let groups: [(&str, &Vec); 8] = [ + let groups: [(&str, &Vec); 9] = [ ("script", &n.scripts), ("route", &n.routes), ("trigger", &n.triggers), @@ -115,6 +115,7 @@ fn render_tree(plan: &TreePlanDto, mode: OutputMode) { ("extension_point", &n.extension_points), ("collection", &n.collections), ("suppression", &n.suppressions), + ("workflow", &n.workflows), ]; for (rk, changes) in groups { for c in changes { @@ -246,6 +247,24 @@ pub fn build_bundle(manifest: &Manifest, base_dir: &Path) -> Result { .collect::>(), "suppress_triggers": manifest.suppress.triggers, "suppress_routes": manifest.suppress.routes, + "workflows": manifest + .workflows + .iter() + .map(workflow_to_wire) + .collect::>>()?, + })) +} + +/// Convert a `[[workflows]]` manifest block to the server `BundleWorkflow` +/// shape `{ name, enabled, definition: { steps } }`. The manifest authors steps +/// at the workflow's top level; the wire nests them under `definition`. +fn workflow_to_wire(w: &crate::manifest::ManifestWorkflow) -> Result { + let steps = serde_json::to_value(&w.steps) + .with_context(|| format!("encoding workflow `{}`", w.name))?; + Ok(json!({ + "name": w.name, + "enabled": w.enabled, + "definition": { "steps": steps }, })) } @@ -276,7 +295,7 @@ fn render(plan: &PlanDto, mode: OutputMode) { ]); } } - let groups: [(&str, &Vec); 8] = [ + let groups: [(&str, &Vec); 9] = [ ("script", &plan.scripts), ("route", &plan.routes), ("trigger", &plan.triggers), @@ -285,6 +304,7 @@ fn render(plan: &PlanDto, mode: OutputMode) { ("extension_point", &plan.extension_points), ("collection", &plan.collections), ("suppression", &plan.suppressions), + ("workflow", &plan.workflows), ]; for (kind, changes) in groups { for c in changes { diff --git a/crates/picloud-cli/src/cmds/pull.rs b/crates/picloud-cli/src/cmds/pull.rs index 935d5d9..62df6ed 100644 --- a/crates/picloud-cli/src/cmds/pull.rs +++ b/crates/picloud-cli/src/cmds/pull.rs @@ -251,6 +251,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> }, vars: manifest_vars, suppress: crate::manifest::ManifestSuppress::default(), + workflows: Vec::new(), }; std::fs::write(&manifest_path, manifest.to_toml()?) diff --git a/crates/picloud-cli/src/manifest.rs b/crates/picloud-cli/src/manifest.rs index d63ddda..3cea8d2 100644 --- a/crates/picloud-cli/src/manifest.rs +++ b/crates/picloud-cli/src/manifest.rs @@ -60,6 +60,10 @@ pub struct Manifest { /// App-only; a `[group]` carrying it is rejected in [`Manifest::parse`]. #[serde(default, skip_serializing_if = "ManifestSuppress::is_empty")] pub suppress: ManifestSuppress, + /// `[[workflows]]` (v1.2 Workflows) — declarative DAG orchestrations. + /// App-owned; a `[group]` carrying them is rejected server-side. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub workflows: Vec, } impl Manifest { @@ -606,6 +610,64 @@ impl ManifestSuppress { } } +/// `[[workflows]]` (v1.2 Workflows) — a named DAG. Steps are authored as a +/// `[[workflows.steps]]` array of tables. The server nests `steps` under the +/// stored `definition` JSONB and validates the graph (acyclic, deps exist). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManifestWorkflow { + pub name: String, + /// Three-state lifecycle (§4.3); omitted ⇒ active. + #[serde( + default = "picloud_shared::default_true", + skip_serializing_if = "is_true" + )] + pub enabled: bool, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub steps: Vec, +} + +/// One step of a `[[workflows]]` DAG. Serializes to the server's +/// `WorkflowStepDef` shape (the CLI nests these under `definition.steps`). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManifestWorkflowStep { + pub name: String, + /// A function step invokes a script by name; exactly one of + /// `function`/`workflow` is set (validated server-side). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub function: Option, + /// A workflow step starts a nested sub-workflow by name. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workflow: Option, + /// Input mapping — an arbitrary TOML table whose string leaves may carry + /// `{{ … }}` references, round-tripped to JSON for the wire. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub input: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub depends_on: Vec, + /// A `when` JSON-predicate condition (false ⇒ the step is skipped). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub when: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry: Option, + /// `fail` (default) or `continue`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub on_error: Option, +} + +/// Per-step retry policy in the manifest (mirrors the server `WorkflowRetry`). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ManifestWorkflowRetry { + pub max_attempts: u32, + /// `constant` | `linear` | `exponential` (default). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub backoff: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_ms: Option, +} + // ---- serde skip/default helpers ---- fn is_endpoint(kind: &ScriptKind) -> bool { @@ -708,6 +770,7 @@ mod tests { ("max-retries".to_string(), toml::Value::Integer(3)), ]), suppress: ManifestSuppress::default(), + workflows: Vec::new(), } } @@ -814,6 +877,7 @@ mod tests { secrets: ManifestSecrets::default(), vars: BTreeMap::new(), suppress: ManifestSuppress::default(), + workflows: Vec::new(), }; let text = m.to_toml().unwrap(); assert!(!text.contains("[[scripts]]"), "got:\n{text}"); diff --git a/crates/picloud-cli/tests/cli.rs b/crates/picloud-cli/tests/cli.rs index 70467c4..7a2d017 100644 --- a/crates/picloud-cli/tests/cli.rs +++ b/crates/picloud-cli/tests/cli.rs @@ -56,3 +56,4 @@ mod suppress; mod tree; mod triggers; mod vars; +mod workflows; diff --git a/crates/picloud-cli/tests/workflows.rs b/crates/picloud-cli/tests/workflows.rs new file mode 100644 index 0000000..4bcc8e1 --- /dev/null +++ b/crates/picloud-cli/tests/workflows.rs @@ -0,0 +1,155 @@ +//! v1.2 Workflows — M1 declarative authoring journey via `pic`. +//! +//! An app applies a `[[workflows]]` manifest block (a DAG referencing its +//! scripts, with `depends_on`, a `when`, and an input template) → the apply +//! report shows the workflow created; re-applying is a NoOp; a manifest with a +//! cyclic DAG is rejected atomically; pruning the block deletes the workflow. +//! +//! The durable execution of a run (the orchestrator) lands in M2; this journey +//! pins the authoring + reconcile path the operator uses. + +use std::fs; + +use tempfile::TempDir; + +use crate::common; +use crate::common::cleanup::AppGuard; + +fn manifest_dir() -> TempDir { + let dir = TempDir::new().expect("tempdir"); + fs::create_dir_all(dir.path().join("scripts")).expect("scripts dir"); + dir +} + +fn seed_scripts(dir: &TempDir) { + fs::write( + dir.path().join("scripts/validate.rhai"), + "let body = #{ ok: true, value: 7 }; body", + ) + .unwrap(); + fs::write( + dir.path().join("scripts/charge.rhai"), + "let body = #{ charged: true }; body", + ) + .unwrap(); +} + +const SCRIPTS_BLOCK: &str = "\ +[[scripts]]\nname = \"validate\"\nfile = \"scripts/validate.rhai\"\n\n\ +[[scripts]]\nname = \"charge\"\nfile = \"scripts/charge.rhai\"\n\n"; + +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn workflow_apply_creates_then_noop_then_prunes() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let slug = common::unique_slug("wf"); + common::pic_as(&env) + .args(["apps", "create", &slug]) + .assert() + .success(); + let _guard = AppGuard::new(&env.url, &env.token, &slug); + + let dir = manifest_dir(); + seed_scripts(&dir); + let with_wf = format!( + "[app]\nslug = \"{slug}\"\nname = \"WF Test\"\n\n{SCRIPTS_BLOCK}\ + [[workflows]]\nname = \"pipeline\"\n\n\ + [[workflows.steps]]\nname = \"v\"\nfunction = \"validate\"\n\ + input = {{ order = \"{{{{ input.order }}}}\" }}\n\n\ + [[workflows.steps]]\nname = \"c\"\nfunction = \"charge\"\n\ + depends_on = [\"v\"]\nwhen = \"steps.v.output.ok == true\"\n" + ); + let path = dir.path().join("picloud.toml"); + fs::write(&path, &with_wf).unwrap(); + + // First apply: creates the two scripts + the workflow. + let out = common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .output() + .expect("apply"); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!( + out.status.success(), + "apply failed: {}\n{stdout}", + String::from_utf8_lossy(&out.stderr) + ); + assert!( + stdout.contains("workflows") && stdout.contains("+1"), + "expected a workflow creation in the report:\n{stdout}" + ); + + // Re-plan is clean (desired state reached — the workflow is a NoOp). + let p = String::from_utf8( + common::pic_as(&env) + .args(["plan", "--file"]) + .arg(&path) + .output() + .unwrap() + .stdout, + ) + .unwrap(); + assert!( + !p.contains("create") && !p.contains("update"), + "expected a clean plan after apply:\n{p}" + ); + + // Drop the workflow from the manifest and prune → it is deleted. + let without_wf = format!("[app]\nslug = \"{slug}\"\nname = \"WF Test\"\n\n{SCRIPTS_BLOCK}"); + fs::write(&path, &without_wf).unwrap(); + let pruned = common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .args(["--prune", "--yes"]) + .output() + .expect("prune apply"); + let ptext = String::from_utf8_lossy(&pruned.stdout); + assert!( + pruned.status.success(), + "prune apply failed: {}\n{ptext}", + String::from_utf8_lossy(&pruned.stderr) + ); + assert!( + ptext.contains("workflows") && ptext.contains("-1"), + "expected a workflow deletion under --prune:\n{ptext}" + ); +} + +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn workflow_with_a_cycle_is_rejected() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let slug = common::unique_slug("wfcycle"); + common::pic_as(&env) + .args(["apps", "create", &slug]) + .assert() + .success(); + let _guard = AppGuard::new(&env.url, &env.token, &slug); + + let dir = manifest_dir(); + seed_scripts(&dir); + // a → b → a is a cycle; the server's topological sort must reject it. + let cyclic = format!( + "[app]\nslug = \"{slug}\"\nname = \"WF Cycle\"\n\n{SCRIPTS_BLOCK}\ + [[workflows]]\nname = \"loop\"\n\n\ + [[workflows.steps]]\nname = \"a\"\nfunction = \"validate\"\ndepends_on = [\"b\"]\n\n\ + [[workflows.steps]]\nname = \"b\"\nfunction = \"charge\"\ndepends_on = [\"a\"]\n" + ); + let path = dir.path().join("picloud.toml"); + fs::write(&path, &cyclic).unwrap(); + + let out = common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .output() + .expect("apply"); + assert!(!out.status.success(), "cyclic workflow should be rejected"); + let err = String::from_utf8_lossy(&out.stderr); + assert!(err.contains("cycle"), "expected a cycle error, got:\n{err}"); +} diff --git a/crates/shared/src/ids.rs b/crates/shared/src/ids.rs index 72bd002..29851fd 100644 --- a/crates/shared/src/ids.rs +++ b/crates/shared/src/ids.rs @@ -59,3 +59,6 @@ id_type!(InvitationId); id_type!(QueueMessageId); id_type!(GroupId); id_type!(ProjectId); +id_type!(WorkflowId); +id_type!(WorkflowRunId); +id_type!(WorkflowRunStepId); diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index e199b33..c89509c 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -46,6 +46,7 @@ pub mod users; pub mod validator; pub mod vars; pub mod version; +pub mod workflow; /// serde `default` for `enabled`-style boolean fields that should default to /// `true` when absent from the wire (bool's own `Default` is `false`). Shared @@ -80,7 +81,7 @@ pub use group_queue::{GroupQueueError, GroupQueueService, NoopGroupQueueService} pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService}; pub use ids::{ AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, ProjectId, - QueueMessageId, RequestId, ScriptId, TriggerId, + QueueMessageId, RequestId, ScriptId, TriggerId, WorkflowId, WorkflowRunId, WorkflowRunStepId, }; pub use inbox::{ InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult, NoopInboxResolver, diff --git a/crates/shared/src/workflow.rs b/crates/shared/src/workflow.rs new file mode 100644 index 0000000..8f84662 --- /dev/null +++ b/crates/shared/src/workflow.rs @@ -0,0 +1,202 @@ +//! Workflow definition + status DTOs (v1.2 Workflows). +//! +//! Pure data shared across three consumers: the `pic` CLI (which builds a +//! [`WorkflowDefinition`] from the TOML manifest), the manager-core apply path +//! (which validates + stores it as JSONB), and the orchestrator (which executes +//! it). There is NO behavior here — the validator (`workflow_expr`), the input +//! resolver (`workflow_template`), and the durable orchestrator all live +//! server-side in `manager-core`, the only crate that touches Postgres. +//! +//! The `WorkflowService` trait (the injected SDK seam for `workflow::start`) +//! is added here in M5. + +use serde::{Deserialize, Serialize}; + +/// A workflow's DAG: an ordered list of steps connected by `depends_on` edges. +/// Serialized to the `workflows.definition` JSONB column. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkflowDefinition { + pub steps: Vec, +} + +/// One node in the DAG. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkflowStepDef { + pub name: String, + /// A function step invokes a script by name (resolved in the app's scope). + /// Exactly one of `function` / `workflow` is set (enforced server-side). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub function: Option, + /// A workflow step starts a nested sub-workflow by name. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workflow: Option, + /// Input mapping — a JSON value whose string leaves may carry `{{ … }}` + /// references resolved at run time against the run input + prior step + /// outputs (see `workflow_template`). Absent = the run input is passed + /// through unchanged. + #[serde(default, skip_serializing_if = "serde_json::Value::is_null")] + pub input: serde_json::Value, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub depends_on: Vec, + /// A JSON-predicate condition (see `workflow_expr`). When it evaluates + /// false the step is `skipped` (and counts as satisfied-but-empty for its + /// dependents). Absent = always run. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub when: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry: Option, + #[serde(default)] + pub on_error: OnError, +} + +impl WorkflowStepDef { + /// The resolved target of this step, or `None` if neither/both are set + /// (a shape error caught by server-side validation). + #[must_use] + pub fn target(&self) -> Option> { + match (self.function.as_deref(), self.workflow.as_deref()) { + (Some(f), None) => Some(StepTarget::Function(f)), + (None, Some(w)) => Some(StepTarget::Workflow(w)), + _ => None, + } + } + + /// Max attempts for this step (1 = no retry), derived from its retry policy. + #[must_use] + pub fn max_attempts(&self) -> u32 { + self.retry.as_ref().map_or(1, |r| r.max_attempts.max(1)) + } +} + +/// A step's target — a function (script by name) or a nested sub-workflow. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StepTarget<'a> { + Function(&'a str), + Workflow(&'a str), +} + +/// What to do when a step exhausts its attempts. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OnError { + /// Fail the whole run (default). + #[default] + Fail, + /// Mark the step failed but let the run continue — dependents proceed and + /// the run can still succeed. + Continue, +} + +/// Per-step retry policy. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkflowRetry { + pub max_attempts: u32, + #[serde(default)] + pub backoff: WorkflowBackoff, + #[serde(default = "default_base_ms")] + pub base_ms: u32, +} + +/// Backoff shape for step retries (self-contained so this DTO stays in `shared`; +/// the orchestrator maps it to its backoff computation). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkflowBackoff { + Constant, + Linear, + #[default] + Exponential, +} + +fn default_base_ms() -> u32 { + 500 +} + +/// Status of a workflow run (mirrors the `workflow_runs.status` CHECK). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunStatus { + Pending, + Running, + Succeeded, + Failed, + Canceled, +} + +impl RunStatus { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Running => "running", + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::Canceled => "canceled", + } + } + + #[must_use] + pub fn is_terminal(self) -> bool { + matches!(self, Self::Succeeded | Self::Failed | Self::Canceled) + } + + #[must_use] + #[allow(clippy::should_implement_trait)] + pub fn from_str(s: &str) -> Option { + Some(match s { + "pending" => Self::Pending, + "running" => Self::Running, + "succeeded" => Self::Succeeded, + "failed" => Self::Failed, + "canceled" => Self::Canceled, + _ => return None, + }) + } +} + +/// Status of a single step within a run (mirrors `workflow_run_steps.status`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StepStatus { + Pending, + Ready, + Running, + Succeeded, + Failed, + Skipped, +} + +impl StepStatus { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Ready => "ready", + Self::Running => "running", + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::Skipped => "skipped", + } + } + + /// A dependency is "satisfied" (a dependent may proceed) once it has + /// succeeded OR been skipped (a false `when` — satisfied-but-empty). + #[must_use] + pub fn is_satisfied(self) -> bool { + matches!(self, Self::Succeeded | Self::Skipped) + } + + #[must_use] + #[allow(clippy::should_implement_trait)] + pub fn from_str(s: &str) -> Option { + Some(match s { + "pending" => Self::Pending, + "ready" => Self::Ready, + "running" => Self::Running, + "succeeded" => Self::Succeeded, + "failed" => Self::Failed, + "skipped" => Self::Skipped, + _ => return None, + }) + } +}