//! `--dispatch`: run the full pipeline locally.
//!
//! Walks the DAG in dependency order, runs each job (local container, remote
//! SSH, or exec-prefix), passes artifacts between jobs, and collects pass/fail
//! results. Tag-based filtering routes jobs to the right runner by matching
//! their `tags:` against runner declarations in `.cui-cui.toml`.
//!
//! Jobs can be skipped (`--skip`), and the dispatch prints a compact status
//! DAG as it progresses.

use std::collections::HashMap;
use std::fmt;
use std::io::{self, IsTerminal, Write as _};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use crossterm::{
    event::{
        self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyModifiers, MouseEventKind,
    },
    execute,
    terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode},
};
use indexmap::IndexMap;
use petgraph::algo::toposort;
use petgraph::visit::EdgeRef;
use ratatui::{
    Terminal,
    backend::CrosstermBackend,
    buffer::Buffer,
    layout::{Constraint, Layout, Rect},
    style::{Color, Modifier, Style},
    text::{Line, Span},
    widgets::{Block, Borders, Paragraph, Widget},
};
use unicode_width::UnicodeWidthStr;

use crate::artifacts;
use crate::config::RunnerConfig;
use crate::container::{self, ActiveContainers, ContainerConfig, ContainerExitStatus};
use crate::graph;
use crate::model::{CiSystem, Pipeline};
use crate::parser::gitlab_fetch::GitlabConfig;
use crate::remote;
use crate::resolve::GitContext;

/// Search state for the TUI log pane.
struct SearchState {
    /// Whether the user is currently typing a search query.
    input_mode: bool,
    /// The current search query being typed.
    input: String,
    /// The active search query (committed on Enter).
    query: String,
    /// Line indices that contain the search query.
    match_lines: Vec<usize>,
    /// Index into `match_lines` for the current match.
    current: usize,
}

impl SearchState {
    fn new() -> Self {
        Self {
            input_mode: false,
            input: String::new(),
            query: String::new(),
            match_lines: Vec::new(),
            current: 0,
        }
    }

    fn start_input(&mut self) {
        self.input_mode = true;
        self.input.clear();
    }

    fn cancel_input(&mut self) {
        self.input_mode = false;
        self.input.clear();
    }

    fn commit(&mut self) {
        self.input_mode = false;
        self.query = self.input.drain(..).collect();
        self.current = 0;
    }

    fn clear(&mut self) {
        self.input_mode = false;
        self.input.clear();
        self.query.clear();
        self.match_lines.clear();
        self.current = 0;
    }

    fn has_query(&self) -> bool {
        !self.query.is_empty()
    }

    fn recompute_matches(&mut self, lines: &[String], filter: Option<&[usize]>) {
        self.match_lines.clear();
        if self.query.is_empty() {
            return;
        }
        let query_lower = self.query.to_ascii_lowercase();
        let indices: Box<dyn Iterator<Item = usize>> = if let Some(visible) = filter {
            Box::new(visible.iter().copied())
        } else {
            Box::new(0..lines.len())
        };
        for i in indices {
            if let Some(line) = lines.get(i) {
                let plain = strip_ansi(line).to_ascii_lowercase();
                if plain.contains(&query_lower) {
                    self.match_lines.push(i);
                }
            }
        }
        if self.current >= self.match_lines.len() {
            self.current = 0;
        }
    }

    fn next_match(&mut self) {
        if !self.match_lines.is_empty() {
            self.current = (self.current + 1) % self.match_lines.len();
        }
    }

    fn prev_match(&mut self) {
        if !self.match_lines.is_empty() {
            self.current = (self.current + self.match_lines.len() - 1) % self.match_lines.len();
        }
    }

    fn current_line(&self) -> Option<usize> {
        self.match_lines.get(self.current).copied()
    }
}

/// Grep filter state for the TUI log pane.
struct FilterState {
    input_mode: bool,
    input: String,
    query: String,
    /// Indices into the full log buffer for lines that match the filter.
    visible_lines: Vec<usize>,
}

impl FilterState {
    fn new() -> Self {
        Self {
            input_mode: false,
            input: String::new(),
            query: String::new(),
            visible_lines: Vec::new(),
        }
    }

    fn start_input(&mut self) {
        self.input_mode = true;
        self.input.clear();
    }

    fn cancel_input(&mut self) {
        self.input_mode = false;
        self.input.clear();
    }

    fn commit(&mut self) {
        self.input_mode = false;
        self.query = self.input.drain(..).collect();
    }

    fn clear(&mut self) {
        self.input_mode = false;
        self.input.clear();
        self.query.clear();
        self.visible_lines.clear();
    }

    fn is_active(&self) -> bool {
        !self.query.is_empty()
    }

    fn recompute(&mut self, lines: &[String]) {
        self.visible_lines.clear();
        if self.query.is_empty() {
            return;
        }
        let query_lower = self.query.to_ascii_lowercase();
        for (i, line) in lines.iter().enumerate() {
            let plain = strip_ansi(line).to_ascii_lowercase();
            if plain.contains(&query_lower) {
                self.visible_lines.push(i);
            }
        }
    }

    fn visible_count(&self, total: usize) -> usize {
        if self.is_active() {
            self.visible_lines.len()
        } else {
            total
        }
    }

    /// Convert an absolute line index to a scroll position in the filtered view.
    /// Returns the index within `visible_lines`, or the absolute index if no filter is active.
    fn scroll_for_line(&self, abs_line: usize) -> usize {
        if self.is_active() {
            self.visible_lines
                .iter()
                .position(|&l| l >= abs_line)
                .unwrap_or(self.visible_lines.len().saturating_sub(1))
        } else {
            abs_line
        }
    }
}

/// Strip ANSI escape sequences from a string.
fn strip_ansi(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut chars = s.chars().peekable();
    while let Some(ch) = chars.next() {
        if ch == '\x1b' {
            if chars.peek() == Some(&'[') {
                chars.next();
                for c in chars.by_ref() {
                    if c.is_ascii_alphabetic() {
                        break;
                    }
                }
            }
        } else {
            out.push(ch);
        }
    }
    out
}

/// Status of a job during dispatch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobStatus {
    Pending,
    Running,
    Passed,
    Failed,
    /// Job failed but has `allow_failure: true` — does not block dependents.
    AllowedFailure,
    Skipped,
}

impl fmt::Display for JobStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            JobStatus::Pending => write!(f, "pending"),
            JobStatus::Running => write!(f, "running"),
            JobStatus::Passed => write!(f, "passed"),
            JobStatus::Failed => write!(f, "failed"),
            JobStatus::AllowedFailure => write!(f, "failed (allowed)"),
            JobStatus::Skipped => write!(f, "skipped"),
        }
    }
}

/// Why a job was skipped.
#[derive(Debug, Clone)]
pub enum SkipReason {
    Manual,
    Never,
    NoImage,
    TagMismatch(Vec<String>),
    DependencyFailed(String),
    UserSkipped,
}

impl SkipReason {
    /// Whether this skip reason should block downstream jobs via implicit stage ordering.
    ///
    /// Manual and Never jobs are intentionally excluded from the pipeline and do not
    /// block downstream jobs in stage ordering (matching GitLab CI behavior).
    /// Other skip reasons (NoImage, TagMismatch, DependencyFailed) represent actual
    /// failures or infrastructure issues that should propagate.
    fn blocks_stage_dependents(&self) -> bool {
        !matches!(self, SkipReason::Manual | SkipReason::Never)
    }
}

impl fmt::Display for SkipReason {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SkipReason::Manual => write!(f, "manual"),
            SkipReason::Never => write!(f, "rules: never"),
            SkipReason::NoImage => write!(f, "no image"),
            SkipReason::TagMismatch(tags) => write!(f, "tag mismatch: {}", tags.join(",")),
            SkipReason::DependencyFailed(dep) => write!(f, "dependency failed: {}", dep),
            SkipReason::UserSkipped => write!(f, "--skip"),
        }
    }
}

/// Tag filtering configuration for dispatch.
pub struct TagFilter {
    /// Include jobs with no tags.
    pub no_tags: bool,
    /// Include jobs whose tags are a subset of this list.
    pub tags: Vec<String>,
}

impl TagFilter {
    /// Check if a job's tags match this filter.
    ///
    /// Returns `None` if the job is eligible, or `Some(SkipReason)` if not.
    fn check(&self, job_tags: &[String]) -> Option<SkipReason> {
        // No filtering configured: all jobs are eligible
        if !self.no_tags && self.tags.is_empty() {
            return None;
        }

        if job_tags.is_empty() {
            // Untagged job: eligible only if --no-tags
            if self.no_tags {
                None
            } else {
                Some(SkipReason::TagMismatch(vec![]))
            }
        } else {
            // Tagged job: eligible if all tags are in the filter list
            if self.tags.is_empty() {
                Some(SkipReason::TagMismatch(job_tags.to_vec()))
            } else {
                let all_match = job_tags
                    .iter()
                    .all(|t| self.tags.iter().any(|f| f.eq_ignore_ascii_case(t)));
                if all_match {
                    None
                } else {
                    let missing: Vec<String> = job_tags
                        .iter()
                        .filter(|t| !self.tags.iter().any(|f| f.eq_ignore_ascii_case(t)))
                        .cloned()
                        .collect();
                    Some(SkipReason::TagMismatch(missing))
                }
            }
        }
    }
}

/// Find the first runner whose tags are a superset of the job's tags.
///
/// - **Tagged jobs**: a runner matches if its tags contain all of the job's tags
///   (case-insensitive).
/// - **Untagged jobs**: a runner matches only if it has `no_tags = true`.
/// - Returns the first matching runner from the list, or `None`.
pub fn find_runner<'a>(
    runners: &'a [RunnerConfig],
    job_tags: &[String],
) -> Option<&'a RunnerConfig> {
    if runners.is_empty() {
        return None;
    }

    for runner in runners {
        if job_tags.is_empty() {
            // Untagged job: only match if runner has no_tags = true
            if runner.no_tags.unwrap_or(false) {
                return Some(runner);
            }
        } else {
            // Tagged job: runner must have all of the job's tags
            let all_match = job_tags
                .iter()
                .all(|jt| runner.tags.iter().any(|rt| rt.eq_ignore_ascii_case(jt)));
            if all_match {
                return Some(runner);
            }
        }
    }

    None
}

/// Extend a `TagFilter` with tags from configured runners so that jobs
/// matching a runner are not skipped by the tag filter.
pub fn extend_tag_filter_with_runners(filter: &mut TagFilter, runners: &[RunnerConfig]) {
    for runner in runners {
        // If any runner has no_tags, extend the filter
        if runner.no_tags.unwrap_or(false) {
            filter.no_tags = true;
        }
        // Add all runner tags to the filter
        for tag in &runner.tags {
            if !filter.tags.iter().any(|t| t.eq_ignore_ascii_case(tag)) {
                filter.tags.push(tag.clone());
            }
        }
    }
}

/// State tracked for each job during dispatch.
struct JobState {
    status: JobStatus,
    skip_reason: Option<SkipReason>,
    exit_code: Option<i32>,
    duration: Option<std::time::Duration>,
    log_path: Option<PathBuf>,
}

/// Dispatch state for the full pipeline.
struct DispatchState {
    /// Job states keyed by job name, in topological order.
    jobs: IndexMap<String, JobState>,
}

impl DispatchState {
    fn new(job_names: &[String]) -> Self {
        let mut jobs = IndexMap::new();
        for name in job_names {
            jobs.insert(
                name.clone(),
                JobState {
                    status: JobStatus::Pending,
                    skip_reason: None,
                    exit_code: None,
                    duration: None,
                    log_path: None,
                },
            );
        }
        DispatchState { jobs }
    }
}

/// Scheduling engine: owns the DAG, topological order, and job states.
///
/// Separates "which job runs next" from "how to run it", enabling future
/// parallel execution by returning multiple ready jobs at once.
struct Scheduler {
    dag: petgraph::graph::DiGraph<String, graph::DepKind>,
    node_indices: IndexMap<String, petgraph::graph::NodeIndex>,
    state: DispatchState,
    topo_names: Vec<String>,
}

impl Scheduler {
    fn new(
        pipeline: &Pipeline,
        ci_system: CiSystem,
        tag_filter: &TagFilter,
        skip_jobs: &[String],
    ) -> Result<Self> {
        let (dag, node_indices) = graph::build_graph(pipeline);
        let topo_order =
            toposort(&dag, None).map_err(|_| anyhow::anyhow!("Cycle detected in DAG"))?;
        let topo_names: Vec<String> = topo_order.iter().map(|idx| dag[*idx].clone()).collect();
        let mut state = DispatchState::new(&topo_names);

        // Pre-scan: mark ineligible jobs
        for name in &topo_names {
            let job = &pipeline.jobs[name];

            if skip_jobs.iter().any(|s| s == name) {
                state.jobs[name].status = JobStatus::Passed;
                state.jobs[name].skip_reason = Some(SkipReason::UserSkipped);
                continue;
            }

            if job.is_never() {
                state.jobs[name].status = JobStatus::Skipped;
                state.jobs[name].skip_reason = Some(SkipReason::Never);
                continue;
            }

            if job.is_manual() {
                state.jobs[name].status = JobStatus::Skipped;
                state.jobs[name].skip_reason = Some(SkipReason::Manual);
                continue;
            }

            let has_image = job.image.is_some()
                || (ci_system == CiSystem::GitHub
                    && container::runs_on_to_image(&job.tags).is_some());
            if !has_image {
                state.jobs[name].status = JobStatus::Skipped;
                state.jobs[name].skip_reason = Some(SkipReason::NoImage);
                continue;
            }

            if let Some(reason) = tag_filter.check(&job.tags) {
                state.jobs[name].status = JobStatus::Skipped;
                state.jobs[name].skip_reason = Some(reason);
            }
        }

        // Propagate skips through the DAG
        for name in &topo_names {
            if state.jobs[name].status != JobStatus::Skipped {
                continue;
            }
            let blocks_stage = state.jobs[name]
                .skip_reason
                .as_ref()
                .is_some_and(|r| r.blocks_stage_dependents());
            if let Some(&node_idx) = node_indices.get(name) {
                for edge in dag.edges_directed(node_idx, petgraph::Direction::Outgoing) {
                    if *edge.weight() == graph::DepKind::Stage && !blocks_stage {
                        continue;
                    }
                    let dep_name = &dag[edge.target()];
                    let dep_state = &state.jobs[dep_name];
                    if dep_state.status == JobStatus::Pending {
                        state.jobs[dep_name].status = JobStatus::Skipped;
                        state.jobs[dep_name].skip_reason =
                            Some(SkipReason::DependencyFailed(name.clone()));
                    }
                }
            }
        }

        Ok(Self {
            dag,
            node_indices,
            state,
            topo_names,
        })
    }

    fn state(&self) -> &DispatchState {
        &self.state
    }

    /// Return all pending jobs whose dependencies are satisfied, in topo order.
    fn ready_jobs(&self) -> Vec<&str> {
        self.topo_names
            .iter()
            .filter(|name| {
                if self.state.jobs[name.as_str()].status != JobStatus::Pending {
                    return false;
                }
                let Some(&node_idx) = self.node_indices.get(name.as_str()) else {
                    return false;
                };
                // All incoming edges must come from completed jobs
                self.dag
                    .edges_directed(node_idx, petgraph::Direction::Incoming)
                    .all(|edge| {
                        let dep_name = &self.dag[edge.source()];
                        let dep = &self.state.jobs[dep_name];
                        matches!(
                            dep.status,
                            JobStatus::Passed | JobStatus::AllowedFailure | JobStatus::Skipped
                        )
                    })
            })
            .map(|s| s.as_str())
            .collect()
    }

    fn mark_running(&mut self, name: &str) {
        self.state.jobs[name].status = JobStatus::Running;
    }

    fn skip(&mut self, name: &str, reason: SkipReason) {
        self.state.jobs[name].status = JobStatus::Skipped;
        self.state.jobs[name].skip_reason = Some(reason);
    }

    fn complete(
        &mut self,
        name: &str,
        exit_code: i32,
        duration: std::time::Duration,
        log_path: PathBuf,
        allow_failure: bool,
    ) {
        self.state.jobs[name].exit_code = Some(exit_code);
        self.state.jobs[name].duration = Some(duration);
        self.state.jobs[name].log_path = Some(log_path);

        if exit_code == 0 {
            self.state.jobs[name].status = JobStatus::Passed;
        } else if allow_failure {
            self.state.jobs[name].status = JobStatus::AllowedFailure;
        } else {
            self.state.jobs[name].status = JobStatus::Failed;
            if let Some(&node_idx) = self.node_indices.get(name) {
                propagate_failure(&self.dag, node_idx, name, &mut self.state);
            }
        }
    }

    fn fail_error(&mut self, name: &str) {
        self.state.jobs[name].status = JobStatus::Failed;
        if let Some(&node_idx) = self.node_indices.get(name) {
            propagate_failure(&self.dag, node_idx, name, &mut self.state);
        }
    }

    fn all_passed(&self) -> bool {
        self.state.jobs.values().all(|s| {
            matches!(
                s.status,
                JobStatus::Passed | JobStatus::AllowedFailure | JobStatus::Skipped
            )
        })
    }
}

/// Run the pipeline in dispatch mode (sequential, non-interactive).
///
/// Returns `true` if all eligible jobs passed.
#[allow(clippy::too_many_arguments)]
pub fn run_dispatch(
    pipeline: &Pipeline,
    ci_system: CiSystem,
    file_path: &Path,
    project_dir: Option<&Path>,
    pass_env: &[String],
    no_overlay: bool,
    privileged: bool,
    git_context: &GitContext,
    gitlab_config: Option<&GitlabConfig>,
    tag_filter: &TagFilter,
    skip_jobs: &[String],
    git_image: &str,
    runners: &[RunnerConfig],
) -> Result<bool> {
    let mut scheduler = Scheduler::new(pipeline, ci_system, tag_filter, skip_jobs)?;

    let log_dir = tempfile::tempdir()
        .context("Failed to create log directory")?
        .keep();

    // Prompt once for uncommitted changes before running any jobs.
    let effective_project_dir =
        project_dir.unwrap_or_else(|| file_path.parent().unwrap_or(file_path));
    let diff_to_apply: Arc<Option<Vec<u8>>> = Arc::new(container::prompt_uncommitted_changes(
        effective_project_dir,
    )?);

    let mut runner_locks: HashMap<String, Arc<std::sync::Mutex<()>>> = HashMap::new();
    for runner in runners {
        runner_locks.insert(runner.name.clone(), Arc::new(std::sync::Mutex::new(())));
    }

    let is_tty = io::stdout().is_terminal();
    let start = Instant::now();

    let (tx, rx) = std::sync::mpsc::channel::<JobCompletion>();
    let mut running_counts: HashMap<String, usize> = HashMap::new();
    let mut running_total: usize = 0;

    // Shared log buffer for TTY mode (job output goes here instead of stdout)
    let log_lines: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));

    // TTY: set up ratatui alternate screen
    let mut terminal = if is_tty {
        enable_raw_mode().context("Failed to enable raw mode")?;
        let mut stdout = io::stdout();
        execute!(stdout, EnterAlternateScreen, EnableMouseCapture)
            .context("Failed to enter alternate screen")?;
        let mut term = Terminal::new(CrosstermBackend::new(stdout))?;
        term.clear()?;
        Some(term)
    } else {
        None
    };

    // Non-TTY: use the inline display
    let mut display = if !is_tty {
        let mut d = DispatchDisplay::new();
        d.show_initial_status(pipeline, scheduler.state());
        Some(d)
    } else {
        None
    };

    let mut quit_requested = false;
    let mut log_scroll: usize = 0;
    let mut log_auto_scroll = true;
    let mut log_pane_height: usize = 0;
    let mut dispatch_done = false;
    let mut dispatch_elapsed: Option<Duration> = None;
    let active_containers = ActiveContainers::new();
    let mut search = SearchState::new();
    let mut filter = FilterState::new();

    let result = (|| -> Result<bool> {
        loop {
            // --- TTY: handle terminal events ---
            if is_tty {
                while event::poll(Duration::from_millis(50))? {
                    match event::read()? {
                        Event::Key(key) if filter.input_mode => match key.code {
                            KeyCode::Esc => {
                                filter.cancel_input();
                            }
                            KeyCode::Enter => {
                                filter.commit();
                                if filter.is_active()
                                    && let Ok(buf) = log_lines.lock()
                                {
                                    filter.recompute(&buf);
                                }
                                log_scroll = 0;
                                log_auto_scroll = true;
                                search.clear();
                            }
                            KeyCode::Backspace => {
                                filter.input.pop();
                            }
                            KeyCode::Char(c) => {
                                filter.input.push(c);
                            }
                            _ => {}
                        },
                        Event::Key(key) if search.input_mode => match key.code {
                            KeyCode::Esc => {
                                search.cancel_input();
                            }
                            KeyCode::Enter => {
                                search.commit();
                                if search.has_query() {
                                    if let Ok(buf) = log_lines.lock() {
                                        search.recompute_matches(
                                            &buf,
                                            if filter.is_active() {
                                                Some(&filter.visible_lines)
                                            } else {
                                                None
                                            },
                                        );
                                    }
                                    if let Some(line) = search.current_line() {
                                        log_scroll = filter.scroll_for_line(line);
                                        log_auto_scroll = false;
                                    }
                                }
                            }
                            KeyCode::Backspace => {
                                search.input.pop();
                            }
                            KeyCode::Char(c) => {
                                search.input.push(c);
                            }
                            _ => {}
                        },
                        Event::Key(key) => match key.code {
                            KeyCode::Char('q') | KeyCode::Char('c')
                                if key.code == KeyCode::Char('q')
                                    || key.modifiers.contains(KeyModifiers::CONTROL) =>
                            {
                                quit_requested = true;
                            }
                            KeyCode::Char('/') => {
                                search.start_input();
                            }
                            KeyCode::Char('n') if search.has_query() => {
                                search.next_match();
                                if let Some(line) = search.current_line() {
                                    log_scroll = filter.scroll_for_line(line);
                                    log_auto_scroll = false;
                                }
                            }
                            KeyCode::Char('N') if search.has_query() => {
                                search.prev_match();
                                if let Some(line) = search.current_line() {
                                    log_scroll = filter.scroll_for_line(line);
                                    log_auto_scroll = false;
                                }
                            }
                            KeyCode::Esc if search.has_query() => {
                                search.clear();
                            }
                            KeyCode::Esc if filter.is_active() => {
                                filter.clear();
                                log_scroll = 0;
                                log_auto_scroll = true;
                            }
                            KeyCode::Char('f') => {
                                filter.start_input();
                            }
                            KeyCode::Up | KeyCode::Char('k') => {
                                if log_scroll > 0 {
                                    log_scroll -= 1;
                                    log_auto_scroll = false;
                                }
                            }
                            KeyCode::Down | KeyCode::Char('j') => {
                                log_scroll += 1;
                                log_auto_scroll = false;
                            }
                            KeyCode::End | KeyCode::Char('G') => {
                                log_auto_scroll = true;
                            }
                            KeyCode::Home | KeyCode::Char('g') => {
                                log_scroll = 0;
                                log_auto_scroll = false;
                            }
                            KeyCode::PageUp => {
                                log_scroll = log_scroll.saturating_sub(20);
                                log_auto_scroll = false;
                            }
                            KeyCode::PageDown => {
                                log_scroll += 20;
                                log_auto_scroll = false;
                            }
                            _ => {}
                        },
                        Event::Mouse(mouse) => match mouse.kind {
                            MouseEventKind::ScrollUp => {
                                log_scroll = log_scroll.saturating_sub(3);
                                log_auto_scroll = false;
                            }
                            MouseEventKind::ScrollDown => {
                                log_scroll += 3;
                                log_auto_scroll = false;
                            }
                            _ => {}
                        },
                        Event::Resize(_, _) => {
                            if let Some(term) = &mut terminal {
                                term.clear()?;
                            }
                        }
                        _ => {}
                    }
                }
                if quit_requested {
                    break;
                }
            }

            // --- Process completed jobs ---
            while let Ok(completion) = rx.try_recv() {
                let mut disp_ref = display.as_mut();
                process_completion(
                    &completion,
                    &mut scheduler,
                    &mut disp_ref,
                    &log_lines,
                    is_tty,
                );
                *running_counts
                    .entry(completion.runner_key.clone())
                    .or_default() -= 1;
                running_total -= 1;
                if let Some(d) = &mut display {
                    d.update_status(scheduler.state(), start);
                }
            }

            // --- Launch ready jobs ---
            let ready: Vec<String> = scheduler
                .ready_jobs()
                .into_iter()
                .map(str::to_string)
                .collect();
            let mut launched = false;

            for name in &ready {
                let job = &pipeline.jobs[name.as_str()];
                let runner = find_runner(runners, &job.tags);
                let runner_key = runner
                    .map(|r| r.name.clone())
                    .unwrap_or_else(|| "(local)".to_string());
                let max_concurrency = runner.map(|r| r.concurrency()).unwrap_or(1);
                let current = *running_counts.get(&runner_key).unwrap_or(&0);

                if current >= max_concurrency {
                    continue;
                }

                let name = name.to_string();

                let mut config = match container::build_config_for_job(
                    job,
                    ci_system,
                    file_path,
                    project_dir,
                    pass_env,
                    no_overlay,
                    privileged,
                    git_context,
                    git_image,
                ) {
                    Some(c) => c,
                    None => {
                        scheduler.skip(name.as_str(), SkipReason::NoImage);
                        continue;
                    }
                };

                resolve_dispatch_artifacts(
                    &mut config,
                    scheduler.state(),
                    git_context,
                    gitlab_config,
                );

                let log_path = log_dir.join(format!("{}.log", sanitize_filename(&name)));

                if let Some(r) = runner
                    && r.host.is_some()
                    && !config.artifact_dirs.is_empty()
                    && let Err(e) =
                        remote::upload_artifacts(r, &config.project_path, &config.artifact_dirs)
                {
                    let msg = format!("failed to upload artifacts for '{}': {}", name, e);
                    if is_tty {
                        if let Ok(mut buf) = log_lines.lock() {
                            buf.push(format!("\x1b[33mWarning: {}\x1b[0m", msg));
                        }
                    } else if let Some(d) = &mut display {
                        d.warning(&msg);
                    }
                }

                if !is_tty {
                    if let Some(d) = &mut display {
                        d.job_starting(&name, runner);
                    }
                } else if let Ok(mut buf) = log_lines.lock() {
                    if let Some(r) = runner {
                        buf.push(format!(
                            "\x1b[36m━━━ {} → runner '{}' ({}) ━━━\x1b[0m",
                            name,
                            r.name,
                            r.host.as_deref().unwrap_or("local"),
                        ));
                    } else {
                        buf.push(format!("\x1b[36m━━━ {} ━━━\x1b[0m", name));
                    }
                }

                if let Some(r) = runner {
                    config.apply_runner_runtime(
                        r.runtime.as_deref(),
                        r.kernel(),
                        r.memory(),
                        r.cpus(),
                    );
                }

                let job_artifacts_paths = config.artifacts_paths.clone();
                let job_artifacts_exclude = config.artifacts_exclude.clone();
                let job_artifacts_when = config.artifacts_when.clone();
                let job_project_path = config.project_path.clone();

                scheduler.mark_running(&name);
                *running_counts.entry(runner_key.clone()).or_default() += 1;
                running_total += 1;
                launched = true;

                if let Some(d) = &mut display {
                    d.update_status(scheduler.state(), start);
                }

                let tx = tx.clone();
                let runner_clone = runner.cloned();
                let diff = diff_to_apply.clone();
                let project_dir_owned = effective_project_dir.to_path_buf();
                let allow_failure = job.allow_failure;
                let output_config = if is_tty {
                    container::OutputConfig {
                        prefix: Some(format!("[{}]", name)),
                        log_sink: Some(log_lines.clone()),
                        active_containers: Some(active_containers.clone()),
                    }
                } else {
                    container::OutputConfig {
                        prefix: Some(format!("\x1b[36m[{}]\x1b[0m", name)),
                        log_sink: None,
                        active_containers: Some(active_containers.clone()),
                    }
                };
                let git_lock = runner_locks.get(&runner_key).cloned();

                std::thread::spawn(move || {
                    let diff_ref = diff.as_ref().as_deref();
                    let lock_ref = git_lock.as_deref();
                    let result: Result<ContainerExitStatus> = if let Some(runner) = &runner_clone {
                        remote::run_on_runner(
                            &config,
                            runner,
                            &log_path,
                            diff_ref,
                            &project_dir_owned,
                            &output_config,
                            lock_ref,
                        )
                    } else {
                        container::run_container(config, &log_path, diff_ref, None, &output_config)
                    };

                    let mut warnings = Vec::new();
                    if let Ok(ref status) = result
                        && let Some(runner) = &runner_clone
                        && runner.host.is_some()
                        && !job_artifacts_paths.is_empty()
                        && container::should_save_artifacts(status.exit_code, &job_artifacts_when)
                        && let Err(e) = remote::download_artifacts(
                            runner,
                            &name,
                            &job_project_path,
                            &job_artifacts_paths,
                            &job_artifacts_exclude,
                            &output_config,
                        )
                    {
                        warnings.push(format!(
                            "failed to download artifacts for '{}': {}",
                            name, e
                        ));
                    }

                    let _ = tx.send(JobCompletion {
                        name,
                        result,
                        allow_failure,
                        runner_key,
                        warnings,
                    });
                });
            }

            // --- Render (TTY) ---
            if let Some(term) = &mut terminal {
                let total_lines = log_lines.lock().map(|b| b.len()).unwrap_or(0);

                // Recompute filter when log buffer grows
                if filter.is_active()
                    && let Ok(buf) = log_lines.lock()
                {
                    filter.recompute(&buf);
                }

                let effective_count = filter.visible_count(total_lines);

                // Recompute search matches when log buffer grows
                if search.has_query()
                    && let Ok(buf) = log_lines.lock()
                {
                    search.recompute_matches(
                        &buf,
                        if filter.is_active() {
                            Some(&filter.visible_lines)
                        } else {
                            None
                        },
                    );
                }

                // Compute auto-scroll position (needs pane height from previous frame)
                if log_auto_scroll && log_pane_height > 0 {
                    log_scroll = effective_count.saturating_sub(log_pane_height);
                }

                // Clamp scroll to valid range
                let max_scroll = effective_count.saturating_sub(log_pane_height.max(1));
                log_scroll = log_scroll.min(max_scroll);

                // Re-enable auto-scroll when user reaches the bottom
                if !log_auto_scroll && log_pane_height > 0 && log_scroll >= max_scroll {
                    log_auto_scroll = true;
                }

                let auto_scroll = log_auto_scroll;
                let scroll_pos = log_scroll;
                let done = dispatch_done;
                let search_query = if search.has_query() {
                    Some(search.query.clone())
                } else {
                    None
                };
                let search_current_line = search.current_line();
                let search_input_mode = search.input_mode;
                let search_input_text = search.input.clone();
                let search_match_count = search.match_lines.len();
                let search_current_idx = search.current;
                let search_has_query = search.has_query();
                let filter_visible = if filter.is_active() {
                    Some(filter.visible_lines.clone())
                } else {
                    None
                };
                let filter_input_mode = filter.input_mode;
                let filter_input_text = filter.input.clone();
                let filter_active = filter.is_active();
                let filter_query = filter.query.clone();
                term.draw(|frame| {
                    let size = frame.area();
                    let status_h = pipeline_status_height(pipeline, scheduler.state());
                    let chunks = Layout::vertical([
                        Constraint::Length(status_h + 2), // +2 for border
                        Constraint::Min(5),
                        Constraint::Length(1),
                    ])
                    .split(size);

                    // Pipeline status
                    let status_block = Block::default()
                        .borders(Borders::ALL)
                        .title(" Pipeline Status ");
                    let inner = status_block.inner(chunks[0]);
                    status_block.render(chunks[0], frame.buffer_mut());
                    PipelineStatusWidget {
                        pipeline,
                        state: scheduler.state(),
                        elapsed: Some(start),
                    }
                    .render(inner, frame.buffer_mut());

                    // Log pane
                    let log_title = if filter_active {
                        format!(" Output [filter: {}] ", filter_query)
                    } else {
                        " Output ".to_string()
                    };
                    let log_block = Block::default().borders(Borders::ALL).title(log_title);
                    let log_inner = log_block.inner(chunks[1]);
                    log_pane_height = log_inner.height as usize;
                    log_block.render(chunks[1], frame.buffer_mut());
                    render_log_pane(
                        &log_lines,
                        scroll_pos,
                        log_inner,
                        frame.buffer_mut(),
                        search_query.as_deref(),
                        search_current_line,
                        filter_visible.as_deref(),
                    );

                    // Status bar
                    let keys_suffix = if filter_active {
                        "  /:search  f:new filter  Esc:clear"
                    } else {
                        "  /:search  f:filter"
                    };
                    let bar = if filter_input_mode {
                        Line::from(vec![
                            Span::styled(" filter: ", Style::default().fg(Color::Magenta)),
                            Span::styled(
                                filter_input_text.as_str(),
                                Style::default().fg(Color::White),
                            ),
                            Span::styled("\u{2588}", Style::default().fg(Color::White)),
                            Span::styled(
                                "  Enter:apply  Esc:cancel",
                                Style::default().fg(Color::DarkGray),
                            ),
                        ])
                    } else if search_input_mode {
                        Line::from(vec![
                            Span::styled(" /", Style::default().fg(Color::Cyan)),
                            Span::styled(
                                search_input_text.as_str(),
                                Style::default().fg(Color::White),
                            ),
                            Span::styled("\u{2588}", Style::default().fg(Color::White)),
                            Span::styled(
                                "  Enter:search  Esc:cancel",
                                Style::default().fg(Color::DarkGray),
                            ),
                        ])
                    } else if search_has_query {
                        let match_info = if search_match_count > 0 {
                            format!(" [{}/{}]", search_current_idx + 1, search_match_count,)
                        } else {
                            " [no matches]".to_string()
                        };
                        Line::from(vec![
                            Span::styled(
                                format!(" /{}", search_query.as_deref().unwrap_or("")),
                                Style::default().fg(Color::Cyan),
                            ),
                            Span::styled(match_info, Style::default().fg(Color::Yellow)),
                            Span::styled(
                                format!("  n:next  N:prev  Esc:clear{}", keys_suffix),
                                Style::default().fg(Color::DarkGray),
                            ),
                        ])
                    } else if done {
                        Line::from(vec![
                            Span::styled(
                                " Done — press q to exit",
                                Style::default().fg(Color::Green),
                            ),
                            Span::styled(keys_suffix, Style::default().fg(Color::DarkGray)),
                        ])
                    } else if !auto_scroll {
                        Line::from(vec![
                            Span::styled(
                                " Paused — press G to resume auto-scroll",
                                Style::default().fg(Color::Yellow),
                            ),
                            Span::styled(
                                format!("  q:quit{}", keys_suffix),
                                Style::default().fg(Color::DarkGray),
                            ),
                        ])
                    } else {
                        Line::from(vec![Span::styled(
                            format!(" q:quit  j/k:scroll  G:bottom  g:top{}", keys_suffix),
                            Style::default().fg(Color::DarkGray),
                        )])
                    };
                    Paragraph::new(bar).render(chunks[2], frame.buffer_mut());
                })?;
            }

            // TTY: stay in the loop after dispatch finishes so the user can browse logs
            if running_total == 0 && !launched {
                dispatch_done = true;
                dispatch_elapsed.get_or_insert_with(|| start.elapsed());
                if !is_tty {
                    break;
                }
            }

            // Non-TTY: block-wait for next completion
            if !is_tty
                && !launched
                && let Ok(completion) = rx.recv()
            {
                let mut disp_ref = display.as_mut();
                process_completion(
                    &completion,
                    &mut scheduler,
                    &mut disp_ref,
                    &log_lines,
                    is_tty,
                );
                *running_counts
                    .entry(completion.runner_key.clone())
                    .or_default() -= 1;
                running_total -= 1;
                if let Some(d) = &mut display {
                    d.update_status(scheduler.state(), start);
                }
            }
        }

        Ok(scheduler.all_passed())
    })();

    // Restore terminal (TTY)
    if let Some(mut term) = terminal.take() {
        disable_raw_mode().ok();
        execute!(io::stdout(), LeaveAlternateScreen, DisableMouseCapture).ok();
        term.show_cursor().ok();
    }

    if quit_requested && !dispatch_done {
        eprintln!("Interrupted — stopping running containers...");
        active_containers.kill_all();
        std::process::exit(130);
    }

    // Print summary to normal stdout
    let mut summary_display = DispatchDisplay::new();
    let total_elapsed = dispatch_elapsed.unwrap_or_else(|| start.elapsed());
    summary_display.print_summary(scheduler.state(), total_elapsed);

    result
}

/// Result sent from a worker thread back to the main dispatch loop.
struct JobCompletion {
    name: String,
    result: Result<ContainerExitStatus>,
    allow_failure: bool,
    runner_key: String,
    warnings: Vec<String>,
}

fn process_completion(
    completion: &JobCompletion,
    scheduler: &mut Scheduler,
    display: &mut Option<&mut DispatchDisplay>,
    log_lines: &Arc<Mutex<Vec<String>>>,
    is_tty: bool,
) {
    for w in &completion.warnings {
        if is_tty {
            if let Ok(mut buf) = log_lines.lock() {
                buf.push(format!("\x1b[33mWarning: {}\x1b[0m", w));
            }
        } else if let Some(d) = display.as_deref_mut() {
            d.warning(w);
        }
    }

    match &completion.result {
        Ok(status) => {
            let msg = if status.exit_code == 0 {
                format!(
                    "\x1b[32m\u{2713} {} passed ({})\x1b[0m",
                    completion.name,
                    format_duration(status.duration)
                )
            } else if completion.allow_failure {
                format!(
                    "\x1b[33m\u{26a0} {} failed (allowed, exit {}, {})\x1b[0m",
                    completion.name,
                    status.exit_code,
                    format_duration(status.duration)
                )
            } else {
                format!(
                    "\x1b[31m\u{2717} {} failed (exit {}, {})\x1b[0m",
                    completion.name,
                    status.exit_code,
                    format_duration(status.duration)
                )
            };

            if is_tty {
                if let Ok(mut buf) = log_lines.lock() {
                    buf.push(msg);
                }
            } else if let Some(d) = display.as_deref_mut() {
                if status.exit_code == 0 {
                    d.job_passed(&completion.name, status.duration);
                } else if completion.allow_failure {
                    d.job_allowed_failure(&completion.name, status.exit_code, status.duration);
                } else {
                    d.job_failed(&completion.name, status.exit_code, status.duration);
                }
            }

            scheduler.complete(
                &completion.name,
                status.exit_code,
                status.duration,
                status.log_path.clone(),
                completion.allow_failure,
            );
        }
        Err(e) => {
            if is_tty {
                if let Ok(mut buf) = log_lines.lock() {
                    buf.push(format!(
                        "\x1b[31m\u{2717} {} error: {}\x1b[0m",
                        completion.name, e
                    ));
                }
            } else if let Some(d) = display.as_deref_mut() {
                d.job_error(&completion.name, e);
            }
            scheduler.fail_error(&completion.name);
        }
    }
}

/// Render log lines into a fixed area, handling ANSI-colored text.
///
/// `scroll` is the index of the first visible line (in the effective view).
/// When `filter` is set, only those line indices are shown.
/// When `search_query` is set, matching substrings are highlighted.
/// `search_current_line` is the absolute line index of the current match.
fn render_log_pane(
    log_lines: &Arc<Mutex<Vec<String>>>,
    scroll: usize,
    area: Rect,
    buf: &mut Buffer,
    search_query: Option<&str>,
    search_current_line: Option<usize>,
    filter: Option<&[usize]>,
) {
    let lines = match log_lines.lock() {
        Ok(b) => b.clone(),
        Err(_) => return,
    };

    let height = area.height as usize;
    if lines.is_empty() || height == 0 {
        return;
    }

    if let Some(visible_indices) = filter {
        let start = scroll.min(visible_indices.len());
        let end = (start + height).min(visible_indices.len());

        for (i, &abs_idx) in visible_indices[start..end].iter().enumerate() {
            let y = area.y + i as u16;
            if y >= area.y + area.height {
                break;
            }
            if let Some(line) = lines.get(abs_idx) {
                render_ansi_line(line, area.x, y, area.width, buf);

                if let Some(query) = search_query
                    && !query.is_empty()
                {
                    let is_current = search_current_line == Some(abs_idx);
                    highlight_search_matches(line, query, area.x, y, area.width, buf, is_current);
                }
            }
        }
    } else {
        let start_idx = scroll.min(lines.len());
        let end = (start_idx + height).min(lines.len());

        for (i, line) in lines[start_idx..end].iter().enumerate() {
            let y = area.y + i as u16;
            if y >= area.y + area.height {
                break;
            }
            let abs_line = start_idx + i;
            render_ansi_line(line, area.x, y, area.width, buf);

            if let Some(query) = search_query
                && !query.is_empty()
            {
                let is_current = search_current_line == Some(abs_line);
                highlight_search_matches(line, query, area.x, y, area.width, buf, is_current);
            }
        }
    }
}

/// Overlay search match highlights on an already-rendered line.
///
/// Builds a char-index → display-column mapping so that multi-width
/// Unicode characters (━, →, ✓, etc.) are highlighted correctly.
fn highlight_search_matches(
    line: &str,
    query: &str,
    x: u16,
    y: u16,
    width: u16,
    buf: &mut Buffer,
    is_current: bool,
) {
    use unicode_width::UnicodeWidthChar;

    let plain = strip_ansi(line);
    let plain_lower = plain.to_ascii_lowercase();
    let query_lower = query.to_ascii_lowercase();
    let max_x = x + width;

    // Build mapping: for each char index, the display column it starts at
    // and how many columns it occupies.
    let char_cols: Vec<(u16, u16)> = plain
        .chars()
        .scan(0u16, |col, ch| {
            let w = ch.width().unwrap_or(0) as u16;
            let start = *col;
            *col += w;
            Some((x + start, w))
        })
        .collect();

    let highlight = if is_current {
        Style::default().fg(Color::Black).bg(Color::Yellow)
    } else {
        Style::default().fg(Color::Black).bg(Color::DarkGray)
    };

    // Search uses char indices into the lowercased plain text.
    // Both plain_lower and plain have the same char count.
    let mut search_start = 0;
    while let Some(pos) = plain_lower[search_start..].find(&query_lower) {
        let byte_start = search_start + pos;
        let byte_end = byte_start + query_lower.len();
        let char_start = plain_lower[..byte_start].chars().count();
        let char_len = plain_lower[byte_start..byte_end].chars().count();

        for ci in char_start..char_start + char_len {
            if ci >= char_cols.len() {
                break;
            }
            let (col_start, col_w) = char_cols[ci];
            for c in col_start..col_start + col_w {
                if c >= max_x {
                    break;
                }
                if let Some(cell) = buf.cell_mut((c, y)) {
                    cell.set_style(highlight);
                }
            }
        }
        search_start = byte_end;
    }
}

/// Render a single line with ANSI color codes into a ratatui buffer.
fn render_ansi_line(line: &str, x: u16, y: u16, width: u16, buf: &mut Buffer) {
    let mut col = x;
    let max_x = x + width;
    let mut style = Style::default();
    let mut chars = line.chars().peekable();

    while let Some(ch) = chars.next() {
        if col >= max_x {
            break;
        }
        if ch == '\x1b' {
            // Parse ANSI escape sequence
            if chars.peek() == Some(&'[') {
                chars.next();
                let mut seq = String::new();
                for c in chars.by_ref() {
                    if c.is_ascii_alphabetic() {
                        if c == 'm' {
                            style = apply_ansi_codes(&seq, style);
                        }
                        break;
                    }
                    seq.push(c);
                }
            }
        } else {
            buf.set_string(col, y, ch.to_string(), style);
            col += 1;
        }
    }
}

fn apply_ansi_codes(seq: &str, mut style: Style) -> Style {
    for code in seq.split(';') {
        match code {
            "0" | "" => style = Style::default(),
            "1" => style = style.add_modifier(Modifier::BOLD),
            "31" => style = style.fg(Color::Red),
            "32" => style = style.fg(Color::Green),
            "33" => style = style.fg(Color::Yellow),
            "34" => style = style.fg(Color::Blue),
            "35" => style = style.fg(Color::Magenta),
            "36" => style = style.fg(Color::Cyan),
            "90" => style = style.fg(Color::DarkGray),
            _ => {}
        }
    }
    style
}

/// Propagate failure: mark all transitive dependents as skipped.
fn propagate_failure(
    dag: &petgraph::graph::DiGraph<String, graph::DepKind>,
    node_idx: petgraph::graph::NodeIndex,
    failed_name: &str,
    state: &mut DispatchState,
) {
    let mut stack = vec![node_idx];
    while let Some(idx) = stack.pop() {
        for edge in dag.edges_directed(idx, petgraph::Direction::Outgoing) {
            let dep_name = &dag[edge.target()];
            if let Some(dep_state) = state.jobs.get_mut(dep_name.as_str())
                && dep_state.status == JobStatus::Pending
            {
                dep_state.status = JobStatus::Skipped;
                dep_state.skip_reason = Some(SkipReason::DependencyFailed(failed_name.to_string()));
                stack.push(edge.target());
            }
        }
    }
}

/// Resolve artifacts for dispatch: prefer local artifacts from earlier jobs in this run.
fn resolve_dispatch_artifacts(
    config: &mut ContainerConfig,
    _state: &DispatchState,
    git_context: &GitContext,
    gitlab_config: Option<&GitlabConfig>,
) {
    let pipeline_id = git_context.pipeline_id.as_deref();
    match artifacts::resolve_artifacts(
        &config.needs,
        &config.project_path,
        pipeline_id,
        gitlab_config,
    ) {
        Ok(resolution) => {
            config.artifact_dirs = resolution.ready;
            // In dispatch mode, auto-prefer local artifacts (no interactive prompt)
            for conflict in &resolution.conflicts {
                config
                    .artifact_dirs
                    .push((conflict.job_name.clone(), conflict.local_path.clone()));
            }
        }
        Err(e) => {
            eprintln!(
                "\x1b[33mWarning: failed to resolve artifacts for '{}': {}\x1b[0m",
                config.job_name, e
            );
        }
    }
}

/// Sanitize a job name for use as a filename.
fn sanitize_filename(name: &str) -> String {
    name.chars()
        .map(|c| {
            if c.is_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

fn format_duration(d: std::time::Duration) -> String {
    let total_secs = d.as_secs();
    let hours = total_secs / 3600;
    let mins = (total_secs % 3600) / 60;
    let secs = total_secs % 60;
    if hours > 0 {
        format!("{}h{:02}m{:02}s", hours, mins, secs)
    } else if mins > 0 {
        format!("{}m{:02}s", mins, secs)
    } else {
        format!("{}s", secs)
    }
}

fn job_suffix(
    status: &JobStatus,
    skip_reason: Option<&SkipReason>,
    duration: Option<std::time::Duration>,
) -> String {
    let mut s = match (status, skip_reason) {
        (_, Some(SkipReason::UserSkipped)) => " (--skip)".to_string(),
        (JobStatus::Skipped, Some(SkipReason::Manual)) => " \u{23f8}".to_string(),
        (JobStatus::Skipped, Some(SkipReason::Never)) => " (never)".to_string(),
        (JobStatus::Skipped, Some(reason)) => format!(" ({})", reason),
        _ => String::new(),
    };
    if let Some(d) = duration {
        s.push_str(&format!(" ({})", format_duration(d)));
    }
    s
}

/// A stage column: (stage_name, [(job_name, status, skip_reason, duration)]).
type StageColumn<'a> = (
    String,
    Vec<(
        String,
        JobStatus,
        Option<&'a SkipReason>,
        Option<std::time::Duration>,
    )>,
);

// ---------------------------------------------------------------------------
// Ratatui widgets for TTY dispatch display
// ---------------------------------------------------------------------------

fn status_style(status: &JobStatus) -> Style {
    match status {
        JobStatus::Passed => Style::default().fg(Color::Green),
        JobStatus::Failed => Style::default().fg(Color::Red),
        JobStatus::AllowedFailure => Style::default().fg(Color::Yellow),
        JobStatus::Running => Style::default().fg(Color::Cyan),
        JobStatus::Pending => Style::default().fg(Color::DarkGray),
        JobStatus::Skipped => Style::default().fg(Color::Yellow),
    }
}

fn status_icon(status: &JobStatus) -> &'static str {
    match status {
        JobStatus::Passed => "\u{2713}",
        JobStatus::Failed => "\u{2717}",
        JobStatus::AllowedFailure => "\u{26a0}",
        JobStatus::Running => "\u{25cf}",
        JobStatus::Pending => "\u{25cb}",
        JobStatus::Skipped => "-",
    }
}

/// Build stage columns from pipeline + dispatch state.
fn build_stage_columns<'a>(
    pipeline: &'a Pipeline,
    state: &'a DispatchState,
) -> Vec<StageColumn<'a>> {
    let mut columns = Vec::new();
    for stage in &pipeline.stages {
        let mut jobs = Vec::new();
        for job_name in &stage.job_names {
            if let Some(job_state) = state.jobs.get(job_name) {
                jobs.push((
                    job_name.clone(),
                    job_state.status,
                    job_state.skip_reason.as_ref(),
                    job_state.duration,
                ));
            }
        }
        if !jobs.is_empty() {
            columns.push((stage.name.clone(), jobs));
        }
    }
    columns
}

/// Ratatui widget that renders the pipeline status table.
struct PipelineStatusWidget<'a> {
    pipeline: &'a Pipeline,
    state: &'a DispatchState,
    elapsed: Option<Instant>,
}

impl Widget for PipelineStatusWidget<'_> {
    fn render(self, area: Rect, buf: &mut Buffer) {
        let columns = build_stage_columns(self.pipeline, self.state);
        if columns.is_empty() {
            buf.set_string(area.x + 2, area.y, "(no jobs)", Style::default());
            return;
        }

        let col_widths: Vec<usize> = columns
            .iter()
            .map(|(name, jobs)| {
                let max_job = jobs
                    .iter()
                    .map(|(n, status, skip_reason, duration)| {
                        let suffix = job_suffix(status, *skip_reason, *duration);
                        n.width() + 2 + suffix.width()
                    })
                    .max()
                    .unwrap_or(0);
                max_job.max(name.width())
            })
            .collect();

        let mut y = area.y;

        // Stage headers
        let mut x = area.x + 2;
        for (i, (name, _)) in columns.iter().enumerate() {
            let w = col_widths[i];
            buf.set_string(
                x,
                y,
                format!("{:<width$}", name, width = w),
                Style::default().add_modifier(Modifier::BOLD),
            );
            x += w as u16 + 4;
        }
        y += 1;
        if y >= area.y + area.height {
            return;
        }

        // Separator
        x = area.x + 2;
        for (i, (name, _)) in columns.iter().enumerate() {
            let w = col_widths[i];
            let dashes: String = "\u{2500}".repeat(name.width().min(w));
            buf.set_string(
                x,
                y,
                format!("{:<width$}", dashes, width = w),
                Style::default(),
            );
            x += w as u16 + 4;
        }
        y += 1;

        // Job rows
        let max_rows = columns
            .iter()
            .map(|(_, jobs)| jobs.len())
            .max()
            .unwrap_or(0);
        for row in 0..max_rows {
            if y >= area.y + area.height {
                break;
            }
            x = area.x + 2;
            for (i, (_, jobs)) in columns.iter().enumerate() {
                let w = col_widths[i];
                if row < jobs.len() {
                    let (ref name, ref status, ref skip_reason, ref duration) = jobs[row];
                    let icon = status_icon(status);
                    let suffix = job_suffix(status, *skip_reason, *duration);
                    let cell = format!("{} {}{}", icon, name, suffix);
                    let display_len = name.width() + 2 + suffix.width();
                    let padding = w.saturating_sub(display_len);
                    let padded = format!("{}{}", cell, " ".repeat(padding));
                    buf.set_string(x, y, &padded, status_style(status));
                }
                x += w as u16 + 4;
            }
            y += 1;
        }

        // Progress line
        if let Some(start) = self.elapsed {
            if y < area.y + area.height {
                y += 1; // blank line
            }
            if y < area.y + area.height {
                let total = self.state.jobs.len();
                let completed = self
                    .state
                    .jobs
                    .values()
                    .filter(|s| {
                        matches!(
                            s.status,
                            JobStatus::Passed | JobStatus::Failed | JobStatus::AllowedFailure
                        )
                    })
                    .count();
                let skipped = self
                    .state
                    .jobs
                    .values()
                    .filter(|s| s.status == JobStatus::Skipped)
                    .count();
                let running: Vec<&str> = self
                    .state
                    .jobs
                    .iter()
                    .filter(|(_, s)| s.status == JobStatus::Running)
                    .map(|(name, _)| name.as_str())
                    .collect();

                let eligible = total - skipped;
                let mut progress = format!("{}/{}", completed, eligible);
                if !running.is_empty() {
                    progress.push_str(&format!(" | running: {}", running.join(", ")));
                }
                progress.push_str(&format!(" | {}", format_duration(start.elapsed())));
                buf.set_string(
                    area.x + 2,
                    y,
                    &progress,
                    Style::default().fg(Color::DarkGray),
                );
            }
        }
    }
}

/// Compute how many rows the pipeline status widget needs.
fn pipeline_status_height(pipeline: &Pipeline, state: &DispatchState) -> u16 {
    let columns = build_stage_columns(pipeline, state);
    let max_rows = columns
        .iter()
        .map(|(_, jobs)| jobs.len())
        .max()
        .unwrap_or(0);
    // stage header + separator + job rows + blank + progress
    (2 + max_rows + 2) as u16
}

/// Manages dispatch output: inline progress on dumb terminals.
struct DispatchDisplay {
    header_lines: usize,
    writer: Box<dyn std::io::Write>,
}

impl DispatchDisplay {
    fn new() -> Self {
        Self {
            header_lines: 0,
            writer: Box::new(io::stdout()),
        }
    }

    #[cfg(test)]
    fn new_test(writer: Box<dyn io::Write>) -> Self {
        Self {
            header_lines: 0,
            writer,
        }
    }

    fn show_initial_status(&mut self, pipeline: &Pipeline, state: &DispatchState) {
        let lines = render_status_lines(pipeline, state, None);
        self.header_lines = lines.len();

        for line in &lines {
            writeln!(self.writer, "{}", line).ok();
        }
    }

    fn update_status(&mut self, state: &DispatchState, start: Instant) {
        self.print_progress(state, start);
    }

    fn job_starting(&mut self, name: &str, runner: Option<&RunnerConfig>) {
        if let Some(r) = runner {
            writeln!(
                self.writer,
                "\n\x1b[36m━━━ {} → runner '{}' ({}) ━━━\x1b[0m\n",
                name,
                r.name,
                r.host.as_deref().unwrap_or("local"),
            )
            .ok();
        } else {
            writeln!(self.writer, "\n\x1b[36m━━━ {} ━━━\x1b[0m\n", name).ok();
        }
    }

    fn job_passed(&mut self, name: &str, duration: std::time::Duration) {
        writeln!(
            self.writer,
            "\n\x1b[32m✓ {} passed ({})\x1b[0m",
            name,
            format_duration(duration)
        )
        .ok();
    }

    fn job_failed(&mut self, name: &str, exit_code: i32, duration: std::time::Duration) {
        writeln!(
            self.writer,
            "\n\x1b[31m✗ {} failed (exit {}, {})\x1b[0m",
            name,
            exit_code,
            format_duration(duration)
        )
        .ok();
    }

    fn job_allowed_failure(&mut self, name: &str, exit_code: i32, duration: std::time::Duration) {
        writeln!(
            self.writer,
            "\n\x1b[33m⚠ {} failed (allowed, exit {}, {})\x1b[0m",
            name,
            exit_code,
            format_duration(duration)
        )
        .ok();
    }

    fn job_error(&mut self, name: &str, error: &anyhow::Error) {
        writeln!(self.writer, "\n\x1b[31m✗ {} error: {}\x1b[0m", name, error).ok();
    }

    fn warning(&mut self, msg: &str) {
        writeln!(self.writer, "\x1b[33mWarning: {}\x1b[0m", msg).ok();
    }

    fn print_progress(&mut self, state: &DispatchState, start: std::time::Instant) {
        let total = state.jobs.len();
        let completed = state
            .jobs
            .values()
            .filter(|s| {
                matches!(
                    s.status,
                    JobStatus::Passed | JobStatus::Failed | JobStatus::AllowedFailure
                )
            })
            .count();
        let skipped = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::Skipped)
            .count();
        let running: Vec<&str> = state
            .jobs
            .iter()
            .filter(|(_, s)| s.status == JobStatus::Running)
            .map(|(name, _)| name.as_str())
            .collect();

        let eligible = total - skipped;
        let mut line = format!("── {}/{}", completed, eligible);
        if !running.is_empty() {
            line.push_str(&format!(" | running: {}", running.join(", ")));
        }
        line.push_str(&format!(" | {}", format_duration(start.elapsed())));
        line.push_str(" ──");
        writeln!(self.writer, "{}", line).ok();
    }

    fn print_summary(&mut self, state: &DispatchState, elapsed: std::time::Duration) {
        writeln!(self.writer, "\n\x1b[1m──── Summary ────\x1b[0m\n").ok();

        let name_width = state.jobs.keys().map(|n| n.len()).max().unwrap_or(10);
        let dur_width = state
            .jobs
            .values()
            .filter_map(|s| s.duration.map(|d| format_duration(d).len()))
            .max()
            .unwrap_or(0);

        for (name, job_state) in &state.jobs {
            let (icon, color) = match job_state.status {
                JobStatus::Passed => ("✓", "\x1b[32m"),
                JobStatus::Failed => ("✗", "\x1b[31m"),
                JobStatus::AllowedFailure => ("⚠", "\x1b[33m"),
                JobStatus::Skipped => ("-", "\x1b[33m"),
                _ => ("?", "\x1b[90m"),
            };

            let duration = job_state.duration.map(format_duration).unwrap_or_default();

            let log = job_state
                .log_path
                .as_ref()
                .map(|p| p.display().to_string())
                .unwrap_or_default();

            let detail = match &job_state.skip_reason {
                Some(reason) => format!("({})", reason),
                None => {
                    if let Some(code) = job_state.exit_code {
                        if code != 0 {
                            format!("(exit {})", code)
                        } else {
                            String::new()
                        }
                    } else {
                        String::new()
                    }
                }
            };

            writeln!(
                self.writer,
                "  {}{} {:<name_w$}\x1b[0m  {:>dur_w$}  {}  {}",
                color,
                icon,
                name,
                duration,
                log,
                detail,
                name_w = name_width,
                dur_w = dur_width,
            )
            .ok();
        }

        let passed = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::Passed)
            .count();
        let failed = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::Failed)
            .count();
        let warnings = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::AllowedFailure)
            .count();
        let skipped = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::Skipped)
            .count();

        writeln!(self.writer).ok();
        if failed > 0 {
            writeln!(
                self.writer,
                "\x1b[31m  {} passed, {} failed, {} warnings, {} skipped\x1b[0m",
                passed, failed, warnings, skipped
            )
            .ok();
        } else if warnings > 0 {
            writeln!(
                self.writer,
                "\x1b[33m  {} passed, {} warnings, {} skipped\x1b[0m",
                passed, warnings, skipped
            )
            .ok();
        } else {
            writeln!(
                self.writer,
                "\x1b[32m  {} passed, {} skipped\x1b[0m",
                passed, skipped
            )
            .ok();
        }

        writeln!(self.writer, "  Elapsed: {}", format_duration(elapsed)).ok();
    }
}

/// Render the status header as lines (no I/O).
fn render_status_lines(
    pipeline: &Pipeline,
    state: &DispatchState,
    elapsed: Option<std::time::Instant>,
) -> Vec<String> {
    let mut out = Vec::new();
    out.push("\x1b[1m──── Pipeline Status ────\x1b[0m".to_string());
    out.push(String::new());

    let mut stage_columns: Vec<StageColumn> = Vec::new();
    for stage in &pipeline.stages {
        let mut jobs = Vec::new();
        for job_name in &stage.job_names {
            if let Some(job_state) = state.jobs.get(job_name) {
                jobs.push((
                    job_name.clone(),
                    job_state.status,
                    job_state.skip_reason.as_ref(),
                    job_state.duration,
                ));
            }
        }
        if !jobs.is_empty() {
            stage_columns.push((stage.name.clone(), jobs));
        }
    }

    if stage_columns.is_empty() {
        out.push("  (no jobs)".to_string());
        return out;
    }

    let col_widths: Vec<usize> = stage_columns
        .iter()
        .map(|(name, jobs)| {
            let max_job = jobs
                .iter()
                .map(|(n, status, skip_reason, duration)| {
                    let suffix = job_suffix(status, *skip_reason, *duration);
                    n.width() + 2 + suffix.width()
                })
                .max()
                .unwrap_or(0);
            max_job.max(name.width())
        })
        .collect();

    // Stage headers
    let mut header_line = String::from("  ");
    for (i, (name, _)) in stage_columns.iter().enumerate() {
        let width = col_widths[i];
        header_line.push_str(&format!("{:<width$}", name, width = width));
        if i < stage_columns.len() - 1 {
            header_line.push_str("    ");
        }
    }
    out.push(format!("\x1b[1m{}\x1b[0m", header_line));

    // Separator
    let mut sep_line = String::from("  ");
    for (i, (name, _)) in stage_columns.iter().enumerate() {
        let width = col_widths[i];
        let dashes: String = "─".repeat(name.width().min(width));
        sep_line.push_str(&format!("{:<width$}", dashes, width = width));
        if i < stage_columns.len() - 1 {
            sep_line.push_str("    ");
        }
    }
    out.push(sep_line);

    // Job rows
    let max_rows = stage_columns
        .iter()
        .map(|(_, jobs)| jobs.len())
        .max()
        .unwrap_or(0);

    for row in 0..max_rows {
        let mut line = String::from("  ");
        for (i, (_, jobs)) in stage_columns.iter().enumerate() {
            let width = col_widths[i];
            if row < jobs.len() {
                let (name, status, skip_reason, duration) = &jobs[row];
                let (icon, color) = match status {
                    JobStatus::Passed => ("✓", "\x1b[32m"),
                    JobStatus::Failed => ("✗", "\x1b[31m"),
                    JobStatus::AllowedFailure => ("⚠", "\x1b[33m"),
                    JobStatus::Running => ("●", "\x1b[36m"),
                    JobStatus::Pending => ("○", "\x1b[90m"),
                    JobStatus::Skipped => ("-", "\x1b[33m"),
                };

                let suffix = job_suffix(status, *skip_reason, *duration);
                let cell = format!("{} {}{}", icon, name, suffix);
                let display_len = name.width() + 2 + suffix.width();
                let padding = if display_len < width {
                    " ".repeat(width - display_len)
                } else {
                    String::new()
                };
                line.push_str(&format!("{}{}{}\x1b[0m", color, cell, padding));
            } else {
                line.push_str(&" ".repeat(width));
            }
            if i < stage_columns.len() - 1 {
                line.push_str("    ");
            }
        }
        out.push(line);
    }

    // Progress line
    if let Some(start) = elapsed {
        out.push(String::new());
        let total = state.jobs.len();
        let completed = state
            .jobs
            .values()
            .filter(|s| {
                matches!(
                    s.status,
                    JobStatus::Passed | JobStatus::Failed | JobStatus::AllowedFailure
                )
            })
            .count();
        let skipped = state
            .jobs
            .values()
            .filter(|s| s.status == JobStatus::Skipped)
            .count();
        let running: Vec<&str> = state
            .jobs
            .iter()
            .filter(|(_, s)| s.status == JobStatus::Running)
            .map(|(name, _)| name.as_str())
            .collect();

        let eligible = total - skipped;
        let mut progress = format!("  {}/{}", completed, eligible);
        if !running.is_empty() {
            progress.push_str(&format!(" | running: {}", running.join(", ")));
        }
        progress.push_str(&format!(" | {}", format_duration(start.elapsed())));
        out.push(progress);
    }

    out
}

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

    fn make_test_job(name: &str, stage: &str) -> crate::model::Job {
        crate::model::Job {
            name: name.to_string(),
            stage: stage.to_string(),
            image: None,
            services: vec![],
            variables: IndexMap::new(),
            before_script: vec![],
            script: vec![],
            after_script: vec![],
            needs: vec![],
            tags: vec![],
            when_val: "on_success".to_string(),
            allow_failure: false,
            rules_never: false,
            unresolved_extends: vec![],
            rules_summary: vec![],
            artifacts_paths: vec![],
            artifacts_exclude: vec![],
            artifacts_when: "on_success".to_string(),
            timeout: None,
            resource_group: None,
            coverage: None,
            resolved_yaml: String::new(),
            trigger_file: None,
            steps: vec![],
            debug_info: crate::model::JobDebugInfo::default(),
        }
    }

    fn make_test_pipeline(stages: &[(&str, &[&str])]) -> (Pipeline, Vec<String>) {
        use crate::model::{Pipeline, Stage};

        let mut pipeline_jobs = IndexMap::new();
        let mut pipeline_stages = Vec::new();
        let mut topo = Vec::new();

        for (stage_name, job_names) in stages {
            let mut stage_job_names = Vec::new();
            for &name in *job_names {
                pipeline_jobs.insert(name.to_string(), make_test_job(name, stage_name));
                stage_job_names.push(name.to_string());
                topo.push(name.to_string());
            }
            pipeline_stages.push(Stage {
                name: stage_name.to_string(),
                job_names: stage_job_names,
            });
        }

        let pipeline = Pipeline {
            stages: pipeline_stages,
            jobs: pipeline_jobs,
        };
        (pipeline, topo)
    }

    #[test]
    fn test_render_status_lines_contains_stages_and_jobs() {
        let (pipeline, topo) =
            make_test_pipeline(&[("compile", &["build"]), ("verify", &["test"])]);

        let mut state = DispatchState::new(&topo);
        state.jobs["build"].status = JobStatus::Passed;

        let lines = render_status_lines(&pipeline, &state, None);
        let text = lines.join("\n");

        assert!(text.contains("compile"));
        assert!(text.contains("verify"));
        assert!(text.contains("build"));
        assert!(text.contains("test"));
        assert!(!text.contains("\x1b[2J"));
    }

    #[test]
    fn test_render_status_lines_with_elapsed() {
        let (pipeline, topo) = make_test_pipeline(&[("s", &["a"])]);

        let mut state = DispatchState::new(&topo);
        state.jobs["a"].status = JobStatus::Running;

        let start = std::time::Instant::now();
        let lines = render_status_lines(&pipeline, &state, Some(start));
        let text = lines.join("\n");

        assert!(text.contains("running: a"));
    }

    #[test]
    fn test_display_writer_captures_output() {
        use std::sync::{Arc, Mutex};

        let buf = Arc::new(Mutex::new(Vec::<u8>::new()));
        let writer = Box::new(WriterAdapter(buf.clone()));
        let mut display = DispatchDisplay::new_test(writer);

        let (pipeline, topo) = make_test_pipeline(&[("build", &["compile"]), ("test", &["unit"])]);
        let state = DispatchState::new(&topo);

        display.show_initial_status(&pipeline, &state);

        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(output.contains("build"));
        assert!(output.contains("test"));
        assert!(output.contains("compile"));
        assert!(output.contains("unit"));
        // Non-TTY: no ANSI scrolling region escape
        assert!(!output.contains("\x1b[2J"));
    }

    #[test]
    fn test_display_job_lifecycle() {
        use std::sync::{Arc, Mutex};

        let buf = Arc::new(Mutex::new(Vec::<u8>::new()));
        let writer = Box::new(WriterAdapter(buf.clone()));
        let mut display = DispatchDisplay::new_test(writer);

        display.job_starting("build", None);
        display.job_passed("build", std::time::Duration::from_secs(90));

        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(output.contains("━━━ build ━━━"));
        assert!(output.contains("✓ build passed (1m30s)"));
    }

    #[test]
    fn test_display_summary_includes_log_paths() {
        use std::sync::{Arc, Mutex};

        let buf = Arc::new(Mutex::new(Vec::<u8>::new()));
        let writer = Box::new(WriterAdapter(buf.clone()));
        let mut display = DispatchDisplay::new_test(writer);

        let topo = vec!["build".to_string()];
        let mut state = DispatchState::new(&topo);
        state.jobs["build"].status = JobStatus::Passed;
        state.jobs["build"].duration = Some(std::time::Duration::from_secs(5));
        state.jobs["build"].log_path = Some(PathBuf::from("/tmp/build.log"));

        display.print_summary(&state, std::time::Duration::from_secs(42));

        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(output.contains("Summary"));
        assert!(output.contains("/tmp/build.log"));
        assert!(output.contains("1 passed"));
    }

    /// Adapter to make Arc<Mutex<Vec<u8>>> implement Write.
    struct WriterAdapter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);

    impl std::io::Write for WriterAdapter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().write(buf)
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    #[test]
    fn test_format_duration() {
        use std::time::Duration;
        assert_eq!(format_duration(Duration::from_secs(0)), "0s");
        assert_eq!(format_duration(Duration::from_secs(5)), "5s");
        assert_eq!(format_duration(Duration::from_secs(59)), "59s");
        assert_eq!(format_duration(Duration::from_secs(60)), "1m00s");
        assert_eq!(format_duration(Duration::from_secs(90)), "1m30s");
        assert_eq!(format_duration(Duration::from_secs(3599)), "59m59s");
        assert_eq!(format_duration(Duration::from_secs(3600)), "1h00m00s");
        assert_eq!(format_duration(Duration::from_secs(3661)), "1h01m01s");
        assert_eq!(format_duration(Duration::from_secs(7384)), "2h03m04s");
    }

    #[test]
    fn test_tag_filter_no_filter() {
        let filter = TagFilter {
            no_tags: false,
            tags: vec![],
        };
        // No filtering: all jobs eligible
        assert!(filter.check(&[]).is_none());
        assert!(filter.check(&["kvm".to_string()]).is_none());
    }

    #[test]
    fn test_tag_filter_no_tags_only() {
        let filter = TagFilter {
            no_tags: true,
            tags: vec![],
        };
        // Untagged: eligible
        assert!(filter.check(&[]).is_none());
        // Tagged: not eligible
        assert!(filter.check(&["kvm".to_string()]).is_some());
    }

    #[test]
    fn test_tag_filter_tags_only() {
        let filter = TagFilter {
            no_tags: false,
            tags: vec!["kvm".to_string(), "x86_64".to_string()],
        };
        // Untagged: not eligible
        assert!(filter.check(&[]).is_some());
        // Matching tags: eligible
        assert!(filter.check(&["kvm".to_string()]).is_none());
        assert!(
            filter
                .check(&["kvm".to_string(), "x86_64".to_string()])
                .is_none()
        );
        // Non-matching: not eligible
        assert!(filter.check(&["aarch64".to_string()]).is_some());
        // Partial match: not eligible (aarch64 is not in filter)
        assert!(
            filter
                .check(&["kvm".to_string(), "aarch64".to_string()])
                .is_some()
        );
    }

    #[test]
    fn test_tag_filter_both() {
        let filter = TagFilter {
            no_tags: true,
            tags: vec!["kvm".to_string()],
        };
        // Untagged: eligible (--no-tags)
        assert!(filter.check(&[]).is_none());
        // Matching tags: eligible
        assert!(filter.check(&["kvm".to_string()]).is_none());
        // Non-matching: not eligible
        assert!(filter.check(&["aarch64".to_string()]).is_some());
    }

    #[test]
    fn test_tag_filter_case_insensitive() {
        let filter = TagFilter {
            no_tags: false,
            tags: vec!["KVM".to_string()],
        };
        assert!(filter.check(&["kvm".to_string()]).is_none());
        assert!(filter.check(&["Kvm".to_string()]).is_none());
    }

    #[test]
    fn test_sanitize_filename() {
        assert_eq!(sanitize_filename("build"), "build");
        assert_eq!(sanitize_filename("build-test"), "build-test");
        assert_eq!(
            sanitize_filename("build [os: ubuntu]"),
            "build__os__ubuntu_"
        );
    }

    #[test]
    fn test_blocks_stage_dependents() {
        // Manual and Never should not block downstream via stage ordering
        assert!(!SkipReason::Manual.blocks_stage_dependents());
        assert!(!SkipReason::Never.blocks_stage_dependents());

        // All other skip reasons should block
        assert!(SkipReason::NoImage.blocks_stage_dependents());
        assert!(SkipReason::TagMismatch(vec!["kvm".into()]).blocks_stage_dependents());
        assert!(SkipReason::DependencyFailed("build".into()).blocks_stage_dependents());
        assert!(SkipReason::UserSkipped.blocks_stage_dependents());
    }

    /// Build a DispatchState + petgraph DAG for skip propagation tests.
    ///
    /// Layout: stage1=[build, manual_job], stage2=[test]
    /// Edges: build->test (Stage), manual_job->test (Stage)
    fn make_stage_propagation_setup(
        skip_reason: SkipReason,
    ) -> (
        petgraph::graph::DiGraph<String, graph::DepKind>,
        IndexMap<String, petgraph::graph::NodeIndex>,
        DispatchState,
        Vec<String>,
    ) {
        let mut dag = petgraph::graph::DiGraph::new();
        let build = dag.add_node("build".to_string());
        let manual = dag.add_node("manual_job".to_string());
        let test = dag.add_node("test".to_string());

        // Implicit stage ordering edges
        dag.add_edge(build, test, graph::DepKind::Stage);
        dag.add_edge(manual, test, graph::DepKind::Stage);

        let mut node_indices = IndexMap::new();
        node_indices.insert("build".to_string(), build);
        node_indices.insert("manual_job".to_string(), manual);
        node_indices.insert("test".to_string(), test);

        let topo_names = vec![
            "build".to_string(),
            "manual_job".to_string(),
            "test".to_string(),
        ];
        let mut state = DispatchState::new(&topo_names);

        // build passed
        state.jobs["build"].status = JobStatus::Passed;
        // manual_job skipped with the given reason
        state.jobs["manual_job"].status = JobStatus::Skipped;
        state.jobs["manual_job"].skip_reason = Some(skip_reason);

        (dag, node_indices, state, topo_names)
    }

    #[test]
    fn test_manual_skip_does_not_propagate_via_stage() {
        let (dag, _, mut state, topo_names) = make_stage_propagation_setup(SkipReason::Manual);

        // Run the pre-scan propagation loop
        for name in &topo_names {
            if state.jobs[name].status != JobStatus::Skipped {
                continue;
            }
            let blocks_stage = state.jobs[name]
                .skip_reason
                .as_ref()
                .is_some_and(|r| r.blocks_stage_dependents());
            if let Some(&node_idx) = node_indices_from_topo(&topo_names, &dag).get(name) {
                for edge in dag.edges_directed(node_idx, petgraph::Direction::Outgoing) {
                    if *edge.weight() == graph::DepKind::Stage && !blocks_stage {
                        continue;
                    }
                    let dep_name = &dag[edge.target()];
                    if state.jobs[dep_name].status == JobStatus::Pending {
                        state.jobs[dep_name].status = JobStatus::Skipped;
                        state.jobs[dep_name].skip_reason =
                            Some(SkipReason::DependencyFailed(name.clone()));
                    }
                }
            }
        }

        // "test" should still be Pending — manual skip doesn't propagate via Stage
        assert_eq!(state.jobs["test"].status, JobStatus::Pending);
    }

    #[test]
    fn test_never_skip_does_not_propagate_via_stage() {
        let (dag, _, mut state, topo_names) = make_stage_propagation_setup(SkipReason::Never);

        for name in &topo_names {
            if state.jobs[name].status != JobStatus::Skipped {
                continue;
            }
            let blocks_stage = state.jobs[name]
                .skip_reason
                .as_ref()
                .is_some_and(|r| r.blocks_stage_dependents());
            if let Some(&node_idx) = node_indices_from_topo(&topo_names, &dag).get(name) {
                for edge in dag.edges_directed(node_idx, petgraph::Direction::Outgoing) {
                    if *edge.weight() == graph::DepKind::Stage && !blocks_stage {
                        continue;
                    }
                    let dep_name = &dag[edge.target()];
                    if state.jobs[dep_name].status == JobStatus::Pending {
                        state.jobs[dep_name].status = JobStatus::Skipped;
                        state.jobs[dep_name].skip_reason =
                            Some(SkipReason::DependencyFailed(name.clone()));
                    }
                }
            }
        }

        assert_eq!(state.jobs["test"].status, JobStatus::Pending);
    }

    #[test]
    fn test_no_image_skip_propagates_via_stage() {
        let (dag, _, mut state, topo_names) = make_stage_propagation_setup(SkipReason::NoImage);

        for name in &topo_names {
            if state.jobs[name].status != JobStatus::Skipped {
                continue;
            }
            let blocks_stage = state.jobs[name]
                .skip_reason
                .as_ref()
                .is_some_and(|r| r.blocks_stage_dependents());
            if let Some(&node_idx) = node_indices_from_topo(&topo_names, &dag).get(name) {
                for edge in dag.edges_directed(node_idx, petgraph::Direction::Outgoing) {
                    if *edge.weight() == graph::DepKind::Stage && !blocks_stage {
                        continue;
                    }
                    let dep_name = &dag[edge.target()];
                    if state.jobs[dep_name].status == JobStatus::Pending {
                        state.jobs[dep_name].status = JobStatus::Skipped;
                        state.jobs[dep_name].skip_reason =
                            Some(SkipReason::DependencyFailed(name.clone()));
                    }
                }
            }
        }

        // NoImage SHOULD propagate — test gets skipped
        assert_eq!(state.jobs["test"].status, JobStatus::Skipped);
    }

    #[test]
    fn test_manual_skip_propagates_via_needs_edge() {
        // If job B explicitly `needs: [manual_job]`, the manual skip should block B
        let mut dag = petgraph::graph::DiGraph::new();
        let manual = dag.add_node("manual_job".to_string());
        let deploy = dag.add_node("deploy".to_string());

        // Explicit needs edge
        dag.add_edge(manual, deploy, graph::DepKind::Needs);

        let topo_names = vec!["manual_job".to_string(), "deploy".to_string()];
        let mut state = DispatchState::new(&topo_names);

        state.jobs["manual_job"].status = JobStatus::Skipped;
        state.jobs["manual_job"].skip_reason = Some(SkipReason::Manual);

        for name in &topo_names {
            if state.jobs[name].status != JobStatus::Skipped {
                continue;
            }
            let blocks_stage = state.jobs[name]
                .skip_reason
                .as_ref()
                .is_some_and(|r| r.blocks_stage_dependents());
            if let Some(&node_idx) = node_indices_from_topo(&topo_names, &dag).get(name) {
                for edge in dag.edges_directed(node_idx, petgraph::Direction::Outgoing) {
                    if *edge.weight() == graph::DepKind::Stage && !blocks_stage {
                        continue;
                    }
                    let dep_name = &dag[edge.target()];
                    if state.jobs[dep_name].status == JobStatus::Pending {
                        state.jobs[dep_name].status = JobStatus::Skipped;
                        state.jobs[dep_name].skip_reason =
                            Some(SkipReason::DependencyFailed(name.clone()));
                    }
                }
            }
        }

        // Explicit needs: manual skip SHOULD propagate
        assert_eq!(state.jobs["deploy"].status, JobStatus::Skipped);
        assert!(matches!(
            state.jobs["deploy"].skip_reason,
            Some(SkipReason::DependencyFailed(_))
        ));
    }

    /// Helper to build node_indices map from topo_names and dag.
    fn node_indices_from_topo(
        topo_names: &[String],
        dag: &petgraph::graph::DiGraph<String, graph::DepKind>,
    ) -> IndexMap<String, petgraph::graph::NodeIndex> {
        let mut map = IndexMap::new();
        for idx in dag.node_indices() {
            map.insert(dag[idx].clone(), idx);
        }
        let _ = topo_names; // used for API consistency
        map
    }

    fn make_runner(name: &str, tags: &[&str], no_tags: bool) -> RunnerConfig {
        RunnerConfig {
            name: name.to_string(),
            host: Some("host.local".to_string()),
            tags: tags.iter().map(|s| s.to_string()).collect(),
            no_tags: if no_tags { Some(true) } else { None },
            ..Default::default()
        }
    }

    #[test]
    fn test_find_runner_tagged_job_matches() {
        let runners = vec![make_runner("r1", &["kvm", "x86_64"], false)];
        // Job tags are a subset of runner tags -> match
        let job_tags = vec!["kvm".to_string()];
        assert_eq!(find_runner(&runners, &job_tags).unwrap().name, "r1");
    }

    #[test]
    fn test_find_runner_tagged_job_all_tags() {
        let runners = vec![make_runner("r1", &["kvm", "x86_64"], false)];
        let job_tags = vec!["kvm".to_string(), "x86_64".to_string()];
        assert_eq!(find_runner(&runners, &job_tags).unwrap().name, "r1");
    }

    #[test]
    fn test_find_runner_tagged_job_no_match() {
        let runners = vec![make_runner("r1", &["kvm"], false)];
        // Job has a tag not in runner -> no match
        let job_tags = vec!["kvm".to_string(), "aarch64".to_string()];
        assert!(find_runner(&runners, &job_tags).is_none());
    }

    #[test]
    fn test_find_runner_untagged_job_no_match() {
        let runners = vec![make_runner("r1", &["kvm"], false)];
        assert!(find_runner(&runners, &[]).is_none());
    }

    #[test]
    fn test_find_runner_untagged_job_with_no_tags() {
        let runners = vec![make_runner("r1", &["kvm"], true)];
        assert_eq!(find_runner(&runners, &[]).unwrap().name, "r1");
    }

    #[test]
    fn test_find_runner_first_wins() {
        let runners = vec![
            make_runner("r1", &["kvm"], false),
            make_runner("r2", &["kvm", "x86_64"], false),
        ];
        let job_tags = vec!["kvm".to_string()];
        assert_eq!(find_runner(&runners, &job_tags).unwrap().name, "r1");
    }

    #[test]
    fn test_find_runner_case_insensitive() {
        let runners = vec![make_runner("r1", &["KVM", "X86_64"], false)];
        let job_tags = vec!["kvm".to_string(), "x86_64".to_string()];
        assert_eq!(find_runner(&runners, &job_tags).unwrap().name, "r1");
    }

    #[test]
    fn test_find_runner_empty_runners() {
        let job_tags = vec!["kvm".to_string()];
        assert!(find_runner(&[], &job_tags).is_none());
    }

    #[test]
    fn test_extend_tag_filter_with_runners() {
        let mut filter = TagFilter {
            no_tags: false,
            tags: vec![],
        };
        let runners = vec![
            make_runner("r1", &["kvm", "x86_64"], false),
            make_runner("r2", &["macos"], true),
        ];
        extend_tag_filter_with_runners(&mut filter, &runners);

        // no_tags should be set because r2 has no_tags
        assert!(filter.no_tags);
        // All unique runner tags should be present
        assert_eq!(filter.tags.len(), 3);
        assert!(filter.tags.iter().any(|t| t == "kvm"));
        assert!(filter.tags.iter().any(|t| t == "x86_64"));
        assert!(filter.tags.iter().any(|t| t == "macos"));
    }

    #[test]
    fn test_extend_tag_filter_no_duplicates() {
        let mut filter = TagFilter {
            no_tags: false,
            tags: vec!["kvm".to_string()],
        };
        let runners = vec![make_runner("r1", &["KVM", "x86_64"], false)];
        extend_tag_filter_with_runners(&mut filter, &runners);

        // "kvm" should not be duplicated (case-insensitive)
        assert_eq!(filter.tags.len(), 2);
    }

    fn make_test_job_with_image(name: &str, stage: &str) -> crate::model::Job {
        let mut job = make_test_job(name, stage);
        job.image = Some("alpine:latest".to_string());
        job
    }

    fn make_imaged_pipeline(stages: &[(&str, &[&str])]) -> Pipeline {
        use crate::model::Stage;

        let mut pipeline_jobs = IndexMap::new();
        let mut pipeline_stages = Vec::new();

        for (stage_name, job_names) in stages {
            let mut stage_job_names = Vec::new();
            for &name in *job_names {
                pipeline_jobs.insert(name.to_string(), make_test_job_with_image(name, stage_name));
                stage_job_names.push(name.to_string());
            }
            pipeline_stages.push(Stage {
                name: stage_name.to_string(),
                job_names: stage_job_names,
            });
        }

        Pipeline {
            stages: pipeline_stages,
            jobs: pipeline_jobs,
        }
    }

    fn no_filter() -> TagFilter {
        TagFilter {
            no_tags: false,
            tags: vec![],
        }
    }

    #[test]
    fn test_scheduler_ready_jobs_linear() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        let ready = scheduler.ready_jobs();
        assert_eq!(ready, vec!["compile"]);
    }

    #[test]
    fn test_scheduler_ready_jobs_parallel_stage() {
        let pipeline = make_imaged_pipeline(&[("test", &["unit", "lint"])]);
        let scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        let ready = scheduler.ready_jobs();
        assert_eq!(ready.len(), 2);
        assert!(ready.contains(&"unit"));
        assert!(ready.contains(&"lint"));
    }

    #[test]
    fn test_scheduler_complete_unlocks_next_stage() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let mut scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        assert_eq!(scheduler.ready_jobs(), vec!["compile"]);
        assert!(!scheduler.ready_jobs().contains(&"verify"));

        scheduler.mark_running("compile");
        assert!(scheduler.ready_jobs().is_empty());

        scheduler.complete(
            "compile",
            0,
            std::time::Duration::from_secs(1),
            PathBuf::from("/tmp/compile.log"),
            false,
        );

        assert_eq!(scheduler.ready_jobs(), vec!["verify"]);
    }

    #[test]
    fn test_scheduler_failure_propagates() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let mut scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        scheduler.mark_running("compile");
        scheduler.complete(
            "compile",
            1,
            std::time::Duration::from_secs(1),
            PathBuf::from("/tmp/compile.log"),
            false,
        );

        // verify should be skipped due to dependency failure
        assert!(scheduler.ready_jobs().is_empty());
        assert_eq!(scheduler.state().jobs["verify"].status, JobStatus::Skipped);
    }

    #[test]
    fn test_scheduler_allow_failure_does_not_block() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let mut scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        scheduler.mark_running("compile");
        scheduler.complete(
            "compile",
            1,
            std::time::Duration::from_secs(1),
            PathBuf::from("/tmp/compile.log"),
            true,
        );

        // allow_failure: verify should still be ready
        assert_eq!(scheduler.ready_jobs(), vec!["verify"]);
    }

    #[test]
    fn test_scheduler_skip_jobs() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let scheduler = Scheduler::new(
            &pipeline,
            CiSystem::GitLab,
            &no_filter(),
            &["compile".to_string()],
        )
        .unwrap();

        // compile is user-skipped (treated as passed), so verify is ready
        assert_eq!(scheduler.state().jobs["compile"].status, JobStatus::Passed);
        assert_eq!(scheduler.ready_jobs(), vec!["verify"]);
    }

    #[test]
    fn test_scheduler_all_passed() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"])]);
        let mut scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        assert!(!scheduler.all_passed());

        scheduler.mark_running("compile");
        scheduler.complete(
            "compile",
            0,
            std::time::Duration::from_secs(1),
            PathBuf::from("/tmp/compile.log"),
            false,
        );

        assert!(scheduler.all_passed());
    }

    #[test]
    fn test_scheduler_fail_error() {
        let pipeline = make_imaged_pipeline(&[("build", &["compile"]), ("test", &["verify"])]);
        let mut scheduler = Scheduler::new(&pipeline, CiSystem::GitLab, &no_filter(), &[]).unwrap();

        scheduler.mark_running("compile");
        scheduler.fail_error("compile");

        assert_eq!(scheduler.state().jobs["compile"].status, JobStatus::Failed);
        assert!(scheduler.ready_jobs().is_empty());
        assert!(!scheduler.all_passed());
    }
}
