kova_api/

engine.rs

//! Engine-facing endpoints: pipeline state, event tail, and research runs.
//!
//! `kova-core` implements [`Engine`] and wires it in via
//! [`router_with_engine`]; the endpoints themselves stay transport-agnostic
//! and are testable against a fake engine.

use std::sync::Arc;

use async_trait::async_trait;
use axum::{
    extract::{Query, State},
    http::StatusCode,
    routing::{get, post},
    Json, Router,
};
use kova_primitives::{KovaError, PipelineEvent, PipelineState};
use serde::Deserialize;
use serde_json::{json, Value};

/// What the engine exposes over the API. Implemented by `kova-core`.
#[async_trait]
pub trait Engine: Send + Sync {
    /// The current pipeline state row.
    fn state(&self) -> PipelineState;

    /// The last `n` pipeline events, oldest first.
    fn recent_events(&self, n: usize) -> Vec<PipelineEvent>;

    /// Run one Research task to completion and return a JSON summary.
    async fn run_research(&self, task: &str) -> Result<Value, String>;

    /// Advance the pipeline to the next phase, or error when the current status
    /// is not `ReadyToAdvance` or the phase is already final.
    fn advance(&self) -> Result<PipelineState, String>;
}

/// Default window for `GET /pipeline/events`.
const DEFAULT_EVENT_WINDOW: usize = 100;
/// Maximum window a client can request.
const MAX_EVENT_WINDOW: usize = 1000;

#[derive(Debug, Deserialize)]
struct EventsQuery {
    #[serde(default)]
    n: Option<usize>,
}

#[derive(Debug, Deserialize)]
struct ResearchRequest {
    task: String,
}

async fn state(State(engine): State<Arc<dyn Engine>>) -> Json<PipelineState> {
    Json(engine.state())
}

async fn events(
    State(engine): State<Arc<dyn Engine>>,
    Query(query): Query<EventsQuery>,
) -> Json<Vec<PipelineEvent>> {
    let n = query
        .n
        .unwrap_or(DEFAULT_EVENT_WINDOW)
        .min(MAX_EVENT_WINDOW);
    Json(engine.recent_events(n))
}

async fn research(
    State(engine): State<Arc<dyn Engine>>,
    Json(request): Json<ResearchRequest>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
    if request.task.trim().is_empty() {
        return Err((
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": "task must not be empty" })),
        ));
    }
    match engine.run_research(&request.task).await {
        Ok(value) => Ok(Json(value)),
        Err(message) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({ "error": message })),
        )),
    }
}

async fn advance(
    State(engine): State<Arc<dyn Engine>>,
) -> Result<Json<PipelineState>, (StatusCode, Json<Value>)> {
    match engine.advance() {
        Ok(state) => Ok(Json(state)),
        Err(message) => Err((StatusCode::CONFLICT, Json(json!({ "error": message })))),
    }
}

/// [`crate::router`] plus the `/pipeline/*` engine endpoints.
pub fn router_with_engine(engine: Arc<dyn Engine>) -> Router {
    Router::new()
        .route("/pipeline/state", get(state))
        .route("/pipeline/events", get(events))
        .route("/pipeline/research", post(research))
        .route("/pipeline/advance", post(advance))
        .with_state(engine)
        .merge(crate::router())
}

/// Bind `addr` and serve `app` until the process exits.
pub async fn serve(addr: String, app: Router) -> Result<(), KovaError> {
    let listener = tokio::net::TcpListener::bind(&addr)
        .await
        .map_err(|e| KovaError::Config(format!("bind {addr}: {e}")))?;
    tracing::info!(addr = %addr, "engine api listening");
    axum::serve(listener, app)
        .await
        .map_err(|e| KovaError::Io(std::io::Error::other(e.to_string())))
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::{to_bytes, Body};
    use axum::http::{header, Method, Request};
    use kova_primitives::{Phase, PhaseStatus};
    use tower::ServiceExt;

    struct FakeEngine;

    #[async_trait]
    impl Engine for FakeEngine {
        fn state(&self) -> PipelineState {
            PipelineState::initial()
        }

        fn recent_events(&self, n: usize) -> Vec<PipelineEvent> {
            (0..n)
                .map(|i| PipelineEvent::Log {
                    level: "info".into(),
                    message: format!("event {i}"),
                })
                .collect()
        }

        async fn run_research(&self, task: &str) -> Result<Value, String> {
            Ok(json!({
                "status": "complete",
                "task": task,
                "summary": "canned summary",
            }))
        }

        fn advance(&self) -> Result<PipelineState, String> {
            Ok(PipelineState {
                phase: Phase::Verification,
                status: PhaseStatus::Idle,
                plan: None,
            })
        }
    }

    fn app() -> Router {
        router_with_engine(Arc::new(FakeEngine))
    }

    async fn get_response(uri: &str) -> axum::http::Response<Body> {
        app()
            .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
            .await
            .unwrap()
    }

    async fn post_json(uri: &str, body: &str) -> axum::http::Response<Body> {
        app()
            .oneshot(
                Request::builder()
                    .method(Method::POST)
                    .uri(uri)
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(body.to_string()))
                    .unwrap(),
            )
            .await
            .unwrap()
    }

    async fn json_body(response: axum::http::Response<Body>) -> Value {
        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    #[tokio::test]
    async fn pipeline_state_reports_research_phase() {
        let response = get_response("/pipeline/state").await;
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(json_body(response).await["phase"], "research");
    }

    #[tokio::test]
    async fn pipeline_events_returns_requested_window() {
        let response = get_response("/pipeline/events?n=5").await;
        assert_eq!(response.status(), StatusCode::OK);
        let events = json_body(response).await;
        let list = events
            .as_array()
            .expect("events endpoint should return a JSON array");
        assert_eq!(list.len(), 5);
    }

    #[tokio::test]
    async fn research_rejects_blank_task() {
        let response = post_json("/pipeline/research", r#"{"task":"  "}"#).await;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        assert_eq!(json_body(response).await["error"], "task must not be empty");
    }

    #[tokio::test]
    async fn research_returns_canned_summary() {
        let response = post_json("/pipeline/research", r#"{"task":"probe"}"#).await;
        assert_eq!(response.status(), StatusCode::OK);
        let body = json_body(response).await;
        assert_eq!(body["task"], "probe");
        assert_eq!(body["summary"], "canned summary");
    }

    #[tokio::test]
    async fn pipeline_advance_returns_next_state() {
        let response = post_json("/pipeline/advance", "").await;
        assert_eq!(response.status(), StatusCode::OK);
        let body = json_body(response).await;
        assert_eq!(body["phase"], "verification");
        assert_eq!(body["status"], "idle");
    }

    #[tokio::test]
    async fn merged_router_serves_base_health() {
        let response = get_response("/health").await;
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(json_body(response).await["status"], "ok");
    }
}