kelora 0.4.0

use anyhow::Result;
use chrono::{DateTime, Utc};
use rhai::Dynamic;
use std::collections::HashMap;

use crate::engine::RhaiEngine;
use crate::event::Event;

// Re-export submodules
pub mod builders;
pub mod defaults;
pub mod multiline;
pub mod prefix_extractor;
pub mod prefix_parser;
pub mod stages;

// Re-export main types for convenience
pub use builders::*;
pub use defaults::*;
pub use multiline::*;
pub use prefix_extractor::*;
pub use prefix_parser::*;
pub use stages::*;

/// Formatted output from the pipeline with optional timestamp metadata
#[derive(Debug, Clone)]
pub struct FormattedOutput {
    pub line: String,
    pub timestamp: Option<DateTime<Utc>>,
}

impl FormattedOutput {
    pub fn new(line: String, timestamp: Option<DateTime<Utc>>) -> Self {
        Self { line, timestamp }
    }
}

/// Helper function to collect discovered levels and keys from an event for stats
fn collect_discovered_levels_and_keys(event: &Event, ctx: &mut PipelineContext) {
    // Collect discovered level
    for level_field_name in crate::event::LEVEL_FIELD_NAMES {
        if let Some(value) = event.fields.get(*level_field_name) {
            if let Ok(level_str) = value.clone().into_string() {
                if !level_str.is_empty() {
                    // Add to both ctx.tracker (for parallel) and thread-local tracking (for sequential)
                    // 1. Add to ctx.tracker
                    let key = "__kelora_stats_discovered_levels".to_string();
                    let current = ctx
                        .tracker
                        .get(&key)
                        .cloned()
                        .unwrap_or_else(|| Dynamic::from(rhai::Array::new()));

                    if let Ok(mut arr) = current.into_array() {
                        let level_dynamic = Dynamic::from(level_str.clone());
                        // Check if level already exists in array
                        if !arr.iter().any(|v| {
                            v.clone().into_string().unwrap_or_default()
                                == level_dynamic.clone().into_string().unwrap_or_default()
                        }) {
                            arr.push(level_dynamic);
                        }
                        ctx.tracker.insert(key.clone(), Dynamic::from(arr));
                        ctx.tracker
                            .insert(format!("__op_{}", key), Dynamic::from("unique"));
                    }

                    // 2. Add to thread-local tracking state (reuse existing track_unique pattern)
                    crate::rhai_functions::tracking::THREAD_TRACKING_STATE.with(|state| {
                        let mut state = state.borrow_mut();
                        let key = "__kelora_stats_discovered_levels";

                        let current = state
                            .get(key)
                            .cloned()
                            .unwrap_or_else(|| Dynamic::from(rhai::Array::new()));

                        if let Ok(mut arr) = current.into_array() {
                            let level_dynamic = Dynamic::from(level_str);
                            if !arr.iter().any(|v| {
                                v.clone().into_string().unwrap_or_default()
                                    == level_dynamic.clone().into_string().unwrap_or_default()
                            }) {
                                arr.push(level_dynamic);
                            }
                            state.insert(key.to_string(), Dynamic::from(arr));
                            state.insert(format!("__op_{}", key), Dynamic::from("unique"));
                        }
                    });
                    break; // Only take the first level field found
                }
            }
        }
    }

    // Collect discovered keys
    let key = "__kelora_stats_discovered_keys".to_string();
    let current = ctx
        .tracker
        .get(&key)
        .cloned()
        .unwrap_or_else(|| Dynamic::from(rhai::Array::new()));

    if let Ok(mut arr) = current.into_array() {
        for field_key in event.fields.keys() {
            let key_dynamic = Dynamic::from(field_key.clone());
            // Check if key already exists in array
            if !arr.iter().any(|v| {
                v.clone().into_string().unwrap_or_default()
                    == key_dynamic.clone().into_string().unwrap_or_default()
            }) {
                arr.push(key_dynamic.clone());
            }
        }
        ctx.tracker.insert(key.clone(), Dynamic::from(arr.clone()));
        ctx.tracker
            .insert(format!("__op_{}", key), Dynamic::from("unique"));

        // Also add to thread-local tracking state
        crate::rhai_functions::tracking::THREAD_TRACKING_STATE.with(|state| {
            let mut state = state.borrow_mut();
            let key = "__kelora_stats_discovered_keys";
            state.insert(key.to_string(), Dynamic::from(arr));
            state.insert(format!("__op_{}", key), Dynamic::from("unique"));
        });
    }
}

/// Core pipeline result types
#[derive(Debug, Clone)]
pub enum ScriptResult {
    Skip,
    Emit(Event),
    #[allow(dead_code)]
    EmitMultiple(Vec<Event>), // For future emit_each() support
    Error(String),
}

impl ScriptResult {
    /// Try to unwrap the event from Emit variant, returns error if not Emit
    #[allow(dead_code)]
    pub fn try_unwrap_emit(self) -> Result<Event> {
        match self {
            ScriptResult::Emit(event) => Ok(event),
            ScriptResult::Skip => Err(anyhow::anyhow!("Expected ScriptResult::Emit, got Skip")),
            ScriptResult::EmitMultiple(_) => Err(anyhow::anyhow!(
                "Expected ScriptResult::Emit, got EmitMultiple"
            )),
            ScriptResult::Error(msg) => Err(anyhow::anyhow!(
                "Expected ScriptResult::Emit, got Error: {}",
                msg
            )),
        }
    }
}

/// Shared context passed between pipeline stages
pub struct PipelineContext {
    pub config: PipelineConfig,
    pub tracker: HashMap<String, Dynamic>,
    pub window: Vec<Event>, // window[0] = current event, rest are previous
    pub rhai: RhaiEngine,
    pub meta: MetaData,
}

/// Pipeline configuration
#[derive(Debug, Clone)]
pub struct PipelineConfig {
    #[allow(dead_code)]
    // LEGACY: Remove during resiliency migration (see dev/resiliency-todos.md #4)
    pub error_report: crate::config::ErrorReportConfig,
    pub brief: bool,
    pub wrap: bool,
    pub color_mode: crate::config::ColorMode,
    /// Timestamp formatting configuration (display-only)
    pub timestamp_formatting: crate::config::TimestampFormatConfig,
    /// Exit on first error (fail-fast behavior) - new resiliency model
    pub strict: bool,
    /// Show detailed error information - new resiliency model (levels: 0-3)
    pub verbose: u8,
    /// Quiet mode level (0=normal, 1=suppress diagnostics, 2=suppress events, 3=suppress script output)
    pub quiet_level: u8,
    /// Disable emoji in error output
    pub no_emoji: bool,
}

/// Metadata about current processing context
#[derive(Debug, Clone, Default)]
pub struct MetaData {
    #[allow(dead_code)]
    pub filename: Option<String>,
    pub line_num: Option<usize>,
}

/// Core pipeline traits
///
/// Parse raw text lines into structured events
pub trait EventParser: Send + Sync {
    fn parse(&self, line: &str) -> Result<Event>;
}

/// Optional line-level filtering before parsing
pub trait LineFilter: Send {
    fn should_keep(&self, line: &str) -> bool;
}

/// Handle multi-line log records (future feature)
pub trait Chunker: Send {
    fn feed_line(&mut self, line: String) -> Option<String>;
    fn flush(&mut self) -> Option<String>;
    fn has_pending(&self) -> bool;
}

/// Manage sliding window of events (future feature)
pub trait WindowManager: Send {
    fn get_window(&self) -> Vec<Event>; // includes current as window[0]
    fn update(&mut self, current: &Event);
}

/// Core script processing stage (filters, execs, etc.)
pub trait ScriptStage: Send {
    fn apply(&mut self, event: Event, ctx: &mut PipelineContext) -> ScriptResult;
}

/// Optional event limiting (--take N)
pub trait EventLimiter: Send {
    fn allow(&mut self) -> bool;
    #[allow(dead_code)] // Used by implementors
    fn is_exhausted(&self) -> bool;
}

/// Format events for output
pub trait Formatter: Send + Sync {
    fn format(&self, event: &Event) -> String;

    /// Flush any pending formatter state at the end of processing
    fn finish(&self) -> Option<String> {
        None
    }
}

/// Write formatted output
#[allow(dead_code)]
pub trait OutputWriter: Send {
    fn write(&mut self, line: &str) -> std::io::Result<()>;
    fn flush(&mut self) -> std::io::Result<()>;
}

/// Main pipeline structure
pub struct Pipeline {
    pub line_filter: Option<Box<dyn LineFilter>>,
    pub chunker: Box<dyn Chunker>,
    pub parser: Box<dyn EventParser>,
    pub script_stages: Vec<Box<dyn ScriptStage>>,
    pub limiter: Option<Box<dyn EventLimiter>>,
    pub formatter: Box<dyn Formatter>,
    #[allow(dead_code)]
    pub output: Box<dyn OutputWriter>,
    pub window_manager: Box<dyn WindowManager>,
}

impl Pipeline {
    /// Process a single line through the entire pipeline
    /// This is the core method used by both sequential and parallel processing
    pub fn process_line(
        &mut self,
        line: String,
        ctx: &mut PipelineContext,
    ) -> Result<Vec<FormattedOutput>> {
        let mut results = Vec::new();

        // Line filter stage
        if let Some(filter) = &self.line_filter {
            if !filter.should_keep(&line) {
                return Ok(results);
            }
        }

        // Chunker stage (for multi-line records)
        if let Some(chunk) = self.chunker.feed_line(line) {
            // Parse stage
            let event = match self.parser.parse(&chunk) {
                Ok(mut e) => {
                    // Event was successfully created from chunk
                    crate::stats::stats_add_event_created();

                    // Collect discovered levels and keys for stats
                    collect_discovered_levels_and_keys(&e, ctx);

                    // Also track in Rhai context for parallel processing
                    ctx.tracker
                        .entry("__kelora_stats_events_created".to_string())
                        .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                        .or_insert(rhai::Dynamic::from(1i64));
                    ctx.tracker.insert(
                        "__op___kelora_stats_events_created".to_string(),
                        rhai::Dynamic::from("count"),
                    );

                    // Copy metadata from context to event
                    if let Some(line_num) = ctx.meta.line_num {
                        e.set_metadata(line_num, ctx.meta.filename.clone());
                    }

                    e
                }
                Err(err) => {
                    // Use unified error tracking system
                    crate::rhai_functions::tracking::track_error(
                        "parse",
                        ctx.meta.line_num,
                        &err.to_string(),
                        Some(&chunk),
                        ctx.meta.filename.as_deref(),
                        ctx.config.verbose,
                        ctx.config.quiet_level,
                        Some(&ctx.config),
                    );

                    // New resiliency model: skip unparseable lines by default,
                    // only propagate errors in strict mode
                    if ctx.config.strict {
                        return Err(err);
                    } else {
                        // Skip this line and continue processing
                        return Ok(results);
                    }
                }
            };

            // Update window manager
            self.window_manager.update(&event);
            ctx.window = self.window_manager.get_window();

            // Apply script stages (filters, execs, etc.)
            let mut result = ScriptResult::Emit(event);

            for stage in &mut self.script_stages {
                result = match result {
                    ScriptResult::Emit(event) => stage.apply(event, ctx),
                    ScriptResult::EmitMultiple(events) => {
                        // Process each event through remaining stages
                        let mut multi_results = Vec::new();
                        for event in events {
                            let original_line = event.original_line.clone(); // Capture before consuming
                            match stage.apply(event, ctx) {
                                ScriptResult::Emit(e) => multi_results.push(e),
                                ScriptResult::EmitMultiple(mut es) => multi_results.append(&mut es),
                                ScriptResult::Skip => {}
                                ScriptResult::Error(msg) => {
                                    // Use unified error tracking system
                                    crate::rhai_functions::tracking::track_error(
                                        "script",
                                        ctx.meta.line_num,
                                        &msg,
                                        Some(&original_line),
                                        ctx.meta.filename.as_deref(),
                                        ctx.config.verbose,
                                        ctx.config.quiet_level,
                                        Some(&ctx.config),
                                    );

                                    // New resiliency model: use strict flag
                                    if ctx.config.strict {
                                        return Err(anyhow::anyhow!(msg));
                                    } else {
                                        // Skip errors in resilient mode and continue processing
                                        return Ok(results);
                                    }
                                }
                            }
                        }
                        ScriptResult::EmitMultiple(multi_results)
                    }
                    other => other, // Skip or Error, stop processing
                };

                match &result {
                    ScriptResult::Skip | ScriptResult::Error(_) => break,
                    _ => {}
                }
            }

            // Handle final result
            match result {
                ScriptResult::Emit(event) => {
                    if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                        // Filter out empty events before formatting
                        if event.fields.is_empty() {
                            // Empty events are counted as filtered
                            crate::stats::stats_add_event_filtered();

                            // Also track in Rhai context for parallel processing
                            ctx.tracker
                                .entry("__kelora_stats_events_filtered".to_string())
                                .and_modify(|v| {
                                    *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                })
                                .or_insert(rhai::Dynamic::from(1i64));
                            ctx.tracker.insert(
                                "__op___kelora_stats_events_filtered".to_string(),
                                rhai::Dynamic::from("count"),
                            );
                        } else {
                            crate::stats::stats_add_event_output();

                            // Also track in Rhai context for parallel processing
                            ctx.tracker
                                .entry("__kelora_stats_events_output".to_string())
                                .and_modify(|v| {
                                    *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                })
                                .or_insert(rhai::Dynamic::from(1i64));
                            ctx.tracker.insert(
                                "__op___kelora_stats_events_output".to_string(),
                                rhai::Dynamic::from("count"),
                            );

                            let formatted = self.formatter.format(&event);
                            let timestamp = event.parsed_ts;
                            results.push(FormattedOutput::new(formatted, timestamp));
                        }
                    } else {
                        crate::stats::stats_add_event_filtered();

                        // Also track in Rhai context for parallel processing
                        ctx.tracker
                            .entry("__kelora_stats_events_filtered".to_string())
                            .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                            .or_insert(rhai::Dynamic::from(1i64));
                        ctx.tracker.insert(
                            "__op___kelora_stats_events_filtered".to_string(),
                            rhai::Dynamic::from("count"),
                        );
                    }
                }
                ScriptResult::EmitMultiple(events) => {
                    for event in events {
                        if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                            // Filter out empty events before formatting
                            if event.fields.is_empty() {
                                // Empty events are counted as filtered
                                crate::stats::stats_add_event_filtered();

                                // Also track in Rhai context for parallel processing
                                ctx.tracker
                                    .entry("__kelora_stats_events_filtered".to_string())
                                    .and_modify(|v| {
                                        *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                    })
                                    .or_insert(rhai::Dynamic::from(1i64));
                                ctx.tracker.insert(
                                    "__op___kelora_stats_events_filtered".to_string(),
                                    rhai::Dynamic::from("count"),
                                );
                            } else {
                                crate::stats::stats_add_event_output();

                                // Also track in Rhai context for parallel processing
                                ctx.tracker
                                    .entry("__kelora_stats_events_output".to_string())
                                    .and_modify(|v| {
                                        *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                    })
                                    .or_insert(rhai::Dynamic::from(1i64));
                                ctx.tracker.insert(
                                    "__op___kelora_stats_events_output".to_string(),
                                    rhai::Dynamic::from("count"),
                                );

                                let formatted = self.formatter.format(&event);
                                let timestamp = event.parsed_ts;
                                results.push(FormattedOutput::new(formatted, timestamp));
                            }
                        } else {
                            crate::stats::stats_add_event_filtered();

                            // Also track in Rhai context for parallel processing
                            ctx.tracker
                                .entry("__kelora_stats_events_filtered".to_string())
                                .and_modify(|v| {
                                    *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                })
                                .or_insert(rhai::Dynamic::from(1i64));
                            ctx.tracker.insert(
                                "__op___kelora_stats_events_filtered".to_string(),
                                rhai::Dynamic::from("count"),
                            );
                        }
                    }
                }
                ScriptResult::Skip => {
                    crate::stats::stats_add_event_filtered();

                    // Also track in Rhai context for parallel processing
                    ctx.tracker
                        .entry("__kelora_stats_events_filtered".to_string())
                        .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                        .or_insert(rhai::Dynamic::from(1i64));
                    ctx.tracker.insert(
                        "__op___kelora_stats_events_filtered".to_string(),
                        rhai::Dynamic::from("count"),
                    );
                }
                ScriptResult::Error(msg) => {
                    // Use unified error tracking system
                    crate::rhai_functions::tracking::track_error(
                        "script",
                        ctx.meta.line_num,
                        &msg,
                        None, // Original line not available at this stage
                        ctx.meta.filename.as_deref(),
                        ctx.config.verbose,
                        ctx.config.quiet_level,
                        Some(&ctx.config),
                    );

                    // New resiliency model: use strict flag
                    if ctx.config.strict {
                        return Err(anyhow::anyhow!(msg));
                    } else {
                        // Skip errors in resilient mode and continue processing
                        return Ok(results);
                    }
                }
            }
        }

        Ok(results)
    }

    /// Flush any remaining chunks from the chunker
    pub fn flush(&mut self, ctx: &mut PipelineContext) -> Result<Vec<FormattedOutput>> {
        if let Some(chunk) = self.chunker.flush() {
            // Process chunk directly, not through feed_line
            self.process_chunk_directly(chunk, ctx)
        } else {
            Ok(Vec::new())
        }
    }

    /// Flush formatter state to emit any remaining buffered output
    pub fn finish_formatter(&self) -> Option<FormattedOutput> {
        self.formatter
            .finish()
            .map(|line| FormattedOutput::new(line, None))
    }

    /// Process a chunk directly without going through the chunker
    fn process_chunk_directly(
        &mut self,
        chunk: String,
        ctx: &mut PipelineContext,
    ) -> Result<Vec<FormattedOutput>> {
        let mut results = Vec::new();

        // This is the same logic as in process_line starting from the "Parse stage" comment
        let event = match self.parser.parse(&chunk) {
            Ok(mut e) => {
                // Event was successfully created from chunk
                crate::stats::stats_add_event_created();

                // Collect discovered levels and keys for stats
                collect_discovered_levels_and_keys(&e, ctx);

                // Also track in Rhai context for parallel processing
                ctx.tracker
                    .entry("__kelora_stats_events_created".to_string())
                    .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                    .or_insert(rhai::Dynamic::from(1i64));
                ctx.tracker.insert(
                    "__op___kelora_stats_events_created".to_string(),
                    rhai::Dynamic::from("count"),
                );

                // Copy metadata from context to event
                if let Some(line_num) = ctx.meta.line_num {
                    e.set_metadata(line_num, ctx.meta.filename.clone());
                }

                e
            }
            Err(err) => {
                // Use unified error tracking system
                crate::rhai_functions::tracking::track_error(
                    "parse",
                    ctx.meta.line_num,
                    &err.to_string(),
                    Some(&chunk),
                    ctx.meta.filename.as_deref(),
                    ctx.config.verbose,
                    ctx.config.quiet_level,
                    Some(&ctx.config),
                );

                // New resiliency model: skip unparseable lines by default,
                // only propagate errors in strict mode
                if ctx.config.strict {
                    return Err(err);
                } else {
                    // Skip this line and continue processing
                    return Ok(results);
                }
            }
        };

        // Update window manager
        self.window_manager.update(&event);
        ctx.window = self.window_manager.get_window();

        // Apply script stages (filters, execs, etc.)
        let mut result = ScriptResult::Emit(event);

        for stage in &mut self.script_stages {
            result = match result {
                ScriptResult::Emit(event) => stage.apply(event, ctx),
                ScriptResult::EmitMultiple(events) => {
                    // Process each event through remaining stages
                    let mut multi_results = Vec::new();
                    for event in events {
                        let original_line = event.original_line.clone(); // Capture before consuming
                        match stage.apply(event, ctx) {
                            ScriptResult::Emit(e) => multi_results.push(e),
                            ScriptResult::EmitMultiple(mut es) => multi_results.append(&mut es),
                            ScriptResult::Skip => {}
                            ScriptResult::Error(msg) => {
                                // Use unified error tracking system
                                crate::rhai_functions::tracking::track_error(
                                    "script",
                                    ctx.meta.line_num,
                                    &msg,
                                    Some(&original_line),
                                    ctx.meta.filename.as_deref(),
                                    ctx.config.verbose,
                                    ctx.config.quiet_level,
                                    Some(&ctx.config),
                                );

                                // New resiliency model: use strict flag
                                if ctx.config.strict {
                                    return Err(anyhow::anyhow!(msg));
                                } else {
                                    // Skip errors in resilient mode and continue processing
                                    return Ok(results);
                                }
                            }
                        }
                    }
                    ScriptResult::EmitMultiple(multi_results)
                }
                other => other, // Skip or Error, stop processing
            };

            match &result {
                ScriptResult::Skip | ScriptResult::Error(_) => break,
                _ => {}
            }
        }

        // Handle final result
        match result {
            ScriptResult::Emit(event) => {
                if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                    // Filter out empty events before formatting
                    if event.fields.is_empty() {
                        // Empty events are counted as filtered
                        crate::stats::stats_add_event_filtered();

                        // Also track in Rhai context for parallel processing
                        ctx.tracker
                            .entry("__kelora_stats_events_filtered".to_string())
                            .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                            .or_insert(rhai::Dynamic::from(1i64));
                        ctx.tracker.insert(
                            "__op___kelora_stats_events_filtered".to_string(),
                            rhai::Dynamic::from("count"),
                        );
                    } else {
                        crate::stats::stats_add_event_output();

                        // Also track in Rhai context for parallel processing
                        ctx.tracker
                            .entry("__kelora_stats_events_output".to_string())
                            .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                            .or_insert(rhai::Dynamic::from(1i64));
                        ctx.tracker.insert(
                            "__op___kelora_stats_events_output".to_string(),
                            rhai::Dynamic::from("count"),
                        );

                        let formatted = self.formatter.format(&event);
                        let timestamp = event.parsed_ts;
                        results.push(FormattedOutput::new(formatted, timestamp));
                    }
                } else {
                    crate::stats::stats_add_event_filtered();

                    // Also track in Rhai context for parallel processing
                    ctx.tracker
                        .entry("__kelora_stats_events_filtered".to_string())
                        .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                        .or_insert(rhai::Dynamic::from(1i64));
                    ctx.tracker.insert(
                        "__op___kelora_stats_events_filtered".to_string(),
                        rhai::Dynamic::from("count"),
                    );
                }
            }
            ScriptResult::EmitMultiple(events) => {
                for event in events {
                    if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                        // Filter out empty events before formatting
                        if event.fields.is_empty() {
                            // Empty events are counted as filtered
                            crate::stats::stats_add_event_filtered();

                            // Also track in Rhai context for parallel processing
                            ctx.tracker
                                .entry("__kelora_stats_events_filtered".to_string())
                                .and_modify(|v| {
                                    *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                })
                                .or_insert(rhai::Dynamic::from(1i64));
                            ctx.tracker.insert(
                                "__op___kelora_stats_events_filtered".to_string(),
                                rhai::Dynamic::from("count"),
                            );
                        } else {
                            crate::stats::stats_add_event_output();

                            // Also track in Rhai context for parallel processing
                            ctx.tracker
                                .entry("__kelora_stats_events_output".to_string())
                                .and_modify(|v| {
                                    *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1)
                                })
                                .or_insert(rhai::Dynamic::from(1i64));
                            ctx.tracker.insert(
                                "__op___kelora_stats_events_output".to_string(),
                                rhai::Dynamic::from("count"),
                            );

                            let formatted = self.formatter.format(&event);
                            let timestamp = event.parsed_ts;
                            results.push(FormattedOutput::new(formatted, timestamp));
                        }
                    } else {
                        crate::stats::stats_add_event_filtered();

                        // Also track in Rhai context for parallel processing
                        ctx.tracker
                            .entry("__kelora_stats_events_filtered".to_string())
                            .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                            .or_insert(rhai::Dynamic::from(1i64));
                        ctx.tracker.insert(
                            "__op___kelora_stats_events_filtered".to_string(),
                            rhai::Dynamic::from("count"),
                        );
                    }
                }
            }
            ScriptResult::Skip => {
                crate::stats::stats_add_event_filtered();

                // Also track in Rhai context for parallel processing
                ctx.tracker
                    .entry("__kelora_stats_events_filtered".to_string())
                    .and_modify(|v| *v = rhai::Dynamic::from(v.as_int().unwrap_or(0) + 1))
                    .or_insert(rhai::Dynamic::from(1i64));
                ctx.tracker.insert(
                    "__op___kelora_stats_events_filtered".to_string(),
                    rhai::Dynamic::from("count"),
                );
            }
            ScriptResult::Error(msg) => {
                // Use unified error tracking system
                crate::rhai_functions::tracking::track_error(
                    "script",
                    ctx.meta.line_num,
                    &msg,
                    None, // Original line not available at this stage
                    ctx.meta.filename.as_deref(),
                    ctx.config.verbose,
                    ctx.config.quiet_level,
                    Some(&ctx.config),
                );

                // New resiliency model: use strict flag
                if ctx.config.strict {
                    return Err(anyhow::anyhow!(msg));
                } else {
                    // Skip errors in resilient mode and continue processing
                    return Ok(results);
                }
            }
        }

        Ok(results)
    }

    /// Check if the event limiter (--take N) is exhausted
    #[allow(dead_code)] // Used by parallel processing logic
    pub fn is_take_limit_exhausted(&self) -> bool {
        self.limiter.as_ref().is_some_and(|l| l.is_exhausted())
    }

    /// Check if the chunker currently holds a partial chunk that hasn't been emitted yet
    pub fn has_pending_chunk(&self) -> bool {
        self.chunker.has_pending()
    }
}