//! Test scheduling and distribution across parallel sandboxes.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Mutex;
use std::time::Duration;

use crate::framework::TestInstance;

/// A scheduled batch of tests ready for execution.
///
/// Wraps a list of test instances. Workers pop these from the scheduler queue;
/// each pop re-queues the batch (split into halves for multi-test batches,
/// or cloned with an incremented retry counter for single-test batches) so
/// the batch can be retried.
#[derive(Clone)]
pub struct ScheduledBatch {
    pub tests: Vec<TestInstance>,
    /// Re-queue count for single-test batches, checked against MAX_SINGLE_TEST_REQUEUES.
    pub single_test_retry_counter: usize,
}

/// Maximum total length (in chars) of all test IDs in a single batch.
///
/// Prevents command lines from exceeding OS or shell limits. A single test
/// whose ID already exceeds this is still placed alone in its own batch.
const MAX_BATCH_COMMAND_LEN: usize = 30_000;

/// Maximum number of times a single-test batch can be re-queued before
/// the scheduler stops re-queuing it.
const MAX_SINGLE_TEST_REQUEUES: usize = 3;

/// A batch of tests being built by the scheduler.
///
/// Tracks the tests, their cumulative expected duration, total command length,
/// and which test IDs are present (to prevent scheduling the same test twice
/// in one batch).
struct Batch {
    tests: Vec<TestInstance>,
    load: Duration,
    command_len: usize,
    test_ids: HashSet<String>,
}

impl Batch {
    fn new() -> Self {
        Self {
            tests: Vec::new(),
            load: Duration::ZERO,
            command_len: 0,
            test_ids: HashSet::new(),
        }
    }

    fn add(&mut self, test: TestInstance, duration: Duration) {
        self.command_len += test.id().len();
        self.test_ids.insert(test.id().to_string());
        self.tests.push(test);
        self.load += duration;
    }

    fn contains(&self, test_id: &str) -> bool {
        self.test_ids.contains(test_id)
    }

    fn is_empty(&self) -> bool {
        self.tests.is_empty()
    }

    fn would_fit(&self, test_id_len: usize) -> bool {
        self.is_empty() || (self.command_len + test_id_len <= MAX_BATCH_COMMAND_LEN)
    }
}

/// Distributes tests across parallel sandboxes.
///
/// Performs LPT scheduling at construction time and exposes the resulting
/// batches through a mutex-protected queue. Workers call [`pop`](Self::pop)
/// to pull batches. Each pop re-queues the batch (split into halves for
/// multi-test batches, or cloned with an incremented retry counter for
/// single-test batches) so the batch can be retried if needed; the
/// `is_decided` check in the spawn loop skips batches whose tests already
/// completed.
pub struct Scheduler {
    queue: Mutex<VecDeque<ScheduledBatch>>,
    notify: tokio::sync::Notify,
    batch_count: usize,
    batch_sizes: Vec<usize>,
}

impl Scheduler {
    /// Creates a new scheduler and schedules the given tests.
    ///
    /// Uses Longest Processing Time First (LPT) algorithm with historical
    /// test durations to minimize total execution time (makespan). Tests are
    /// sorted by duration descending and assigned to the worker with the
    /// smallest current total workload.
    ///
    /// Batches are sorted by total duration descending, so the heaviest batch
    /// is first. This ensures it gets scheduled first with Modal.
    ///
    /// # Arguments
    ///
    /// * `max_parallel` - Maximum number of parallel batches/sandboxes.
    ///   Minimum is 1 (values below 1 are clamped).
    /// * `tests` - Tests to schedule
    /// * `durations` - Historical test durations from previous runs.
    ///   Tests not in the map use the per-group average from `group_to_default_duration`.
    /// * `group_to_default_duration` - Per-group average duration for tests without historical data.
    ///   Falls back to 1 second if the group has no entry.
    pub fn new(
        max_parallel: usize,
        tests: &[TestInstance],
        durations: &HashMap<String, Duration>,
        group_to_default_duration: &HashMap<String, Duration>,
    ) -> Self {
        let max_parallel = max_parallel.max(1);

        if tests.is_empty() {
            return Self {
                queue: Mutex::new(VecDeque::new()),
                notify: tokio::sync::Notify::new(),
                batch_count: 0,
                batch_sizes: Vec::new(),
            };
        }

        // Partition: individually-scheduled tests get their own batches, others go through LPT
        let (individual_tests, normal_tests): (Vec<_>, Vec<_>) =
            tests.iter().cloned().partition(|t| t.schedule_individual());

        // Build individual batches (one test per batch)
        let individual_batches: Vec<ScheduledBatch> = individual_tests
            .into_iter()
            .map(|t| ScheduledBatch {
                tests: vec![t],
                single_test_retry_counter: 0,
            })
            .collect();

        if normal_tests.is_empty() {
            let batch_count = individual_batches.len();
            let batch_sizes = individual_batches.iter().map(|b| b.tests.len()).collect();
            return Self {
                queue: Mutex::new(VecDeque::from(individual_batches)),
                notify: tokio::sync::Notify::new(),
                batch_count,
                batch_sizes,
            };
        }

        // Look up durations for each test, sorted longest-first
        let mut tests_with_duration: Vec<_> = normal_tests
            .iter()
            .map(|t| {
                let duration = match durations.get(t.id()) {
                    Some(&d) => d,
                    None => {
                        let fallback = group_to_default_duration
                            .get(t.group())
                            .copied()
                            .unwrap_or(Duration::from_secs(1));
                        tracing::debug!(
                            "No historical duration for test '{}', using group '{}' default {:?}",
                            t.id(),
                            t.group(),
                            fallback,
                        );
                        fallback
                    }
                };
                (t.clone(), duration)
            })
            .collect();
        tests_with_duration.sort_by_key(|b| std::cmp::Reverse(b.1));

        // Initialize batches
        let num_batches = max_parallel.min(normal_tests.len());
        let mut batches: Vec<Batch> = (0..num_batches).map(|_| Batch::new()).collect();

        // LPT assignment: assign each test to the lightest eligible batch
        for (test, duration) in tests_with_duration {
            let test_id = test.id();

            let target_idx = (0..batches.len())
                .filter(|&i| !batches[i].contains(test_id) && batches[i].would_fit(test_id.len()))
                .min_by_key(|&i| batches[i].load);

            let idx = target_idx.unwrap_or_else(|| {
                batches.push(Batch::new());
                batches.len() - 1
            });

            batches[idx].add(test, duration);
        }

        // Sort by load descending (heaviest first) for optimal Modal scheduling
        batches.sort_by_key(|b| std::cmp::Reverse(b.load));

        // Prepend individually-scheduled batches before LPT batches
        let mut result = individual_batches;
        result.extend(
            batches
                .into_iter()
                .filter(|b| !b.is_empty())
                .map(|b| ScheduledBatch {
                    tests: b.tests,
                    single_test_retry_counter: 0,
                }),
        );

        let batch_count = result.len();
        let batch_sizes = result.iter().map(|b| b.tests.len()).collect();
        Self {
            queue: Mutex::new(VecDeque::from(result)),
            notify: tokio::sync::Notify::new(),
            batch_count,
            batch_sizes,
        }
    }

    /// Removes and returns the next batch from the queue, blocking if empty.
    ///
    /// Multi-test batches are split into two halves when re-queued; single-test
    /// batches are re-queued with an incremented retry counter up to
    /// `MAX_SINGLE_TEST_REQUEUES` times. The `is_decided` check in the spawn
    /// loop skips batches whose tests already completed.
    ///
    /// Returns `None` if the mutex is poisoned.
    ///
    /// This future is cancel-safe: callers should race it against a
    /// cancellation token via `tokio::select!`.
    pub async fn pop(&self) -> Option<ScheduledBatch> {
        loop {
            // Register notified BEFORE checking the queue to avoid missed wakeups
            let notified = self.notify.notified();

            match self.queue.lock() {
                Ok(mut q) => {
                    if let Some(batch) = q.pop_front() {
                        drop(q);
                        if batch.tests.len() > 1 {
                            let mid = batch.tests.len() / 2;
                            self.push(ScheduledBatch {
                                tests: batch.tests[..mid].to_vec(),
                                single_test_retry_counter: 0,
                            });
                            self.push(ScheduledBatch {
                                tests: batch.tests[mid..].to_vec(),
                                single_test_retry_counter: 0,
                            });
                        } else if batch.single_test_retry_counter < MAX_SINGLE_TEST_REQUEUES {
                            self.push(ScheduledBatch {
                                tests: batch.tests.clone(),
                                single_test_retry_counter: batch.single_test_retry_counter + 1,
                            });
                        }
                        return Some(batch);
                    }
                }
                Err(e) => {
                    tracing::error!("batch queue mutex poisoned: {}", e);
                    return None;
                }
            }

            // Queue is empty — wait for a notification
            notified.await;
        }
    }

    /// Pushes a batch to the back of the queue and wakes one waiting worker.
    fn push(&self, batch: ScheduledBatch) {
        match self.queue.lock() {
            Ok(mut q) => {
                q.push_back(batch);
                self.notify.notify_one();
            }
            Err(e) => {
                tracing::error!("batch queue mutex poisoned on push: {}", e);
            }
        }
    }

    /// Number of batches created during scheduling.
    pub fn batch_count(&self) -> usize {
        self.batch_count
    }

    /// Number of tests in each batch, in schedule order.
    pub fn batch_sizes(&self) -> &[usize] {
        &self.batch_sizes
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::framework::TestRecord;

    /// Pops exactly `scheduler.batch_count()` batches.
    async fn drain_batches(scheduler: &Scheduler) -> Vec<Vec<TestInstance>> {
        let n = scheduler.batch_count();
        let mut batches = Vec::with_capacity(n);
        for _ in 0..n {
            if let Some(batch) = scheduler.pop().await {
                batches.push(batch.tests);
            }
        }
        batches
    }

    #[test]
    fn test_schedule_empty() {
        let scheduler = Scheduler::new(4, &[], &HashMap::new(), &HashMap::new());
        assert_eq!(scheduler.batch_count(), 0);
    }

    #[tokio::test]
    async fn test_schedule_balances_load() {
        let records = [
            TestRecord::new("slow_test", "test-group"),
            TestRecord::new("medium_test", "test-group"),
            TestRecord::new("fast_test", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let mut durations = HashMap::new();
        durations.insert("slow_test".to_string(), Duration::from_secs(10));
        durations.insert("medium_test".to_string(), Duration::from_secs(5));
        durations.insert("fast_test".to_string(), Duration::from_secs(1));

        let scheduler = Scheduler::new(2, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // With LPT:
        // 1. Assign slow_test (10s) to worker 0 -> loads: [10, 0]
        // 2. Assign medium_test (5s) to worker 1 -> loads: [10, 5]
        // 3. Assign fast_test (1s) to worker 1 -> loads: [10, 6]
        // Batches sorted by load: batch 0 (10s), batch 1 (6s)
        assert_eq!(batches.len(), 2);
        // Heaviest batch first
        assert_eq!(batches[0].len(), 1);
        assert_eq!(batches[0][0].id(), "slow_test");
        // Second batch has medium and fast
        assert_eq!(batches[1].len(), 2);
    }

    #[tokio::test]
    async fn test_schedule_heaviest_batch_first() {
        let records = [
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_b", "test-group"),
            TestRecord::new("test_c", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let mut durations = HashMap::new();
        durations.insert("test_a".to_string(), Duration::from_secs(1));
        durations.insert("test_b".to_string(), Duration::from_secs(5));
        durations.insert("test_c".to_string(), Duration::from_secs(3));

        let scheduler = Scheduler::new(3, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Each test in its own batch (3 workers, 3 tests)
        // Sorted by duration: test_b (5s), test_c (3s), test_a (1s)
        assert_eq!(batches.len(), 3);
        assert_eq!(batches[0][0].id(), "test_b"); // Heaviest first
        assert_eq!(batches[1][0].id(), "test_c");
        assert_eq!(batches[2][0].id(), "test_a");
    }

    #[tokio::test]
    async fn test_schedule_uses_default_for_unknown() {
        let records = [
            TestRecord::new("known_slow", "test-group"),
            TestRecord::new("unknown_test", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let mut durations = HashMap::new();
        durations.insert("known_slow".to_string(), Duration::from_secs(10));
        // unknown_test will use default of 1 second

        let scheduler = Scheduler::new(2, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        assert_eq!(batches.len(), 2);
        // known_slow (10s) should be in heaviest batch
        assert_eq!(batches[0][0].id(), "known_slow");
        assert_eq!(batches[1][0].id(), "unknown_test");
    }

    #[tokio::test]
    async fn test_schedule_duplicate_prevention() {
        // Simulate retry scenario: same test appears multiple times
        let records = [
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_a", "test-group"), // retry 1
            TestRecord::new("test_a", "test-group"), // retry 2
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let mut durations = HashMap::new();
        durations.insert("test_a".to_string(), Duration::from_secs(5));

        let scheduler = Scheduler::new(3, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Each instance of test_a must be in a different batch
        assert_eq!(batches.len(), 3);
        assert_eq!(batches[0].len(), 1);
        assert_eq!(batches[1].len(), 1);
        assert_eq!(batches[2].len(), 1);

        // Verify no batch contains duplicate test IDs
        for batch in &batches {
            let ids: Vec<_> = batch.iter().map(|t| t.id()).collect();
            let unique: std::collections::HashSet<_> = ids.iter().collect();
            assert_eq!(ids.len(), unique.len(), "Batch contains duplicate test IDs");
        }
    }

    #[tokio::test]
    async fn test_schedule_mixed_duplicates_and_unique() {
        // Mix of retried and unique tests
        let records = [
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_a", "test-group"), // retry
            TestRecord::new("test_b", "test-group"),
            TestRecord::new("test_c", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let mut durations = HashMap::new();
        durations.insert("test_a".to_string(), Duration::from_secs(10));
        durations.insert("test_b".to_string(), Duration::from_secs(5));
        durations.insert("test_c".to_string(), Duration::from_secs(1));

        let scheduler = Scheduler::new(3, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Verify no batch contains duplicate test IDs
        for batch in &batches {
            let ids: Vec<_> = batch.iter().map(|t| t.id()).collect();
            let unique: std::collections::HashSet<_> = ids.iter().collect();
            assert_eq!(ids.len(), unique.len(), "Batch contains duplicate test IDs");
        }

        // Both instances of test_a should exist across batches
        let all_ids: Vec<_> = batches
            .iter()
            .flat_map(|b| b.iter().map(|t| t.id()))
            .collect();
        assert_eq!(all_ids.iter().filter(|&&id| id == "test_a").count(), 2);
    }

    #[tokio::test]
    async fn test_schedule_creates_extra_batches_for_retries() {
        // 2 workers but 3 instances of same test — creates 3 batches (one per instance)
        let records = [
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_a", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let durations = HashMap::new();
        let scheduler = Scheduler::new(2, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Each instance must be in a separate batch
        assert_eq!(batches.len(), 3);
        for batch in &batches {
            assert_eq!(batch.len(), 1);
            assert_eq!(batch[0].id(), "test_a");
        }
    }

    #[tokio::test]
    async fn test_schedule_splits_on_command_length() {
        // Create tests whose IDs together exceed MAX_BATCH_COMMAND_LEN
        let long_name = "a".repeat(MAX_BATCH_COMMAND_LEN / 2 + 1);
        let records = [
            TestRecord::new(format!("{long_name}_1"), "test-group"),
            TestRecord::new(format!("{long_name}_2"), "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let scheduler = Scheduler::new(1, &tests, &HashMap::new(), &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Two tests that each use >half the command length budget must be in separate batches
        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].len(), 1);
        assert_eq!(batches[1].len(), 1);
    }

    #[tokio::test]
    async fn test_schedule_groups_short_commands() {
        // Create many tests with short IDs that fit in one batch
        let records: Vec<_> = (0..100)
            .map(|i| TestRecord::new(format!("t{i}"), "test-group"))
            .collect();
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let scheduler = Scheduler::new(1, &tests, &HashMap::new(), &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Total command length is ~400 chars, well under 30k — should be 1 batch
        assert_eq!(batches.len(), 1);
        assert_eq!(batches[0].len(), 100);
    }

    #[tokio::test]
    async fn test_schedule_individual_tests_get_own_batch() {
        let mut records = [
            TestRecord::new("fast_1", "fast-group"),
            TestRecord::new("fast_2", "fast-group"),
            TestRecord::new("slow_1", "slow-group"),
            TestRecord::new("slow_2", "slow-group"),
        ];
        records[2].schedule_individual = true;
        records[3].schedule_individual = true;

        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();
        let durations = HashMap::new();
        let scheduler = Scheduler::new(2, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Each individually-scheduled test must be alone in its batch
        for batch in &batches {
            let has_individual = batch.iter().any(|t| t.schedule_individual());
            if has_individual {
                assert_eq!(
                    batch.len(),
                    1,
                    "Individually-scheduled test must be alone in its batch"
                );
            }
        }
        // All 4 tests should be scheduled
        let total: usize = batches.iter().map(|b| b.len()).sum();
        assert_eq!(total, 4);
    }

    #[tokio::test]
    async fn test_schedule_individual_tests_preserves_interleaved_order() {
        // Simulate already-interleaved individual instances (as orchestrator would produce)
        let mut records = vec![
            TestRecord::new("slow_a", "slow-group"),
            TestRecord::new("slow_b", "slow-group"),
            TestRecord::new("slow_a", "slow-group"),
            TestRecord::new("slow_b", "slow-group"),
            TestRecord::new("slow_a", "slow-group"),
        ];
        for r in &mut records {
            r.schedule_individual = true;
        }

        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();
        let durations = HashMap::new();
        let scheduler = Scheduler::new(4, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Each individually-scheduled test in its own batch, order preserved
        let ids: Vec<&str> = batches
            .iter()
            .map(|b| {
                assert_eq!(b.len(), 1);
                b[0].id()
            })
            .collect();
        assert_eq!(ids, vec!["slow_a", "slow_b", "slow_a", "slow_b", "slow_a"]);
    }

    #[tokio::test]
    async fn test_schedule_individual_tests_at_front() {
        let mut records = [
            TestRecord::new("fast_1", "fast-group"),
            TestRecord::new("slow_1", "slow-group"),
            TestRecord::new("fast_2", "fast-group"),
        ];
        records[1].schedule_individual = true;

        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();
        let durations = HashMap::new();
        let scheduler = Scheduler::new(4, &tests, &durations, &HashMap::new());
        let batches = drain_batches(&scheduler).await;

        // Individually-scheduled batches come first
        assert!(
            batches[0].iter().any(|t| t.schedule_individual()),
            "First batch should contain individually-scheduled test"
        );
        assert_eq!(batches[0].len(), 1);

        // Total tests scheduled
        let total: usize = batches.iter().map(|b| b.len()).sum();
        assert_eq!(total, 3);
    }

    #[tokio::test]
    async fn test_pop_requeues_batch() -> anyhow::Result<()> {
        // --- Multi-test batch: should split into two halves ---
        let records = [
            TestRecord::new("test_a", "test-group"),
            TestRecord::new("test_b", "test-group"),
        ];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let scheduler = Scheduler::new(1, &tests, &HashMap::new(), &HashMap::new());
        assert_eq!(scheduler.batch_count(), 1);

        // Pop the original 2-test batch
        let batch = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for batch"))?
            .ok_or_else(|| anyhow::anyhow!("expected a batch"))?;
        assert_eq!(batch.tests.len(), 2);
        assert_eq!(batch.single_test_retry_counter, 0);

        // First re-queued half: [test_a] — single-test, counter starts at 0
        let first_half = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for first half"))?
            .ok_or_else(|| anyhow::anyhow!("expected first half"))?;
        assert_eq!(first_half.tests.len(), 1);
        assert_eq!(first_half.tests[0].id(), "test_a");
        assert_eq!(first_half.single_test_retry_counter, 0);

        // Second re-queued half: [test_b] — single-test, counter starts at 0
        let second_half = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for second half"))?
            .ok_or_else(|| anyhow::anyhow!("expected second half"))?;
        assert_eq!(second_half.tests.len(), 1);
        assert_eq!(second_half.tests[0].id(), "test_b");
        assert_eq!(second_half.single_test_retry_counter, 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_pop_requeues_single_test_batch_as_clone() -> anyhow::Result<()> {
        // --- Single-test batch: should clone with incrementing counter ---
        let records = [TestRecord::new("only_test", "test-group")];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let scheduler = Scheduler::new(1, &tests, &HashMap::new(), &HashMap::new());
        assert_eq!(scheduler.batch_count(), 1);

        // Pop the original single-test batch (counter = 0)
        let batch = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for batch"))?
            .ok_or_else(|| anyhow::anyhow!("expected a batch"))?;
        assert_eq!(batch.tests.len(), 1);
        assert_eq!(batch.tests[0].id(), "only_test");
        assert_eq!(batch.single_test_retry_counter, 0);

        // Re-queued batch should have counter incremented to 1
        let requeued = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for requeued batch"))?
            .ok_or_else(|| anyhow::anyhow!("expected requeued batch"))?;
        assert_eq!(requeued.tests.len(), 1);
        assert_eq!(requeued.tests[0].id(), "only_test");
        assert_eq!(requeued.single_test_retry_counter, 1);

        Ok(())
    }

    #[tokio::test]
    async fn test_pop_stops_requeuing_single_test_after_max() -> anyhow::Result<()> {
        let records = [TestRecord::new("only_test", "test-group")];
        let tests: Vec<_> = records.iter().map(|r| r.test()).collect();

        let scheduler = Scheduler::new(1, &tests, &HashMap::new(), &HashMap::new());

        // Pop the initial batch plus MAX_SINGLE_TEST_REQUEUES re-queued copies.
        // The initial batch has counter=0, each re-queue increments by 1.
        // After popping the batch with counter=MAX_SINGLE_TEST_REQUEUES, no
        // further re-queue occurs and the queue should be empty.
        for i in 0..=MAX_SINGLE_TEST_REQUEUES {
            let batch = tokio::time::timeout(Duration::from_millis(100), scheduler.pop())
                .await
                .map_err(|_| anyhow::anyhow!("timed out at iteration {i}"))?
                .ok_or_else(|| anyhow::anyhow!("expected batch at iteration {i}"))?;
            assert_eq!(batch.tests.len(), 1);
            assert_eq!(batch.single_test_retry_counter, i);
        }

        // Queue should now be empty — next pop should time out
        let result = tokio::time::timeout(Duration::from_millis(100), scheduler.pop()).await;
        assert!(
            result.is_err(),
            "expected timeout (empty queue) after max requeues"
        );

        Ok(())
    }
}
