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

klieo_workflow_api/
lib.rs
