#![allow(dead_code)] // Pipeline API exposes embedding/legacy hooks not all used by the current binary
use anyhow::{anyhow, Result};
use chrono::{DateTime, Utc};
use rhai::Dynamic;
use std::collections::HashMap;

use crate::engine::RhaiEngine;
use crate::event::{Event, SpanStatus};
use crate::rhai_functions::file_ops::{self, FileOp};
use crate::rhai_functions::tracking;
use span::SpanProcessor;

// Re-export submodules
pub mod builders;
pub mod defaults;
pub mod multiline;
pub mod prefix_extractor;
pub mod prefix_parser;
pub mod section_selector;
mod span;
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 section_selector::*;
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>>,
    pub file_ops: Vec<FileOp>,
}

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

    pub fn with_ops(line: String, timestamp: Option<DateTime<Utc>>, file_ops: Vec<FileOp>) -> Self {
        Self {
            line,
            timestamp,
            file_ops,
        }
    }
}

/// 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.internal_tracker (for parallel) and thread-local tracking (for sequential)
                    // 1. Add to ctx.internal_tracker
                    let key = "__kelora_stats_discovered_levels".to_string();
                    let current = ctx
                        .internal_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.internal_tracker.insert(key.clone(), Dynamic::from(arr));
                        ctx.internal_tracker
                            .insert(format!("__op_{}", key), Dynamic::from("unique"));
                    }

                    // 2. Add to thread-local tracking state (reuse existing track_unique pattern)
                    tracking::with_internal_tracking(|state| {
                        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.clone());
                            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
        .internal_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.internal_tracker
            .insert(key.clone(), Dynamic::from(arr.clone()));
        ctx.internal_tracker
            .insert(format!("__op_{}", key), Dynamic::from("unique"));

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

/// Core pipeline result types
#[derive(Debug, Clone)]
pub enum ScriptResult {
    Skip,
    Emit(Event),
    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
    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 internal_tracker: HashMap<String, Dynamic>,
    pub window: Vec<Event>, // window[0] = current event, rest are previous
    pub rhai: RhaiEngine,
    pub meta: MetaData,
    pub pending_file_ops: Vec<FileOp>,
}

/// Pipeline configuration
#[derive(Debug, Clone)]
pub struct PipelineConfig {
    // LEGACY: Remove during resiliency migration (see dev/resiliency-todos.md #4)
    pub error_report: crate::config::ErrorReportConfig,
    pub brief: bool,
    pub wrap: bool,
    pub pretty: 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,
    /// Suppress formatter/event output
    pub quiet_events: bool,
    /// Suppress diagnostics and summaries
    pub suppress_diagnostics: bool,
    /// Suppress all stdout/stderr emitters except the fatal line
    pub silent: bool,
    /// Suppress Rhai print/eprint and side-effect warnings
    pub suppress_script_output: bool,
    /// Legacy quiet level (derived)
    pub quiet_level: u8,
    /// Disable emoji in error output
    pub no_emoji: bool,
    /// Input files for smart error message formatting
    pub input_files: Vec<String>,
    /// Allow Rhai scripts to create directories and write files on disk
    pub allow_fs_writes: bool,
}

/// Metadata about current processing context
#[derive(Debug, Clone, Default)]
pub struct MetaData {
    pub filename: Option<String>,
    pub line_num: Option<usize>,
    pub span_status: Option<crate::event::SpanStatus>,
    pub span_id: Option<String>,
    pub span_start: Option<DateTime<Utc>>,
    pub span_end: Option<DateTime<Utc>>,
}

/// 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;
    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
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>,
    pub output: Box<dyn OutputWriter>,
    pub window_manager: Box<dyn WindowManager>,
    pub span_processor: Option<SpanProcessor>,
    pub ts_config: crate::timestamp::TsConfig,
}

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 mut event = match self.parser.parse(&chunk) {
                Ok(mut e) => {
                    // Event was successfully created from chunk
                    crate::stats::stats_add_event_created();

                    // Track timestamp for time span statistics
                    if let Some(ts) = e.parsed_ts {
                        crate::stats::stats_update_timestamp(ts);
                    }

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

                    // Also track in Rhai context for parallel processing
                    ctx.internal_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.internal_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);
                    }
                }
            };

            if let Some(span_processor) = self.span_processor.as_mut() {
                span_processor.prepare_event(&mut event, ctx)?;
            }

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

            file_ops::clear_pending_ops();
            ctx.pending_file_ops.clear();

            // 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
            let remaining_ops = file_ops::take_pending_ops();
            if !remaining_ops.is_empty() {
                ctx.pending_file_ops.extend(remaining_ops);
            }

            self.apply_script_result(result, ctx, &mut 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())
        }
    }

    /// Process a complete event string (for pre-chunked multiline events)
    /// Skips the chunking stage and goes directly to parsing
    pub fn process_event_string(
        &mut self,
        event_string: String,
        ctx: &mut PipelineContext,
    ) -> Result<Vec<FormattedOutput>> {
        self.process_chunk_directly(event_string, ctx)
    }

    /// 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))
    }

    pub fn finish_spans(&mut self, ctx: &mut PipelineContext) -> Result<()> {
        if let Some(span_processor) = self.span_processor.as_mut() {
            span_processor.finish(ctx)?;
        }
        Ok(())
    }

    fn apply_script_result(
        &mut self,
        result: ScriptResult,
        ctx: &mut PipelineContext,
        outputs: &mut Vec<FormattedOutput>,
    ) -> Result<()> {
        match result {
            ScriptResult::Emit(mut event) => {
                if let Some(span) = self.span_processor.as_mut() {
                    span.prepare_emitted_event(&mut event);
                }

                let ops = std::mem::take(&mut ctx.pending_file_ops);

                if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                    if event.fields.is_empty() {
                        event.span.status = Some(SpanStatus::Filtered);
                        crate::stats::stats_add_event_filtered();

                        ctx.internal_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.internal_tracker.insert(
                            "__op___kelora_stats_events_filtered".to_string(),
                            rhai::Dynamic::from("count"),
                        );

                        if let Some(span) = self.span_processor.as_mut() {
                            span.handle_skip(ctx);
                        }

                        if !ops.is_empty() {
                            outputs.push(FormattedOutput::with_ops(String::new(), None, ops));
                        }
                    } else {
                        crate::stats::stats_add_event_output();

                        // Track result timestamp for time span statistics
                        let mut result_event = event.clone();
                        result_event.parsed_ts = None; // Clear to force re-extraction
                        result_event.extract_timestamp_with_config(None, &self.ts_config);
                        if let Some(result_ts) = result_event.parsed_ts {
                            crate::stats::stats_update_result_timestamp(result_ts);
                        }

                        ctx.internal_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.internal_tracker.insert(
                            "__op___kelora_stats_events_output".to_string(),
                            rhai::Dynamic::from("count"),
                        );

                        if let Some(span) = self.span_processor.as_mut() {
                            span.record_emitted_event(&event, ctx)?;
                        }

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

                    ctx.internal_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.internal_tracker.insert(
                        "__op___kelora_stats_events_filtered".to_string(),
                        rhai::Dynamic::from("count"),
                    );

                    event.span.status = Some(SpanStatus::Filtered);
                    if let Some(span) = self.span_processor.as_mut() {
                        span.handle_skip(ctx);
                    }

                    if !ops.is_empty() {
                        outputs.push(FormattedOutput::with_ops(String::new(), None, ops));
                    }
                }

                if let Some(span) = self.span_processor.as_mut() {
                    span.complete_pending();
                }
            }
            ScriptResult::EmitMultiple(events) => {
                let mut ops = std::mem::take(&mut ctx.pending_file_ops);

                for (idx, mut event) in events.into_iter().enumerate() {
                    if let Some(span) = self.span_processor.as_mut() {
                        span.prepare_emitted_event(&mut event);
                    }

                    if self.limiter.as_mut().is_none_or(|l| l.allow()) {
                        if event.fields.is_empty() {
                            event.span.status = Some(SpanStatus::Filtered);
                            crate::stats::stats_add_event_filtered();

                            ctx.internal_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.internal_tracker.insert(
                                "__op___kelora_stats_events_filtered".to_string(),
                                rhai::Dynamic::from("count"),
                            );

                            if let Some(span) = self.span_processor.as_mut() {
                                span.handle_skip(ctx);
                            }

                            if idx == 0 && !ops.is_empty() {
                                outputs.push(FormattedOutput::with_ops(
                                    String::new(),
                                    None,
                                    std::mem::take(&mut ops),
                                ));
                            }
                        } else {
                            crate::stats::stats_add_event_output();

                            // Track result timestamp for time span statistics
                            let mut result_event = event.clone();
                            result_event.parsed_ts = None; // Clear to force re-extraction
                            result_event.extract_timestamp_with_config(None, &self.ts_config);
                            if let Some(result_ts) = result_event.parsed_ts {
                                crate::stats::stats_update_result_timestamp(result_ts);
                            }

                            ctx.internal_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.internal_tracker.insert(
                                "__op___kelora_stats_events_output".to_string(),
                                rhai::Dynamic::from("count"),
                            );

                            if let Some(span) = self.span_processor.as_mut() {
                                span.record_emitted_event(&event, ctx)?;
                            }

                            let formatted = self.formatter.format(&event);
                            let timestamp = event.parsed_ts;
                            let event_ops = if idx == 0 {
                                std::mem::take(&mut ops)
                            } else {
                                Vec::new()
                            };
                            outputs
                                .push(FormattedOutput::with_ops(formatted, timestamp, event_ops));
                        }
                    } else {
                        event.span.status = Some(SpanStatus::Filtered);
                        crate::stats::stats_add_event_filtered();

                        ctx.internal_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.internal_tracker.insert(
                            "__op___kelora_stats_events_filtered".to_string(),
                            rhai::Dynamic::from("count"),
                        );

                        if let Some(span) = self.span_processor.as_mut() {
                            span.handle_skip(ctx);
                        }

                        if idx == 0 && !ops.is_empty() {
                            outputs.push(FormattedOutput::with_ops(
                                String::new(),
                                None,
                                std::mem::take(&mut ops),
                            ));
                        }
                    }
                }

                if !ops.is_empty() {
                    outputs.push(FormattedOutput::with_ops(String::new(), None, ops));
                }

                if let Some(span) = self.span_processor.as_mut() {
                    span.complete_pending();
                }
            }
            ScriptResult::Skip => {
                crate::stats::stats_add_event_filtered();

                ctx.internal_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.internal_tracker.insert(
                    "__op___kelora_stats_events_filtered".to_string(),
                    rhai::Dynamic::from("count"),
                );

                if let Some(span) = self.span_processor.as_mut() {
                    span.handle_skip(ctx);
                    span.complete_pending();
                }

                let ops = std::mem::take(&mut ctx.pending_file_ops);
                if !ops.is_empty() {
                    outputs.push(FormattedOutput::with_ops(String::new(), None, ops));
                }
            }
            ScriptResult::Error(msg) => {
                ctx.pending_file_ops.clear();
                file_ops::clear_pending_ops();

                if let Some(span) = self.span_processor.as_mut() {
                    span.complete_pending();
                }

                crate::rhai_functions::tracking::track_error(
                    "script",
                    ctx.meta.line_num,
                    &msg,
                    None,
                    ctx.meta.filename.as_deref(),
                    ctx.config.verbose,
                    ctx.config.quiet_level,
                    Some(&ctx.config),
                );

                if ctx.config.strict {
                    return Err(anyhow!(msg));
                } else {
                    return Ok(());
                }
            }
        }

        Ok(())
    }

    /// 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 mut event = match self.parser.parse(&chunk) {
            Ok(mut e) => {
                // Event was successfully created from chunk
                crate::stats::stats_add_event_created();

                // Track timestamp for time span statistics
                if let Some(ts) = e.parsed_ts {
                    crate::stats::stats_update_timestamp(ts);
                }

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

                // Also track in Rhai context for parallel processing
                ctx.internal_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.internal_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);
                }
            }
        };

        if let Some(span_processor) = self.span_processor.as_mut() {
            span_processor.prepare_event(&mut event, ctx)?;
        }

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

        file_ops::clear_pending_ops();
        ctx.pending_file_ops.clear();

        // 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
        let remaining_ops = file_ops::take_pending_ops();
        if !remaining_ops.is_empty() {
            ctx.pending_file_ops.extend(remaining_ops);
        }

        self.apply_script_result(result, ctx, &mut results)?;

        Ok(results)
    }

    /// Check if the event limiter (--take N) is exhausted
    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()
    }
}
