fabro/lib/crates/fabro-server/tests/it/api/system.rs

#![expect(
    clippy::disallowed_methods,
    reason = "integration tests stage fixtures with sync std::fs; test infrastructure, not Tokio-hot path"
)]

use std::path::PathBuf;

use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_config::Storage;
use fabro_types::RunId;
use fabro_types::settings::interp::InterpString;
use tempfile::tempdir;
use tower::ServiceExt;

use crate::helpers::{
    MINIMAL_DOT, POLL_ATTEMPTS, POLL_INTERVAL, TestAppSettings, api, checked_response,
    minimal_manifest_json, minimal_manifest_json_with_dry_run, response_json, response_status,
    test_app_state_with_options, test_app_with_scheduler, test_settings, wait_for_run_status,
};

const HUMAN_GATE_DOT: &str = r#"digraph GateTest {
    graph [goal="Test gate"]
    start [shape=Mdiamond]
    exit  [shape=Msquare]
    work  [shape=box, prompt="Do work"]
    gate  [shape=hexagon, type="human", label="Approve?"]
    done  [shape=box, prompt="Finish"]
    revise [shape=box, prompt="Revise"]

    start -> work -> gate
    gate -> done   [label="[A] Approve"]
    gate -> revise [label="[R] Revise"]
    done -> exit
    revise -> gate
}"#;

fn temp_storage_settings() -> (tempfile::TempDir, TestAppSettings, PathBuf) {
    let temp = tempdir().expect("tempdir should create");
    let mut settings = test_settings();
    let storage_dir = temp.path().join("storage");
    settings.server_settings.server.storage.root =
        InterpString::parse(&storage_dir.to_string_lossy());
    (temp, settings, storage_dir)
}

async fn create_run(app: &axum::Router, manifest: serde_json::Value) -> String {
    let request = Request::builder()
        .method("POST")
        .uri(api("/runs"))
        .header("content-type", "application/json")
        .body(Body::from(
            serde_json::to_vec(&manifest).expect("manifest fixture should serialize"),
        ))
        .expect("create-run request should build");
    let response = app.clone().oneshot(request).await.unwrap();
    let body = response_json(response, StatusCode::CREATED, "POST /api/v1/runs").await;
    body["id"]
        .as_str()
        .expect("create-run response should include an id")
        .to_string()
}

async fn start_run(app: &axum::Router, run_id: &str) {
    let request = Request::builder()
        .method("POST")
        .uri(api(&format!("/runs/{run_id}/start")))
        .body(Body::empty())
        .expect("start-run request should build");
    let response = app.clone().oneshot(request).await.unwrap();
    response_status(
        response,
        StatusCode::OK,
        format!("POST /api/v1/runs/{run_id}/start"),
    )
    .await;
}

async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Value {
    for _ in 0..POLL_ATTEMPTS {
        let request = Request::builder()
            .method("GET")
            .uri(api(&format!("/runs/{run_id}/questions")))
            .body(Body::empty())
            .expect("questions request should build");
        let response = app.clone().oneshot(request).await.unwrap();
        let body = response_json(
            response,
            StatusCode::OK,
            format!("GET /api/v1/runs/{run_id}/questions"),
        )
        .await;
        if let Some(question) = body["data"].as_array().and_then(|items| items.first()) {
            return question.clone();
        }
        tokio::time::sleep(POLL_INTERVAL).await;
    }
    panic!("question should have appeared for {run_id}");
}

async fn load_questions(app: &axum::Router, run_id: &str) -> serde_json::Value {
    let request = Request::builder()
        .method("GET")
        .uri(api(&format!("/runs/{run_id}/questions")))
        .body(Body::empty())
        .expect("questions request should build");
    let response = app.clone().oneshot(request).await.unwrap();
    response_json(
        response,
        StatusCode::OK,
        format!("GET /api/v1/runs/{run_id}/questions"),
    )
    .await
}

#[tokio::test]
async fn get_system_info_returns_runtime_fields() {
    let (_temp, settings, expected_storage_dir) = temp_storage_settings();
    let configured_server_url = settings.server_settings.server.web.url.as_source();
    let app =
        fabro_server::test_support::build_test_router(test_app_state_with_options(settings, 5));

    let request = Request::builder()
        .method("GET")
        .uri(api("/system/info"))
        .body(Body::empty())
        .unwrap();
    let response = app.oneshot(request).await.unwrap();

    let body = response_json(response, StatusCode::OK, "GET /api/v1/system/info").await;
    assert!(body["version"].as_str().is_some());
    assert_eq!(body["server_url"], configured_server_url);
    assert_eq!(body["storage_engine"], "slatedb");
    assert_eq!(
        body["storage_dir"],
        expected_storage_dir.display().to_string()
    );
    assert_eq!(body["runs"]["total"], 0);
    assert_eq!(body["runs"]["active"], 0);
    assert!(body["uptime_secs"].as_i64().is_some());
    assert!(
        body.get("features").is_none(),
        "system info should not include a features field"
    );
}

#[tokio::test]
async fn get_system_resources_returns_server_visible_metrics() {
    let (_temp, settings, expected_storage_dir) = temp_storage_settings();
    let app =
        fabro_server::test_support::build_test_router(test_app_state_with_options(settings, 5));

    let request = Request::builder()
        .method("GET")
        .uri(api("/system/resources"))
        .body(Body::empty())
        .unwrap();
    let response = app.oneshot(request).await.unwrap();

    let body = response_json(response, StatusCode::OK, "GET /api/v1/system/resources").await;
    let sampled_at = body["sampled_at"]
        .as_str()
        .expect("sampled_at should be an RFC3339 timestamp");
    chrono::DateTime::parse_from_rfc3339(sampled_at).expect("sampled_at should parse as RFC3339");

    assert_eq!(body["cpu"]["supported"], true);
    assert_eq!(body["cpu"]["scope"], "server_environment");
    assert!(
        body["cpu"]["sample_window_ms"].is_null()
            || body["cpu"]["sample_window_ms"]
                .as_i64()
                .is_some_and(|value| value >= 0),
        "sample window should be null until a delta sample is available or nonnegative: {body}"
    );
    assert!(
        body["cpu"]["logical_cpus"].as_i64().unwrap_or_default() > 0,
        "logical CPU count should be positive"
    );
    assert_percent_if_present(&body["cpu"]["usage_percent"]);

    assert_eq!(body["memory"]["supported"], true);
    assert!(matches!(
        body["memory"]["scope"].as_str(),
        Some("host" | "cgroup")
    ));
    assert_nonnegative_i64(&body["memory"]["total_bytes"], "memory.total_bytes");
    assert_nonnegative_i64(&body["memory"]["used_bytes"], "memory.used_bytes");
    assert_nonnegative_i64(&body["memory"]["available_bytes"], "memory.available_bytes");
    assert_percent_if_present(&body["memory"]["used_percent"]);

    assert_eq!(body["disk"]["supported"], true);
    assert_eq!(body["disk"]["scope"], "storage_filesystem");
    assert_eq!(
        body["disk"]["storage_path"],
        expected_storage_dir.display().to_string()
    );
    assert!(body["disk"]["mount_point"].as_str().is_some());
    assert_nonnegative_i64(&body["disk"]["total_bytes"], "disk.total_bytes");
    assert_nonnegative_i64(&body["disk"]["used_bytes"], "disk.used_bytes");
    assert_nonnegative_i64(&body["disk"]["available_bytes"], "disk.available_bytes");
    assert_percent_if_present(&body["disk"]["used_percent"]);
    assert_nonnegative_i64(
        &body["disk"]["fabro_managed_bytes"],
        "disk.fabro_managed_bytes",
    );
    assert_nonnegative_i64(
        &body["disk"]["fabro_reclaimable_bytes"],
        "disk.fabro_reclaimable_bytes",
    );
    assert!(
        body["disk"]["fabro_managed_bytes"]
            .as_i64()
            .unwrap_or_default()
            >= body["disk"]["fabro_reclaimable_bytes"]
                .as_i64()
                .unwrap_or_default(),
        "managed bytes must include everything reclaimable: {body}"
    );
    assert!(
        body["notes"]
            .as_array()
            .is_some_and(std::vec::Vec::is_empty),
        "initial test app resources should not need notes: {body}"
    );
}

fn assert_nonnegative_i64(value: &serde_json::Value, name: &str) {
    assert!(
        value.as_i64().is_some_and(|value| value >= 0),
        "{name} should be a nonnegative integer: {value:?}"
    );
}

fn assert_percent_if_present(value: &serde_json::Value) {
    if value.is_null() {
        return;
    }
    let percent = value
        .as_f64()
        .expect("percent should be numeric when present");
    assert!(
        (0.0..=100.0).contains(&percent),
        "percent should be between 0 and 100: {percent}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_app_state_with_options_respects_max_concurrent_runs() {
    let app = test_app_with_scheduler(test_app_state_with_options(test_settings(), 1));

    let first_run = create_run(&app, minimal_manifest_json(HUMAN_GATE_DOT)).await;
    let second_run = create_run(&app, minimal_manifest_json(HUMAN_GATE_DOT)).await;

    start_run(&app, &first_run).await;
    start_run(&app, &second_run).await;

    let question = wait_for_question(&app, &first_run).await;
    assert_eq!(question["stage"], "gate");

    tokio::time::sleep(POLL_INTERVAL * 5).await;

    let second_questions = load_questions(&app, &second_run).await;
    assert!(
        second_questions["data"]
            .as_array()
            .is_some_and(std::vec::Vec::is_empty),
        "second run should still be waiting for scheduler capacity while the first waits at the human gate: {second_questions}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn get_system_disk_usage_returns_summary_and_verbose_rows() {
    let (_temp, settings, storage_dir) = temp_storage_settings();
    let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));

    let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
    start_run(&app, &run_id).await;
    let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
    assert_eq!(status, "succeeded");

    let logs_dir = storage_dir.join("logs");
    std::fs::create_dir_all(&logs_dir).unwrap();
    std::fs::write(logs_dir.join("server.log"), b"log line\n").unwrap();

    let request = Request::builder()
        .method("GET")
        .uri(api("/system/df?verbose=true"))
        .body(Body::empty())
        .unwrap();
    let response = app.oneshot(request).await.unwrap();

    let body = response_json(
        response,
        StatusCode::OK,
        "GET /api/v1/system/df?verbose=true",
    )
    .await;
    let summary = body["summary"].as_array().expect("summary array");
    assert!(summary.iter().any(|row| row["type"] == "other"));

    let row_size = |type_: &str| {
        summary
            .iter()
            .find(|row| row["type"] == type_)
            .and_then(|row| row["size_bytes"].as_i64())
            .unwrap_or_default()
    };
    let total_size = body["total_size_bytes"].as_i64().unwrap_or_default();
    assert!(total_size > 0);
    assert_eq!(
        row_size("runs") + row_size("logs") + row_size("other"),
        total_size,
    );
    assert!(
        body["runs"]
            .as_array()
            .is_some_and(|runs| runs.iter().any(|entry| entry["run_id"] == run_id))
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn prune_runs_supports_dry_run_and_deletion() {
    let (_temp, settings, storage_dir) = temp_storage_settings();
    let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));

    let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
    start_run(&app, &run_id).await;
    let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
    assert_eq!(status, "succeeded");

    let run_id_parsed: RunId = run_id.parse().unwrap();
    let run_dir = Storage::new(&storage_dir)
        .run_scratch(&run_id_parsed)
        .root()
        .to_path_buf();
    assert!(run_dir.exists());

    let dry_run_request = Request::builder()
        .method("POST")
        .uri(api("/system/prune/runs"))
        .header("content-type", "application/json")
        .body(Body::from(r#"{"before":"9999"}"#))
        .unwrap();
    let dry_run_response = app.clone().oneshot(dry_run_request).await.unwrap();
    let dry_run_body = response_json(
        dry_run_response,
        StatusCode::OK,
        "POST /api/v1/system/prune/runs",
    )
    .await;
    assert_eq!(dry_run_body["dry_run"], true);
    assert_eq!(dry_run_body["total_count"], 1);
    assert_eq!(dry_run_body["runs"][0]["run_id"], run_id);
    assert!(run_dir.exists());

    let delete_request = Request::builder()
        .method("POST")
        .uri(api("/system/prune/runs"))
        .header("content-type", "application/json")
        .body(Body::from(r#"{"dry_run":false,"before":"9999"}"#))
        .unwrap();
    let delete_response = app.clone().oneshot(delete_request).await.unwrap();
    let delete_body = response_json(
        delete_response,
        StatusCode::OK,
        "POST /api/v1/system/prune/runs",
    )
    .await;
    assert_eq!(delete_body["dry_run"], false);
    assert_eq!(delete_body["deleted_count"], 1);
    assert!(!run_dir.exists());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attach_events_returns_sse_stream() {
    let (_temp, settings, _storage_dir) = temp_storage_settings();
    let app = test_app_with_scheduler(test_app_state_with_options(settings, 5));
    let run_id = RunId::new();

    let request = Request::builder()
        .method("GET")
        .uri(api(&format!("/attach?run_id={run_id}")))
        .body(Body::empty())
        .unwrap();
    let response = checked_response(
        app.clone().oneshot(request).await.unwrap(),
        StatusCode::OK,
        format!("GET /api/v1/attach?run_id={run_id}"),
    )
    .await;
    let content_type = response
        .headers()
        .get("content-type")
        .expect("content-type should be present")
        .to_str()
        .unwrap();
    assert!(content_type.contains("text/event-stream"));
}