From 5a630e1d9f4208bb2c6bb93ca2e8157268666d1c Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sun, 12 Jul 2026 17:48:38 +0200 Subject: [PATCH] =?UTF-8?q?feat(workflows):=20M5=20=E2=80=94=20SDK=20workf?= =?UTF-8?q?low::start=20+=20admin=20run=20API=20+=20pic=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The user-facing surface for the durable DAG engine. A workflow run can now be started three ways, all resolving the workflow by name in the caller's app (`cx.app_id` / the resolved app — never a passed-in arg, the isolation boundary), seeding a run the orchestrator advances. SDK seam (shared): - `WorkflowService` trait + `NoopWorkflowService` + `WorkflowError` in `shared::workflow` (mirrors `InvokeService`); added to `Services` as a noop-defaulted `with_workflow` builder (all `Services::new` call sites unchanged). - `WorkflowServiceImpl` (manager-core): resolves + seeds a run; authenticated callers gated on `AppInvoke` (anonymous skips, script-as-gate), the same gate `invoke()` uses. - `executor-core::sdk::workflow` — the Rhai bridge `workflow::start(name, input)` / `workflow::start(name)`, registered in `register_all`. Returns the run id. Admin API (manager-core `workflows_api`, mounted in the binary): - `GET /apps/{app}/workflows` — definitions (AppRead) - `POST /apps/{app}/workflows/{name}/runs` — start a run (AppInvoke) - `GET /apps/{app}/workflows/{name}/runs` — run history (AppRead) - `GET /apps/{app}/workflow-runs/{run_id}` — one run + its steps (AppRead) `app_id` scopes every query; a foreign run 404s. CLI (`pic workflows`): `ls` · `run [--input JSON]` · `runs ` · `run-status ` — all `--app`-scoped; run definitions stay declarative (`[[workflows]]` → `pic apply`). Also: `list_runs_for_workflow` repo reader. Tests: `workflow_service_start_seeds_a_run` (DB-gated — the SDK/API-shared entry) + `workflow_run_executes_end_to_end` CLI journey (apply → `pic workflows run` → poll `run-status` → succeeded, through the real binary + orchestrator). fmt + clippy -D warnings clean, 449 manager-core lib tests, 14 orchestrator DB tests, 18 executor-core lib tests, 3 workflow CLI journeys green. Co-Authored-By: Claude Opus 4.8 --- crates/executor-core/src/sdk/mod.rs | 2 + crates/executor-core/src/sdk/workflow.rs | 83 +++++ crates/manager-core/src/lib.rs | 4 + crates/manager-core/src/workflow_repo.rs | 20 ++ crates/manager-core/src/workflow_service.rs | 87 ++++++ crates/manager-core/src/workflows_api.rs | 292 ++++++++++++++++++ .../tests/workflow_orchestrator.rs | 62 +++- crates/picloud-cli/src/client.rs | 94 ++++++ crates/picloud-cli/src/cmds/mod.rs | 1 + crates/picloud-cli/src/cmds/workflows.rs | 79 +++++ crates/picloud-cli/src/main.rs | 54 ++++ crates/picloud-cli/tests/workflows.rs | 105 +++++++ crates/picloud/src/lib.rs | 15 +- crates/shared/src/lib.rs | 1 + crates/shared/src/services.rs | 19 +- crates/shared/src/workflow.rs | 54 +++- 16 files changed, 965 insertions(+), 7 deletions(-) create mode 100644 crates/executor-core/src/sdk/workflow.rs create mode 100644 crates/manager-core/src/workflow_service.rs create mode 100644 crates/manager-core/src/workflows_api.rs create mode 100644 crates/picloud-cli/src/cmds/workflows.rs diff --git a/crates/executor-core/src/sdk/mod.rs b/crates/executor-core/src/sdk/mod.rs index 649fb56..911a469 100644 --- a/crates/executor-core/src/sdk/mod.rs +++ b/crates/executor-core/src/sdk/mod.rs @@ -27,6 +27,7 @@ pub mod secrets; pub mod stdlib; pub mod users; pub mod vars; +pub mod workflow; pub use bridge::{dynamic_to_json, json_to_dynamic}; pub use cx::SdkCallCx; @@ -66,5 +67,6 @@ pub fn register_all( vars::register(engine, services, cx.clone()); email::register(engine, services, cx.clone()); users::register(engine, services, cx.clone()); + workflow::register(engine, services, cx.clone()); invoke::register(engine, services, cx, limits, self_engine); } diff --git a/crates/executor-core/src/sdk/workflow.rs b/crates/executor-core/src/sdk/workflow.rs new file mode 100644 index 0000000..1dd1af1 --- /dev/null +++ b/crates/executor-core/src/sdk/workflow.rs @@ -0,0 +1,83 @@ +//! `workflow::` Rhai bridge — start a durable workflow run (v1.2 M5). +//! +//! ```rhai +//! let run_id = workflow::start("nightly_report", #{ date: "2026-07-12" }); +//! let run_id = workflow::start("cleanup"); // no input +//! ``` +//! +//! Returns the run id as a string. The owning workflow resolves from +//! `cx.app_id` inside the service — never a script arg (the isolation +//! boundary). The durable orchestrator advances the run asynchronously; the +//! script does not block on completion. + +use std::sync::Arc; + +use picloud_shared::{SdkCallCx, Services, WorkflowService}; +use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Module}; +use tokio::runtime::Handle as TokioHandle; + +use crate::sdk::bridge::dynamic_to_json; + +pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc) { + let svc = services.workflow.clone(); + let mut module = Module::new(); + + // `workflow::start(name, input)` + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "start", + move |name: &str, input: Dynamic| -> Result> { + start_blocking(&svc, &cx, name, &input) + }, + ); + } + + // `workflow::start(name)` — no input (passes JSON null). + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "start", + move |name: &str| -> Result> { + start_blocking(&svc, &cx, name, &Dynamic::UNIT) + }, + ); + } + + engine.register_static_module("workflow", module.into()); +} + +fn start_blocking( + svc: &Arc, + cx: &Arc, + name: &str, + input: &Dynamic, +) -> Result> { + let input_json = if input.is_unit() { + serde_json::Value::Null + } else { + dynamic_to_json(input) + }; + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("workflow::start: no tokio runtime available: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let svc = svc.clone(); + let cx = cx.clone(); + let name = name.to_string(); + let run_id = handle + .block_on(async move { svc.start(&cx, &name, input_json).await }) + .map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("workflow::start: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + Ok(run_id.to_string()) +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 2080a60..7e60635 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -111,7 +111,9 @@ pub mod vars_service; pub mod workflow_expr; pub mod workflow_orchestrator; pub mod workflow_repo; +pub mod workflow_service; pub mod workflow_template; +pub mod workflows_api; pub use abandoned_repo::{ AbandonedRepo, AbandonedRepoError, NewAbandonedExecution, PostgresAbandonedRepo, @@ -269,3 +271,5 @@ 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}; +pub use workflow_service::WorkflowServiceImpl; +pub use workflows_api::{workflows_router, WorkflowsState}; diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs index 6723fab..9d3f5f1 100644 --- a/crates/manager-core/src/workflow_repo.rs +++ b/crates/manager-core/src/workflow_repo.rs @@ -865,6 +865,26 @@ 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"; +/// Most-recent runs of a workflow (newest first), capped at `limit`. Backs the +/// admin/CLI run-history list. +pub async fn list_runs_for_workflow( + pool: &PgPool, + app_id: AppId, + workflow_id: WorkflowId, + limit: i64, +) -> Result, WorkflowRepoError> { + let rows: Vec = sqlx::query_as(&format!( + "SELECT {RUN_COLS} FROM workflow_runs \ + WHERE app_id = $1 AND workflow_id = $2 ORDER BY created_at DESC LIMIT $3" + )) + .bind(app_id.into_inner()) + .bind(workflow_id.into_inner()) + .bind(limit) + .fetch_all(pool) + .await?; + rows.into_iter().map(TryInto::try_into).collect() +} + /// Read a single run by id (app-scoped for isolation). pub async fn get_run( pool: &PgPool, diff --git a/crates/manager-core/src/workflow_service.rs b/crates/manager-core/src/workflow_service.rs new file mode 100644 index 0000000..32ed114 --- /dev/null +++ b/crates/manager-core/src/workflow_service.rs @@ -0,0 +1,87 @@ +//! `WorkflowServiceImpl` — backs the Rhai `workflow::start(name, input)` SDK +//! call (v1.2 Workflows M5). +//! +//! Resolves a workflow **by name in the caller's app** (`cx.app_id` — never a +//! script-passed arg, the isolation boundary), seeds a durable run, and returns +//! its id. The dedicated orchestrator advances the run from there. Same-app +//! only, mirroring `InvokeServiceImpl`. +//! +//! Like `invoke()`, an authenticated caller is gated on `AppInvoke`; an +//! anonymous caller skips the check under the script-as-gate convention (the +//! public script IS the authorization boundary). + +use std::sync::Arc; + +use async_trait::async_trait; +use picloud_shared::{SdkCallCx, WorkflowError, WorkflowRunId, WorkflowService}; +use sqlx::PgPool; + +use crate::authz::{self, AuthzRepo, Capability}; +use crate::workflow_repo::{get_workflow_by_name, start_run, NewWorkflowRun}; + +pub struct WorkflowServiceImpl { + pool: PgPool, + /// Optional. When set, `start` requires `AppInvoke` for authenticated + /// callers; anonymous callers skip the check (script-as-gate). + authz: Option>, +} + +impl WorkflowServiceImpl { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool, authz: None } + } + + /// Attach the authz repo so authenticated callers are gated on `AppInvoke`. + #[must_use] + pub fn with_authz(mut self, authz: Arc) -> Self { + self.authz = Some(authz); + self + } + + async fn check(&self, cx: &SdkCallCx) -> Result<(), WorkflowError> { + let Some(authz_repo) = self.authz.as_ref() else { + return Ok(()); + }; + authz::script_gate( + authz_repo.as_ref(), + cx, + Capability::AppInvoke(cx.app_id), + || WorkflowError::Forbidden, + WorkflowError::Backend, + ) + .await + } +} + +#[async_trait] +impl WorkflowService for WorkflowServiceImpl { + async fn start( + &self, + cx: &SdkCallCx, + name: &str, + input: serde_json::Value, + ) -> Result { + self.check(cx).await?; + let wf = get_workflow_by_name(&self.pool, cx.app_id, name) + .await + .map_err(|e| WorkflowError::Backend(e.to_string()))? + .ok_or_else(|| WorkflowError::NotFound(name.to_string()))?; + if !wf.enabled { + return Err(WorkflowError::Disabled(name.to_string())); + } + let new = NewWorkflowRun { + workflow_id: wf.id, + app_id: cx.app_id, + input, + // Correlate the run's step logs under the caller's execution chain. + root_execution_id: cx.root_execution_id.into_inner(), + workflow_depth: 0, + parent_run_id: None, + parent_step_id: None, + }; + start_run(&self.pool, new, &wf.definition) + .await + .map_err(|e| WorkflowError::Backend(e.to_string())) + } +} diff --git a/crates/manager-core/src/workflows_api.rs b/crates/manager-core/src/workflows_api.rs new file mode 100644 index 0000000..53ec4e2 --- /dev/null +++ b/crates/manager-core/src/workflows_api.rs @@ -0,0 +1,292 @@ +//! Admin API for v1.2 Workflows (M5): list definitions, start runs, and read +//! run history. +//! +//! - `GET /apps/{app}/workflows` — list the app's workflows +//! - `POST /apps/{app}/workflows/{name}/runs` — start a run (body: input JSON) +//! - `GET /apps/{app}/workflows/{name}/runs` — recent runs of a workflow +//! - `GET /apps/{app}/workflow-runs/{run_id}` — one run + its steps +//! +//! Reads are gated on `AppRead`; starting a run is gated on `AppInvoke` — the +//! same capability the SDK `workflow::start` uses. `app_id` scopes every query +//! (the isolation boundary); a run of another app 404s. + +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Json, Response}; +use axum::routing::get; +use axum::{Extension, Router}; +use chrono::{DateTime, Utc}; +use picloud_shared::{AppId, Principal, WorkflowRunId}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use uuid::Uuid; + +use crate::app_repo::AppRepository; +use crate::authz::{require, AuthzDenied, AuthzError, AuthzRepo, Capability}; +use crate::workflow_repo::{ + get_run, get_workflow_by_name, list_run_steps, list_runs_for_workflow, list_workflows_for_app, + start_run, NewWorkflowRun, WorkflowRun, WorkflowRunStep, +}; +use sqlx::PgPool; + +/// Cap on the run-history list length. +const RUN_LIST_LIMIT: i64 = 100; + +#[derive(Clone)] +pub struct WorkflowsState { + pub pool: PgPool, + pub apps: Arc, + pub authz: Arc, +} + +pub fn workflows_router(state: WorkflowsState) -> Router { + Router::new() + .route("/apps/{app_id}/workflows", get(list_workflows)) + .route( + "/apps/{app_id}/workflows/{name}/runs", + get(list_runs).post(start_run_handler), + ) + .route("/apps/{app_id}/workflow-runs/{run_id}", get(get_run_detail)) + .with_state(state) +} + +// ---- response DTOs -------------------------------------------------------- + +#[derive(Serialize)] +struct WorkflowDto { + name: String, + enabled: bool, + steps: usize, +} + +#[derive(Serialize)] +struct RunDto { + id: Uuid, + workflow_id: Uuid, + status: String, + input: Value, + output: Option, + error: Option, + workflow_depth: i32, + root_execution_id: Uuid, + created_at: DateTime, + started_at: Option>, + finished_at: Option>, +} + +impl From for RunDto { + fn from(r: WorkflowRun) -> Self { + Self { + id: r.id.into_inner(), + workflow_id: r.workflow_id.into_inner(), + status: r.status.as_str().to_string(), + input: r.input, + output: r.output, + error: r.error, + workflow_depth: r.workflow_depth, + root_execution_id: r.root_execution_id, + created_at: r.created_at, + started_at: r.started_at, + finished_at: r.finished_at, + } + } +} + +#[derive(Serialize)] +struct StepDto { + name: String, + status: String, + attempt: i32, + max_attempts: i32, + output: Option, + error: Option, + child_run_id: Option, +} + +impl From for StepDto { + fn from(s: WorkflowRunStep) -> Self { + Self { + name: s.step_name, + status: s.status.as_str().to_string(), + attempt: s.attempt, + max_attempts: s.max_attempts, + output: s.output, + error: s.error, + child_run_id: s.child_run_id.map(WorkflowRunId::into_inner), + } + } +} + +#[derive(Deserialize, Default)] +struct StartRunReq { + #[serde(default)] + input: Value, +} + +// ---- handlers ------------------------------------------------------------- + +async fn list_workflows( + State(s): State, + Extension(principal): Extension, + Path(id_or_slug): Path, +) -> Result>, WorkflowsApiError> { + let app_id = resolve_app(&*s.apps, &id_or_slug).await?; + require(s.authz.as_ref(), &principal, Capability::AppRead(app_id)).await?; + let workflows = list_workflows_for_app(&s.pool, app_id) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + Ok(Json( + workflows + .into_iter() + .map(|w| WorkflowDto { + name: w.name, + enabled: w.enabled, + steps: w.definition.steps.len(), + }) + .collect(), + )) +} + +async fn start_run_handler( + State(s): State, + Extension(principal): Extension, + Path((id_or_slug, name)): Path<(String, String)>, + body: Option>, +) -> Result<(StatusCode, Json), WorkflowsApiError> { + let app_id = resolve_app(&*s.apps, &id_or_slug).await?; + require(s.authz.as_ref(), &principal, Capability::AppInvoke(app_id)).await?; + + let wf = get_workflow_by_name(&s.pool, app_id, &name) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + .ok_or_else(|| WorkflowsApiError::NotFound(format!("workflow {name:?}")))?; + if !wf.enabled { + return Err(WorkflowsApiError::Invalid(format!( + "workflow {name:?} is disabled" + ))); + } + let input = body.map_or(Value::Null, |Json(b)| b.input); + let new = NewWorkflowRun { + workflow_id: wf.id, + app_id, + input, + root_execution_id: Uuid::new_v4(), + workflow_depth: 0, + parent_run_id: None, + parent_step_id: None, + }; + let run_id = start_run(&s.pool, new, &wf.definition) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + // Read it back for the response (status = pending until the orchestrator ticks). + let run = get_run(&s.pool, app_id, run_id) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + .ok_or_else(|| WorkflowsApiError::Backend("run vanished after start".into()))?; + Ok((StatusCode::CREATED, Json(run.into()))) +} + +async fn list_runs( + State(s): State, + Extension(principal): Extension, + Path((id_or_slug, name)): Path<(String, String)>, +) -> Result>, WorkflowsApiError> { + let app_id = resolve_app(&*s.apps, &id_or_slug).await?; + require(s.authz.as_ref(), &principal, Capability::AppRead(app_id)).await?; + let wf = get_workflow_by_name(&s.pool, app_id, &name) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + .ok_or_else(|| WorkflowsApiError::NotFound(format!("workflow {name:?}")))?; + let runs = list_runs_for_workflow(&s.pool, app_id, wf.id, RUN_LIST_LIMIT) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + Ok(Json(runs.into_iter().map(Into::into).collect())) +} + +async fn get_run_detail( + State(s): State, + Extension(principal): Extension, + Path((id_or_slug, run_id)): Path<(String, Uuid)>, +) -> Result, WorkflowsApiError> { + let app_id = resolve_app(&*s.apps, &id_or_slug).await?; + require(s.authz.as_ref(), &principal, Capability::AppRead(app_id)).await?; + let run = get_run(&s.pool, app_id, WorkflowRunId::from(run_id)) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + .ok_or_else(|| WorkflowsApiError::NotFound(format!("run {run_id}")))?; + let steps = list_run_steps(&s.pool, run.id) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + let run_dto: RunDto = run.into(); + let step_dtos: Vec = steps.into_iter().map(Into::into).collect(); + Ok(Json(json!({ "run": run_dto, "steps": step_dtos }))) +} + +async fn resolve_app(apps: &dyn AppRepository, ident: &str) -> Result { + crate::app_repo::resolve_app(apps, ident) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + .map(|l| l.app.id) + .ok_or_else(|| WorkflowsApiError::NotFound(format!("app {ident:?}"))) +} + +// ---- errors --------------------------------------------------------------- + +#[derive(Debug, thiserror::Error)] +pub enum WorkflowsApiError { + #[error("{0}")] + NotFound(String), + #[error("{0}")] + Invalid(String), + #[error("forbidden")] + Forbidden, + #[error("authorization repo error: {0}")] + AuthzRepo(String), + #[error("workflow backend: {0}")] + Backend(String), +} + +impl From for WorkflowsApiError { + fn from(d: AuthzDenied) -> Self { + match d { + AuthzDenied::Denied => Self::Forbidden, + AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()), + } + } +} + +impl From for WorkflowsApiError { + fn from(e: AuthzError) -> Self { + Self::AuthzRepo(e.to_string()) + } +} + +impl IntoResponse for WorkflowsApiError { + fn into_response(self) -> Response { + let (status, body) = match &self { + Self::NotFound(_) => (StatusCode::NOT_FOUND, json!({ "error": self.to_string() })), + Self::Invalid(_) => ( + StatusCode::UNPROCESSABLE_ENTITY, + json!({ "error": self.to_string() }), + ), + Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })), + Self::AuthzRepo(e) => { + tracing::error!(error = %e, "workflows authz repo error"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": "internal error" }), + ) + } + Self::Backend(e) => { + tracing::error!(error = %e, "workflows api backend error"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": "internal error" }), + ) + } + }; + (status, Json(body)).into_response() + } +} diff --git a/crates/manager-core/tests/workflow_orchestrator.rs b/crates/manager-core/tests/workflow_orchestrator.rs index 8d90c29..02cea49 100644 --- a/crates/manager-core/tests/workflow_orchestrator.rs +++ b/crates/manager-core/tests/workflow_orchestrator.rs @@ -22,15 +22,17 @@ 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, + claim_ready_step, complete_step_and_advance, get_run, list_run_steps, list_runs_for_workflow, + reclaim_stale_steps, start_run, AdvanceResult, ClaimedStep, NewWorkflowRun, StepOutcome, }; +use picloud_manager_core::workflow_service::WorkflowServiceImpl; 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 picloud_shared::{RequestId, ScriptId, WorkflowService}; use serde_json::{json, Value}; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; @@ -927,3 +929,59 @@ async fn nesting_depth_ceiling_fails_the_run() { "self-nesting run fails at the depth ceiling, not infinitely" ); } + +// =========================================================================== +// M5 — the SDK/API-shared service entry (`WorkflowServiceImpl::start`). +// =========================================================================== + +fn anon_cx(app: AppId) -> picloud_shared::SdkCallCx { + let execution_id = picloud_shared::ExecutionId::new(); + picloud_shared::SdkCallCx { + app_id: app, + script_id: ScriptId::new(), + principal: None, + execution_id, + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: execution_id, + is_dead_letter_handler: false, + event: None, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn workflow_service_start_seeds_a_run() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let _claim_guard = lock_and_reset(&pool).await; + let app = seed_app(&pool).await; + let def = WorkflowDefinition { + steps: vec![step("only", "fn_only", &[])], + }; + seed_named_workflow(&pool, app, "svc_wf", &def).await; + + let svc = WorkflowServiceImpl::new(pool.clone()); + let cx = anon_cx(app); + + // start seeds a durable run with the given input + its steps. + let run_id = svc + .start(&cx, "svc_wf", json!({ "k": "v" })) + .await + .expect("start"); + let r = get_run(&pool, app, run_id).await.unwrap().unwrap(); + assert_eq!(r.input, json!({ "k": "v" })); + let steps = list_run_steps(&pool, run_id).await.unwrap(); + assert_eq!(steps.len(), 1); + assert_eq!(steps[0].step_name, "only"); + assert_eq!(steps[0].status, StepStatus::Ready); // a root + + // it shows up in the run-history list. + let runs = list_runs_for_workflow(&pool, app, r.workflow_id, 10) + .await + .unwrap(); + assert_eq!(runs.len(), 1); + + // an unknown workflow name is a not-found error. + assert!(svc.start(&cx, "ghost", Value::Null).await.is_err()); +} diff --git a/crates/picloud-cli/src/client.rs b/crates/picloud-cli/src/client.rs index 1cef39c..5ac8f66 100644 --- a/crates/picloud-cli/src/client.rs +++ b/crates/picloud-cli/src/client.rs @@ -624,6 +624,65 @@ impl Client { decode(resp).await } + /// `GET /api/v1/admin/apps/{id}/workflows` + pub async fn workflows_list(&self, app: &str) -> Result> { + let app = seg(app); + let resp = self + .request(Method::GET, &format!("/api/v1/admin/apps/{app}/workflows")) + .send() + .await?; + decode(resp).await + } + + /// `POST /api/v1/admin/apps/{id}/workflows/{name}/runs` — start a run. + pub async fn workflow_run_start( + &self, + app: &str, + name: &str, + input: serde_json::Value, + ) -> Result { + let (app, name) = (seg(app), seg(name)); + let resp = self + .request( + Method::POST, + &format!("/api/v1/admin/apps/{app}/workflows/{name}/runs"), + ) + .json(&serde_json::json!({ "input": input })) + .send() + .await?; + decode(resp).await + } + + /// `GET /api/v1/admin/apps/{id}/workflows/{name}/runs` — recent runs. + pub async fn workflow_runs_list(&self, app: &str, name: &str) -> Result> { + let (app, name) = (seg(app), seg(name)); + let resp = self + .request( + Method::GET, + &format!("/api/v1/admin/apps/{app}/workflows/{name}/runs"), + ) + .send() + .await?; + decode(resp).await + } + + /// `GET /api/v1/admin/apps/{id}/workflow-runs/{run_id}` — one run + steps. + pub async fn workflow_run_status( + &self, + app: &str, + run_id: &str, + ) -> Result { + let (app, run_id) = (seg(app), seg(run_id)); + let resp = self + .request( + Method::GET, + &format!("/api/v1/admin/apps/{app}/workflow-runs/{run_id}"), + ) + .send() + .await?; + decode(resp).await + } + /// `DELETE /api/v1/admin/apps/{id}/triggers/{trigger_id}` pub async fn triggers_delete(&self, app: &str, trigger_id: &str) -> Result<()> { let (app, trigger_id) = (seg(app), seg(trigger_id)); @@ -2202,6 +2261,41 @@ struct UpdateScriptBody<'a> { // columns are a one-line add), but the TSV/KvBlock renderers don't print // every field today. +#[derive(Debug, Deserialize)] +pub struct WorkflowDto { + pub name: String, + pub enabled: bool, + pub steps: usize, +} + +#[derive(Debug, Deserialize)] +pub struct WorkflowRunDto { + pub id: String, + pub status: String, + #[serde(default)] + pub error: Option, + pub workflow_depth: i32, + pub created_at: String, +} + +#[derive(Debug, Deserialize)] +pub struct WorkflowRunStepDto { + pub name: String, + pub status: String, + pub attempt: i32, + pub max_attempts: i32, + #[serde(default)] + pub error: Option, + #[serde(default)] + pub child_run_id: Option, +} + +#[derive(Debug, Deserialize)] +pub struct WorkflowRunDetailDto { + pub run: WorkflowRunDto, + pub steps: Vec, +} + #[derive(Debug, Deserialize)] pub struct AppMemberDto { pub user_id: AdminUserId, diff --git a/crates/picloud-cli/src/cmds/mod.rs b/crates/picloud-cli/src/cmds/mod.rs index af03e7a..6213d88 100644 --- a/crates/picloud-cli/src/cmds/mod.rs +++ b/crates/picloud-cli/src/cmds/mod.rs @@ -30,6 +30,7 @@ pub mod triggers; pub mod users; pub mod vars; pub mod whoami; +pub mod workflows; /// A resolved owner reference for a command that targets EITHER an app or a /// group (the `--app` / `--group` XOR). See [`require_one_owner`]. diff --git a/crates/picloud-cli/src/cmds/workflows.rs b/crates/picloud-cli/src/cmds/workflows.rs new file mode 100644 index 0000000..a840624 --- /dev/null +++ b/crates/picloud-cli/src/cmds/workflows.rs @@ -0,0 +1,79 @@ +//! `pic workflows` subcommands (v1.2 Workflows M5). +//! +//! Read-only listing + starting/inspecting runs. Workflow *definitions* are +//! authored declaratively (`[[workflows]]` in `picloud.toml` → `pic apply`); +//! this surface starts runs and reads run history. + +use anyhow::{Context, Result}; +use serde_json::Value; + +use crate::client::Client; +use crate::config; +use crate::output::{KvBlock, OutputMode, Table}; + +/// `pic workflows ls --app ` — the app's workflow definitions. +pub async fn ls(app: &str, mode: OutputMode) -> Result<()> { + let creds = config::resolve()?; + let client = Client::from_creds(&creds)?; + let workflows = client.workflows_list(app).await?; + let mut table = Table::new(["name", "enabled", "steps"]); + for w in workflows { + table.row([w.name, w.enabled.to_string(), w.steps.to_string()]); + } + table.print(mode); + Ok(()) +} + +/// `pic workflows run --app [--input JSON]` — start a run. +pub async fn run(app: &str, name: &str, input: Option<&str>, mode: OutputMode) -> Result<()> { + let input_json: Value = match input { + Some(s) => serde_json::from_str(s) + .context("--input must be valid JSON (e.g. '{\"date\":\"2026-07-12\"}')")?, + None => Value::Null, + }; + let creds = config::resolve()?; + let client = Client::from_creds(&creds)?; + let run = client.workflow_run_start(app, name, input_json).await?; + KvBlock::new() + .field("run_id", run.id) + .field("status", run.status) + .print(mode); + Ok(()) +} + +/// `pic workflows runs --app ` — recent runs of a workflow. +pub async fn runs(app: &str, name: &str, mode: OutputMode) -> Result<()> { + let creds = config::resolve()?; + let client = Client::from_creds(&creds)?; + let runs = client.workflow_runs_list(app, name).await?; + let mut table = Table::new(["run_id", "status", "depth", "created_at"]); + for r in runs { + table.row([r.id, r.status, r.workflow_depth.to_string(), r.created_at]); + } + table.print(mode); + Ok(()) +} + +/// `pic workflows run-status --app ` — one run + its steps. +pub async fn run_status(app: &str, run_id: &str, mode: OutputMode) -> Result<()> { + let creds = config::resolve()?; + let client = Client::from_creds(&creds)?; + let detail = client.workflow_run_status(app, run_id).await?; + KvBlock::new() + .field("run_id", detail.run.id) + .field("status", detail.run.status) + .field("error", detail.run.error.unwrap_or_default()) + .print(mode); + let mut table = Table::new(["step", "status", "attempt", "error", "child_run"]); + for s in detail.steps { + table.row([ + s.name, + s.status, + format!("{}/{}", s.attempt, s.max_attempts), + s.error.unwrap_or_default(), + s.child_run_id.unwrap_or_default(), + ]); + } + table.print(mode); + Ok(()) +} diff --git a/crates/picloud-cli/src/main.rs b/crates/picloud-cli/src/main.rs index 104c130..1acbda4 100644 --- a/crates/picloud-cli/src/main.rs +++ b/crates/picloud-cli/src/main.rs @@ -113,6 +113,14 @@ enum Cmd { cmd: TriggersCmd, }, + /// Durable DAG workflows (v1.2). Definitions are authored declaratively + /// (`[[workflows]]` in `picloud.toml` → `pic apply`); this surface starts + /// runs and reads run history. + Workflows { + #[command(subcommand)] + cmd: WorkflowsCmd, + }, + /// Pub/sub topic registry — control which topics external clients /// can subscribe to over SSE (`/realtime/topics/{name}`). Publishing /// from a script needs no registration; only outside subscribers do. @@ -958,6 +966,40 @@ enum RoutesCmd { }, } +#[derive(Subcommand)] +enum WorkflowsCmd { + /// List an app's workflow definitions. + Ls { + #[arg(long)] + app: String, + }, + /// Start a run of a workflow. + Run { + #[arg(long)] + app: String, + /// Workflow name. + name: String, + /// Run input as a JSON object (e.g. '{"date":"2026-07-12"}'). + #[arg(long)] + input: Option, + }, + /// List recent runs of a workflow. + Runs { + #[arg(long)] + app: String, + /// Workflow name. + name: String, + }, + /// Show one run's status + its per-step detail. + #[command(name = "run-status")] + RunStatus { + #[arg(long)] + app: String, + /// Run id (from `pic workflows run` / `runs`). + run_id: String, + }, +} + #[derive(Subcommand)] enum TriggersCmd { /// List an app's triggers, or a group's trigger TEMPLATES (`--group`). @@ -1796,6 +1838,18 @@ async fn main() -> ExitCode { Cmd::Admins { cmd: AdminsCmd::Rm { id }, } => cmds::admins::rm(&id).await, + Cmd::Workflows { + cmd: WorkflowsCmd::Ls { app }, + } => cmds::workflows::ls(&app, mode).await, + Cmd::Workflows { + cmd: WorkflowsCmd::Run { app, name, input }, + } => cmds::workflows::run(&app, &name, input.as_deref(), mode).await, + Cmd::Workflows { + cmd: WorkflowsCmd::Runs { app, name }, + } => cmds::workflows::runs(&app, &name, mode).await, + Cmd::Workflows { + cmd: WorkflowsCmd::RunStatus { app, run_id }, + } => cmds::workflows::run_status(&app, &run_id, mode).await, Cmd::Triggers { cmd: TriggersCmd::Ls { app, group }, } => match cmds::require_one_owner(app.as_deref(), group.as_deref()) { diff --git a/crates/picloud-cli/tests/workflows.rs b/crates/picloud-cli/tests/workflows.rs index 4bcc8e1..63e4dd2 100644 --- a/crates/picloud-cli/tests/workflows.rs +++ b/crates/picloud-cli/tests/workflows.rs @@ -153,3 +153,108 @@ fn workflow_with_a_cycle_is_rejected() { let err = String::from_utf8_lossy(&out.stderr); assert!(err.contains("cycle"), "expected a cycle error, got:\n{err}"); } + +/// M5 end-to-end: `pic workflows run` starts a run, the durable orchestrator +/// (running inside the fixture's picloud) executes both steps, and +/// `run-status` shows it succeeded. +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn workflow_run_executes_end_to_end() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let slug = common::unique_slug("wfe2e"); + 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); + // validate → charge; charge only runs when validate's output.ok is true. + let manifest = format!( + "[app]\nslug = \"{slug}\"\nname = \"WF E2E\"\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, &manifest).unwrap(); + common::pic_as(&env) + .args(["apply", "--file"]) + .arg(&path) + .assert() + .success(); + + // `pic workflows ls` shows the definition. + let ls = String::from_utf8( + common::pic_as(&env) + .args(["workflows", "ls", "--app", &slug]) + .output() + .unwrap() + .stdout, + ) + .unwrap(); + assert!( + ls.contains("pipeline"), + "workflows ls missing pipeline:\n{ls}" + ); + + // Start a run (JSON output so we can parse the run id). + let run_out = common::pic_as(&env) + .args([ + "--output", + "json", + "workflows", + "run", + "--app", + &slug, + "pipeline", + "--input", + "{\"order\":\"o1\"}", + ]) + .output() + .expect("run"); + assert!( + run_out.status.success(), + "workflows run failed: {}", + String::from_utf8_lossy(&run_out.stderr) + ); + let started: serde_json::Value = + serde_json::from_slice(&run_out.stdout).expect("run output is JSON"); + let run_id = started["run_id"] + .as_str() + .expect("run_id in output") + .to_string(); + + // Poll run-status until the run reaches a terminal state. + let mut final_status = String::new(); + for _ in 0..40 { + let status_out = String::from_utf8( + common::pic_as(&env) + .args(["workflows", "run-status", "--app", &slug, &run_id]) + .output() + .unwrap() + .stdout, + ) + .unwrap(); + final_status = status_out; + if final_status.contains("succeeded") || final_status.contains("failed") { + break; + } + std::thread::sleep(std::time::Duration::from_millis(300)); + } + assert!( + final_status.contains("succeeded"), + "run did not succeed in time:\n{final_status}" + ); + // Both steps ran (validate always; charge because when was true). + assert!( + final_status.contains('v') && final_status.contains('c'), + "expected both steps in the status detail:\n{final_status}" + ); +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 99b0a24..9413267 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -411,6 +411,11 @@ pub async fn build_app( ); let vars: Arc = Arc::new(VarsServiceImpl::new(pool.clone(), authz.clone())); + // v1.2 Workflows (M5): `workflow::start(name, input)`. Gated on AppInvoke + // for authenticated callers, like invoke() (anonymous skips it). + let workflow: Arc = Arc::new( + picloud_manager_core::WorkflowServiceImpl::new(pool.clone()).with_authz(authz.clone()), + ); let services = Services::new( kv, docs, @@ -431,7 +436,8 @@ pub async fn build_app( .with_group_docs(group_docs) .with_group_files(group_files) .with_group_pubsub(group_pubsub) - .with_group_queue(group_queue); + .with_group_queue(group_queue) + .with_workflow(workflow); // v1.1.9: keep the invoke depth bound aligned with the dispatcher's // trigger-depth bound (same counter under the hood). let engine_limits = Limits { @@ -569,6 +575,12 @@ pub async fn build_app( config: trigger_config, master_key: master_key.clone(), }; + // v1.2 Workflows (M5): list definitions, start runs, read run history. + let workflows_state = picloud_manager_core::WorkflowsState { + pool: pool.clone(), + apps: apps_repo.clone(), + authz: authz.clone(), + }; // Declarative reconcile engine (pic plan / apply). Trait-object repos // for the read/diff path; shares the same handles as the CRUD routers. let apply_service = ApplyService { @@ -738,6 +750,7 @@ pub async fn build_app( )) .merge(api_keys_router(api_keys_state)) .merge(triggers_router(triggers_state)) + .merge(picloud_manager_core::workflows_router(workflows_state)) .merge(apply_router(apply_service)) .merge(projects_router(ProjectsState { projects: projects_repo.clone(), diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index c89509c..c8a3658 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -120,3 +120,4 @@ pub use users::{ pub use validator::{ScriptValidator, ValidatedScript, ValidationError}; pub use vars::{NoopVarsService, VarsError, VarsService}; pub use version::{API_VERSION, PRODUCT_VERSION, SDK_VERSION, WIRE_VERSION}; +pub use workflow::{NoopWorkflowService, WorkflowError, WorkflowService}; diff --git a/crates/shared/src/services.rs b/crates/shared/src/services.rs index fc366d5..e66de8b 100644 --- a/crates/shared/src/services.rs +++ b/crates/shared/src/services.rs @@ -26,8 +26,9 @@ use crate::{ NoopEmailService, NoopEventEmitter, NoopFilesService, NoopGroupDocsService, NoopGroupFilesService, NoopGroupKvService, NoopGroupPubsubService, NoopGroupQueueService, NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource, NoopPubsubService, - NoopQueueService, NoopSecretsService, NoopUsersService, NoopVarsService, PubsubService, - QueueService, SecretsService, ServiceEventEmitter, UsersService, VarsService, + NoopQueueService, NoopSecretsService, NoopUsersService, NoopVarsService, NoopWorkflowService, + PubsubService, QueueService, SecretsService, ServiceEventEmitter, UsersService, VarsService, + WorkflowService, }; /// SDK service bundle. See module docs for the lifecycle and the v1.1.x @@ -146,6 +147,11 @@ pub struct Services { /// drained by competing per-descendant consumers. Wired via /// [`Services::with_group_queue`]; defaults to `NoopGroupQueueService`. pub group_queue: Arc, + + /// v1.2 Workflows (M5). Scripts get `workflow::start(name, input)` — start a + /// run of a named workflow in the caller's app. Wired via + /// [`Services::with_workflow`]; defaults to `NoopWorkflowService`. + pub workflow: Arc, } impl Services { @@ -193,9 +199,18 @@ impl Services { group_files: Arc::new(NoopGroupFilesService), group_pubsub: Arc::new(NoopGroupPubsubService), group_queue: Arc::new(NoopGroupQueueService), + workflow: Arc::new(NoopWorkflowService), } } + /// Set the v1.2 Workflows service (picloud binary wires the Postgres-backed + /// impl; tests leave the noop default). + #[must_use] + pub fn with_workflow(mut self, workflow: Arc) -> Self { + self.workflow = workflow; + self + } + /// Set the §11.6 shared group-KV service (the picloud binary wires the /// Postgres-backed impl; tests leave the noop default). #[must_use] diff --git a/crates/shared/src/workflow.rs b/crates/shared/src/workflow.rs index 990d940..ec83279 100644 --- a/crates/shared/src/workflow.rs +++ b/crates/shared/src/workflow.rs @@ -7,10 +7,60 @@ //! 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. +//! The `WorkflowService` trait is the injected SDK seam for `workflow::start` +//! (M5): a script (or the admin API) starts a run of a named workflow in the +//! caller's app. Same-app only — the owning workflow resolves from `cx.app_id`, +//! never a script-passed arg (the isolation boundary), mirroring `InvokeService`. +use async_trait::async_trait; use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::{SdkCallCx, WorkflowRunId}; + +/// Backs the Rhai `workflow::start(name, input)` call and the admin run API. +#[async_trait] +pub trait WorkflowService: Send + Sync { + /// Start a run of the workflow named `name` in `cx.app_id`, seeded with + /// `input`. Returns the new run id. The durable orchestrator advances it. + async fn start( + &self, + cx: &SdkCallCx, + name: &str, + input: serde_json::Value, + ) -> Result; +} + +#[derive(Debug, Error)] +pub enum WorkflowError { + #[error("workflow {0:?} not found")] + NotFound(String), + #[error("workflow {0:?} is disabled")] + Disabled(String), + /// The authenticated caller lacks `AppInvoke` on the app. + #[error("forbidden")] + Forbidden, + #[error("workflow backend error: {0}")] + Backend(String), +} + +/// Default no-op: every call errors so a misconfigured bundle can't silently +/// no-op a `workflow::start`. +pub struct NoopWorkflowService; + +#[async_trait] +impl WorkflowService for NoopWorkflowService { + async fn start( + &self, + _cx: &SdkCallCx, + _name: &str, + _input: serde_json::Value, + ) -> Result { + Err(WorkflowError::Backend( + "workflow service is not configured".into(), + )) + } +} /// A workflow's DAG: an ordered list of steps connected by `depends_on` edges. /// Serialized to the `workflows.definition` JSONB column.