// ============================================================
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,
    /// Per-inference random token (used for end-marker, separators, planning tags).
    /// Empty when constructed without an inference context (e.g. direct channel calls).
    pub token: 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(), token: String::new() }
    }

    /// Returns the end-marker string for this prompt's token, or empty if no token.
    pub fn end_marker(&self) -> String {
        if self.token.is_empty() {
            String::new()
        } else {
            format!("end-marker-{}", self.token)
        }
    }

    /// 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<StreamChunk>, 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<StreamChunk>, 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<StreamChunk>, 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,
    #[serde(default)]
    finish_reason: Option<String>,
}

#[derive(serde::Deserialize)]
struct SseDelta {
    content: Option<String>,
    #[serde(default)]
    reasoning: Option<String>,
    #[serde(default)]
    reasoning_content: Option<String>,
    #[serde(default)]
    thinking: Option<String>,
    #[serde(default)]
    thought: Option<String>,
    #[serde(default)]
    reasoning_details: Vec<SseReasoningDetail>,
}

#[derive(serde::Deserialize)]
struct SseReasoningDetail {
    #[serde(default)]
    summary: Option<String>,
    #[serde(default)]
    text: Option<String>,
}

impl SseDelta {
    fn reasoning_text(&self) -> String {
        let mut out = String::new();
        for part in [
            self.reasoning.as_deref(),
            self.reasoning_content.as_deref(),
            self.thinking.as_deref(),
            self.thought.as_deref(),
        ]
        .into_iter()
        .flatten()
        {
            out.push_str(part);
        }
        for detail in &self.reasoning_details {
            if let Some(summary) = detail.summary.as_deref() {
                out.push_str(summary);
            }
            if let Some(text) = detail.text.as_deref() {
                out.push_str(text);
            }
        }
        out
    }
}

// ============================================================
// StreamChunk: typed messages on the SSE channel
// ============================================================

/// Typed message carried on the internal SSE channel.
///
/// Replaces the old string-tag in-band signalling
/// (`\n[ERROR] ...`, `\n[USAGE:...]`) with explicit variants.
#[derive(Debug, Clone)]
pub enum StreamChunk {
    /// Raw content delta from LLM (text). This feeds the Parseable buffer.
    Content(String),
    /// Non-body reasoning/thinking delta from OpenAI-compatible providers.
    ///
    /// This is forwarded to `on_text` for real-time display, but it is NOT
    /// appended to the Parseable content buffer.
    Reasoning(String),
    /// Token usage snapshot captured from the SSE stream.
    Usage(InferUsage),
    /// Error signal — upstream transport/HTTP/parse failure.
    Error(String),
    /// Stream-end signal with explicit reason.
    /// - clean = true when `finish_reason == "stop"` (content complete)
    /// - clean = false when `finish_reason == "length"` (token-budget truncated)
    Eof { clean: bool },
}

// ============================================================
// 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,
}

static LLM_REQUEST_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);

fn next_llm_request_id() -> u64 {
    LLM_REQUEST_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
}

#[derive(Clone)]
struct CacheLogContext {
    request_id: u64,
    model: String,
    endpoint: String,
    key_tail: String,
}

impl CacheLogContext {
    fn new(request_id: u64, model: &str, endpoint: &str, api_key: &str) -> Self {
        Self {
            request_id,
            model: model.to_owned(),
            endpoint: endpoint.to_owned(),
            key_tail: secret_tail(api_key),
        }
    }

    fn prefix(&self) -> String {
        format!(
            "[LLM-CACHE req={} model={} endpoint={} key_tail={}]",
            self.request_id, self.model, self.endpoint, self.key_tail
        )
    }
}

fn secret_tail(secret: &str) -> String {
    let chars: Vec<char> = secret.chars().collect();
    let len = chars.len();
    let start = len.saturating_sub(4);
    chars[start..].iter().collect()
}

fn fnv1a64_hex_bytes(bytes: &[u8]) -> String {
    let mut hash: u64 = 0xcbf29ce484222325;
    for b in bytes {
        hash ^= u64::from(*b);
        hash = hash.wrapping_mul(0x100000001b3);
    }
    format!("{:016x}", hash)
}

fn fnv1a64_hex_str(text: &str) -> String {
    fnv1a64_hex_bytes(text.as_bytes())
}

fn json_body_digest(body: &serde_json::Value) -> (usize, String) {
    let bytes = serde_json::to_vec(body).unwrap_or_default();
    (bytes.len(), fnv1a64_hex_bytes(&bytes))
}

fn message_shape(body: &serde_json::Value) -> String {
    let stream = body.get("stream").and_then(|v| v.as_bool()).unwrap_or(false);
    let has_stream_options = body.get("stream_options").is_some();
    let has_max_tokens = body.get("max_tokens").is_some();
    let has_stop = body.get("stop").is_some();
    let messages = body
        .get("messages")
        .and_then(|v| v.as_array())
        .map(|items| {
            items
                .iter()
                .map(|m| {
                    let role = m.get("role").and_then(|v| v.as_str()).unwrap_or("?");
                    let content = m.get("content").unwrap_or(&serde_json::Value::Null);
                    let content_shape = if let Some(s) = content.as_str() {
                        format!("str:{}", s.len())
                    } else if let Some(arr) = content.as_array() {
                        let blocks = arr
                            .iter()
                            .map(|b| {
                                let typ = b.get("type").and_then(|v| v.as_str()).unwrap_or("?");
                                let text_len = b.get("text").and_then(|v| v.as_str()).map(|t| t.len()).unwrap_or(0);
                                let cache = b.get("cache_control").is_some();
                                format!("{}:{}:cache={}", typ, text_len, cache)
                            })
                            .collect::<Vec<_>>()
                            .join("+");
                        format!("array[{}]", blocks)
                    } else {
                        content.to_string()
                    };
                    format!("{}={}", role, content_shape)
                })
                .collect::<Vec<_>>()
                .join("|")
        })
        .unwrap_or_else(|| "no_messages".to_string());
    format!(
        "stream={} stream_options={} max_tokens={} stop={} messages={}",
        stream, has_stream_options, has_max_tokens, has_stop, messages
    )
}

fn log_body_shape(context: &CacheLogContext, label: &str, body: &serde_json::Value) {
    let (body_bytes, body_hash) = json_body_digest(body);
    eprintln!(
        "{} Body: label={} body_bytes={} body_hash={} shape={}",
        context.prefix(),
        label,
        body_bytes,
        body_hash,
        message_shape(body)
    );
}

/// Log SSE usage statistics including cache hit/miss information.
fn log_sse_usage(usage: &Option<SseUsage>, context: &CacheLogContext) {
    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!(
                "{} Usage: prompt={}, completion={}, cache_write={}, cache_read={}, prompt_tokens_details={:?}",
                context.prefix(),
                u.prompt_tokens, u.completion_tokens,
                cache_write, cache_read, u.prompt_tokens_details
            );
        } else {
            eprintln!(
                "{} Usage: prompt={}, completion={} (no cache activity), prompt_tokens_details={:?}",
                context.prefix(),
                u.prompt_tokens, u.completion_tokens, u.prompt_tokens_details
            );
        }
    } else {
        eprintln!("{} Usage: none", context.prefix());
    }
}

/// 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, end_marker: Option<String>, cache_context: CacheLogContext) -> tokio::sync::mpsc::UnboundedReceiver<StreamChunk> {
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

    tokio::spawn(async move {
        const END_MARKER_TAIL_ABORT_BYTES: usize = 1024;

        let mut last_usage: Option<SseUsage> = None;
        let mut last_finish_reason: Option<String> = None;
        let client = shared_http_client();
        // Stream control is independent of receiver state.
        // We keep reading until [DONE] or stream ends even if receiver dropped.
        // After the global end-marker appears, content is suppressed from callers.
        // We still prefer the provider's graceful stop signal; only if suppressed
        // tail exceeds 1KB do we close the response early to save tokens.
        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(StreamChunk::Error(format!("HTTP request failed: {}", e)));
                return;
            }
            Err(_) => {
                eprintln!("[SSE-ERROR] HTTP request timed out after 60s waiting for response from {}", url);
                let _ = tx.send(StreamChunk::Error(format!("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(StreamChunk::Error(format!("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();
        let mut pending_content = String::new();
        let mut end_marker_seen = false;
        let mut suppressed_tail_bytes: usize = 0;
        let end_marker_keep = end_marker
            .as_ref()
            .map(|marker| marker.len().saturating_sub(1))
            .unwrap_or(0);

        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(StreamChunk::Error(
                        "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(StreamChunk::Error(format!(
                        "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(StreamChunk::Error(
                            "SSE stream completed with [DONE] but 0 content chunks received".to_string(),
                        ));
                    }
                    if receiver_alive && !pending_content.is_empty() {
                        let content_to_send = std::mem::take(&mut pending_content);
                        if tx.send(StreamChunk::Content(content_to_send)).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;
                        }
                    }
                    log_sse_usage(&last_usage, &cache_context);
                    if receiver_alive {
                        send_usage_via_channel(&tx, &last_usage);
                        let fr = last_finish_reason.as_deref();
                        // Truncation risk: end-marker stop was injected, but finish_reason != "stop"
                        // → model was cut off (length/disconnect/etc), output must be discarded.
                        if end_marker.is_some() && fr != Some("stop") {
                            let _ = tx.send(StreamChunk::Error(format!(
                                "Truncation risk: finish_reason={:?} (expected \"stop\"), discarding output",
                                fr
                            )));
                        } else {
                            // fr=="stop" accepted regardless of whether model emitted end-marker
                            // (user's tolerance: optional end-marker, mandatory [stop]).
                            let clean = fr == Some("stop");
                            let _ = tx.send(StreamChunk::Eof { clean });
                        }
                    }
                    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(fr) = &choice.finish_reason {
                                last_finish_reason = Some(fr.clone());
                            }

                            let reasoning = choice.delta.reasoning_text();
                            if !reasoning.is_empty() && receiver_alive {
                                if tx.send(StreamChunk::Reasoning(reasoning)).is_err() {
                                    eprintln!("[SSE-DIAG] Receiver dropped after {}s, {} chunks, {} bytes while sending reasoning. Stream control continues until [DONE].",
                                        stream_start.elapsed().as_secs(), chunks_sent, total_bytes);
                                    receiver_alive = false;
                                }
                            }

                            if let Some(content) = &choice.delta.content {
                                if !content.is_empty() {
                                    chunks_sent += 1;
                                    last_content_time = std::time::Instant::now();

                                    if let Some(marker) = end_marker.as_deref() {
                                        if end_marker_seen {
                                            // Content after global end-marker is provider tail garbage:
                                            // do not yield it to on_text/caller. Keep reading for the
                                            // graceful provider stop, but abort if it grows too large.
                                            suppressed_tail_bytes += content.len();
                                            if suppressed_tail_bytes > END_MARKER_TAIL_ABORT_BYTES {
                                                eprintln!("[SSE-DIAG] Global end-marker tail exceeded {} bytes ({} bytes); closing response early to save tokens.",
                                                    END_MARKER_TAIL_ABORT_BYTES, suppressed_tail_bytes);
                                                log_sse_usage(&last_usage, &cache_context);
                                                if receiver_alive {
                                                    send_usage_via_channel(&tx, &last_usage);
                                                    let _ = tx.send(StreamChunk::Eof { clean: true });
                                                }
                                                return;
                                            }
                                            continue;
                                        }

                                        pending_content.push_str(content);

                                        if let Some(marker_pos) = pending_content.find(marker) {
                                            let content_before_marker = pending_content[..marker_pos].to_string();
                                            let tail_start = marker_pos + marker.len();
                                            let tail_after_marker = pending_content[tail_start..].to_string();
                                            pending_content.clear();
                                            end_marker_seen = true;
                                            suppressed_tail_bytes += tail_after_marker.len();

                                            if receiver_alive && !content_before_marker.is_empty() {
                                                if tx.send(StreamChunk::Content(content_before_marker)).is_err() {
                                                    eprintln!("[SSE-DIAG] Receiver dropped after {}s, {} chunks, {} bytes while sending content before end-marker.",
                                                        stream_start.elapsed().as_secs(), chunks_sent, total_bytes);
                                                    receiver_alive = false;
                                                }
                                            }

                                            eprintln!("[SSE-DIAG] Global end-marker observed after {}s, {} chunks, {} bytes; suppressing tail and waiting for provider stop.",
                                                stream_start.elapsed().as_secs(), chunks_sent, total_bytes);

                                            if suppressed_tail_bytes > END_MARKER_TAIL_ABORT_BYTES {
                                                eprintln!("[SSE-DIAG] Global end-marker tail exceeded {} bytes ({} bytes); closing response early to save tokens.",
                                                    END_MARKER_TAIL_ABORT_BYTES, suppressed_tail_bytes);
                                                log_sse_usage(&last_usage, &cache_context);
                                                if receiver_alive {
                                                    send_usage_via_channel(&tx, &last_usage);
                                                    let _ = tx.send(StreamChunk::Eof { clean: true });
                                                }
                                                return;
                                            }
                                            continue;
                                        }

                                        // Keep a suffix long enough to detect an end-marker split across SSE chunks.
                                        // Flush only the prefix that can no longer participate in a future marker match.
                                        if pending_content.len() > end_marker_keep {
                                            let mut flush_len = pending_content.len() - end_marker_keep;
                                            while flush_len > 0 && !pending_content.is_char_boundary(flush_len) {
                                                flush_len -= 1;
                                            }
                                            if flush_len > 0 {
                                                let content_to_send = pending_content[..flush_len].to_string();
                                                pending_content = pending_content[flush_len..].to_string();
                                                if receiver_alive {
                                                    if tx.send(StreamChunk::Content(content_to_send)).is_err() {
                                                        eprintln!("[SSE-DIAG] Receiver dropped after {}s, {} chunks, {} bytes. Stream control continues until [DONE] or end-marker.",
                                                            stream_start.elapsed().as_secs(), chunks_sent, total_bytes);
                                                        receiver_alive = false;
                                                    }
                                                }
                                            }
                                        }
                                    } else if receiver_alive {
                                        if tx.send(StreamChunk::Content(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;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }

            // 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(StreamChunk::Error(format!(
                    "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(StreamChunk::Error(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);
        }
        if receiver_alive && !pending_content.is_empty() {
            let content_to_send = std::mem::take(&mut pending_content);
            if tx.send(StreamChunk::Content(content_to_send)).is_err() {
                receiver_alive = false;
            }
        }
        log_sse_usage(&last_usage, &cache_context);
        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<StreamChunk>, 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(StreamChunk::Usage(InferUsage {
            prompt_tokens: u.prompt_tokens,
            completion_tokens: u.completion_tokens,
            cache_read_tokens: cr,
            cache_write_tokens: 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<StreamChunk>, String> {
        let url = self.build_url();
        let request_id = next_llm_request_id();
        let cache_context = CacheLogContext::new(request_id, &self.model, &url, &self.api_key);
        eprintln!(
            "{} Request: prompt_bytes={} prompt_hash={} (plain)",
            cache_context.prefix(),
            prompt.len(),
            fnv1a64_hex_str(&prompt)
        );
        let body = self.build_body(serde_json::json!(prompt));
        log_body_shape(&cache_context, "plain", &body);
        // No token available on this path, no end-marker stop injected
        Ok(spawn_sse_reader(url, self.api_key.clone(), body, None, cache_context))
    }

    fn start_stream_cached(&self, parts: CachedPrompt) -> Result<tokio::sync::mpsc::UnboundedReceiver<StreamChunk>, String> {
        let url = self.build_url();
        let request_id = next_llm_request_id();
        let cache_context = CacheLogContext::new(request_id, &self.model, &url, &self.api_key);
        let end_marker = parts.end_marker();
        let end_marker_opt = if end_marker.is_empty() { None } else { Some(end_marker.clone()) };
        let mut body = if parts.has_volatile() {
            eprintln!(
                "{} Request: stable={} bytes stable_hash={} delta={} bytes delta_hash={} token_hash={} (volatile enabled)",
                cache_context.prefix(),
                parts.stable.len(),
                fnv1a64_hex_str(&parts.stable),
                parts.delta.len(),
                fnv1a64_hex_str(&parts.delta),
                fnv1a64_hex_str(&parts.token)
            );
            // 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!(
                "{} Request: stable={} bytes stable_hash={} token_hash={} (no volatile, cache passthrough)",
                cache_context.prefix(),
                parts.stable.len(),
                fnv1a64_hex_str(&parts.stable),
                fnv1a64_hex_str(&parts.token)
            );
            self.build_body(serde_json::json!(parts.stable))
        };
        if !end_marker.is_empty() {
            body["stop"] = serde_json::json!([&end_marker]);
        }
        log_body_shape(&cache_context, "cached", &body);
        Ok(spawn_sse_reader(url, self.api_key.clone(), body, end_marker_opt, cache_context))
    }

    fn start_stream_with_resume(&self, parts: CachedPrompt, prefill: &str) -> Result<tokio::sync::mpsc::UnboundedReceiver<StreamChunk>, String> {
        let url = self.build_url();
        let request_id = next_llm_request_id();
        let cache_context = CacheLogContext::new(request_id, &self.model, &url, &self.api_key);
        let mut messages = if parts.has_volatile() {
            eprintln!(
                "{} Resume request: stable={} bytes stable_hash={} delta={} bytes delta_hash={} prefill={} bytes prefill_hash={} token_hash={}",
                cache_context.prefix(),
                parts.stable.len(),
                fnv1a64_hex_str(&parts.stable),
                parts.delta.len(),
                fnv1a64_hex_str(&parts.delta),
                prefill.len(),
                fnv1a64_hex_str(prefill),
                fnv1a64_hex_str(&parts.token)
            );
            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!(
                "{} Resume request: stable={} bytes stable_hash={} prefill={} bytes prefill_hash={} token_hash={}",
                cache_context.prefix(),
                parts.stable.len(),
                fnv1a64_hex_str(&parts.stable),
                prefill.len(),
                fnv1a64_hex_str(prefill),
                fnv1a64_hex_str(&parts.token)
            );
            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);
        }
        let end_marker = parts.end_marker();
        let end_marker_opt = if end_marker.is_empty() { None } else { Some(end_marker.clone()) };
        if !end_marker.is_empty() {
            body["stop"] = serde_json::json!([&end_marker]);
        }
        log_body_shape(&cache_context, "resume", &body);
        Ok(spawn_sse_reader(url, self.api_key.clone(), body, end_marker_opt, cache_context))
    }

}

/// 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, multi_output: bool) -> CachedPrompt {
    let parts = request.to_markdown_cached();
    let schema = Resp::schema_markdown(token);
    let suffix = if allow_planning {
        format!(
            "\n\n### Output format ###\nReturn only text that follows the schema below. Do not add a preface or explanation before the first schema line.\n\n{schema}\n\nOptional planning: if needed, write brief planning inside <planning-{token}>...</planning-{token}> before the action block. Planning is optional. After planning, output the action block exactly as specified by the schema.\nExample:\n<planning-{token}>\nBrief plan...\n</planning-{token}>\nThen output the action block.",
            schema = schema,
            token = token,
        )
    } else {
        format!(
            "\n\n### Output format ###\nReturn exactly the schema below. Do not include planning, preface, or explanation.\n\n{schema}",
            schema = schema,
        )
    };

    let output_hint = if multi_output {
        format!("\n\nYou may output multiple {} blocks. Each block must follow the schema above.", Resp::type_name())
    } else {
        format!("\n\nOutput exactly one {} block.", Resp::type_name())
    };
    let suffix = format!("{}{}", suffix, output_hint);

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

/// 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 {
    // StreamChunk::Error contents are already raw messages (no "\n[ERROR] " prefix).
    let msg = chunk;
    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, true, None).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,
    multi_output: bool,
    fixed_token: Option<String>,
) -> Result<StreamInfer<Resp>, InferError>
where
    Req: Promptable,
    Resp: Parseable,
{
    let token = fixed_token.unwrap_or_else(generate_token);
    let cached_prompt = build_cached_prompt::<Req, Resp>(request, &token, allow_planning, multi_output);
    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;
                    }
                }
                // Content and non-body reasoning both go to on_text for live display.
                // Only Content is later appended to the Parseable buffer; Reasoning is
                // side-channel text and must not pollute structured output parsing.
                match &chunk {
                    StreamChunk::Content(text) | StreamChunk::Reasoning(text) => {
                        on_text_cb(text);
                    }
                    StreamChunk::Usage(_) | StreamChunk::Error(_) | StreamChunk::Eof { .. } => {}
                }
                if drain_tx.send(chunk).is_err() {
                    break; // StreamInfer dropped
                }
            }
        });
        drain_rx
    } else {
        receiver
    };

    Ok(StreamInfer {
        receiver,
        clean_eof: false,
        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<StreamChunk>,
    /// Clean EOF seen from server (`finish_reason == "stop"` + `[DONE]`).
    /// When false and receiver closes, the last unterminated block is NOT yielded.
    clean_eof: bool,
    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);

        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;
                    // Content before the first planning tag is model preface noise.
                    // Allow and skip it; the planning block itself remains the first
                    // semantic part of the response. Preamble before the action
                    // separator is still rejected when no planning block is present.
                    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. Do not conclude that
                    // planning is absent just because the current partial
                    // stream chunk starts with arbitrary preface text: models
                    // may emit "preface" first and the <planning-{token}> tag
                    // in a later chunk. If there is no planning block, a real
                    // preamble before the action separator is still rejected by
                    // the first-separator fallback check below.
                }
            }

            // 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: expected start with `{}`, got: `{}`.\n\nTwo common causes:\n1. You wrote a different token after the type name. The token is per-turn; copy it exactly from the schema.\n2. You wrote commentary/preamble before the separator. The output must begin with the separator, nothing before it.",
                                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: expected start with `{}`, got: `{}`.\n\nTwo common causes:\n1. You wrote a different token after the type name. The token is per-turn; copy it exactly from the schema.\n2. You wrote commentary/preamble before the separator. The output must begin with the separator, nothing before it.",
                                    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);

                match next_sep {
                    Some(s) => {
                        let boundary_pos = after_first + s;
                        let element_text = self.buffer[after_first..boundary_pos].trim();
                        if !element_text.is_empty() {
                            let parse_input = format!("{}\n{}", separator, element_text);
                            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;
                        }
                    }
                    None => {
                        // No next separator found yet - need more data
                    }
                }
            }
            // Need more data - read from async receiver
            match self.receiver.recv().await {
                Some(StreamChunk::Content(text)) => {
                    self.buffer.push_str(&text);
                }
                Some(StreamChunk::Reasoning(_)) => {
                    // Reasoning/thinking deltas are delivered through on_text only.
                    // They are intentionally ignored by the Parseable buffer so that
                    // non-body thinking cannot corrupt structured action parsing.
                    continue;
                }
                Some(StreamChunk::Usage(usage)) => {
                    self.last_usage = Some(usage);
                    continue;
                }
                Some(StreamChunk::Error(msg)) => {
                    self.done = true;
                    return Some(Err(parse_channel_error(&msg)));
                }
                Some(StreamChunk::Eof { clean }) => {
                    self.clean_eof = clean;
                    continue;
                }
                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
                            ),
                        )));
                    }

                    // Try to parse remaining content after last separator
                    // ONLY when clean_eof was signaled (finish_reason==stop + [DONE]).
                    // Truncated or abnormally-closed streams must NOT yield the
                    // unterminated last block — it may be partially received.
                    if self.clean_eof {
                        if let Some(sep_pos) = find_line_match(&self.buffer, &separator) {
                            let after = sep_pos + separator.len();
                            let remaining = self.buffer[after..].trim();
                            if !remaining.is_empty() {
                                let parse_input = format!("{}\n{}", separator, remaining);
                                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)));
                                    }
                                    _ => {}
                                }
                            }
                        }
                        if self.parsed_count > 0 {
                            return None;
                        }
                    } else if self.parsed_count > 0 {
                        // Abnormal/truncated close after some actions already yielded:
                        // surface as Truncated so the upper layer can resume/retry,
                        // rather than a silent graceful end.
                        let tail = safe_tail(&self.buffer, 200);
                        return Some(Err(InferError::new(
                            InferErrorCode::Truncated,
                            format!(
                                "[{}] Stream ended without clean EOF after {} action(s). Buffer tail: {}",
                                self.type_name, self.parsed_count, tail
                            ),
                        )));
                    }
                    let tail = safe_tail(&self.buffer, 200);
                    return Some(Err(InferError::new(
                        InferErrorCode::Truncated,
                        format!(
                            "[{}] Stream ended without complete output. Buffer tail (up to 200 chars): {}",
                            self.type_name, 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 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, None).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 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,
    fixed_token: Option<String>,
) -> 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, fixed_token,
    ).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,
    fixed_token: Option<String>,
) -> 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, false);
            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,
                clean_eof: false,
                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, false, fixed_token.clone(),
            ).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));
    }


    #[derive(Debug, PartialEq)]
    struct PlanningPrefaceAction {
        content: String,
    }

    impl Reflectable for PlanningPrefaceAction {}

    impl Promptable for PlanningPrefaceAction {
        fn to_markdown_depth(&self, _depth: usize) -> String {
            self.content.clone()
        }

        fn schema_markdown(token: &str) -> String
        where
            Self: Sized,
        {
            format!("PlanningPrefaceAction-{token}\nreply\ncontent-{token}\n<value>\n")
        }
    }

    impl Parseable for PlanningPrefaceAction {
        fn type_name() -> &'static str {
            "PlanningPrefaceAction"
        }

        fn from_markdown(text: &str, token: &str) -> Result<Vec<Self>, String>
        where
            Self: Sized,
        {
            let content_marker = format!("content-{token}\n");
            let Some(pos) = text.find(&content_marker) else {
                return Err(format!("missing content marker in: {text}"));
            };
            let content = text[pos + content_marker.len()..].trim();
            if content.is_empty() {
                return Err("empty content".to_string());
            }
            Ok(vec![PlanningPrefaceAction {
                content: content.to_string(),
            }])
        }
    }

    #[tokio::test]
    async fn test_preface_before_planning_is_skipped() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let token = "tok";
        tx.send(StreamChunk::Content(format!(
            "Sure, I will think first.\n<planning-{token}>\nbrief plan\n</planning-{token}>\nPlanningPrefaceAction-{token}\nreply\ncontent-{token}\nok"
        )))
        .unwrap();
        tx.send(StreamChunk::Eof { clean: true }).unwrap();
        drop(tx);

        let mut stream = StreamInfer::<PlanningPrefaceAction> {
            receiver: rx,
            clean_eof: false,
            buffer: String::new(),
            token: token.to_string(),
            type_name: "PlanningPrefaceAction".to_string(),
            done: false,
            parsed_count: 0,
            cancel: None,
            on_thinking: None,
            planning_started: false,
            thinking_done: false,
            planning_seen: false,
            allow_planning: true,
            last_usage: None,
            _phantom: std::marker::PhantomData,
        };

        let item = stream.next().await.expect("expected parsed action").unwrap();
        assert_eq!(
            item,
            PlanningPrefaceAction {
                content: "ok".to_string()
            }
        );
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_preface_then_planning_split_across_chunks_is_skipped() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let token = "tok";
        tx.send(StreamChunk::Content("Sure, I will think first.".to_string()))
            .unwrap();
        tx.send(StreamChunk::Content(format!(
            "\n<planning-{token}>\nbrief plan\n</planning-{token}>\nPlanningPrefaceAction-{token}\nreply\ncontent-{token}\nok"
        )))
        .unwrap();
        tx.send(StreamChunk::Eof { clean: true }).unwrap();
        drop(tx);

        let mut stream = StreamInfer::<PlanningPrefaceAction> {
            receiver: rx,
            clean_eof: false,
            buffer: String::new(),
            token: token.to_string(),
            type_name: "PlanningPrefaceAction".to_string(),
            done: false,
            parsed_count: 0,
            cancel: None,
            on_thinking: None,
            planning_started: false,
            thinking_done: false,
            planning_seen: false,
            allow_planning: true,
            last_usage: None,
            _phantom: std::marker::PhantomData,
        };

        let item = stream.next().await.expect("expected parsed action").unwrap();
        assert_eq!(
            item,
            PlanningPrefaceAction {
                content: "ok".to_string()
            }
        );
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn test_preface_without_planning_is_still_rejected() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let token = "tok";
        tx.send(StreamChunk::Content(format!(
            "Sure, here is the action.\nPlanningPrefaceAction-{token}\nreply\ncontent-{token}\nok\n"
        )))
        .unwrap();
        drop(tx);

        let mut stream = StreamInfer::<PlanningPrefaceAction> {
            receiver: rx,
            clean_eof: false,
            buffer: String::new(),
            token: token.to_string(),
            type_name: "PlanningPrefaceAction".to_string(),
            done: false,
            parsed_count: 0,
            cancel: None,
            on_thinking: None,
            planning_started: false,
            thinking_done: false,
            planning_seen: false,
            allow_planning: true,
            last_usage: None,
            _phantom: std::marker::PhantomData,
        };

        let err = stream.next().await.expect("expected format violation").unwrap_err();
        assert_eq!(err.code, InferErrorCode::FormatViolation);
    }

    #[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"));
    }
}