use super::support::*;
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicUsize, Ordering};

use crate::GenerationOptions;
use crate::llm::types::{LlmToolChoice, LlmUsage};

#[derive(Clone, Debug, Default)]
struct MutatingProvider {
    options: ProviderOptions,
    marker: String,
}

#[derive(Clone, Debug)]
struct FailingProvider {
    options: ProviderOptions,
    attempts: Arc<AtomicUsize>,
    fail_until: usize,
    retryable: bool,
}

impl FailingProvider {
    fn into_components(self) -> ProviderComponents {
        ProviderComponents::new(Box::new(self), Arc::new(StaticModelPolicy::new()))
    }
}

impl MutatingProvider {
    fn into_components(self, policy: Arc<dyn ProviderModelPolicy>) -> ProviderComponents {
        ProviderComponents::new(Box::new(self), policy)
    }
}

#[async_trait::async_trait]
impl Provider for MutatingProvider {
    fn kind(&self) -> &'static str {
        "mutating"
    }

    fn options(&self) -> ProviderOptions {
        self.options.clone()
    }

    fn set_options(&mut self, options: ProviderOptions) {
        self.options = options;
    }

    fn serialize_config(&self) -> serde_json::Value {
        serde_json::json!({ "marker": self.marker })
    }

    async fn complete(&mut self, _request: LlmRequest) -> Result<LlmResponse, LlmTransportError> {
        self.marker = "complete".to_string();
        Ok(LlmResponse {
            full_text: "ok".to_string(),
            parts: Vec::new(),
            usage: LlmUsage::default(),
            terminal_reason: crate::LlmTerminalReason::Stop,
            terminal_diagnostic: None,
            provider_usage: None,
            request_body: None,
            http_summary: None,
        })
    }

    fn clone_boxed(&self) -> Box<dyn Provider> {
        Box::new(self.clone())
    }
}

#[async_trait::async_trait]
impl Provider for FailingProvider {
    fn kind(&self) -> &'static str {
        "failing"
    }

    fn options(&self) -> ProviderOptions {
        self.options.clone()
    }

    fn set_options(&mut self, options: ProviderOptions) {
        self.options = options;
    }

    fn serialize_config(&self) -> serde_json::Value {
        serde_json::Value::Object(Default::default())
    }

    async fn complete(&mut self, _request: LlmRequest) -> Result<LlmResponse, LlmTransportError> {
        let attempt = self.attempts.fetch_add(1, Ordering::SeqCst) + 1;
        if attempt <= self.fail_until {
            let kind = if self.retryable {
                ProviderFailureKind::Transport
            } else {
                ProviderFailureKind::Validation
            };
            return Err(LlmTransportError::new("temporary failure")
                .with_kind(kind)
                .retryable(self.retryable));
        }
        Ok(LlmResponse {
            full_text: "ok".to_string(),
            parts: Vec::new(),
            usage: LlmUsage::default(),
            terminal_reason: crate::LlmTerminalReason::Stop,
            terminal_diagnostic: None,
            provider_usage: None,
            request_body: None,
            http_summary: None,
        })
    }

    fn clone_boxed(&self) -> Box<dyn Provider> {
        Box::new(self.clone())
    }
}

/// Fails `fail_until` calls with an HTTP-status failure (plus optional
/// `Retry-After`), leaving kind/retryability to the classifier — the shape
/// every wire provider produces for 429/5xx responses.
#[derive(Clone, Debug)]
struct StatusFailingProvider {
    options: ProviderOptions,
    attempts: Arc<AtomicUsize>,
    fail_until: usize,
    status: u16,
    retry_after: Option<Duration>,
}

impl StatusFailingProvider {
    fn into_components(self) -> ProviderComponents {
        ProviderComponents::new(Box::new(self), Arc::new(StaticModelPolicy::new()))
    }
}

#[async_trait::async_trait]
impl Provider for StatusFailingProvider {
    fn kind(&self) -> &'static str {
        "status-failing"
    }

    fn options(&self) -> ProviderOptions {
        self.options.clone()
    }

    fn set_options(&mut self, options: ProviderOptions) {
        self.options = options;
    }

    fn serialize_config(&self) -> serde_json::Value {
        serde_json::Value::Object(Default::default())
    }

    async fn complete(&mut self, _request: LlmRequest) -> Result<LlmResponse, LlmTransportError> {
        let attempt = self.attempts.fetch_add(1, Ordering::SeqCst) + 1;
        if attempt <= self.fail_until {
            let message = if self.status == 429 {
                "throttled by provider"
            } else {
                "upstream unavailable"
            };
            let mut failure = LlmTransportError::new(message).with_status(self.status);
            if let Some(retry_after) = self.retry_after {
                failure = failure.with_retry_after(retry_after);
            }
            return Err(failure);
        }
        Ok(LlmResponse {
            full_text: "ok".to_string(),
            parts: Vec::new(),
            usage: LlmUsage::default(),
            terminal_reason: crate::LlmTerminalReason::Stop,
            terminal_diagnostic: None,
            provider_usage: None,
            request_body: None,
            http_summary: None,
        })
    }

    fn clone_boxed(&self) -> Box<dyn Provider> {
        Box::new(self.clone())
    }
}

/// Injected [`crate::Clock`] that resolves sleeps immediately while recording
/// the total requested wait, so retry-ladder tests assert real durations
/// without real waits.
#[derive(Debug, Default)]
struct RecordingClock {
    slept_ms: std::sync::atomic::AtomicU64,
}

impl RecordingClock {
    fn slept(&self) -> Duration {
        Duration::from_millis(self.slept_ms.load(Ordering::SeqCst))
    }
}

#[async_trait::async_trait]
impl crate::Clock for RecordingClock {
    fn now(&self) -> std::time::Instant {
        std::time::Instant::now()
    }

    fn timestamp_ms(&self) -> u64 {
        0
    }

    fn timestamp_rfc3339(&self) -> String {
        self.timestamp_datetime().to_rfc3339()
    }

    fn timestamp_datetime(&self) -> chrono::DateTime<chrono::Utc> {
        chrono::DateTime::<chrono::Utc>::from(std::time::UNIX_EPOCH)
    }

    async fn sleep(&self, duration: Duration) {
        self.slept_ms
            .fetch_add(duration.as_millis() as u64, Ordering::SeqCst);
    }

    async fn sleep_until(&self, _deadline: std::time::Instant) {}
}

#[derive(Debug)]
struct MetricsTransport {
    inner: Box<dyn Provider>,
    hits: Arc<AtomicUsize>,
}

impl Clone for MetricsTransport {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone_boxed(),
            hits: Arc::clone(&self.hits),
        }
    }
}

#[async_trait::async_trait]
impl Provider for MetricsTransport {
    fn kind(&self) -> &'static str {
        self.inner.kind()
    }

    fn options(&self) -> ProviderOptions {
        self.inner.options()
    }

    fn set_options(&mut self, options: ProviderOptions) {
        self.inner.set_options(options);
    }

    fn serialize_config(&self) -> serde_json::Value {
        self.inner.serialize_config()
    }

    async fn complete(&mut self, request: LlmRequest) -> Result<LlmResponse, LlmTransportError> {
        self.hits.fetch_add(1, Ordering::SeqCst);
        self.inner.complete(request).await
    }

    fn requires_streaming(&self) -> bool {
        self.inner.requires_streaming()
    }

    fn clone_boxed(&self) -> Box<dyn Provider> {
        Box::new(self.clone())
    }
}

fn empty_request() -> LlmRequest {
    LlmRequest {
        model: "model".to_string(),
        messages: Vec::new(),
        attachments: Vec::new(),
        tools: Arc::new(Vec::new()),
        tool_choice: LlmToolChoice::None,
        model_variant: None,
        scope: crate::LlmRequestScope::new(
            "provider-test",
            "provider-test:frame",
            "provider-test:request",
        ),
        output_spec: None,
        stream_events: None,
        generation: GenerationOptions::default(),
        provider_trace: None,
    }
}

#[test]
fn provider_spec_roundtrips_as_flat_object() {
    let spec = ProviderSpec {
        kind: "anthropic".to_string(),
        config: serde_json::json!({
            "api_key": "sk-ant-test",
            "base_url": null
        }),
    };
    let serialized = serde_json::to_value(&spec).expect("serialize");
    assert_eq!(serialized["type"], serde_json::json!("anthropic"));
    assert_eq!(serialized["api_key"], serde_json::json!("sk-ant-test"));
    let roundtripped: ProviderSpec = serde_json::from_value(serialized).expect("deserialize");
    assert_eq!(roundtripped.kind, spec.kind);
    assert_eq!(roundtripped.config["api_key"], spec.config["api_key"]);
}

#[test]
fn provider_options_serialize_only_reliability_shape() {
    let options = ProviderOptions {
        reliability: ProviderReliability::default()
            .request_timeout(Some(RequestTimeout::Millis(1_234)))
            .stream_chunk_timeout_ms(Some(567))
            .max_attempts(2),
        ..ProviderOptions::default()
    };

    let value = serde_json::to_value(options).expect("serialize");
    assert!(value.get("timeout").is_none());
    assert!(value.get("chunk_timeout").is_none());
    assert_eq!(
        value["reliability"]["request_timeout"],
        serde_json::json!(1234)
    );
    assert_eq!(
        value["reliability"]["chunk_timeout"],
        serde_json::json!(567)
    );
}

#[test]
fn provider_options_roundtrip_output_limit_and_cache_retention() {
    let options = ProviderOptions {
        max_output_tokens: Some(16_384),
        cache_retention: CacheRetention::Long,
        ..ProviderOptions::default()
    };

    let value = serde_json::to_value(&options).expect("serialize");
    assert_eq!(value["max_output_tokens"], serde_json::json!(16_384));
    assert_eq!(value["cache_retention"], serde_json::json!("long"));

    let roundtripped: ProviderOptions = serde_json::from_value(value).expect("deserialize");
    assert_eq!(roundtripped, options);
}

#[test]
fn provider_options_default_omits_and_restores_shared_output_fields() {
    let value = serde_json::to_value(ProviderOptions::default()).expect("serialize");
    assert!(value.get("max_output_tokens").is_none());
    assert!(value.get("cache_retention").is_none());

    let restored: ProviderOptions = serde_json::from_value(serde_json::json!({})).expect("default");
    assert_eq!(restored.max_output_tokens, None);
    assert_eq!(restored.cache_retention, CacheRetention::Short);
    assert!(restored.is_default());
}

#[test]
fn generation_policy_prefers_request_then_provider_then_default() {
    let provider_options = ProviderOptions {
        max_output_tokens: Some(8_192),
        cache_retention: CacheRetention::Long,
        expose_thinking: true,
        ..ProviderOptions::default()
    };
    let defaulted = resolve_generation_policy(
        &GenerationOptions::default(),
        &ProviderOptions::default(),
        32_768,
        "thinking",
    );
    assert_eq!(defaulted.max_output_tokens, 32_768);
    assert_eq!(defaulted.cache_retention, CacheRetention::Short);
    assert!(!defaulted.expose_thinking);
    assert_eq!(defaulted.thinking, "thinking");

    let provider_limited =
        resolve_generation_policy(&GenerationOptions::default(), &provider_options, 32_768, ());
    assert_eq!(provider_limited.max_output_tokens, 8_192);
    assert_eq!(provider_limited.cache_retention, CacheRetention::Long);
    assert!(provider_limited.expose_thinking);

    let request_generation = GenerationOptions {
        output_token_cap: NonZeroUsize::new(2_048),
    };
    let request_limited =
        resolve_generation_policy(&request_generation, &provider_options, 32_768, ());
    assert_eq!(request_limited.max_output_tokens, 2_048);
}

#[test]
fn provider_handle_delegates_variant_policy() {
    static VARIANTS: &[&str] = &["low", "high"];
    let handle = ProviderHandle::new(
        MutatingProvider::default()
            .into_components(Arc::new(StaticModelPolicy::with_variants(VARIANTS))),
    );

    assert_eq!(handle.supported_variants("model-a"), VARIANTS);
    assert!(handle.validate_variant("model-a", "low").is_ok());
    assert!(handle.validate_variant("model-a", "medium").is_err());
}

#[tokio::test]
async fn transport_mutations_are_visible_after_completion_returns() {
    let mut handle = ProviderHandle::new(
        MutatingProvider::default().into_components(Arc::new(StaticModelPolicy::new())),
    );

    handle.complete(empty_request()).await.expect("complete");

    assert_eq!(
        handle.to_spec().config["marker"],
        serde_json::json!("complete")
    );
}

#[tokio::test]
async fn map_provider_installs_transport_decorator() {
    let hits = Arc::new(AtomicUsize::new(0));
    let components = MutatingProvider::default()
        .into_components(Arc::new(StaticModelPolicy::new()))
        .map_provider({
            let hits = Arc::clone(&hits);
            move |inner| Box::new(MetricsTransport { inner, hits })
        });
    let mut handle = ProviderHandle::new(components);

    handle.complete(empty_request()).await.expect("complete");

    assert_eq!(hits.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn provider_handle_retries_retryable_failures_in_shared_executor() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let provider = FailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(3)
                .base_delay_ms(0)
                .max_delay_ms(0),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 2,
        retryable: true,
    };
    let mut handle = ProviderHandle::new(provider.into_components());

    handle
        .complete(empty_request())
        .await
        .expect("eventual success");

    assert_eq!(attempts.load(Ordering::SeqCst), 3);
}

#[tokio::test]
async fn provider_handle_stops_on_non_retryable_failure() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let provider = FailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(3)
                .base_delay_ms(0)
                .max_delay_ms(0),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 3,
        retryable: false,
    };
    let mut handle = ProviderHandle::new(provider.into_components());

    let err = handle
        .complete(empty_request())
        .await
        .expect_err("non retryable");

    assert!(!err.retryable);
    assert_eq!(attempts.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn provider_handle_set_options_affects_retry_behavior() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let provider = FailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::disabled(),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 1,
        retryable: true,
    };
    let mut handle = ProviderHandle::new(provider.into_components());
    handle.set_options(ProviderOptions {
        reliability: ProviderReliability::default()
            .max_attempts(2)
            .base_delay_ms(0)
            .max_delay_ms(0),
        ..ProviderOptions::default()
    });

    handle
        .complete(empty_request())
        .await
        .expect("retry after set_options");

    assert_eq!(attempts.load(Ordering::SeqCst), 2);
}

#[tokio::test]
async fn provider_handle_throttle_with_retry_after_does_not_consume_attempts() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let clock = Arc::new(RecordingClock::default());
    let provider = StatusFailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(2)
                .base_delay_ms(0)
                .max_delay_ms(0),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        // Three throttles in a row: more failures than the two-attempt
        // budget could ever absorb if throttles consumed attempts.
        fail_until: 3,
        status: 429,
        retry_after: Some(Duration::from_secs(5)),
    };
    let mut handle =
        ProviderHandle::new(provider.into_components()).with_clock(Arc::clone(&clock) as _);

    handle
        .complete(empty_request())
        .await
        .expect("success after deferred throttle waits");

    assert_eq!(attempts.load(Ordering::SeqCst), 4);
    // Each deference honored the provider-stated 5s Retry-After.
    assert_eq!(clock.slept(), Duration::from_secs(15));
}

#[tokio::test]
async fn provider_handle_throttle_budget_exhaustion_degrades_to_attempt_counting() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let clock = Arc::new(RecordingClock::default());
    let provider = StatusFailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(2)
                .base_delay_ms(0)
                .max_delay_ms(0)
                // Room for exactly two 4s deferences; the third throttle
                // overflows the budget and counts as a normal failure.
                .throttle_wait_budget_ms(10_000),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 100,
        status: 429,
        retry_after: Some(Duration::from_secs(4)),
    };
    let mut handle =
        ProviderHandle::new(provider.into_components()).with_clock(Arc::clone(&clock) as _);

    let err = handle
        .complete(empty_request())
        .await
        .expect_err("throttle storm outlives budget and attempts");

    // Two free deferences, then the two-attempt ladder (one counted retry
    // that still waits the provider-stated Retry-After before re-calling).
    assert_eq!(attempts.load(Ordering::SeqCst), 4);
    assert_eq!(clock.slept(), Duration::from_secs(12));
    assert!(err.retryable);
    assert_eq!(err.kind, ProviderFailureKind::Quota);
}

#[tokio::test]
async fn provider_handle_throttle_without_retry_after_consumes_attempts() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let provider = StatusFailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(2)
                .base_delay_ms(0)
                .max_delay_ms(0),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 100,
        status: 429,
        retry_after: None,
    };
    let mut handle = ProviderHandle::new(provider.into_components());

    let err = handle
        .complete(empty_request())
        .await
        .expect_err("no server-stated wait, so the normal ladder applies");

    assert_eq!(attempts.load(Ordering::SeqCst), 2);
    assert_eq!(err.kind, ProviderFailureKind::Quota);
}

#[tokio::test]
async fn provider_handle_server_error_with_retry_after_still_consumes_attempts() {
    let attempts = Arc::new(AtomicUsize::new(0));
    let provider = StatusFailingProvider {
        options: ProviderOptions {
            reliability: ProviderReliability::default()
                .max_attempts(2)
                .base_delay_ms(0)
                .max_delay_ms(0),
            ..ProviderOptions::default()
        },
        attempts: Arc::clone(&attempts),
        fail_until: 100,
        status: 503,
        retry_after: Some(Duration::from_secs(1)),
    };
    let mut handle = ProviderHandle::new(provider.into_components())
        .with_clock(Arc::new(RecordingClock::default()) as _);

    let err = handle
        .complete(empty_request())
        .await
        .expect_err("5xx is a failure, not a throttle");

    assert_eq!(attempts.load(Ordering::SeqCst), 2);
    assert!(err.retryable);
    assert_eq!(err.kind, ProviderFailureKind::Http);
}

#[test]
fn default_failure_classifier_classifies_429_as_retryable_throttle() {
    let classifier = DefaultProviderFailureClassifier;
    let failure = classifier.classify(
        ProviderFailure::new("Rate limit reached for requests")
            .with_status(429)
            .with_retry_after(Duration::from_secs(7)),
    );
    assert_eq!(failure.kind, ProviderFailureKind::Quota);
    assert!(failure.retryable);
    assert_eq!(failure.retry_after, Some(Duration::from_secs(7)));
}

#[test]
fn default_failure_classifier_keeps_quota_exhaustion_non_retryable() {
    let classifier = DefaultProviderFailureClassifier;
    for message in [
        "insufficient_quota",
        "usage_limit_reached",
        "usage_not_included in your plan",
    ] {
        let failure = classifier.classify(ProviderFailure::new(message).with_status(429));
        assert_eq!(failure.kind, ProviderFailureKind::Quota);
        assert!(!failure.retryable);
    }
}

#[test]
fn default_failure_classifier_marks_context_overflow() {
    let classifier = DefaultProviderFailureClassifier;
    for message in [
        "Anthropic request failed: prompt is too long",
        "context_length_exceeded",
        "This model's maximum context length is 128000 tokens",
        "Google says input is too long",
        "OpenRouter error: too many tokens",
        "Together: request too large",
        "Copilot: exceeds the maximum number of tokens",
        "local model: context window exceeded",
        "Anthropic: request_too_large",
        "OpenAI: Your input exceeds the context window of this model",
        "Google: The input token count (1196265) exceeds the maximum number of tokens allowed",
        "xAI: This model's maximum prompt length is 131072 but the request contains more",
        "Groq: Please reduce the length of the messages",
        "OpenRouter: This endpoint's maximum context length is 128000 tokens",
        "Together: The input (150000 tokens) is longer than the model's context length (128000 tokens).",
        "llama.cpp: the request exceeds the available context size",
        "LM Studio: tokens to keep from the initial prompt is greater than the context length",
        "MiniMax: invalid params, context window exceeds limit",
        "Kimi: Your request exceeded model token limit: 131072 (requested: 150000)",
        "Mistral: Prompt contains too many tokens; too large for model with 128000 maximum context length",
        "z.ai: model_context_window_exceeded",
    ] {
        let failure = classifier.classify(ProviderFailure::new(message).with_status(400));
        assert_eq!(
            failure.terminal_reason,
            crate::LlmTerminalReason::ContextOverflow
        );
        assert!(!failure.retryable);
    }
}

#[test]
fn default_failure_classifier_marks_content_filter() {
    let classifier = DefaultProviderFailureClassifier;
    for message in [
        "Provider finish_reason: content_filter",
        "blocked by prohibited_content",
        "safety filter tripped",
        "sensitive content refused",
    ] {
        let failure = classifier.classify(ProviderFailure::new(message).with_status(400));
        assert_eq!(
            failure.terminal_reason,
            crate::LlmTerminalReason::ContentFilter
        );
    }
}

#[test]
fn default_failure_classifier_does_not_treat_rate_limits_as_context_overflow() {
    let classifier = DefaultProviderFailureClassifier;
    for message in [
        "rate limit: too many tokens per minute",
        "Too many requests",
        "throttling because token rate exceeded",
        "insufficient_quota",
    ] {
        let failure = classifier.classify(ProviderFailure::new(message).with_status(429));
        assert_ne!(
            failure.terminal_reason,
            crate::LlmTerminalReason::ContextOverflow
        );
    }
}
