klieo_workflow_api/
lib.rs

1#![deny(missing_docs)]
2#![deny(rustdoc::broken_intra_doc_links)]
3//! `klieo-workflow-api` — embeddable Axum run-service fronting
4//! [`klieo_workflow`].
5//!
6//! A caller submits a `WorkflowDef` to `POST /runs`; the service compiles it
7//! against a [`klieo_workflow::Registry`] allow-list, runs it asynchronously,
8//! and returns a run id to poll at `GET /runs/{id}`. Auth-gated, with a
9//! server-stamped author and a bounded, supervised executor. `GET /health`
10//! is an unauthenticated liveness probe, for an orchestrator that has no
11//! credential to present.
12//!
13//! Mount the router produced by [`WorkflowRunRouterBuilder::build`].
14//!
15//! # Deploying
16//!
17//! This crate — not `klieo-server` — is the deployable run-service surface:
18//! `klieo-server` is a `publish = false` reference demo, not meant to serve
19//! real traffic. For container/orchestration topology (Docker Compose,
20//! Helm, Terraform) fronting a service built on this router, use the
21//! `klieo-compliance` repo's `deploy/` tree as the reference: swap its
22//! compliance-server container image for a binary that mounts
23//! [`WorkflowRunRouterBuilder::build`]'s router; the surrounding topology
24//! (reverse proxy, secrets) carries over. klieo is pre-production —
25//! `deploy/` does not itself prescribe autoscaling or canary rollout; add
26//! those only if your deployment needs them.
27
28mod auth;
29mod error;
30mod events;
31mod executor;
32mod handlers;
33mod progress;
34mod provenance;
35mod run_store;
36mod state;
37mod sweep;
38
39pub use auth::{WORKFLOW_READ_ALL_SCOPE, WORKFLOW_READ_SCOPE, WORKFLOW_RUN_SCOPE};
40pub use error::{ErrorCode, WorkflowApiError};
41pub use run_store::{ClaimOutcome, RunRecordView, RunStatus, RunStore, StoreError, RUNS_BUCKET};
42pub use sweep::{sweep_orphaned_runs, SweepError, SweepSummary};
43
44use crate::state::WorkflowRunState;
45use axum::{
46    middleware,
47    routing::{get, post},
48    Router,
49};
50use klieo::App;
51use klieo_auth_common::Authenticator;
52use klieo_provenance::ProvenanceRepository;
53use klieo_workflow::Registry;
54use std::sync::Arc;
55use std::time::Duration;
56use tokio::sync::Semaphore;
57use tower_http::limit::RequestBodyLimitLayer;
58
59/// Maximum number of runs executing at once; excess submits get `429`.
60pub const MAX_CONCURRENT_RUNS: usize = 32;
61
62/// Wall-clock cap on a single run; exceeding it aborts the run.
63pub const MAX_RUN_DURATION: Duration = Duration::from_secs(300);
64
65/// Maximum accepted `POST /runs` body size, bounding the definition + input.
66pub const MAX_DEF_BYTES: usize = 256 * 1024;
67
68/// Why [`WorkflowRunRouterBuilder::build`] could not assemble a router.
69#[non_exhaustive]
70#[derive(Debug, thiserror::Error)]
71pub enum WorkflowRunBuilderError {
72    /// [`WorkflowRunRouterBuilder::with_authenticator`] was not called.
73    #[error("authenticator not configured — call with_authenticator() before build()")]
74    MissingAuthenticator,
75    /// [`WorkflowRunRouterBuilder::with_app`] was not called.
76    #[error("app not configured — call with_app() before build()")]
77    MissingApp,
78    /// [`WorkflowRunRouterBuilder::with_registry`] was not called.
79    #[error("registry not configured — call with_registry() before build()")]
80    MissingRegistry,
81    /// [`WorkflowRunRouterBuilder::with_run_store`] was not called.
82    #[error("run store not configured — call with_run_store() before build()")]
83    MissingRunStore,
84    /// [`WorkflowRunRouterBuilder::with_provenance`] was not called. Audit is
85    /// mandatory — the service refuses to run without a durable chain.
86    #[error("provenance repository not configured — call with_provenance() before build()")]
87    MissingProvenance,
88}
89
90/// Builds the run-service router (`POST /runs`, `GET /runs/{id}`).
91#[derive(Default)]
92pub struct WorkflowRunRouterBuilder {
93    authenticator: Option<Arc<dyn Authenticator>>,
94    app: Option<Arc<App>>,
95    registry: Option<Registry>,
96    run_store: Option<RunStore>,
97    provenance: Option<Arc<dyn ProvenanceRepository>>,
98    run_timeout: Option<Duration>,
99    max_concurrent_runs: Option<usize>,
100}
101
102impl WorkflowRunRouterBuilder {
103    /// Start an empty builder.
104    pub fn new() -> Self {
105        Self::default()
106    }
107
108    /// Required — the credential verifier for every route.
109    pub fn with_authenticator(mut self, authenticator: Arc<dyn Authenticator>) -> Self {
110        self.authenticator = Some(authenticator);
111        self
112    }
113
114    /// Required — the App that mints an `AgentContext` per run. Its LLM is
115    /// unused (each agent node swaps to its registry model); it supplies the
116    /// bus, kv, tools, and memory a run needs.
117    pub fn with_app(mut self, app: Arc<App>) -> Self {
118        self.app = Some(app);
119        self
120    }
121
122    /// Required — the allow-list a submitted definition compiles against.
123    pub fn with_registry(mut self, registry: Registry) -> Self {
124        self.registry = Some(registry);
125        self
126    }
127
128    /// Required — the run-status poll cache.
129    pub fn with_run_store(mut self, run_store: RunStore) -> Self {
130        self.run_store = Some(run_store);
131        self
132    }
133
134    /// Required — the durable, authoritative audit chain. Audit is mandatory;
135    /// [`Self::build`] errors without it. Supply a durable
136    /// `SqliteProvenanceRepository::open(path)` in production.
137    pub fn with_provenance(mut self, provenance: Arc<dyn ProvenanceRepository>) -> Self {
138        self.provenance = Some(provenance);
139        self
140    }
141
142    /// Override the per-run wall-clock cap. Defaults to [`MAX_RUN_DURATION`].
143    pub fn with_run_timeout(mut self, timeout: Duration) -> Self {
144        self.run_timeout = Some(timeout);
145        self
146    }
147
148    /// Override the concurrency cap. Defaults to [`MAX_CONCURRENT_RUNS`].
149    pub fn with_max_concurrent_runs(mut self, max: usize) -> Self {
150        self.max_concurrent_runs = Some(max);
151        self
152    }
153
154    /// Assemble the router.
155    ///
156    /// # Errors
157    /// Returns a [`WorkflowRunBuilderError`] variant naming the first
158    /// required port that was not configured.
159    pub fn build(self) -> Result<Router, WorkflowRunBuilderError> {
160        let max_concurrent = self.max_concurrent_runs.unwrap_or(MAX_CONCURRENT_RUNS);
161        let state = Arc::new(WorkflowRunState {
162            authenticator: self
163                .authenticator
164                .ok_or(WorkflowRunBuilderError::MissingAuthenticator)?,
165            app: self.app.ok_or(WorkflowRunBuilderError::MissingApp)?,
166            registry: self
167                .registry
168                .ok_or(WorkflowRunBuilderError::MissingRegistry)?,
169            run_store: self
170                .run_store
171                .ok_or(WorkflowRunBuilderError::MissingRunStore)?,
172            provenance: self
173                .provenance
174                .ok_or(WorkflowRunBuilderError::MissingProvenance)?,
175            progress: progress::ProgressHub::default(),
176            permits: Arc::new(Semaphore::new(max_concurrent)),
177            run_timeout: self.run_timeout.unwrap_or(MAX_RUN_DURATION),
178        });
179
180        let submit_route = post(handlers::submit)
181            .layer(RequestBodyLimitLayer::new(MAX_DEF_BYTES))
182            .route_layer(middleware::from_fn_with_state(
183                Arc::clone(&state),
184                auth::require_run_scope,
185            ));
186        let read_layer =
187            middleware::from_fn_with_state(Arc::clone(&state), auth::require_read_scope);
188        let get_route = get(handlers::get_run).route_layer(read_layer.clone());
189        let events_route = get(events::get_run_events).route_layer(read_layer.clone());
190        let registry_route = get(handlers::get_registry).route_layer(read_layer);
191
192        Ok(Router::new()
193            .route("/health", get(handlers::health))
194            .route("/runs", submit_route)
195            .route("/runs/:run_id", get_route)
196            .route("/runs/:run_id/events", events_route)
197            .route("/registry", registry_route)
198            .with_state(state))
199    }
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205
206    #[test]
207    fn build_without_authenticator_returns_missing_authenticator() {
208        let err = WorkflowRunRouterBuilder::new().build().unwrap_err();
209        assert!(matches!(err, WorkflowRunBuilderError::MissingAuthenticator));
210    }
211}
klieo_workflow_api/lib.rs

klieo_workflow_api/
lib.rs
