klieo-workflow-api 3.6.0

#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]
//! `klieo-workflow-api` — embeddable Axum run-service fronting
//! [`klieo_workflow`].
//!
//! A caller submits a `WorkflowDef` to `POST /runs`; the service compiles it
//! against a [`klieo_workflow::Registry`] allow-list, runs it asynchronously,
//! and returns a run id to poll at `GET /runs/{id}`. Auth-gated, with a
//! server-stamped author and a bounded, supervised executor. `GET /health`
//! is an unauthenticated liveness probe, for an orchestrator that has no
//! credential to present.
//!
//! Mount the router produced by [`WorkflowRunRouterBuilder::build`].
//!
//! # Deploying
//!
//! This crate — not `klieo-server` — is the deployable run-service surface:
//! `klieo-server` is a `publish = false` reference demo, not meant to serve
//! real traffic. For container/orchestration topology (Docker Compose,
//! Helm, Terraform) fronting a service built on this router, use the
//! `klieo-compliance` repo's `deploy/` tree as the reference: swap its
//! compliance-server container image for a binary that mounts
//! [`WorkflowRunRouterBuilder::build`]'s router; the surrounding topology
//! (reverse proxy, secrets) carries over. klieo is pre-production —
//! `deploy/` does not itself prescribe autoscaling or canary rollout; add
//! those only if your deployment needs them.

mod auth;
mod error;
mod events;
mod executor;
mod handlers;
mod progress;
mod provenance;
mod run_store;
mod state;
mod sweep;

pub use auth::{WORKFLOW_READ_ALL_SCOPE, WORKFLOW_READ_SCOPE, WORKFLOW_RUN_SCOPE};
pub use error::{ErrorCode, WorkflowApiError};
pub use run_store::{ClaimOutcome, RunRecordView, RunStatus, RunStore, StoreError, RUNS_BUCKET};
pub use sweep::{sweep_orphaned_runs, SweepError, SweepSummary};

use crate::state::WorkflowRunState;
use axum::{
    middleware,
    routing::{get, post},
    Router,
};
use klieo::App;
use klieo_auth_common::Authenticator;
use klieo_provenance::ProvenanceRepository;
use klieo_workflow::Registry;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tower_http::limit::RequestBodyLimitLayer;

/// Maximum number of runs executing at once; excess submits get `429`.
pub const MAX_CONCURRENT_RUNS: usize = 32;

/// Wall-clock cap on a single run; exceeding it aborts the run.
pub const MAX_RUN_DURATION: Duration = Duration::from_secs(300);

/// Maximum accepted `POST /runs` body size, bounding the definition + input.
pub const MAX_DEF_BYTES: usize = 256 * 1024;

/// Why [`WorkflowRunRouterBuilder::build`] could not assemble a router.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum WorkflowRunBuilderError {
    /// [`WorkflowRunRouterBuilder::with_authenticator`] was not called.
    #[error("authenticator not configured — call with_authenticator() before build()")]
    MissingAuthenticator,
    /// [`WorkflowRunRouterBuilder::with_app`] was not called.
    #[error("app not configured — call with_app() before build()")]
    MissingApp,
    /// [`WorkflowRunRouterBuilder::with_registry`] was not called.
    #[error("registry not configured — call with_registry() before build()")]
    MissingRegistry,
    /// [`WorkflowRunRouterBuilder::with_run_store`] was not called.
    #[error("run store not configured — call with_run_store() before build()")]
    MissingRunStore,
    /// [`WorkflowRunRouterBuilder::with_provenance`] was not called. Audit is
    /// mandatory — the service refuses to run without a durable chain.
    #[error("provenance repository not configured — call with_provenance() before build()")]
    MissingProvenance,
}

/// Builds the run-service router (`POST /runs`, `GET /runs/{id}`).
#[derive(Default)]
pub struct WorkflowRunRouterBuilder {
    authenticator: Option<Arc<dyn Authenticator>>,
    app: Option<Arc<App>>,
    registry: Option<Registry>,
    run_store: Option<RunStore>,
    provenance: Option<Arc<dyn ProvenanceRepository>>,
    run_timeout: Option<Duration>,
    max_concurrent_runs: Option<usize>,
}

impl WorkflowRunRouterBuilder {
    /// Start an empty builder.
    pub fn new() -> Self {
        Self::default()
    }

    /// Required — the credential verifier for every route.
    pub fn with_authenticator(mut self, authenticator: Arc<dyn Authenticator>) -> Self {
        self.authenticator = Some(authenticator);
        self
    }

    /// Required — the App that mints an `AgentContext` per run. Its LLM is
    /// unused (each agent node swaps to its registry model); it supplies the
    /// bus, kv, tools, and memory a run needs.
    pub fn with_app(mut self, app: Arc<App>) -> Self {
        self.app = Some(app);
        self
    }

    /// Required — the allow-list a submitted definition compiles against.
    pub fn with_registry(mut self, registry: Registry) -> Self {
        self.registry = Some(registry);
        self
    }

    /// Required — the run-status poll cache.
    pub fn with_run_store(mut self, run_store: RunStore) -> Self {
        self.run_store = Some(run_store);
        self
    }

    /// Required — the durable, authoritative audit chain. Audit is mandatory;
    /// [`Self::build`] errors without it. Supply a durable
    /// `SqliteProvenanceRepository::open(path)` in production.
    pub fn with_provenance(mut self, provenance: Arc<dyn ProvenanceRepository>) -> Self {
        self.provenance = Some(provenance);
        self
    }

    /// Override the per-run wall-clock cap. Defaults to [`MAX_RUN_DURATION`].
    pub fn with_run_timeout(mut self, timeout: Duration) -> Self {
        self.run_timeout = Some(timeout);
        self
    }

    /// Override the concurrency cap. Defaults to [`MAX_CONCURRENT_RUNS`].
    pub fn with_max_concurrent_runs(mut self, max: usize) -> Self {
        self.max_concurrent_runs = Some(max);
        self
    }

    /// Assemble the router.
    ///
    /// # Errors
    /// Returns a [`WorkflowRunBuilderError`] variant naming the first
    /// required port that was not configured.
    pub fn build(self) -> Result<Router, WorkflowRunBuilderError> {
        let max_concurrent = self.max_concurrent_runs.unwrap_or(MAX_CONCURRENT_RUNS);
        let state = Arc::new(WorkflowRunState {
            authenticator: self
                .authenticator
                .ok_or(WorkflowRunBuilderError::MissingAuthenticator)?,
            app: self.app.ok_or(WorkflowRunBuilderError::MissingApp)?,
            registry: self
                .registry
                .ok_or(WorkflowRunBuilderError::MissingRegistry)?,
            run_store: self
                .run_store
                .ok_or(WorkflowRunBuilderError::MissingRunStore)?,
            provenance: self
                .provenance
                .ok_or(WorkflowRunBuilderError::MissingProvenance)?,
            progress: progress::ProgressHub::default(),
            permits: Arc::new(Semaphore::new(max_concurrent)),
            run_timeout: self.run_timeout.unwrap_or(MAX_RUN_DURATION),
        });

        let submit_route = post(handlers::submit)
            .layer(RequestBodyLimitLayer::new(MAX_DEF_BYTES))
            .route_layer(middleware::from_fn_with_state(
                Arc::clone(&state),
                auth::require_run_scope,
            ));
        let read_layer =
            middleware::from_fn_with_state(Arc::clone(&state), auth::require_read_scope);
        let get_route = get(handlers::get_run).route_layer(read_layer.clone());
        let events_route = get(events::get_run_events).route_layer(read_layer.clone());
        let registry_route = get(handlers::get_registry).route_layer(read_layer);

        Ok(Router::new()
            .route("/health", get(handlers::health))
            .route("/runs", submit_route)
            .route("/runs/:run_id", get_route)
            .route("/runs/:run_id/events", events_route)
            .route("/registry", registry_route)
            .with_state(state))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn build_without_authenticator_returns_missing_authenticator() {
        let err = WorkflowRunRouterBuilder::new().build().unwrap_err();
        assert!(matches!(err, WorkflowRunBuilderError::MissingAuthenticator));
    }
}