klieo-ops-api 3.3.0

use axum::{
    extract::{Query, State},
    Json,
};
use klieo_a2a::types::Task;
use serde::Deserialize;
use std::sync::Arc;

use crate::error::OpsError;
use crate::pagination::clamp_limit;
use crate::state::OpsState;

#[derive(Debug, Deserialize)]
pub(crate) struct ListTasksParams {
    pub(crate) context_id: Option<String>,
    pub(crate) limit: Option<usize>,
    pub(crate) offset: Option<usize>,
}

/// The store list is already context-scoped (bounded per context), so a
/// handler-level slice is a complete bound on the response — no trait change
/// is needed to keep a single context's task page from growing unboundedly.
pub(crate) async fn list_tasks(
    State(state): State<Arc<OpsState>>,
    Query(params): Query<ListTasksParams>,
) -> Result<Json<Vec<Task>>, OpsError> {
    let store = state
        .task_store
        .as_ref()
        .ok_or(OpsError::StoreNotConfigured)?;

    let context_id = params.context_id.ok_or(OpsError::ContextIdRequired)?;

    let tasks = store
        .list(Some(&context_id))
        .await
        .map_err(|e| {
            tracing::error!(operation = "list_tasks", context_id = %context_id, detail = %e, "store error");
            OpsError::Internal(Box::new(e))
        })?;

    let offset = params.offset.unwrap_or(0);
    let limit = clamp_limit(params.limit);
    let page = tasks.into_iter().skip(offset).take(limit).collect();

    Ok(Json(page))
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::{
        body::Body,
        http::{Request, StatusCode},
        routing::get,
        Router,
    };
    use klieo_a2a::task_store::{A2aTaskStore, DEFAULT_BUCKET};
    use klieo_auth_common::AllowAnonymous;
    use klieo_bus_memory::MemoryBus;
    use std::sync::Arc;
    use tower::ServiceExt;

    fn no_task_store_state() -> Arc<OpsState> {
        Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: None,
            resume_buffer: None,
        })
    }

    fn task_router(state: Arc<OpsState>) -> Router {
        Router::new()
            .route("/tasks", get(list_tasks))
            .with_state(state)
    }

    #[tokio::test]
    async fn list_tasks_no_store_returns_501() {
        let app = task_router(no_task_store_state());
        let req = Request::builder()
            .uri("/tasks?context_id=ctx-1")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED);
    }

    #[tokio::test]
    async fn list_tasks_missing_context_id_returns_400() {
        let bus = MemoryBus::new();
        let store = Arc::new(A2aTaskStore::new(bus.kv.clone(), DEFAULT_BUCKET.into()));
        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: Some(store),
            resume_buffer: None,
        });
        let app = task_router(state);
        let req = Request::builder()
            .uri("/tasks")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn list_tasks_returns_seeded_task() {
        use klieo_a2a::types::{Task, TaskStatus};

        let bus = MemoryBus::new();
        let store = Arc::new(A2aTaskStore::new(bus.kv.clone(), DEFAULT_BUCKET.into()));

        let task = Task {
            id: "task-001".to_string(),
            contextId: "ctx-test".to_string(),
            status: TaskStatus::Completed,
            artifacts: vec![],
            history: vec![],
            metadata: None,
        };
        store.put(&task).await.unwrap();

        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: Some(store),
            resume_buffer: None,
        });
        let app = task_router(state);

        let req = Request::builder()
            .uri("/tasks?context_id=ctx-test")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let tasks: Vec<serde_json::Value> = serde_json::from_slice(&body).unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0]["id"], "task-001");
    }

    #[tokio::test]
    async fn list_tasks_returns_only_the_requested_page() {
        use klieo_a2a::types::{Task, TaskStatus};

        let bus = MemoryBus::new();
        let store = Arc::new(A2aTaskStore::new(bus.kv.clone(), DEFAULT_BUCKET.into()));
        for index in 0..5 {
            let task = Task {
                id: format!("task-{index}"),
                contextId: "ctx-page".to_string(),
                status: TaskStatus::Completed,
                artifacts: vec![],
                history: vec![],
                metadata: None,
            };
            store.put(&task).await.unwrap();
        }

        let state = Arc::new(OpsState {
            authenticator: Arc::new(AllowAnonymous),
            run_log_store: None,
            task_store: Some(store),
            resume_buffer: None,
        });
        let app = task_router(state);
        let req = Request::builder()
            .uri("/tasks?context_id=ctx-page&limit=2&offset=2")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let tasks: Vec<serde_json::Value> = serde_json::from_slice(&body).unwrap();
        assert_eq!(tasks.len(), 2, "limit caps the page to two tasks");
        let ids: Vec<&str> = tasks.iter().map(|t| t["id"].as_str().unwrap()).collect();
        assert_eq!(ids, ["task-2", "task-3"], "offset selects the 2nd page");
        // Negative fixtures: items outside the page must be absent.
        assert!(!ids.contains(&"task-1"), "before-page item must not appear");
        assert!(!ids.contains(&"task-4"), "after-page item must not appear");
    }
}