//! SPIKE 3 — cancellation + bounded completion-order scheduler.
//!
//! Builds the parent→round→child token chain and a bounded (concurrency 8),
//! completion-order scheduler over `futures::stream::buffer_unordered`, and
//! asserts the cancel-gate invariant from `02_MODULE_CONVERSION_GUIDE.md` §1.5 /
//! risk A1:
//!
//! - A slow task that observes cancel resolves to a **typed** outcome
//!   (`is_error: true`), never an `Err` and never a panic.
//! - A mid-stream abort causes **no deadlock** — the scheduler stream drains and
//!   the test completes.
//!
//! The full production scheduler lives in `indusagi-runtime`; this spike proves
//! the core primitives (`CancellationToken` chain + `buffer_unordered`) compose
//! into the contract before that crate is built.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use futures::{StreamExt, stream};
use indusagi_core::cancel::{CancellationToken, child_token, root_token, round_token};

/// The typed outcome a runner produces — the cooperative contract's "tool
/// returns `is_error`, never `Err`" shape, in miniature.
#[derive(Debug, Clone, PartialEq)]
struct Outcome {
    id: String,
    is_error: bool,
    output: String,
}

/// One call's id + how long its work takes if not cancelled.
struct Call {
    id: String,
    work_ms: u64,
}

/// Default per-round concurrency, matching the runtime scheduler's `DEFAULT_CONCURRENCY`.
const DEFAULT_CONCURRENCY: usize = 8;

/// Run one call under a per-call child token chained to `round`. The cooperative
/// contract: if the token cancels first, return a TYPED `is_error` outcome — we
/// never `panic!`, never return `Result::Err`, never tear the round down by
/// surfacing an exception.
async fn run_one(call: Call, round: &CancellationToken) -> Outcome {
    let child = child_token(round);
    tokio::select! {
        biased;
        () = child.cancelled() => Outcome {
            id: call.id,
            is_error: true,
            output: "cancelled".to_string(),
        },
        () = tokio::time::sleep(Duration::from_millis(call.work_ms)) => Outcome {
            id: call.id,
            is_error: false,
            output: "ok".to_string(),
        },
    }
}

/// Drive a batch with bounded concurrency, delivering outcomes in **completion
/// order**. `buffer_unordered(limit)` gives the cap and completion-order
/// delivery for free — exactly the scheduler's contract.
async fn run_batch(calls: Vec<Call>, round: CancellationToken, limit: usize) -> Vec<Outcome> {
    stream::iter(calls)
        .map(|call| {
            let round = round.clone();
            async move { run_one(call, &round).await }
        })
        .buffer_unordered(limit.max(1))
        .collect()
        .await
}

#[tokio::test]
async fn slow_task_observing_cancel_resolves_to_typed_error_no_panic() {
    let parent = root_token();
    let round = round_token(&parent);

    // One fast call, one slow call. Abort the round mid-stream (after the fast
    // call would finish but well before the slow one).
    let calls = vec![
        Call {
            id: "fast".into(),
            work_ms: 5,
        },
        Call {
            id: "slow".into(),
            work_ms: 60_000,
        }, // would hang for a minute
    ];

    let round_for_abort = round.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(20)).await;
        // Cancel the PARENT — propagates down parent→round→child.
        // (Abort via the parent proves down-propagation through the full chain.)
        round_for_abort.cancel();
    });

    // No deadlock: this must complete. If buffer_unordered or the select! were
    // wrong, this would hang for 60s and the test-runner timeout would fail it.
    let outcomes = tokio::time::timeout(
        Duration::from_secs(5),
        run_batch(calls, round, DEFAULT_CONCURRENCY),
    )
    .await
    .expect("scheduler must not deadlock on mid-stream abort");

    assert_eq!(outcomes.len(), 2, "every call yields exactly one outcome");

    let slow = outcomes.iter().find(|o| o.id == "slow").unwrap();
    // The slow task OBSERVED cancel and returned a TYPED is_error outcome.
    assert!(
        slow.is_error,
        "cancelled task is a typed is_error outcome, not Err/panic"
    );
    assert_eq!(slow.output, "cancelled");

    let fast = outcomes.iter().find(|o| o.id == "fast").unwrap();
    assert!(
        !fast.is_error,
        "the fast task that finished before abort is a clean success"
    );
    assert_eq!(fast.output, "ok");
}

#[tokio::test]
async fn parent_abort_propagates_to_every_in_flight_child() {
    let parent = root_token();
    let round = round_token(&parent);

    // A batch larger than the concurrency cap, all slow. Aborting the PARENT
    // must cancel every in-flight child AND skip the queued ones — the whole
    // round tears down with no deadlock.
    let calls: Vec<Call> = (0..20)
        .map(|i| Call {
            id: format!("c{i}"),
            work_ms: 60_000,
        })
        .collect();

    let parent_for_abort = parent.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(15)).await;
        parent_for_abort.cancel();
    });

    let outcomes = tokio::time::timeout(
        Duration::from_secs(5),
        run_batch(calls, round, DEFAULT_CONCURRENCY),
    )
    .await
    .expect("aborting the parent must drain the whole round without deadlock");

    assert_eq!(outcomes.len(), 20);
    assert!(
        outcomes
            .iter()
            .all(|o| o.is_error && o.output == "cancelled"),
        "every call (in-flight or queued) resolves to a typed cancel outcome"
    );
}

#[tokio::test]
async fn no_abort_runs_all_to_completion_in_completion_order() {
    let parent = root_token();
    let round = round_token(&parent);

    // Staggered durations so completion order differs from request order.
    let calls = vec![
        Call {
            id: "a".into(),
            work_ms: 30,
        },
        Call {
            id: "b".into(),
            work_ms: 5,
        },
        Call {
            id: "c".into(),
            work_ms: 15,
        },
    ];

    let outcomes = run_batch(calls, round, DEFAULT_CONCURRENCY).await;

    assert_eq!(outcomes.len(), 3);
    assert!(outcomes.iter().all(|o| !o.is_error));
    // Completion order: b (5ms) then c (15ms) then a (30ms), NOT request order.
    let order: Vec<&str> = outcomes.iter().map(|o| o.id.as_str()).collect();
    assert_eq!(order, vec!["b", "c", "a"], "delivery is completion order");
}

#[tokio::test]
async fn concurrency_cap_is_respected() {
    // Prove at most `limit` runners execute simultaneously. Each runner flips a
    // shared "active" counter up on entry and down on exit; we assert the peak
    // never exceeds the cap.
    let limit = DEFAULT_CONCURRENCY;
    let active = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let peak = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let exceeded = Arc::new(AtomicBool::new(false));

    let calls: Vec<usize> = (0..40).collect();
    let active_c = active.clone();
    let peak_c = peak.clone();
    let exceeded_c = exceeded.clone();

    stream::iter(calls)
        .map(move |_| {
            let active = active_c.clone();
            let peak = peak_c.clone();
            let exceeded = exceeded_c.clone();
            async move {
                let now = active.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                if now > limit {
                    exceeded.store(true, Ordering::SeqCst);
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
                active.fetch_sub(1, Ordering::SeqCst);
            }
        })
        .buffer_unordered(limit)
        .collect::<Vec<()>>()
        .await;

    assert!(
        !exceeded.load(Ordering::SeqCst),
        "never exceeded the concurrency cap"
    );
    assert!(
        peak.load(Ordering::SeqCst) <= limit,
        "peak concurrency {} exceeded cap {}",
        peak.load(Ordering::SeqCst),
        limit
    );
    // And the cap was actually reached (the batch is far larger than the cap).
    assert_eq!(
        peak.load(Ordering::SeqCst),
        limit,
        "cap should be saturated"
    );
}
