feat(workflows): M5 — SDK workflow::start + admin run API + pic CLI

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 <name> [--input JSON]` · `runs <name>` ·
`run-status <run_id>` — 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 <noreply@anthropic.com>

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};

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<Vec<WorkflowRun>, WorkflowRepoError> {
+    let rows: Vec<RunRow> = 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,

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<Arc<dyn AuthzRepo>>,
+}
+
+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<dyn AuthzRepo>) -> 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<WorkflowRunId, WorkflowError> {
+        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()))
+    }
+}

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<dyn AppRepository>,
+    pub authz: Arc<dyn AuthzRepo>,
+}
+
+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<Value>,
+    error: Option<String>,
+    workflow_depth: i32,
+    root_execution_id: Uuid,
+    created_at: DateTime<Utc>,
+    started_at: Option<DateTime<Utc>>,
+    finished_at: Option<DateTime<Utc>>,
+}
+
+impl From<WorkflowRun> 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<Value>,
+    error: Option<String>,
+    child_run_id: Option<Uuid>,
+}
+
+impl From<WorkflowRunStep> 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<WorkflowsState>,
+    Extension(principal): Extension<Principal>,
+    Path(id_or_slug): Path<String>,
+) -> Result<Json<Vec<WorkflowDto>>, 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<WorkflowsState>,
+    Extension(principal): Extension<Principal>,
+    Path((id_or_slug, name)): Path<(String, String)>,
+    body: Option<Json<StartRunReq>>,
+) -> Result<(StatusCode, Json<RunDto>), 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<WorkflowsState>,
+    Extension(principal): Extension<Principal>,
+    Path((id_or_slug, name)): Path<(String, String)>,
+) -> Result<Json<Vec<RunDto>>, 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<WorkflowsState>,
+    Extension(principal): Extension<Principal>,
+    Path((id_or_slug, run_id)): Path<(String, Uuid)>,
+) -> Result<Json<Value>, 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<StepDto> = 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<AppId, WorkflowsApiError> {
+    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<AuthzDenied> for WorkflowsApiError {
+    fn from(d: AuthzDenied) -> Self {
+        match d {
+            AuthzDenied::Denied => Self::Forbidden,
+            AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()),
+        }
+    }
+}
+
+impl From<AuthzError> 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()
+    }
+}

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