//! Sequential pipeline executor.
//!
//! Runs a loaded [`Pipeline`] step by step: resolve each step's references
//! against earlier output, evaluate its `when:` guard, call the tool, store
//! the result, and apply the `on_error` policy on failure. Budget gating sits
//! in the same loop so a cumulative cost or token cap stops the run before the
//! step that would breach it.
//!
//! The executor is transport-agnostic above [`Connection`]: a real run calls
//! tools over HTTP; a `--dry-run` never connects and instead reports the plan
//! and a cost projection.

use std::collections::HashMap;
use std::time::Instant;

use serde_json::Value;

use super::budget::{BreachKind, BudgetTracker, OnBudgetExceeded, StepCost};
use super::conn::Connection;
use super::model::{OnError, Pipeline, Step};
use super::pricing::PricingTable;
use super::reference::{evaluate_when, resolve_args, ResolveContext};

/// Outcome of a single executed (or skipped) step.
#[derive(Debug, Clone)]
pub struct StepOutcome {
    /// Step id.
    pub id: String,
    /// Tool name.
    pub tool: String,
    /// What happened.
    pub status: StepStatus,
    /// The resolved output, when the step ran and succeeded.
    pub output: Option<Value>,
    /// Cost/token/timing record fed to the budget ledger.
    pub cost: StepCost,
}

/// The disposition of a step after the executor handled it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StepStatus {
    /// Ran and the tool returned a result.
    Ok,
    /// `when:` guard was false, so the step did not run.
    Skipped,
    /// Ran (possibly after retries) and failed, but `on_error: continue`
    /// let the pipeline carry on.
    FailedContinued(String),
}

/// The full result of a pipeline run.
#[derive(Debug)]
pub struct RunResult {
    /// Per-step outcomes in execution order.
    pub steps: Vec<StepOutcome>,
    /// The final budget ledger.
    pub budget: BudgetTracker,
    /// The last successful step's output, surfaced to stdout.
    pub last_output: Option<Value>,
    /// A breach that aborted the run (fail-fast) or was flagged at the end.
    pub breach: Option<(String, BreachKind)>,
}

impl RunResult {
    /// Whether the run should exit non-zero.
    pub fn is_failure(&self) -> bool {
        if self
            .steps
            .iter()
            .any(|s| matches!(s.status, StepStatus::FailedContinued(_)))
        {
            return true;
        }
        match &self.breach {
            Some(_) => self.budget.on_exceeded() != OnBudgetExceeded::Warn,
            None => false,
        }
    }
}

/// Run a pipeline against a live server connection.
///
/// Drives the step loop, threading the growing output map and the budget
/// tracker through each step. A breach under [`OnBudgetExceeded::Stop`]
/// returns early; under `continue`/`warn` the loop runs to completion and the
/// breach is flagged afterward.
pub async fn run(
    pipeline: &Pipeline,
    conn: &Connection,
    vars: &HashMap<String, String>,
    mut budget: BudgetTracker,
    pricing: &PricingTable,
) -> RunResult {
    let mut outputs: HashMap<String, Value> = HashMap::new();
    let mut steps: Vec<StepOutcome> = Vec::new();
    let mut last_output: Option<Value> = None;
    let mut breach: Option<(String, BreachKind)> = None;

    for step in &pipeline.steps {
        let ctx = ResolveContext {
            outputs: &outputs,
            vars,
        };
        // when: guard.
        match guard_allows(step, &ctx) {
            Ok(true) => {}
            Ok(false) => {
                steps.push(skipped(step));
                continue;
            }
            Err(msg) => {
                steps.push(failed_now(step, msg));
                breach = None;
                break;
            }
        }

        // Budget pre-check: a plain tool call costs the table's per-call rate.
        let projected = pricing.tool_call_cost();
        if let Some(kind) = budget.would_exceed(projected, 0, projected) {
            if budget.on_exceeded() == OnBudgetExceeded::Stop {
                breach = Some((step.id.clone(), kind));
                break;
            }
        }

        let resolved = match resolve_args(&step.args, &ctx) {
            Ok(v) => v,
            Err(e) => {
                steps.push(failed_now(step, e.to_string()));
                break;
            }
        };

        let started = Instant::now();
        let call = call_with_retry(conn, step, &resolved).await;
        let duration = started.elapsed();
        match call {
            Ok(output) => {
                let cost = StepCost {
                    step_id: step.id.clone(),
                    cost_usd: projected,
                    input_tokens: 0,
                    output_tokens: 0,
                    duration,
                    is_llm: false,
                };
                budget.record(cost.clone());
                outputs.insert(step.id.clone(), output.clone());
                last_output = Some(output.clone());
                steps.push(StepOutcome {
                    id: step.id.clone(),
                    tool: step.tool.clone(),
                    status: StepStatus::Ok,
                    output: Some(output),
                    cost,
                });
            }
            Err(msg) => match step.on_error {
                OnError::Continue => {
                    let cost = zero_cost(step, duration);
                    budget.record(cost.clone());
                    steps.push(StepOutcome {
                        id: step.id.clone(),
                        tool: step.tool.clone(),
                        status: StepStatus::FailedContinued(msg),
                        output: None,
                        cost,
                    });
                }
                _ => {
                    steps.push(failed_now(step, msg));
                    break;
                }
            },
        }
    }

    // Under continue/warn, surface a post-run breach.
    if breach.is_none() {
        if let Some(kind) = budget.post_run_breach() {
            let id = steps.last().map(|s| s.id.clone()).unwrap_or_default();
            breach = Some((id, kind));
        }
    }

    RunResult {
        steps,
        budget,
        last_output,
        breach,
    }
}

fn guard_allows(step: &Step, ctx: &ResolveContext) -> Result<bool, String> {
    match &step.when {
        None => Ok(true),
        Some(expr) => evaluate_when(expr, ctx)
            .map_err(|e| format!("step `{}` when expression failed: {e}", step.id)),
    }
}

async fn call_with_retry(conn: &Connection, step: &Step, args: &Value) -> Result<Value, String> {
    let attempts = match step.on_error {
        OnError::Retry(n) => n + 1,
        _ => 1,
    };
    let mut last_err = String::new();
    for _ in 0..attempts {
        match conn.call_tool(&step.tool, args).await {
            Ok(v) => return Ok(v),
            Err(e) => last_err = e.to_string(),
        }
    }
    Err(last_err)
}

fn skipped(step: &Step) -> StepOutcome {
    StepOutcome {
        id: step.id.clone(),
        tool: step.tool.clone(),
        status: StepStatus::Skipped,
        output: None,
        cost: zero_cost(step, std::time::Duration::ZERO),
    }
}

fn failed_now(step: &Step, msg: String) -> StepOutcome {
    StepOutcome {
        id: step.id.clone(),
        tool: step.tool.clone(),
        status: StepStatus::FailedContinued(msg),
        output: None,
        cost: zero_cost(step, std::time::Duration::ZERO),
    }
}

fn zero_cost(step: &Step, duration: std::time::Duration) -> StepCost {
    StepCost {
        step_id: step.id.clone(),
        cost_usd: 0.0,
        input_tokens: 0,
        output_tokens: 0,
        duration,
        is_llm: false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pipe::budget::BudgetLimits;

    #[test]
    fn run_result_is_failure_when_a_step_failed_continued() {
        let result = RunResult {
            steps: vec![StepOutcome {
                id: "a".into(),
                tool: "t".into(),
                status: StepStatus::FailedContinued("boom".into()),
                output: None,
                cost: StepCost {
                    step_id: "a".into(),
                    cost_usd: 0.0,
                    input_tokens: 0,
                    output_tokens: 0,
                    duration: std::time::Duration::ZERO,
                    is_llm: false,
                },
            }],
            budget: BudgetTracker::new(BudgetLimits::default()),
            last_output: None,
            breach: None,
        };
        assert!(result.is_failure());
    }

    #[test]
    fn run_result_warn_breach_is_not_a_failure() {
        let result = RunResult {
            steps: vec![],
            budget: BudgetTracker::new(BudgetLimits {
                max_cost: Some(0.01),
                on_exceeded: OnBudgetExceeded::Warn,
                ..Default::default()
            }),
            last_output: None,
            breach: Some(("a".into(), BreachKind::Cost)),
        };
        assert!(!result.is_failure());
    }

    #[test]
    fn run_result_stop_breach_is_a_failure() {
        let result = RunResult {
            steps: vec![],
            budget: BudgetTracker::new(BudgetLimits {
                max_cost: Some(0.01),
                on_exceeded: OnBudgetExceeded::Stop,
                ..Default::default()
            }),
            last_output: None,
            breach: Some(("a".into(), BreachKind::Cost)),
        };
        assert!(result.is_failure());
    }
}
