// ============================================================
use crate::critical::Critical;
// New trait hierarchy: Reflectable → Promptable → Parseable
// ============================================================

/// Reflectable: bottom-layer trait for struct metadata capture.
/// Like Java reflection — makes struct fields, types, and annotations accessible.
/// Currently empty; metadata API will be added as the framework matures.
pub trait Reflectable {}

/// Promptable: renderable as LLM prompt. Inherits Reflectable.
/// When used as request, renders request content via to_markdown().
/// When used as response.class, renders output format via schema_markdown().
pub trait Promptable: Reflectable {
    fn to_markdown(&self) -> String {
        self.to_markdown_depth(2)
    }
    fn to_markdown_depth(&self, depth: usize) -> String;
    fn to_markdown_item(&self) -> String {
        String::new()
    }
    /// Render output schema for LLM format specification.
    /// Default returns empty string (for request-only structs).
    fn schema_markdown(_token: &str) -> String where Self: Sized {
        String::new()
    }
    /// Render as cache-segmented prompt. Default: all content in stable segment.
    /// Override with `#[prompt(volatile)]` on a field to split at that boundary.
    fn to_markdown_cached(&self) -> CachedPrompt {
        CachedPrompt::single(self.to_markdown())
    }
}

/// Parseable: can parse LLM output into structured data. Inherits Promptable.
/// Must be able to render output schema (Promptable) to tell LLM the expected format,
/// then parse the LLM response back into a struct.
pub trait Parseable: Promptable {
    fn from_markdown(text: &str, token: &str) -> Result<Vec<Self>, String> where Self: Sized;
    fn type_name() -> &'static str where Self: Sized;
}


/// Marker trait: only derived on structs via #[derive(Promptable)]
/// Used to enforce struct-only constraint on infer/stream_infer input
pub trait StructInput {}

/// Marker trait: only derived on enums/structs via #[derive(Parseable)]
/// Used to enforce struct-only constraint on infer/stream_infer output
pub trait StructOutput {}

// ============================================================
// CachedPrompt: structured prompt with cache-friendly segmentation
// ============================================================

/// Prompt split into stable (cacheable) and volatile (changing) segments.
///
/// When sent to LLM providers that support prompt caching (e.g. Anthropic),
/// the stable segment is marked with `cache_control` for prefix caching.
/// Providers that don't support caching simply merge both segments.
///
/// Use `#[prompt(volatile)]` on a struct field to mark where the volatile
/// section begins. Fields before it render into `stable`, the rest into `delta`.
#[derive(Clone)]
pub struct CachedPrompt {
    /// Stable prefix — content that rarely changes between requests.
    /// Marked with cache_control for LLM provider caching.
    pub stable: String,
    /// Volatile suffix — content that changes every request.
    /// Not cached.
    pub delta: String,
}

impl CachedPrompt {
    /// Create a prompt with all content in the stable segment (no volatile split).
    pub fn single(text: String) -> Self {
        Self { stable: text, delta: String::new() }
    }

    /// Merge stable and delta into a single string (for non-cache-aware channels).
    pub fn merge(&self) -> String {
        if self.delta.is_empty() {
            self.stable.clone()
        } else {
            format!("{}{}", self.stable, self.delta)
        }
    }

    /// Returns true if this prompt has a volatile segment.
    pub fn has_volatile(&self) -> bool {
        !self.delta.is_empty()
    }
}

// ============================================================
// Chrono date type support — inline rendering for LLM prompts
// ============================================================

impl<Tz: chrono::TimeZone> Reflectable for chrono::DateTime<Tz> {}
impl Reflectable for chrono::NaiveDateTime {}
impl Reflectable for chrono::NaiveDate {}

impl<Tz: chrono::TimeZone> Promptable for chrono::DateTime<Tz>
where
    Tz::Offset: std::fmt::Display,
{
    fn to_markdown_depth(&self, _depth: usize) -> String {
        self.format("%Y%m%d%H%M%S").to_string()
    }
    fn to_markdown_item(&self) -> String {
        self.format("%Y%m%d%H%M%S").to_string()
    }
}

impl Promptable for chrono::NaiveDateTime {
    fn to_markdown_depth(&self, _depth: usize) -> String {
        self.format("%Y%m%d%H%M%S").to_string()
    }
    fn to_markdown_item(&self) -> String {
        self.format("%Y%m%d%H%M%S").to_string()
    }
}

impl Promptable for chrono::NaiveDate {
    fn to_markdown_depth(&self, _depth: usize) -> String {
        self.format("%Y%m%d").to_string()
    }
    fn to_markdown_item(&self) -> String {
        self.format("%Y%m%d").to_string()
    }
}

// ============================================================
// InferError: structured error type for LLM inference
// ============================================================

// Retry constants for transient inference failures
const INFER_MAX_RETRIES: u32 = 2;
const INFER_RETRY_BASE_SECS: u64 = 2;
const INFER_RETRY_MAX_EXPONENT: u32 = 3;
const INFER_RETRY_CAP_SECS: u64 = 30;

/// Error code for LLM inference errors.
///
/// Two categories:
/// - **Channel-level** (`is_channel() == true`): network/provider issues, caller should switch channel and retry
/// - **Output-level** (`is_channel() == false`): LLM output quality issues, channel is working normally
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InferErrorCode {
    // Channel-level errors (switch channel)
    /// Network failure: connection refused, DNS resolution, timeout
    HttpRequestFailed,
    /// Non-200 HTTP response with status code: 402 Payment Required, 429 Rate Limited, 5xx Server Error
    HttpStatus(u16),
    /// LlmChannel::start_stream() returned Err
    StartStreamFailed,

    // Output-level errors (channel is fine, output is bad)
    /// LLM output preamble before first separator (zero tolerance)
    FormatViolation,
    /// Parseable parsing failed: unknown variant, field mismatch
    ParseError,
    /// Stream ended without end marker (truncated output)
    Truncated,
}

impl InferErrorCode {
    /// Returns true if this is a channel/network-level error.
    /// Caller should switch to a different LLM channel and retry.
    pub fn is_channel(&self) -> bool {
        matches!(self, Self::HttpRequestFailed | Self::HttpStatus(_) | Self::StartStreamFailed)
    }

    /// Returns true if this is a 402 quota exceeded error.
    pub fn is_quota(&self) -> bool {
        matches!(self, Self::HttpStatus(402))
    }

    /// Returns true if this error is worth retrying (transient network/server issue).
    /// Channel-level errors excluding quota exhaustion, plus truncated output (server-side TCP close).
    pub fn is_retryable(&self) -> bool {
        (self.is_channel() && !self.is_quota()) || matches!(self, Self::Truncated)
    }
}

impl std::fmt::Display for InferErrorCode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::HttpRequestFailed => write!(f, "HttpRequestFailed"),
            Self::HttpStatus(status) => write!(f, "HttpStatus({})", status),
            Self::StartStreamFailed => write!(f, "StartStreamFailed"),
            Self::FormatViolation => write!(f, "FormatViolation"),
            Self::ParseError => write!(f, "ParseError"),
            Self::Truncated => write!(f, "Truncated"),
        }
    }
}

/// Structured error type for LLM inference.
///
/// Contains an error code (for programmatic matching) and a human-readable message.
#[derive(Debug)]
pub struct InferError {
    pub code: InferErrorCode,
    pub message: String,
}

impl InferError {
    pub fn new(code: InferErrorCode, message: impl Into<String>) -> Self {
        Self { code, message: message.into() }
    }
}

impl std::fmt::Display for InferError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "[{}] {}", self.code, self.message)
    }
}

impl std::error::Error for InferError {}

// ============================================================
// Utilities
// ============================================================

/// Find `pattern` appearing as a complete line (after trimming) in `text`.
/// Returns the byte offset where the pattern starts (not line start).
/// This prevents substring matches inside content lines.
/// Safe tail slice that respects UTF-8 char boundaries.
/// Returns the last `max_bytes` bytes (adjusted to char boundary) with "..." prefix,
/// or the full string if shorter than `max_bytes`.
#[allow(clippy::string_slice)]
fn safe_tail(s: &str, max_bytes: usize) -> String {
    if s.len() <= max_bytes {
        s.to_string()
    } else {
        let mut start = s.len() - max_bytes;
        while !s.is_char_boundary(start) {
            start += 1;
        }
        format!("...{}", &s[start..])
    }
}

fn find_line_match(text: &str, pattern: &str) -> Option<usize> {
    let mut offset = 0;
    for line in text.split('\n') {
        let trimmed = line.trim();
        if trimmed == pattern {
            let leading = line.find(pattern).unwrap_or(0);
            return Some(offset + leading);
        }
        offset += line.len() + 1;
    }
    None
}

/// LlmChannel: abstraction for LLM text streaming (async)
///
/// Returns an unbounded receiver that yields text chunks from the LLM stream.
/// Implementations should spawn an async task to read the stream and send chunks.
pub trait LlmChannel: Send + Sync {
    #[allow(clippy::string_slice)]
    fn start_stream(&self, prompt: String) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String>;

    /// Start a stream with cache-segmented prompt.
    /// Default: merges segments and falls back to start_stream().
    /// Override for providers that support prompt caching (e.g. Anthropic via cache_control).
    fn start_stream_cached(&self, parts: CachedPrompt) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String> {
        self.start_stream(parts.merge())
    }

    /// Start a stream with resume support (断点续传).
    /// When retrying after truncation, `prefill` contains the partial output
    /// received so far. Providers that support assistant prefill can include it
    /// in the request so the LLM continues from where it left off.
    /// Default: ignores prefill and falls back to start_stream_cached().
    fn start_stream_with_resume(&self, parts: CachedPrompt, _prefill: &str) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String> {
        self.start_stream_cached(parts)
    }

}

/// OpenAI-compatible LLM channel configuration
pub struct OpenAiChannel {
    pub endpoint: String,
    pub model: String,
    pub api_key: String,
    pub max_tokens: Option<u32>,
}

impl OpenAiChannel {
    pub fn new(endpoint: impl Into<String>, model: impl Into<String>, api_key: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            model: model.into(),
            api_key: api_key.into(),
            max_tokens: None,
        }
    }

    pub fn with_max_tokens(mut self, max_tokens: u32) -> Self {
        self.max_tokens = Some(max_tokens);
        self
    }
}

/// SSE response types for OpenAI streaming
#[derive(serde::Deserialize)]
struct SseResponse {
    choices: Vec<SseChoice>,
    #[serde(default)]
    usage: Option<SseUsage>,
}

#[derive(serde::Deserialize, Clone, Debug)]
struct SseUsage {
    #[serde(default)]
    prompt_tokens: u64,
    #[serde(default)]
    completion_tokens: u64,
    #[serde(default)]
    prompt_tokens_details: Option<PromptTokensDetails>,
}

/// SSE usage prompt_tokens_details (nested cache statistics from ZenMux/OpenAI-compatible providers).
#[derive(serde::Deserialize, Clone, Debug)]
struct PromptTokensDetails {
    #[serde(default)]
    cached_tokens: u64,
    #[serde(default, rename = "cacheCreationInputTokens")]
    cache_creation_input_tokens: u64,
}

#[derive(serde::Deserialize)]
struct SseChoice {
    delta: SseDelta,
}

#[derive(serde::Deserialize)]
struct SseDelta {
    content: Option<String>,
}

// ============================================================
// Cache statistics: automatic collection + query API
// ============================================================

/// Maximum data retention duration (48 hours in seconds).
const CACHE_RETENTION_SECS: u64 = 48 * 3600;

/// Single inference cache data point.
struct CacheDataPoint {
    recorded_at: std::time::Instant,
    prompt_tokens: u64,
    completion_tokens: u64,
    cache_read_tokens: u64,
    cache_write_tokens: u64,
}

/// Global cache statistics collector.
fn cache_collector() -> &'static std::sync::Mutex<std::collections::VecDeque<CacheDataPoint>> {
    static COLLECTOR: std::sync::OnceLock<std::sync::Mutex<std::collections::VecDeque<CacheDataPoint>>> = std::sync::OnceLock::new();
    COLLECTOR.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new()))
}

/// Record a cache data point from SSE usage.
fn record_cache_data(prompt_tokens: u64, completion_tokens: u64, cache_read_tokens: u64, cache_write_tokens: u64) {
    let now = std::time::Instant::now();
    let mut collector = cache_collector().lock().unwrap_or_else(|e| e.into_inner());
    // Evict entries older than retention window
    while let Some(front) = collector.front() {
        if now.duration_since(front.recorded_at).as_secs() > CACHE_RETENTION_SECS {
            collector.pop_front();
        } else {
            break;
        }
    }
    collector.push_back(CacheDataPoint {
        recorded_at: now,
        prompt_tokens,
        completion_tokens,
        cache_read_tokens,
        cache_write_tokens,
    });
}

/// Aggregated cache statistics over a time window.
#[derive(Debug, Clone, serde::Serialize)]
pub struct CacheStats {
    /// Total inference requests in the window.
    pub total_requests: u64,
    /// Requests where cache was hit.
    pub cache_hit_requests: u64,
    /// Total input tokens (prompt + cached). The true total sent to the model.
    pub total_input_tokens: u64,
    /// Tokens served from cache (the portion that was "free" or discounted).
    pub cached_tokens: u64,
    /// Total output tokens generated.
    pub total_output_tokens: u64,
    /// Token-level cache hit rate: cached_tokens / total_input_tokens (0.0–1.0).
    pub hit_rate: f64,
}

/// Query aggregated cache statistics for a given time window.
///
/// # Arguments
/// * `window_secs` - Time window in seconds (e.g. 86400 for 24h). Pass 0 for all available history.
///
/// # Returns
/// Aggregated `CacheStats` for the specified window.
pub fn cache_stats(window_secs: u64) -> CacheStats {
    let now = std::time::Instant::now();
    let collector = cache_collector().lock().unwrap_or_else(|e| e.into_inner());
    let mut stats = CacheStats {
        total_requests: 0,
        cache_hit_requests: 0,
        total_input_tokens: 0,
        cached_tokens: 0,
        total_output_tokens: 0,
        hit_rate: 0.0,
    };
    for point in collector.iter() {
        if window_secs > 0 && now.duration_since(point.recorded_at).as_secs() > window_secs {
            continue;
        }
        stats.total_requests += 1;
        if point.cache_read_tokens > 0 {
            stats.cache_hit_requests += 1;
        }
        stats.total_input_tokens += point.prompt_tokens + point.cache_read_tokens + point.cache_write_tokens;
        stats.cached_tokens += point.cache_read_tokens;
        stats.total_output_tokens += point.completion_tokens;
    }
    if stats.total_input_tokens > 0 {
        stats.hit_rate = stats.cached_tokens as f64 / stats.total_input_tokens as f64;
    }
    stats
}

/// Structured usage data returned from inference calls.
#[derive(Debug, Clone, serde::Serialize)]
pub struct InferUsage {
    pub prompt_tokens: u64,
    pub completion_tokens: u64,
    pub cache_read_tokens: u64,
    pub cache_write_tokens: u64,
}

/// Log SSE usage statistics including cache hit/miss information.
fn log_sse_usage(usage: &Option<SseUsage>) {
    if let Some(u) = usage {
        let (cache_write, cache_read) = match &u.prompt_tokens_details {
            Some(d) => (d.cache_creation_input_tokens, d.cached_tokens),
            None => (0, 0),
        };
        // Record data point for statistics
        record_cache_data(u.prompt_tokens, u.completion_tokens, cache_read, cache_write);
        if cache_write > 0 || cache_read > 0 {
            eprintln!(
                "[LLM-CACHE] Usage: prompt={}, completion={}, cache_write={}, cache_read={}",
                u.prompt_tokens, u.completion_tokens,
                cache_write, cache_read
            );
        } else {
            eprintln!(
                "[LLM-CACHE] Usage: prompt={}, completion={} (no cache activity)",
                u.prompt_tokens, u.completion_tokens
            );
        }
    }
}

/// Shared reqwest::Client for connection pooling across all SSE streams.
/// reqwest::Client is internally Arc-wrapped, so clone is cheap.
fn shared_http_client() -> reqwest::Client {
    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
    CLIENT.get_or_init(|| {
        reqwest::Client::builder()
            .connect_timeout(std::time::Duration::from_secs(30))
            .build()
            .expect("Failed to build HTTP client")
    }).clone()
}

/// Spawn an async SSE stream reader task.
/// Shared between start_stream and start_stream_cached to avoid code duplication.
#[allow(clippy::string_slice)]
fn spawn_sse_reader(url: String, api_key: String, body: serde_json::Value) -> tokio::sync::mpsc::UnboundedReceiver<String> {
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

    tokio::spawn(async move {
        let mut last_usage: Option<SseUsage> = None;
        let client = shared_http_client();
        // Stream control is independent of receiver state.
        // We always read until [DONE] or stream ends — never cut short because receiver dropped.
        let mut receiver_alive = true;
        let response = match tokio::time::timeout(
            std::time::Duration::from_secs(60),
            client
                .post(&url)
                .header("Authorization", format!("Bearer {}", api_key))
                .header("Content-Type", "application/json")
                .json(&body)
                .send()
        ).await {
            Ok(Ok(resp)) => resp,
            Ok(Err(e)) => {
                eprintln!("[SSE-ERROR] HTTP request failed: {}", e);
                let _ = tx.send(format!("\n[ERROR] HTTP request failed: {}", e));
                return;
            }
            Err(_) => {
                eprintln!("[SSE-ERROR] HTTP request timed out after 60s waiting for response from {}", url);
                let _ = tx.send(format!("\n[ERROR] HTTP request timed out after 60s waiting for response from {}", url));
                return;
            }
        };

        if !response.status().is_success() {
            let status = response.status();
            let body_text = response.text().await.unwrap_or_default();
            eprintln!("[SSE-ERROR] HTTP {}: {}", status.as_u16(), body_text);
            let _ = tx.send(format!("\n[ERROR] HTTP {}: {}", status.as_u16(), body_text));
            return;
        }

        use futures_util::StreamExt;
        let mut byte_stream = response.bytes_stream();
        let mut line_buf = String::new();
        let mut chunks_sent: usize = 0;
        let mut total_bytes: usize = 0;
        let stream_start = std::time::Instant::now();
        let mut last_content_time = std::time::Instant::now();
        let mut last_diag_time = std::time::Instant::now();

        loop {
            // Stream control: always use full timeout. Never shorten because of receiver state.
            let chunk_result = match tokio::time::timeout(
                std::time::Duration::from_secs(120),
                byte_stream.next(),
            )
            .await
            {
                Ok(Some(chunk_result)) => chunk_result,
                Ok(None) => break,
                Err(_) => {
                    eprintln!("[SSE-ERROR] SSE stream idle timeout (120s no data), {} chunks, {} bytes",
                        chunks_sent, total_bytes);
                    let _ = tx.send(
                        "\n[ERROR] HTTP request failed: SSE stream idle timeout (120s no data)"
                            .to_string(),
                    );
                    return;
                }
            };
            let chunk = match chunk_result {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("[SSE-ERROR] SSE stream read error: {}", e);
                    let _ = tx.send(format!(
                        "\n[ERROR] HTTP request failed: SSE stream read error: {}",
                        e
                    ));
                    return;
                }
            };
            total_bytes += chunk.len();
            let text = String::from_utf8_lossy(&chunk);
            line_buf.push_str(&text);

            while let Some(newline_pos) = line_buf.find('\n') {
                let line = line_buf[..newline_pos].trim().to_string();
                line_buf = line_buf[newline_pos + 1..].to_string();

                if line.is_empty() {
                    continue;
                }
                if line == "data: [DONE]" || line == "data:[DONE]" {
                    if chunks_sent == 0 && receiver_alive {
                        eprintln!("[SSE-ERROR] SSE stream completed with [DONE] but 0 content chunks received");
                        let _ = tx.send(
                            "\n[ERROR] HTTP request failed: SSE stream completed with [DONE] but 0 content chunks received"
                                .to_string(),
                        );
                    }
                    log_sse_usage(&last_usage);
                    if receiver_alive {
                        send_usage_via_channel(&tx, &last_usage);
                    }
                    return;
                }
                if let Some(data) = line
                    .strip_prefix("data: ")
                    .or_else(|| line.strip_prefix("data:"))
                {
                    if let Ok(sse) = serde_json::from_str::<SseResponse>(data) {
                        if let Some(usage) = sse.usage {
                            last_usage = Some(usage);
                        }
                        if let Some(choice) = sse.choices.first() {
                            if let Some(content) = &choice.delta.content {
                                if !content.is_empty() {
                                    chunks_sent += 1;
                                    if receiver_alive {
                                        if tx.send(content.clone()).is_err() {
                                            eprintln!("[SSE-DIAG] Receiver dropped after {}s, {} chunks, {} bytes. Stream control continues until [DONE].",
                                                stream_start.elapsed().as_secs(), chunks_sent, total_bytes);
                                            receiver_alive = false;
                                        }
                                    }
                                    last_content_time = std::time::Instant::now();
                                }
                            }
                        }
                    }
                }
            }

            // Content-level idle timeout: 300s no content produced
            if last_content_time.elapsed().as_secs() >= 300 {
                eprintln!("[SSE-ERROR] SSE content idle timeout (300s no content produced, {} bytes received, {} chunks sent)",
                    total_bytes, chunks_sent);
                let _ = tx.send(format!(
                    "\n[ERROR] HTTP request failed: SSE content idle timeout (300s no content produced, {} bytes received, {} content chunks sent)",
                    total_bytes, chunks_sent
                ));
                return;
            }

            // Periodic diagnostic: every 60s
            if last_diag_time.elapsed().as_secs() >= 60 {
                eprintln!("[SSE-DIAG] Stream alive: {}s, {} content chunks sent, {} bytes received{}",
                    stream_start.elapsed().as_secs(), chunks_sent, total_bytes,
                    if receiver_alive { "" } else { " (receiver dropped, awaiting [DONE])" });
                last_diag_time = std::time::Instant::now();
            }
        }

        if chunks_sent == 0 {
            let diagnostic = if line_buf.trim().is_empty() {
                "SSE stream ended with no content (empty response body)".to_string()
            } else {
                let preview: String = line_buf.chars().take(500).collect();
                format!(
                    "SSE stream ended with no content. Unprocessed buffer ({} bytes): {}",
                    line_buf.len(),
                    preview
                )
            };
            eprintln!("[SSE-ERROR] {}", diagnostic);
            let _ = tx.send(format!("\n[ERROR] HTTP request failed: {}", diagnostic));
        } else {
            eprintln!("[SSE-DIAG] Stream ended without [DONE]: {}s, {} chunks, {} bytes (possible server-side close)",
                stream_start.elapsed().as_secs(), chunks_sent, total_bytes);
        }
        log_sse_usage(&last_usage);
        if receiver_alive {
            send_usage_via_channel(&tx, &last_usage);
        }
    });

    rx
}

/// Send usage data as a tagged message through the content channel.
fn send_usage_via_channel(tx: &tokio::sync::mpsc::UnboundedSender<String>, usage: &Option<SseUsage>) {
    if let Some(u) = usage {
        let (cw, cr) = match &u.prompt_tokens_details {
            Some(d) => (d.cache_creation_input_tokens, d.cached_tokens),
            None => (0, 0),
        };
        let _ = tx.send(format!("\n[USAGE:{}:{}:{}:{}]", u.prompt_tokens, u.completion_tokens, cr, cw));
    }
}

impl OpenAiChannel {
    fn build_url(&self) -> String {
        let endpoint = self.endpoint.trim_end_matches('/').to_string();
        if endpoint.ends_with("/v1/chat/completions") {
            endpoint
        } else {
            format!("{}/v1/chat/completions", endpoint)
        }
    }

    fn build_body(&self, content: serde_json::Value) -> serde_json::Value {
        let mut body = serde_json::json!({
            "model": &self.model,
            "messages": [{"role": "user", "content": content}],
            "stream": true,
            "stream_options": {"include_usage": true}
        });
        if let Some(max_tokens) = self.max_tokens {
            body["max_tokens"] = serde_json::json!(max_tokens);
        }
        body
    }
}

impl LlmChannel for OpenAiChannel {
    fn start_stream(&self, prompt: String) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String> {
        let url = self.build_url();
        let body = self.build_body(serde_json::json!(prompt));
        Ok(spawn_sse_reader(url, self.api_key.clone(), body))
    }

    fn start_stream_cached(&self, parts: CachedPrompt) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String> {
        let url = self.build_url();
        let body = if parts.has_volatile() {
            eprintln!(
                "[LLM-CACHE] Request: stable={} bytes, delta={} bytes (volatile enabled)",
                parts.stable.len(), parts.delta.len()
            );
            // Stable content in system message with cache_control, delta in user message
            let mut body = serde_json::json!({
                "model": &self.model,
                "messages": [
                    {
                        "role": "system",
                        "content": [
                            {"type": "text", "text": parts.stable, "cache_control": {"type": "ephemeral"}}
                        ]
                    },
                    {
                        "role": "user",
                        "content": parts.delta
                    }
                ],
                "stream": true,
                "stream_options": {"include_usage": true}
            });
            if let Some(max_tokens) = self.max_tokens {
                body["max_tokens"] = serde_json::json!(max_tokens);
            }
            body
        } else {
            eprintln!(
                "[LLM-CACHE] Request: {} bytes (no volatile, cache passthrough)",
                parts.stable.len()
            );
            self.build_body(serde_json::json!(parts.stable))
        };
        Ok(spawn_sse_reader(url, self.api_key.clone(), body))
    }

    fn start_stream_with_resume(&self, parts: CachedPrompt, prefill: &str) -> Result<tokio::sync::mpsc::UnboundedReceiver<String>, String> {
        let url = self.build_url();
        let mut messages = if parts.has_volatile() {
            eprintln!(
                "[LLM-CACHE] Resume request: stable={} bytes, delta={} bytes, prefill={} bytes",
                parts.stable.len(), parts.delta.len(), prefill.len()
            );
            vec![
                serde_json::json!({
                    "role": "system",
                    "content": [
                        {"type": "text", "text": parts.stable, "cache_control": {"type": "ephemeral"}}
                    ]
                }),
                serde_json::json!({
                    "role": "user",
                    "content": parts.delta
                }),
            ]
        } else {
            eprintln!(
                "[LLM-CACHE] Resume request: {} bytes, prefill={} bytes",
                parts.stable.len(), prefill.len()
            );
            vec![serde_json::json!({
                "role": "user",
                "content": parts.stable
            })]
        };
        // Append assistant prefill so LLM continues from where it left off
        messages.push(serde_json::json!({
            "role": "assistant",
            "content": prefill
        }));
        let mut body = serde_json::json!({
            "model": &self.model,
            "messages": messages,
            "stream": true,
            "stream_options": {"include_usage": true}
        });
        if let Some(max_tokens) = self.max_tokens {
            body["max_tokens"] = serde_json::json!(max_tokens);
        }
        Ok(spawn_sse_reader(url, self.api_key.clone(), body))
    }

}

/// Apply regex extraction to a parsed value.
/// Tries each pattern in order; if a pattern matches, returns capture group 1
/// (if present) or the full match. If no pattern matches, returns original value.
pub fn apply_regex_extraction(value: &str, patterns: &[&str]) -> String {
    for pat in patterns {
        if let Ok(re) = regex::Regex::new(pat) {
            if let Some(caps) = re.captures(value) {
                if let Some(m) = caps.get(1) {
                    return m.as_str().to_string();
                }
                return caps.get(0).unwrap().as_str().to_string();
            }
        }
    }
    value.to_string()
}

/// Strip markdown code block wrapper (```...```) from LLM output
#[allow(clippy::string_slice)]
pub fn strip_code_block(text: &str) -> String {
    let trimmed = text.trim();
    if trimmed.starts_with("```") && trimmed.ends_with("```") {
        let inner = &trimmed[3..trimmed.len() - 3];
        // Skip optional language tag on first line
        let inner = if let Some(newline_pos) = inner.find('\n') {
            let first_line = &inner[..newline_pos];
            if first_line.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '-') {
                &inner[newline_pos + 1..]
            } else {
                inner
            }
        } else {
            inner
        };
        inner.trim().to_string()
    } else {
        text.to_string()
    }
}

/// Generate a random token from nanosecond timestamp
pub fn generate_token() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("{:06x}", nanos % 0x1000000)
}

/// Build the full prompt for LLM inference
fn build_cached_prompt<Req: Promptable, Resp: Parseable>(request: &Req, token: &str, allow_planning: bool) -> CachedPrompt {
    let parts = request.to_markdown_cached();
    let schema = Resp::schema_markdown(token);
    let suffix = if allow_planning {
        format!(
            "\n\n### 输出规范 ###\n你必须严格按照以下格式输出，不要输出任何额外的解释或前言，直接从第一行开始按格式输出。\n\n{schema}\n\n如果你需要在输出前思考，可以使用 <planning-{token}>...</planning-{token}> 标签包裹你的思考过程（用中文思考）。思考内容是可选的。\n可以在实施action之前先记录planning（思考计划），也可以在关键action之后记录reflection（观察结果）。\n思考结束后必须严格按照上面的格式输出。示例：\n<planning-{token}>\n分析一下这个问题...\n</planning-{token}>\n（然后直接按格式输出）",
            schema = schema,
            token = token,
        )
    } else {
        format!(
            "\n\n### 输出规范 ###\n你必须严格按照以下格式输出。直接输出内容，不要输出任何思考过程、前言或解释。\n\n{schema}",
            schema = schema,
        )
    };

    if parts.has_volatile() {
        // Volatile split: stable = request prefix, delta = volatile fields + schema/planning
        CachedPrompt {
            stable: parts.stable,
            delta: format!("{}{}", parts.delta, suffix),
        }
    } else {
        // No volatile: stable = all request content, delta = schema/planning
        CachedPrompt {
            stable: parts.stable,
            delta: suffix,
        }
    }
}

/// Parse an error chunk from the spawned SSE task into an InferError.
///
/// The chunk format is `\n[ERROR] {message}` where message is either:
/// - `HTTP request failed: {details}` → HttpRequestFailed
/// - `HTTP {status}: {body}` → HttpStatus
#[allow(clippy::string_slice)]
fn parse_channel_error(chunk: &str) -> InferError {
    let msg = &chunk[9..]; // skip "\n[ERROR] "
    if msg.starts_with("HTTP request failed:") {
        InferError::new(InferErrorCode::HttpRequestFailed, msg)
    } else if msg.starts_with("HTTP ") {
        let status = msg[5..].split(':').next()
            .and_then(|s| s.trim().parse::<u16>().ok())
            .unwrap_or(0);
        InferError::new(InferErrorCode::HttpStatus(status), msg)
    } else {
        // Fallback: treat unknown error format as channel error
        InferError::new(InferErrorCode::HttpRequestFailed, msg)
    }
}

/// Stream inference: yields parsed elements one by one as they arrive (async)
///
/// Each time a complete element is detected in the stream (delimited by
/// `{TypeName}-{token}`), it is parsed and yielded immediately.
pub(crate) async fn stream_infer<Req, Resp>(
    channel: &dyn LlmChannel,
    request: &Req,
    allow_planning: bool,
) -> Result<StreamInfer<Resp>, InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    stream_infer_with_on_text(channel, request, None, None, None, None, allow_planning).await
}

/// Stream inference with optional text callback (async)
///
/// Like `stream_infer()`, but accepts an optional callback that receives
/// each raw text chunk as it arrives from the LLM stream. This enables
/// real-time logging or forwarding of the raw LLM output.
///
/// # Arguments
/// * `on_text` - Optional callback invoked with each raw chunk before parsing.
///   Pass `None` for no callback (equivalent to `stream_infer()`).
pub(crate) async fn stream_infer_with_on_text<Req, Resp>(
    channel: &dyn LlmChannel,
    request: &Req,
    on_text: Option<Box<dyn FnMut(&str) + Send>>,
    on_input: Option<Box<dyn FnOnce(&str) + Send>>,
    cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
    on_thinking: Option<Box<dyn FnMut(&str) + Send>>,
    allow_planning: bool,
) -> Result<StreamInfer<Resp>, InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    let token = generate_token();
    let cached_prompt = build_cached_prompt::<Req, Resp>(request, &token, allow_planning);
    if let Some(cb) = on_input {
        let merged = cached_prompt.merge();
        cb(&merged);
    }
    let receiver = {
        let mut last_err = String::new();
        let mut result = None;
        for start_attempt in 0..=INFER_MAX_RETRIES {
            if start_attempt > 0 {
                let backoff_secs = crate::ioutil::exponential_backoff(
                    start_attempt, INFER_RETRY_BASE_SECS, INFER_RETRY_MAX_EXPONENT, INFER_RETRY_CAP_SECS
                );
                eprintln!(
                    "[INFER-RETRY] start_stream attempt {}/{}: backing off {}s",
                    start_attempt + 1, INFER_MAX_RETRIES + 1, backoff_secs
                );
                tokio::time::sleep(std::time::Duration::from_secs(backoff_secs)).await;
            }
            match channel.start_stream_cached(cached_prompt.clone()) {
                Ok(rx) => { result = Some(rx); break; }
                Err(e) => { last_err = e; }
            }
        }
        result.ok_or_else(|| InferError::new(InferErrorCode::StartStreamFailed, last_err))?
    };

    // If on_text callback provided, spawn independent drain task.
    // Drain task continuously receives chunks and calls on_text, even when
    // the caller of StreamInfer is awaiting other work (e.g. shell execution).
    // Without this, on_text only fires inside stream.next() — which stalls
    // when nobody polls it.
    let receiver = if let Some(mut on_text_cb) = on_text {
        let (drain_tx, drain_rx) = tokio::sync::mpsc::unbounded_channel();
        let cancel_clone = cancel.clone();
        tokio::spawn(async move {
            let mut receiver = receiver;
            while let Some(chunk) = receiver.recv().await {
                if let Some(ref c) = cancel_clone {
                    if c.load(std::sync::atomic::Ordering::Relaxed) {
                        break;
                    }
                }
                // Usage messages go to StreamInfer but not to on_text callback
                if chunk.starts_with("\n[USAGE:") {
                    if drain_tx.send(chunk).is_err() { break; }
                    continue;
                }
                on_text_cb(&chunk);
                if drain_tx.send(chunk).is_err() {
                    break; // StreamInfer dropped
                }
            }
        });
        drain_rx
    } else {
        receiver
    };

    Ok(StreamInfer {
        receiver,
        buffer: String::new(),
        token,
        type_name: Resp::type_name().to_string(),
        done: false,
        parsed_count: 0,
        cancel,
        on_thinking,
        planning_started: false,
        thinking_done: !allow_planning,
        planning_seen: false,
        allow_planning,
        last_usage: None,
        _phantom: std::marker::PhantomData,
    })
}

/// Async stream that yields parsed elements from a streaming LLM response
pub struct StreamInfer<T: Parseable> {
    receiver: tokio::sync::mpsc::UnboundedReceiver<String>,
    buffer: String,
    token: String,
    type_name: String,
    done: bool,
    parsed_count: usize,
    cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
    on_thinking: Option<Box<dyn FnMut(&str) + Send>>,
    planning_started: bool,
    thinking_done: bool,
    planning_seen: bool,
    allow_planning: bool,
    last_usage: Option<InferUsage>,
    _phantom: std::marker::PhantomData<T>,
}

impl<T: Parseable> StreamInfer<T> {
    /// Get the token used for this inference session
    pub fn token(&self) -> &str {
        &self.token
    }

    /// Get usage data captured from the stream (available after stream completes).
    pub fn usage(&self) -> Option<&InferUsage> {
        self.last_usage.as_ref()
    }

    /// Take the accumulated buffer content (for resume on truncation).
    /// After calling this, the internal buffer is empty.
    pub fn take_buffer(&mut self) -> String {
        std::mem::take(&mut self.buffer)
    }


    /// Async next: yields the next parsed element from the stream
    #[allow(clippy::string_slice)]
    pub async fn next(&mut self) -> Option<Result<T, InferError>> {
        if self.done {
            return None;
        }

        let separator = format!("{}-{}", self.type_name, self.token);
        let end_marker = format!("{}-end-{}", self.type_name, self.token);

        loop {
            // Check cancel signal before processing
            if let Some(ref cancel) = self.cancel {
                if cancel.load(std::sync::atomic::Ordering::Relaxed) {
                    self.done = true;
                    return None;
                }
            }

            // Thinking detection: before preamble check, handle <planning-{token}>...</planning-{token}>
            if self.parsed_count == 0 && !self.thinking_done {
                let planning_open = format!("<planning-{}>", self.token);
                let planning_close = format!("</planning-{}>", self.token);
                if let Some(planning_start) = self.buffer.find(&planning_open) {
                    self.planning_started = true;
                    // Check that content before planning tag is only whitespace
                    let before_planning = self.buffer[..planning_start].trim();
                    if !before_planning.is_empty() {
                        // Non-empty content before planning tag = preamble
                        self.done = true;
                        let display = if before_planning.len() > 200 { &before_planning[..200] } else { before_planning };
                        return Some(Err(InferError::new(
                            InferErrorCode::FormatViolation,
                            format!(
                                "FORMAT VIOLATION: Output REJECTED. You MUST start with the separator `{}`, not garbage text. You wrote: `{}`. NO preamble, NO commentary before the separator. Your output gets thrown away every single time you do this.",
                                separator, display
                            ),
                        )));
                    }
                    if let Some(planning_end) = find_line_match(&self.buffer, &planning_close) {
                        // Complete planning block found
                        let thinking_content = self.buffer[planning_start + planning_open.len()..planning_end].trim().to_string();
                        if !thinking_content.is_empty() {
                            if let Some(ref mut cb) = self.on_thinking {
                                cb(&thinking_content);
                            }
                        }
                        // Remove planning block from buffer
                        self.buffer = self.buffer[planning_end + planning_close.len()..].to_string();
                        self.thinking_done = true;
                        self.planning_seen = true;
                        continue;
                    } else {
                        // Thinking open tag found but no close tag yet — need more data, skip preamble check
                        // Fall through to receiver read below
                    }
                } else {
                    // No planning open tag found yet
                    // If buffer could be a partial planning tag prefix, wait for more data
                    let trimmed = self.buffer.trim();
                    if !trimmed.is_empty() && !planning_open.starts_with(trimmed) {
                        self.thinking_done = true;
                    }
                    // Otherwise buffer is empty/whitespace or partial planning tag prefix — need more data
                }
            }

            // Hallucination detection: planning tag after planning completed or actions yielded
            let hallucination_tag = format!("<planning-{}>", self.token);
            if (self.planning_seen || self.parsed_count > 0) && self.buffer.contains(&hallucination_tag) {
                self.done = true;
                return Some(Err(InferError::new(
                    InferErrorCode::FormatViolation,
                    format!(
                        "[HALLUCINATION] Planning tag detected after {} action(s). Planning is only allowed before the first action. This output is rejected.",
                        self.parsed_count
                    ),
                )));
            }

            // Early preamble detection: check completed lines before first separator
            // Uses line-level matching to avoid substring false positives
            if self.allow_planning && self.parsed_count == 0 && self.thinking_done && find_line_match(&self.buffer, &separator).is_none() {
                if let Some(last_newline) = self.buffer.rfind('\n') {
                    let completed = &self.buffer[..last_newline];
                    for line in completed.lines() {
                        let t = line.trim();
                        if t.is_empty() || t.starts_with("```") {
                            continue;
                        }
                        // Non-empty, non-exempt line before any separator = preamble
                        self.done = true;
                        let display = if t.len() > 200 { &t[..200] } else { t };
                        return Some(Err(InferError::new(
                            InferErrorCode::FormatViolation,
                            format!(
                                "FORMAT VIOLATION: Output REJECTED. You MUST start with the separator `{}`, not garbage text. You wrote: `{}`. NO preamble, NO commentary before the separator. Your output gets thrown away every single time you do this.",
                                separator, display
                            ),
                        )));
                    }
                }
            }

            // Priority 1: Try to extract a complete element between two separators
            // Uses line-level matching to prevent substring matches in content
            let first_sep = find_line_match(&self.buffer, &separator);
            if let Some(first_pos) = first_sep {
                if self.parsed_count == 0 && first_pos > 0 {
                    if self.allow_planning {
                        // Fallback preamble check for content before first separator
                        let before = self.buffer[..first_pos].trim();
                        if !before.is_empty() && !before.lines().all(|l| {
                            let t = l.trim();
                            t.is_empty() || t.starts_with("```")
                        }) {
                            self.done = true;
                            let display = if before.len() > 200 { &before[..200] } else { before };
                            return Some(Err(InferError::new(
                                InferErrorCode::FormatViolation,
                                format!(
                                    "FORMAT VIOLATION: Output REJECTED. You MUST start with the separator `{}`, not garbage text. You wrote: `{}`. NO preamble, NO commentary before the separator. Your output gets thrown away every single time you do this.",
                                    separator, display
                                ),
                            )));
                        }
                    }
                    self.buffer = self.buffer[first_pos..].to_string();
                    continue;
                }

                let after_first = first_pos + separator.len();
                let next_sep = find_line_match(&self.buffer[after_first..], &separator);
                let next_end = find_line_match(&self.buffer[after_first..], &end_marker);

                let boundary = match (next_sep, next_end) {
                    (Some(s), Some(e)) => {
                        if s <= e {
                            Some(("sep", after_first + s))
                        } else {
                            Some(("end", after_first + e))
                        }
                    }
                    (Some(s), None) => Some(("sep", after_first + s)),
                    (None, Some(e)) => Some(("end", after_first + e)),
                    (None, None) => None,
                };

                match boundary {
                    Some(("sep", boundary_pos)) => {
                        let element_text = self.buffer[after_first..boundary_pos].trim();
                        if !element_text.is_empty() {
                            let parse_input = format!("{}\n{}\n{}", separator, element_text, end_marker);
                            self.buffer = self.buffer[boundary_pos..].to_string();
                            match T::from_markdown(&parse_input, &self.token) {
                                Ok(mut items) if !items.is_empty() => {
                                    self.parsed_count += 1;
                                    return Some(Ok(items.remove(0)));
                                }
                                Ok(_) => { continue; }
                                Err(e) => return Some(Err(InferError::new(InferErrorCode::ParseError, e))),
                            }
                        } else {
                            self.buffer = self.buffer[boundary_pos..].to_string();
                            continue;
                        }
                    }
                    Some(("end", boundary_pos)) => {
                        let element_text = self.buffer[after_first..boundary_pos].trim();
                        self.done = true;
                        if !element_text.is_empty() {
                            let parse_input = format!("{}\n{}\n{}", separator, element_text, end_marker);
                            match T::from_markdown(&parse_input, &self.token) {
                                Ok(mut items) if !items.is_empty() => {
                                    self.parsed_count += 1;
                                    return Some(Ok(items.remove(0)));
                                }
                                Ok(_) => return None,
                                Err(e) => return Some(Err(InferError::new(InferErrorCode::ParseError, e))),
                            }
                        }
                        return None;
                    }
                    _ => {
                        // No boundary found yet — need more data
                    }
                }
            }

            // Need more data - read from async receiver
            match self.receiver.recv().await {
                Some(chunk) => {
                    // Capture usage data from the stream
                    if chunk.starts_with("\n[USAGE:") {
                        if let Some(data) = chunk.strip_prefix("\n[USAGE:").and_then(|s| s.strip_suffix("]")) {
                            let parts: Vec<&str> = data.split(':').collect();
                            if parts.len() == 4 {
                                self.last_usage = Some(InferUsage {
                                    prompt_tokens: parts[0].parse().unwrap_or(0),
                                    completion_tokens: parts[1].parse().unwrap_or(0),
                                    cache_read_tokens: parts[2].parse().unwrap_or(0),
                                    cache_write_tokens: parts[3].parse().unwrap_or(0),
                                });
                            }
                        }
                        continue;
                    }

                    // Check for error signals from the spawned task
                    if chunk.starts_with("\n[ERROR] ") {
                        self.done = true;
                        return Some(Err(parse_channel_error(&chunk)));
                    }

                    self.buffer.push_str(&chunk);
                }
                None => {
                    // Channel closed — stream ended
                    self.done = true;

                    // State machine check: planning opened but never closed → Truncated
                    if self.planning_started && !self.thinking_done {
                        let tail = safe_tail(&self.buffer, 200);
                        return Some(Err(InferError::new(
                            InferErrorCode::Truncated,
                            format!(
                                "[{}] Planning block opened but never closed. Buffer tail (up to 200 chars): {}",
                                self.type_name, tail
                            ),
                        )));
                    }

                    if find_line_match(&self.buffer, &end_marker).is_some() {
                        return None;
                    }
                    let tail = safe_tail(&self.buffer, 200);
                    return Some(Err(InferError::new(
                        InferErrorCode::Truncated,
                        format!(
                            "[{}] Missing end marker '{}' after {} element(s). Buffer tail (up to 200 chars): {}",
                            self.type_name, end_marker, self.parsed_count, tail
                        ),
                    )));
                }
            }
        }
    }
}

/// Full inference: collects all elements from stream (async)
///
/// Sends request to LLM and parses the complete response.
/// For streaming element-by-element processing, use `stream_infer()`.
pub(crate) async fn infer<Req, Resp>(
    channel: &dyn LlmChannel,
    request: &Req,
    allow_planning: bool,
) -> Result<(Resp, Option<InferUsage>), InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    infer_with_on_text(channel, request, None, None, None, None, allow_planning).await
}

/// Extract short type name from full path (e.g. "my_crate::BeatRequest" -> "BeatRequest")
fn short_type_name<T>() -> &'static str {
    let full = std::any::type_name::<T>();
    full.rsplit("::").next().unwrap_or(full)
}

/// Full inference with optional text callback (async)
///
/// Like `infer()`, but accepts an optional callback that receives each raw
/// text chunk as it arrives from the LLM stream.
///
/// Automatically logs `[INFER-RESULT]` with request type name, success/failure,
/// token breakdown, and elapsed time.
pub(crate) async fn infer_with_on_text<Req, Resp>(
    channel: &dyn LlmChannel,
    request: &Req,
    on_text: Option<Box<dyn FnMut(&str) + Send>>,
    on_input: Option<Box<dyn FnOnce(&str) + Send>>,
    cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
    on_thinking: Option<Box<dyn FnMut(&str) + Send>>,
    allow_planning: bool,
) -> Result<(Resp, Option<InferUsage>), InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    let label = short_type_name::<Req>();
    let start = std::time::Instant::now();
    let result = infer_with_on_text_inner::<Req, Resp>(
        channel, request, on_text, on_input, cancel, on_thinking, allow_planning,
    ).await;
    let elapsed = start.elapsed().as_secs_f64();
    match &result {
        Ok((_, usage)) => {
            if let Some(u) = usage {
                let total_input = u.prompt_tokens + u.cache_read_tokens + u.cache_write_tokens;
                eprintln!(
                    "[INFER-RESULT] {} SUCCESS: total_input={}(prompt={}+cache_read={}+cache_write={}) completion={}, elapsed={:.1}s",
                    label, total_input, u.prompt_tokens, u.cache_read_tokens, u.cache_write_tokens, u.completion_tokens, elapsed,
                );
            } else {
                eprintln!("[INFER-RESULT] {} SUCCESS: elapsed={:.1}s", label, elapsed);
            }
        }
        Err(e) => {
            eprintln!("[INFER-RESULT] {} FAILED: {}, elapsed={:.1}s", label, e, elapsed);
        }
    }
    result
}

async fn infer_with_on_text_inner<Req, Resp>(
    channel: &dyn LlmChannel,
    request: &Req,
    mut on_text: Option<Box<dyn FnMut(&str) + Send>>,
    mut on_input: Option<Box<dyn FnOnce(&str) + Send>>,
    cancel: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
    mut on_thinking: Option<Box<dyn FnMut(&str) + Send>>,
    allow_planning: bool,
) -> Result<(Resp, Option<InferUsage>), InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    let max_attempts = INFER_MAX_RETRIES + 1;
    let mut last_error: Option<InferError> = None;
    let mut resume_state: Option<(String, String)> = None; // (token, buffer) for resume
    let mut resume_grace: u32 = 1; // allow 1 free fallback when resume fails (e.g. 400 prefill not supported)

    let mut attempt: u32 = 0;
    while attempt < max_attempts {
        if attempt > 0 {
            let backoff_secs = crate::ioutil::exponential_backoff(
                attempt, INFER_RETRY_BASE_SECS, INFER_RETRY_MAX_EXPONENT, INFER_RETRY_CAP_SECS
            );
            eprintln!(
                "[INFER-RETRY] Attempt {}/{}: backing off {}s{}(last error: {})",
                attempt + 1, max_attempts, backoff_secs,
                if resume_state.is_some() { " [RESUME] " } else { " " },
                last_error.as_ref().map(|e| e.to_string()).unwrap_or_default(),
            );
            tokio::time::sleep(std::time::Duration::from_secs(backoff_secs)).await;
        }

        let was_resume = resume_state.is_some();

        // Build stream: resume path or normal path
        let mut stream: StreamInfer<Resp> = if let Some((ref saved_token, ref saved_buffer)) = resume_state {
            // Resume path: reuse saved token and buffer, send prefill to LLM
            let cached_prompt = build_cached_prompt::<Req, Resp>(request, saved_token, allow_planning);
            let receiver = {
                let mut last_err = String::new();
                let mut result = None;
                for start_attempt in 0..=INFER_MAX_RETRIES {
                    if start_attempt > 0 {
                        let bs = crate::ioutil::exponential_backoff(
                            start_attempt, INFER_RETRY_BASE_SECS, INFER_RETRY_MAX_EXPONENT, INFER_RETRY_CAP_SECS
                        );
                        eprintln!(
                            "[INFER-RETRY] resume start_stream attempt {}/{}: backing off {}s",
                            start_attempt + 1, INFER_MAX_RETRIES + 1, bs
                        );
                        tokio::time::sleep(std::time::Duration::from_secs(bs)).await;
                    }
                    match channel.start_stream_with_resume(cached_prompt.clone(), saved_buffer) {
                        Ok(rx) => { result = Some(rx); break; }
                        Err(e) => { last_err = e; }
                    }
                }
                match result {
                    Some(rx) => rx,
                    None => {
                        let e = InferError::new(InferErrorCode::StartStreamFailed, last_err);
                        if e.code.is_retryable() && attempt + 1 < max_attempts {
                            last_error = Some(e);
                            resume_state = None;
                            if resume_grace > 0 {
                                resume_grace -= 1;
                                eprintln!("[INFER-RESUME] Resume start_stream failed, falling back to fresh retry (not consuming attempt)");
                                // Don't increment attempt - resume failure is free
                            } else {
                                attempt += 1;
                            }
                            continue;
                        }
                        return Err(e);
                    }
                }
            };
            // Reconstruct StreamInfer with pre-filled buffer
            let planning_open = format!("<planning-{}>", saved_token);
            let planning_close = format!("</planning-{}>", saved_token);
            let has_planning_open = saved_buffer.contains(&planning_open);
            let has_planning_close = saved_buffer.contains(&planning_close);
            StreamInfer {
                receiver,
                buffer: saved_buffer.clone(),
                token: saved_token.clone(),
                type_name: Resp::type_name().to_string(),
                done: false,
                parsed_count: 0,
                cancel: cancel.clone(),
                on_thinking: None, // callbacks consumed on first attempt
                planning_started: has_planning_open,
                thinking_done: if allow_planning { has_planning_close } else { true },
                planning_seen: has_planning_open,
                allow_planning,
                last_usage: None,
                _phantom: std::marker::PhantomData,
            }
        } else {
            // Normal path (first attempt or non-truncation retry)
            let text_cb = on_text.take();
            let input_cb = on_input.take();
            let thinking_cb = on_thinking.take();
            match stream_infer_with_on_text::<Req, Resp>(
                channel, request, text_cb, input_cb, cancel.clone(), thinking_cb, allow_planning,
            ).await {
                Ok(s) => s,
                Err(e) => {
                    if e.code.is_retryable() && attempt + 1 < max_attempts {
                        last_error = Some(e);
                        attempt += 1;
                        continue;
                    }
                    return Err(e);
                }
            }
        };

        match stream.next().await {
            Some(Ok(resp)) => return Ok((resp, stream.usage().cloned())),
            Some(Err(e)) => {
                if e.code == InferErrorCode::Truncated && attempt + 1 < max_attempts {
                    // Save state for resume on next attempt
                    let token = stream.token().to_string();
                    let buffer = stream.take_buffer();
                    eprintln!(
                        "[INFER-RESUME] Truncated, saving {} bytes for resume (token={})",
                        buffer.len(), token
                    );
                    resume_state = Some((token, buffer));
                    last_error = Some(e);
                    attempt += 1;
                    continue;
                }
                if was_resume && e.code != InferErrorCode::Truncated && resume_grace > 0 {
                    // Resume path failed with non-truncation error (e.g. 400 prefill not supported)
                    // Don't consume attempt - fall back to fresh retry for free
                    resume_grace -= 1;
                    eprintln!(
                        "[INFER-RESUME] Resume failed ({}), falling back to fresh retry (not consuming attempt)",
                        e.code
                    );
                    resume_state = None;
                    last_error = Some(e);
                    continue;
                }
                if e.code.is_retryable() && attempt + 1 < max_attempts {
                    last_error = Some(e);
                    resume_state = None; // clear resume state for non-truncation retry
                    attempt += 1;
                    continue;
                }
                return Err(e);
            }
            None => return Err(InferError::new(InferErrorCode::ParseError, "Expected one output but got none")),
        }
    }

    Err(last_error.unwrap_or_else(|| InferError::new(InferErrorCode::ParseError, "Retry exhausted")))
}

// ============================================================
// Cancel Registry —帽子层推理中断管理
// ============================================================

/// Global cancel registry. Maps key (typically instance_id) to cancel token.
/// Framework-level singleton — managed by hat, not engine.
static CANCEL_REGISTRY: std::sync::OnceLock<
    std::sync::RwLock<std::collections::HashMap<String, std::sync::Arc<std::sync::atomic::AtomicBool>>>
> = std::sync::OnceLock::new();

fn get_registry() -> &'static std::sync::RwLock<std::collections::HashMap<String, std::sync::Arc<std::sync::atomic::AtomicBool>>> {
    CANCEL_REGISTRY.get_or_init(|| std::sync::RwLock::new(std::collections::HashMap::new()))
}

/// RAII cancel scope. Registers a cancel token on creation, unregisters on drop.
///
/// Create at instance/session level. Pass `token()` to inference APIs.
/// Cancel from API handlers via `cancel_infer()`.
pub struct CancelScope {
    key: String,
    token: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl CancelScope {
    /// Register a new cancel scope with the given key.
    pub fn new(key: impl Into<String>) -> Self {
        let key = key.into();
        let token = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        get_registry().write().unwrap().insert(key.clone(), token.clone());
        Self { key, token }
    }

    /// Get the cancel token to pass to inference APIs.
    pub fn token(&self) -> std::sync::Arc<std::sync::atomic::AtomicBool> {
        self.token.clone()
    }

    /// Check if cancellation was requested, and clear the flag.
    /// Returns true if cancel was requested since last check.
    pub fn is_cancelled(&self) -> bool {
        self.token.swap(false, std::sync::atomic::Ordering::SeqCst)
    }
}

impl Drop for CancelScope {
    fn drop(&mut self) {
        if let Some(registry) = CANCEL_REGISTRY.get() {
            if let Ok(mut map) = registry.write() {
                map.remove(&self.key);
            }
        }
    }
}

/// Cancel an ongoing inference by key. Returns true if a matching scope was found.
///
/// Called by API handlers to interrupt inference for a specific instance.
pub fn cancel_infer(key: &str) -> bool {
    if let Some(registry) = CANCEL_REGISTRY.get() {
        if let Ok(map) = registry.read() {
            if let Some(token) = map.get(key) {
                token.store(true, std::sync::atomic::Ordering::SeqCst);
                return true;
            }
        }
    }
    false
}



/// HTTP POST JSON request via shared connection pool.
/// Returns `Critical` wrapping the parsed JSON response.
/// Non-2xx status codes are treated as errors.
pub(crate) async fn http_post_json(
    url: &str,
    headers: &[(&str, &str)],
    body: &serde_json::Value,
) -> Critical<serde_json::Value> {
    let client = shared_http_client();
    let mut req = client.post(url);
    for &(k, v) in headers {
        req = req.header(k, v);
    }
    let response = match req.json(body).send().await {
        Ok(r) => r,
        Err(e) => return Critical::error("http_post_json", format!("request failed: {}", e)),
    };
    let status = response.status();
    let text = match response.text().await {
        Ok(t) => t,
        Err(e) => return Critical::error("http_post_json", format!("response read: {}", e)),
    };
    if !status.is_success() {
        let preview_end = text.char_indices().nth(500).map_or(text.len(), |(i, _)| i);
        return Critical::error(
            "http_post_json",
            format!("HTTP {} — {}", status.as_u16(), &text[..preview_end]),
        );
    }
    match serde_json::from_str(&text) {
        Ok(json) => Critical::ok(json),
        Err(e) => {
            let preview_end = text.char_indices().nth(500).map_or(text.len(), |(i, _)| i);
            Critical::error(
                "http_post_json",
                format!("JSON parse: {} — {}", e, &text[..preview_end]),
            )
        }
    }
}


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

    #[test]
    fn test_find_line_match_exact_line() {
        let text = "hello\nAction-f54ce2\nworld";
        assert_eq!(find_line_match(text, "Action-f54ce2"), Some(6));
    }

    #[test]
    fn test_find_line_match_substring_no_match() {
        // Key scenario: separator appears as substring within a line, should NOT match
        let text = "hello\nsome text Action-f54ce2 more text\nworld";
        assert_eq!(find_line_match(text, "Action-f54ce2"), None);
    }

    #[test]
    fn test_find_line_match_with_whitespace() {
        let text = "hello\n  Action-f54ce2  \nworld";
        // trim() should match, returns offset of pattern within the line
        assert!(find_line_match(text, "Action-f54ce2").is_some());
    }

    #[test]
    fn test_find_line_match_first_line() {
        let text = "Action-f54ce2\nhello\nworld";
        assert_eq!(find_line_match(text, "Action-f54ce2"), Some(0));
    }

    #[test]
    fn test_find_line_match_last_line_no_newline() {
        let text = "hello\nAction-f54ce2";
        assert_eq!(find_line_match(text, "Action-f54ce2"), Some(6));
    }

    #[test]
    fn test_find_line_match_no_match() {
        let text = "hello\nworld\nfoo";
        assert_eq!(find_line_match(text, "Action-f54ce2"), None);
    }

    #[test]
    fn test_find_line_match_empty_text() {
        assert_eq!(find_line_match("", "Action-f54ce2"), None);
    }

    #[test]
    fn test_find_line_match_multiple_occurrences() {
        // Should return the first match
        let text = "Action-f54ce2\nhello\nAction-f54ce2";
        assert_eq!(find_line_match(text, "Action-f54ce2"), Some(0));
    }

    #[test]
    fn test_cache_stats_record_and_query() {
        // Baseline: capture current state (global collector may have data from other tests)
        let baseline = cache_stats(0);

        // Record some data points
        record_cache_data(1000, 200, 800, 0);   // cache hit: 800 read
        record_cache_data(500, 100, 0, 500);     // cache miss: 500 write
        record_cache_data(1500, 300, 1200, 0);   // cache hit: 1200 read

        let stats = cache_stats(0);
        let added_requests = stats.total_requests - baseline.total_requests;
        let added_input = stats.total_input_tokens - baseline.total_input_tokens;
        let added_cached = stats.cached_tokens - baseline.cached_tokens;
        let added_output = stats.total_output_tokens - baseline.total_output_tokens;
        let added_hits = stats.cache_hit_requests - baseline.cache_hit_requests;

        assert_eq!(added_requests, 3);
        assert_eq!(added_hits, 2);          // 2 of 3 had cache_read > 0
        assert_eq!(added_input, 5500);      // (1000+800+0)+(500+0+500)+(1500+1200+0)
        assert_eq!(added_cached, 2000);     // 800+1200
        assert_eq!(added_output, 600);      // 200+100+300

        // hit_rate should be > 0 (cache_read > 0 globally)
        assert!(stats.hit_rate > 0.0);
    }

    #[test]
    fn test_cache_stats_window_filter() {
        // Record a data point, then query with window_secs > 0
        record_cache_data(100, 50, 80, 0);
        let stats = cache_stats(1);
        // Should include at least the point we just recorded (within 1 second)
        assert!(stats.total_requests > 0);
        assert!(stats.total_input_tokens > 0);
    }

    #[test]
    fn test_cache_stats_serializable() {
        let stats = cache_stats(0);
        let json = serde_json::to_string(&stats).unwrap();
        assert!(json.contains("total_requests"));
        assert!(json.contains("hit_rate"));
    }

    #[test]
    fn test_find_line_match_realistic_scenario() {
        // Realistic: thinking content contains separator as substring
        let text = "Action-abc123\nthinking\nseparator格式统一（Action-abc123 + 字段-abc123）✅\nAction-abc123\nsend_msg\n";
        // First match should be at position 0 (the actual separator line)
        let first = find_line_match(text, "Action-abc123");
        assert_eq!(first, Some(0));

        // After first separator + "thinking\n...", find next separator
        // Skip past first separator line
        let after_first = "Action-abc123\n".len();
        let rest = &text[after_first..];
        let second = find_line_match(rest, "Action-abc123");
        // Should find the second real separator, NOT the substring in thinking
        assert!(second.is_some());
        // The substring line should be skipped
        let matched_pos = second.unwrap();
        let matched_line_start = after_first + matched_pos;
        assert!(text[matched_line_start..].starts_with("Action-abc123\nsend_msg"));
    }
}