klieo-workflow-api 3.8.2

//! `POST /runs` behaviour — including the crucible invariants: panic →
//! Failed, timeout → Aborted, concurrent duplicate → one run, cap → 429.

mod common;

use std::time::Duration;

use axum::body::Body;
use axum::http::{Request, StatusCode};
use common::*;
use klieo_workflow::test_support::const_subflow;
use klieo_workflow::Registry;
use klieo_workflow_api::RunStatus;
use tower::ServiceExt;

const GENEROUS_TIMEOUT: Duration = Duration::from_secs(5);
const HIGH_CAP: usize = 8;

fn echo_registry() -> Registry {
    Registry::new().with_subflow("echo", const_subflow("OK"))
}

#[tokio::test]
async fn valid_submit_returns_202_then_succeeds_with_server_stamped_author() {
    let router = build_router(echo_registry(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let response = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("echo"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::ACCEPTED);

    let run_id = run_id_of(response).await;
    let view = poll_until_terminal(router, &run_id).await;
    assert_eq!(view.status, RunStatus::Succeeded);
    assert_eq!(view.result, Some(serde_json::json!({ "tag": "OK" })));
    assert_eq!(
        view.author, TEST_TOKEN,
        "author is the authenticated identity"
    );
}

#[tokio::test]
async fn bad_definition_returns_400() {
    let router = build_router(Registry::new(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let response = router
        .oneshot(submit_request(
            single_subflow_def("ghost"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn non_object_input_returns_400() {
    let router = build_router(echo_registry(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let response = router
        .oneshot(submit_request(
            single_subflow_def("echo"),
            serde_json::json!([1, 2, 3]),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn missing_credential_returns_401() {
    let router = build_router(echo_registry(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let body = serde_json::json!({
        "definition": single_subflow_def("echo"), "input": {}
    });
    let request = Request::builder()
        .method("POST")
        .uri("/runs")
        .header("content-type", "application/json")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap();
    let response = router.oneshot(request).await.unwrap();
    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn duplicate_concurrent_submits_collapse_to_one_run() {
    let router = build_router(echo_registry(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let def = single_subflow_def("echo");
    let input = serde_json::json!({ "same": true });
    let (first, second) = tokio::join!(
        router
            .clone()
            .oneshot(submit_request(def.clone(), input.clone())),
        router
            .clone()
            .oneshot(submit_request(def.clone(), input.clone())),
    );
    let id_a = run_id_of(first.unwrap()).await;
    let id_b = run_id_of(second.unwrap()).await;
    assert_eq!(id_a, id_b, "identical concurrent submits share one run id");
}

#[tokio::test]
async fn reordered_input_keys_collapse_to_one_run() {
    let router = build_router(echo_registry(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let def = single_subflow_def("echo");
    // Same input, keys in different source order — canonical input hashing
    // must fold both to one idempotency key.
    let first = router
        .clone()
        .oneshot(submit_request(
            def.clone(),
            serde_json::json!({"a": 1, "b": 2}),
        ))
        .await
        .unwrap();
    let second = router
        .clone()
        .oneshot(submit_request(
            def.clone(),
            serde_json::json!({"b": 2, "a": 1}),
        ))
        .await
        .unwrap();
    assert_eq!(
        run_id_of(first).await,
        run_id_of(second).await,
        "key-order-only differences must share one run id"
    );
}

#[tokio::test]
async fn panicking_flow_ends_failed_not_stuck_running() {
    let registry = Registry::new().with_subflow("boom", panicking_subflow());
    let router = build_router(registry, GENEROUS_TIMEOUT, HIGH_CAP).await;
    let response = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("boom"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::ACCEPTED);

    let view = poll_until_terminal(router, &run_id_of(response).await).await;
    assert_eq!(view.status, RunStatus::Failed);
}

#[tokio::test]
async fn slow_flow_past_timeout_ends_aborted() {
    let registry = Registry::new().with_subflow("slow", slow_subflow(Duration::from_secs(30)));
    let router = build_router(registry, Duration::from_millis(50), HIGH_CAP).await;
    let response = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("slow"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    let view = poll_until_terminal(router, &run_id_of(response).await).await;
    assert_eq!(view.status, RunStatus::Aborted);
}

#[tokio::test]
async fn concurrency_cap_exhaustion_returns_429_with_retry_after() {
    let registry = Registry::new().with_subflow("slow", slow_subflow(Duration::from_secs(5)));
    let router = build_router(registry, Duration::from_secs(30), 1).await;

    let first = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("slow"),
            serde_json::json!({ "n": 1 }),
        ))
        .await
        .unwrap();
    assert_eq!(
        first.status(),
        StatusCode::ACCEPTED,
        "first run takes the only permit"
    );

    let second = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("slow"),
            serde_json::json!({ "n": 2 }),
        ))
        .await
        .unwrap();
    assert_eq!(second.status(), StatusCode::TOO_MANY_REQUESTS);
    assert!(second
        .headers()
        .contains_key(axum::http::header::RETRY_AFTER));
}