ftui_runtime/

resize_coalescer.rs

//! Adaptive resize stream coalescer.
//!
//! This module implements the resize coalescing behavior specified in
//! `docs/spec/resize-scheduler.md`. It provides:
//!
//! - **Latest-wins semantics**: Only the final size in a burst is rendered
//! - **Bounded latency**: Hard deadline guarantees render within max wait
//! - **Regime awareness**: Adapts behavior between steady and burst modes
//! - **Decision logging**: JSONL-compatible evidence for each decision
//!
//! # Usage
//!
//! ```ignore
//! use ftui_runtime::resize_coalescer::{ResizeCoalescer, CoalescerConfig};
//!
//! let config = CoalescerConfig::default();
//! let mut coalescer = ResizeCoalescer::new(config, (80, 24));
//!
//! // On resize event
//! let action = coalescer.handle_resize(100, 40);
//!
//! // On tick (called each frame)
//! let action = coalescer.tick();
//! ```
//!
//! # Regime Detection
//!
//! The coalescer uses a simplified regime model with two states:
//! - **Steady**: Single resize or slow sequence — prioritize responsiveness
//! - **Burst**: Rapid resize events — prioritize coalescing to reduce work
//!
//! Regime transitions are detected via event rate tracking with hysteresis.
//!
//! # Invariants
//!
//! - **Latest-wins**: the final resize in a burst is never dropped.
//! - **Bounded latency**: pending resizes apply within `hard_deadline_ms`.
//! - **Deterministic**: identical event sequences yield identical decisions.
//!
//! # Failure Modes
//!
//! | Condition | Behavior | Rationale |
//! |-----------|----------|-----------|
//! | `hard_deadline_ms = 0` | Apply immediately | Avoids zero-latency stall |
//! | `rate_window_size < 2` | `event_rate = 0` | No divide-by-zero in rate |
//! | No pending size | Return `None` | Avoids spurious applies |
//!
//! # Decision Rule (Explainable)
//!
//! 1) If `time_since_render ≥ hard_deadline_ms`, **apply** (forced).
//! 2) If `dt ≥ delay_ms`, **apply** when in **Steady** (or when BOCPD is
//!    enabled). (`delay_ms` = steady/burst delay, or BOCPD posterior-interpolated
//!    delay when enabled.)
//! 3) If `event_rate ≥ burst_enter_rate`, switch to **Burst**.
//! 4) If in **Burst** and `event_rate < burst_exit_rate` for `cooldown_frames`,
//!    switch to **Steady**.
//! 5) Otherwise, **coalesce** and optionally show a placeholder.

#![forbid(unsafe_code)]

use std::collections::VecDeque;
use web_time::{Duration, Instant};

use crate::bocpd::{BocpdConfig, BocpdDetector, BocpdRegime};
use crate::evidence_sink::{EVIDENCE_SCHEMA_VERSION, EvidenceSink};
use crate::terminal_writer::ScreenMode;

/// FNV-1a 64-bit offset basis.
/// Bound on retained cycle-time samples: percentiles are computed over the
/// most recent window, and unbounded retention (one sample per applied
/// resize) is a slow leak in long-lived sessions.
const MAX_CYCLE_TIME_SAMPLES: usize = 1024;

/// Bound on retained regime-transition evidence entries (one per regime
/// flip, otherwise never truncated by the runtime).
const MAX_TRANSITION_LOGS: usize = 256;

const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
/// FNV-1a 64-bit prime.
const FNV_PRIME: u64 = 0x100000001b3;

fn fnv_hash_bytes(hash: &mut u64, bytes: &[u8]) {
    for byte in bytes {
        *hash ^= *byte as u64;
        *hash = hash.wrapping_mul(FNV_PRIME);
    }
}

#[inline]
fn duration_since_or_zero(now: Instant, earlier: Instant) -> Duration {
    now.saturating_duration_since(earlier)
}

fn default_resize_run_id() -> String {
    format!("resize-{}", std::process::id())
}

fn screen_mode_str(mode: ScreenMode) -> &'static str {
    match mode {
        ScreenMode::Inline { .. } => "inline",
        ScreenMode::InlineAuto { .. } => "inline_auto",
        ScreenMode::AltScreen => "altscreen",
    }
}

#[inline]
fn json_escape(value: &str) -> String {
    let mut out = String::with_capacity(value.len());
    for ch in value.chars() {
        match ch {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            '\r' => out.push_str("\\r"),
            '\t' => out.push_str("\\t"),
            c if c.is_control() => {
                use std::fmt::Write as _;
                let _ = write!(out, "\\u{:04X}", c as u32);
            }
            _ => out.push(ch),
        }
    }
    out
}

fn evidence_prefix(
    run_id: &str,
    screen_mode: ScreenMode,
    cols: u16,
    rows: u16,
    event_idx: u64,
) -> String {
    format!(
        r#""schema_version":"{}","run_id":"{}","event_idx":{},"screen_mode":"{}","cols":{},"rows":{}"#,
        EVIDENCE_SCHEMA_VERSION,
        json_escape(run_id),
        event_idx,
        screen_mode_str(screen_mode),
        cols,
        rows,
    )
}

/// Configuration for the resize coalescer.
#[derive(Debug, Clone)]
pub struct CoalescerConfig {
    /// Maximum coalesce delay in steady regime (ms).
    /// In steady state, we want quick response.
    pub steady_delay_ms: u64,

    /// Maximum coalesce delay in burst regime (ms).
    /// During bursts, we coalesce more aggressively.
    pub burst_delay_ms: u64,

    /// Hard deadline — always render within this time (ms).
    /// Guarantees bounded worst-case latency.
    pub hard_deadline_ms: u64,

    /// Event rate threshold to enter burst mode (events/second).
    pub burst_enter_rate: f64,

    /// Event rate threshold to exit burst mode (events/second).
    /// Lower than enter_rate for hysteresis.
    pub burst_exit_rate: f64,

    /// Number of frames to hold in burst mode after rate drops.
    pub cooldown_frames: u32,

    /// Window size for rate calculation (number of events).
    pub rate_window_size: usize,

    /// Enable decision logging (JSONL format).
    pub enable_logging: bool,

    /// Enable BOCPD (Bayesian Online Change-Point Detection) for regime detection.
    ///
    /// When enabled, the coalescer uses a Bayesian posterior over run-lengths to
    /// detect regime changes (steady vs burst), replacing the simple rate threshold
    /// heuristics. BOCPD provides:
    /// - Principled uncertainty quantification via P(burst)
    /// - Automatic adaptation without hand-tuned thresholds
    /// - Evidence logging for decision explainability
    ///
    /// When disabled, falls back to rate threshold heuristics.
    pub enable_bocpd: bool,

    /// BOCPD configuration (used when `enable_bocpd` is true).
    pub bocpd_config: Option<BocpdConfig>,
}

impl Default for CoalescerConfig {
    fn default() -> Self {
        Self {
            steady_delay_ms: 16, // ~60fps responsiveness
            burst_delay_ms: 40,  // Aggressive coalescing
            hard_deadline_ms: 100,
            burst_enter_rate: 10.0, // 10 events/sec to enter burst
            burst_exit_rate: 5.0,   // 5 events/sec to exit burst
            cooldown_frames: 3,
            rate_window_size: 8,
            enable_logging: false,
            enable_bocpd: false,
            bocpd_config: None,
        }
    }
}

impl CoalescerConfig {
    /// Enable or disable decision logging.
    #[must_use]
    pub fn with_logging(mut self, enabled: bool) -> Self {
        self.enable_logging = enabled;
        self
    }

    /// Enable BOCPD-based regime detection with default configuration.
    #[must_use]
    pub fn with_bocpd(mut self) -> Self {
        self.enable_bocpd = true;
        self.bocpd_config = Some(BocpdConfig::default());
        self
    }

    /// Enable BOCPD-based regime detection with custom configuration.
    #[must_use]
    pub fn with_bocpd_config(mut self, config: BocpdConfig) -> Self {
        self.enable_bocpd = true;
        self.bocpd_config = Some(config);
        self
    }

    /// Serialize configuration to JSONL format.
    #[must_use]
    pub fn to_jsonl(
        &self,
        run_id: &str,
        screen_mode: ScreenMode,
        cols: u16,
        rows: u16,
        event_idx: u64,
    ) -> String {
        let prefix = evidence_prefix(run_id, screen_mode, cols, rows, event_idx);
        format!(
            r#"{{{prefix},"event":"config","steady_delay_ms":{},"burst_delay_ms":{},"hard_deadline_ms":{},"burst_enter_rate":{:.3},"burst_exit_rate":{:.3},"cooldown_frames":{},"rate_window_size":{},"logging_enabled":{}}}"#,
            self.steady_delay_ms,
            self.burst_delay_ms,
            self.hard_deadline_ms,
            self.burst_enter_rate,
            self.burst_exit_rate,
            self.cooldown_frames,
            self.rate_window_size,
            self.enable_logging
        )
    }
}

/// Action returned by the coalescer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CoalesceAction {
    /// No action needed.
    None,

    /// Show a placeholder/skeleton while coalescing.
    ShowPlaceholder,

    /// Apply the resize with the given dimensions.
    ApplyResize {
        width: u16,
        height: u16,
        /// Time spent coalescing.
        coalesce_time: Duration,
        /// Whether this was forced by hard deadline.
        forced_by_deadline: bool,
    },
}

/// Detected regime for resize events.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Regime {
    /// Single resize or slow sequence.
    #[default]
    Steady,
    /// Rapid resize events (storm).
    Burst,
}

impl Regime {
    /// Get the stable string representation.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Steady => "steady",
            Self::Burst => "burst",
        }
    }
}

/// Structured reason codes for regime transitions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransitionReasonCode {
    /// Heuristic detector entered burst because event rate crossed enter threshold.
    HeuristicEnterBurstRate,
    /// Heuristic detector exited burst after cooldown with low event rate.
    HeuristicExitBurstCooldown,
    /// BOCPD posterior crossed burst threshold.
    BocpdPosteriorBurst,
    /// BOCPD posterior crossed steady threshold.
    BocpdPosteriorSteady,
}

impl TransitionReasonCode {
    /// Stable string form for JSONL evidence.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::HeuristicEnterBurstRate => "heuristic_enter_burst_rate",
            Self::HeuristicExitBurstCooldown => "heuristic_exit_burst_cooldown",
            Self::BocpdPosteriorBurst => "bocpd_posterior_burst",
            Self::BocpdPosteriorSteady => "bocpd_posterior_steady",
        }
    }
}

/// Event emitted when a resize operation is applied (bd-bksf.6 stub).
///
/// Used by [`ResizeSlaMonitor`](crate::resize_sla::ResizeSlaMonitor) for latency tracking.
#[derive(Debug, Clone)]
pub struct ResizeAppliedEvent {
    /// New terminal size after resize.
    pub new_size: (u16, u16),
    /// Previous terminal size.
    pub old_size: (u16, u16),
    /// Time elapsed from resize request to apply.
    pub elapsed: Duration,
    /// Whether the apply was forced (hard deadline).
    pub forced: bool,
}

/// Event emitted when regime changes between Steady and Burst (bd-bksf.6 stub).
///
/// Used by SLA monitoring for regime-aware alerting.
#[derive(Debug, Clone)]
pub struct RegimeChangeEvent {
    /// Previous regime.
    pub from: Regime,
    /// New regime.
    pub to: Regime,
    /// Event index when transition occurred.
    pub event_idx: u64,
    /// Structured reason for the transition.
    pub reason_code: TransitionReasonCode,
    /// Transition confidence in [0, 1].
    pub confidence: f64,
}

// =============================================================================
// Evidence Ledger for Scheduler Decisions (bd-1rz0.27)
// =============================================================================

/// Evidence supporting a scheduler decision with Bayes factors.
///
/// This captures the mathematical reasoning behind coalesce/apply decisions,
/// providing explainability for the regime-adaptive scheduler.
///
/// # Bayes Factor Interpretation
///
/// The `log_bayes_factor` represents log10(P(evidence|apply_now) / P(evidence|coalesce)):
/// - Positive values favor immediate apply (respond quickly)
/// - Negative values favor coalescing (wait for more events)
/// - |LBF| > 1 is "strong" evidence, |LBF| > 2 is "decisive"
///
/// # Example
///
/// ```ignore
/// let evidence = DecisionEvidence {
///     log_bayes_factor: 1.5,  // Strong evidence to apply now
///     regime_contribution: 0.8,
///     timing_contribution: 0.5,
///     rate_contribution: 0.2,
///     explanation: "Steady regime with long idle interval".to_string(),
/// };
/// ```
#[derive(Debug, Clone)]
pub struct DecisionEvidence {
    /// Log10 Bayes factor: positive favors apply, negative favors coalesce.
    pub log_bayes_factor: f64,

    /// Contribution from regime detection (Steady vs Burst).
    pub regime_contribution: f64,

    /// Contribution from timing (dt_ms, time since last render).
    pub timing_contribution: f64,

    /// Contribution from event rate.
    pub rate_contribution: f64,

    /// Human-readable explanation of the decision.
    pub explanation: String,
}

impl DecisionEvidence {
    /// Evidence strongly favoring immediate apply (Steady regime, long idle).
    #[must_use]
    pub fn favor_apply(regime: Regime, dt_ms: f64, event_rate: f64) -> Self {
        let regime_contrib = if regime == Regime::Steady { 1.0 } else { -0.5 };
        let timing_contrib = (dt_ms / 50.0).min(2.0); // Higher dt -> favor apply
        let rate_contrib = if event_rate < 5.0 { 0.5 } else { -0.3 };

        let lbf = regime_contrib + timing_contrib + rate_contrib;

        Self {
            log_bayes_factor: lbf,
            regime_contribution: regime_contrib,
            timing_contribution: timing_contrib,
            rate_contribution: rate_contrib,
            explanation: format!(
                "Regime={:?} (contrib={:.2}), dt={:.1}ms (contrib={:.2}), rate={:.1}/s (contrib={:.2})",
                regime, regime_contrib, dt_ms, timing_contrib, event_rate, rate_contrib
            ),
        }
    }

    /// Evidence favoring coalescing (Burst regime, high rate).
    #[must_use]
    pub fn favor_coalesce(regime: Regime, dt_ms: f64, event_rate: f64) -> Self {
        let regime_contrib = if regime == Regime::Burst { 1.0 } else { -0.5 };
        let timing_contrib = (20.0 / dt_ms.max(1.0)).min(2.0); // Lower dt -> favor coalesce
        let rate_contrib = if event_rate > 10.0 { 0.5 } else { -0.3 };

        let lbf = -(regime_contrib + timing_contrib + rate_contrib);

        Self {
            log_bayes_factor: lbf,
            regime_contribution: regime_contrib,
            timing_contribution: timing_contrib,
            rate_contribution: rate_contrib,
            explanation: format!(
                "Regime={:?} (contrib={:.2}), dt={:.1}ms (contrib={:.2}), rate={:.1}/s (contrib={:.2})",
                regime, regime_contrib, dt_ms, timing_contrib, event_rate, rate_contrib
            ),
        }
    }

    /// Evidence for a forced deadline decision.
    #[must_use]
    pub fn forced_deadline(deadline_ms: f64) -> Self {
        Self {
            log_bayes_factor: f64::INFINITY,
            regime_contribution: 0.0,
            timing_contribution: deadline_ms,
            rate_contribution: 0.0,
            explanation: format!("Forced by hard deadline ({:.1}ms)", deadline_ms),
        }
    }

    /// Serialize to JSONL format.
    #[must_use]
    pub fn to_jsonl(
        &self,
        run_id: &str,
        screen_mode: ScreenMode,
        cols: u16,
        rows: u16,
        event_idx: u64,
    ) -> String {
        let lbf_str = if self.log_bayes_factor.is_infinite() {
            "\"inf\"".to_string()
        } else {
            format!("{:.3}", self.log_bayes_factor)
        };
        let prefix = evidence_prefix(run_id, screen_mode, cols, rows, event_idx);
        format!(
            r#"{{{prefix},"event":"decision_evidence","log_bayes_factor":{},"regime_contribution":{:.3},"timing_contribution":{:.3},"rate_contribution":{:.3},"explanation":"{}"}}"#,
            lbf_str,
            self.regime_contribution,
            self.timing_contribution,
            self.rate_contribution,
            json_escape(&self.explanation)
        )
    }

    /// Check if evidence strongly supports the action (|LBF| > 1).
    #[must_use]
    pub fn is_strong(&self) -> bool {
        self.log_bayes_factor.abs() > 1.0
    }

    /// Check if evidence decisively supports the action (|LBF| > 2).
    #[must_use]
    pub fn is_decisive(&self) -> bool {
        self.log_bayes_factor.abs() > 2.0 || self.log_bayes_factor.is_infinite()
    }
}

/// Decision log entry for observability.
#[derive(Debug, Clone)]
pub struct DecisionLog {
    /// Timestamp of the decision.
    pub timestamp: Instant,
    /// Elapsed time since logging started (ms).
    pub elapsed_ms: f64,
    /// Event index in session.
    pub event_idx: u64,
    /// Time since last event (ms).
    pub dt_ms: f64,
    /// Current event rate (events/sec).
    pub event_rate: f64,
    /// Detected regime.
    pub regime: Regime,
    /// Chosen action.
    pub action: &'static str,
    /// Pending size (if any).
    pub pending_size: Option<(u16, u16)>,
    /// Applied size (for apply decisions).
    pub applied_size: Option<(u16, u16)>,
    /// Time since last render (ms).
    pub time_since_render_ms: f64,
    /// Time spent coalescing until apply (ms).
    pub coalesce_ms: Option<f64>,
    /// Was forced by deadline.
    pub forced: bool,
    /// Transition reason code if this decision coincided with a regime transition.
    pub transition_reason_code: Option<TransitionReasonCode>,
    /// Transition confidence if this decision coincided with a regime transition.
    pub transition_confidence: Option<f64>,
}

impl DecisionLog {
    /// Serialize decision log to JSONL format.
    #[must_use]
    pub fn to_jsonl(&self, run_id: &str, screen_mode: ScreenMode, cols: u16, rows: u16) -> String {
        let (pending_w, pending_h) = match self.pending_size {
            Some((w, h)) => (w.to_string(), h.to_string()),
            None => ("null".to_string(), "null".to_string()),
        };
        let (applied_w, applied_h) = match self.applied_size {
            Some((w, h)) => (w.to_string(), h.to_string()),
            None => ("null".to_string(), "null".to_string()),
        };
        let coalesce_ms = match self.coalesce_ms {
            Some(ms) => format!("{:.3}", ms),
            None => "null".to_string(),
        };
        let transition_reason_code = self
            .transition_reason_code
            .map(TransitionReasonCode::as_str)
            .map(|code| format!(r#""{code}""#))
            .unwrap_or_else(|| "null".to_string());
        let transition_confidence = self
            .transition_confidence
            .map(|confidence| format!("{confidence:.6}"))
            .unwrap_or_else(|| "null".to_string());
        let prefix = evidence_prefix(run_id, screen_mode, cols, rows, self.event_idx);

        format!(
            r#"{{{prefix},"event":"decision","idx":{},"elapsed_ms":{:.3},"dt_ms":{:.3},"event_rate":{:.3},"regime":"{}","action":"{}","pending_w":{},"pending_h":{},"applied_w":{},"applied_h":{},"time_since_render_ms":{:.3},"coalesce_ms":{},"forced":{},"transition_reason_code":{},"transition_confidence":{}}}"#,
            self.event_idx,
            self.elapsed_ms,
            self.dt_ms,
            self.event_rate,
            self.regime.as_str(),
            self.action,
            pending_w,
            pending_h,
            applied_w,
            applied_h,
            self.time_since_render_ms,
            coalesce_ms,
            self.forced,
            transition_reason_code,
            transition_confidence
        )
    }
}

#[derive(Debug, Clone, Copy)]
struct PendingTransitionEvidence {
    reason_code: TransitionReasonCode,
    confidence: f64,
}

/// Transition evidence entry emitted when the controller changes regime.
#[derive(Debug, Clone)]
pub struct RegimeTransitionLog {
    /// Timestamp when transition occurred.
    pub timestamp: Instant,
    /// Event index when transition occurred.
    pub event_idx: u64,
    /// Previous regime.
    pub from_regime: Regime,
    /// New regime.
    pub to_regime: Regime,
    /// Structured reason code.
    pub reason_code: TransitionReasonCode,
    /// Confidence in [0, 1].
    pub confidence: f64,
    /// Event rate at transition time.
    pub event_rate: f64,
    /// BOCPD posterior probability of burst, if available.
    pub p_burst: Option<f64>,
    /// Current cooldown counter after transition accounting.
    pub cooldown_remaining: u32,
}

impl RegimeTransitionLog {
    /// Serialize transition evidence to JSONL format.
    #[must_use]
    pub fn to_jsonl(&self, run_id: &str, screen_mode: ScreenMode, cols: u16, rows: u16) -> String {
        let prefix = evidence_prefix(run_id, screen_mode, cols, rows, self.event_idx);
        let p_burst = self
            .p_burst
            .map(|value| format!("{value:.6}"))
            .unwrap_or_else(|| "null".to_string());
        format!(
            r#"{{{prefix},"event":"regime_transition","from_regime":"{}","to_regime":"{}","reason_code":"{}","confidence":{:.6},"event_rate":{:.3},"p_burst":{},"cooldown_remaining":{}}}"#,
            self.from_regime.as_str(),
            self.to_regime.as_str(),
            self.reason_code.as_str(),
            self.confidence,
            self.event_rate,
            p_burst,
            self.cooldown_remaining,
        )
    }
}

/// Adaptive resize stream coalescer.
///
/// Implements latest-wins coalescing with regime-aware behavior.
#[derive(Debug)]
pub struct ResizeCoalescer {
    config: CoalescerConfig,

    /// Currently pending size (latest wins).
    pending_size: Option<(u16, u16)>,

    /// Last applied size.
    last_applied: (u16, u16),

    /// Timestamp of first event in current coalesce window.
    window_start: Option<Instant>,

    /// Timestamp of last resize event.
    last_event: Option<Instant>,

    /// Timestamp of last render.
    last_render: Instant,

    /// Current detected regime.
    regime: Regime,

    /// Frames remaining in cooldown (for burst exit hysteresis).
    cooldown_remaining: u32,

    /// Recent event timestamps for rate calculation.
    event_times: VecDeque<Instant>,

    /// Total event count.
    event_count: u64,

    /// Logging start time for elapsed timestamps.
    log_start: Option<Instant>,

    /// Decision logs (if logging enabled).
    logs: Vec<DecisionLog>,
    /// Regime transition evidence logs.
    transition_logs: Vec<RegimeTransitionLog>,
    /// Pending transition evidence to attach to the next decision row.
    pending_transition_evidence: Option<PendingTransitionEvidence>,
    /// Evidence sink for JSONL decision logs.
    evidence_sink: Option<EvidenceSink>,
    /// Whether config has been logged to the evidence sink.
    config_logged: bool,
    /// Run identifier for evidence logs.
    evidence_run_id: String,
    /// Screen mode label for evidence logs.
    evidence_screen_mode: ScreenMode,

    // --- Telemetry integration (bd-1rz0.7) ---
    /// Telemetry hooks for external observability.
    telemetry_hooks: Option<TelemetryHooks>,

    /// Count of regime transitions during this session.
    regime_transitions: u64,

    /// Events coalesced in current window.
    events_in_window: u64,

    /// History of cycle times (ms) for percentile calculation.
    cycle_times: Vec<f64>,

    /// BOCPD detector for Bayesian regime detection (when enabled).
    bocpd: Option<BocpdDetector>,
}

/// Cycle time percentiles for reflow diagnostics (bd-1rz0.7).
#[derive(Debug, Clone, Copy)]
pub struct CycleTimePercentiles {
    /// 50th percentile (median) cycle time in ms.
    pub p50_ms: f64,
    /// 95th percentile cycle time in ms.
    pub p95_ms: f64,
    /// 99th percentile cycle time in ms.
    pub p99_ms: f64,
    /// Number of samples.
    pub count: usize,
    /// Mean cycle time in ms.
    pub mean_ms: f64,
}

impl CycleTimePercentiles {
    /// Serialize to JSONL format.
    #[must_use]
    pub fn to_jsonl(&self) -> String {
        format!(
            r#"{{"event":"cycle_time_percentiles","p50_ms":{:.3},"p95_ms":{:.3},"p99_ms":{:.3},"mean_ms":{:.3},"count":{}}}"#,
            self.p50_ms, self.p95_ms, self.p99_ms, self.mean_ms, self.count
        )
    }
}

impl ResizeCoalescer {
    /// Create a new coalescer with the given configuration and initial size.
    pub fn new(config: CoalescerConfig, initial_size: (u16, u16)) -> Self {
        let bocpd = if config.enable_bocpd {
            let mut bocpd_cfg = config.bocpd_config.clone().unwrap_or_default();
            if config.enable_logging {
                bocpd_cfg.enable_logging = true;
            }
            Some(BocpdDetector::new(bocpd_cfg))
        } else {
            None
        };

        Self {
            config,
            pending_size: None,
            last_applied: initial_size,
            window_start: None,
            last_event: None,
            last_render: Instant::now(),
            regime: Regime::Steady,
            cooldown_remaining: 0,
            event_times: VecDeque::new(),
            event_count: 0,
            log_start: None,
            logs: Vec::new(),
            transition_logs: Vec::new(),
            pending_transition_evidence: None,
            evidence_sink: None,
            config_logged: false,
            evidence_run_id: default_resize_run_id(),
            evidence_screen_mode: ScreenMode::AltScreen,
            telemetry_hooks: None,
            regime_transitions: 0,
            events_in_window: 0,
            cycle_times: Vec::new(),
            bocpd,
        }
    }

    /// Attach telemetry hooks for external observability.
    #[must_use]
    pub fn with_telemetry_hooks(mut self, hooks: TelemetryHooks) -> Self {
        self.telemetry_hooks = Some(hooks);
        self
    }

    /// Attach an evidence sink for JSONL decision logs.
    #[must_use]
    pub fn with_evidence_sink(mut self, sink: EvidenceSink) -> Self {
        self.evidence_sink = Some(sink);
        self.config_logged = false;
        self
    }

    /// Set the run identifier used in evidence logs.
    #[must_use]
    pub fn with_evidence_run_id(mut self, run_id: impl Into<String>) -> Self {
        self.evidence_run_id = run_id.into();
        self
    }

    /// Set the screen mode label used in evidence logs.
    #[must_use]
    pub fn with_screen_mode(mut self, screen_mode: ScreenMode) -> Self {
        self.evidence_screen_mode = screen_mode;
        self
    }

    /// Set or clear the evidence sink.
    pub fn set_evidence_sink(&mut self, sink: Option<EvidenceSink>) {
        self.evidence_sink = sink;
        self.config_logged = false;
    }

    /// Set the last render time (for deterministic testing).
    #[must_use]
    pub fn with_last_render(mut self, time: Instant) -> Self {
        self.last_render = time;
        self
    }

    /// Record an externally-applied resize (immediate path).
    pub fn record_external_apply(&mut self, width: u16, height: u16, now: Instant) {
        self.event_count += 1;
        self.event_times.push_back(now);
        while self.event_times.len() > self.config.rate_window_size {
            self.event_times.pop_front();
        }
        self.update_regime(now);

        self.pending_size = None;
        self.window_start = None;
        self.last_event = Some(now);
        self.last_applied = (width, height);
        self.last_render = now;
        self.events_in_window = 0;
        self.cooldown_remaining = 0;

        self.log_decision(now, "apply_immediate", false, Some(0.0), Some(0.0));

        if let Some(ref hooks) = self.telemetry_hooks
            && let Some(entry) = self.logs.last()
        {
            hooks.fire_resize_applied(entry);
        }
    }

    /// Get current regime transition count.
    #[must_use]
    pub fn regime_transition_count(&self) -> u64 {
        self.regime_transitions
    }

    /// Get cycle time percentiles (p50, p95, p99) in milliseconds.
    /// Returns None if no cycle times recorded.
    #[must_use]
    pub fn cycle_time_percentiles(&self) -> Option<CycleTimePercentiles> {
        if self.cycle_times.is_empty() {
            return None;
        }

        let mut sorted = self.cycle_times.clone();
        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

        let len = sorted.len();
        let p50_idx = len / 2;
        let p95_idx = (len * 95) / 100;
        let p99_idx = (len * 99) / 100;

        Some(CycleTimePercentiles {
            p50_ms: sorted[p50_idx],
            p95_ms: sorted[p95_idx.min(len - 1)],
            p99_ms: sorted[p99_idx.min(len - 1)],
            count: len,
            mean_ms: sorted.iter().sum::<f64>() / len as f64,
        })
    }

    /// Handle a resize event.
    ///
    /// Returns the action to take immediately.
    pub fn handle_resize(&mut self, width: u16, height: u16) -> CoalesceAction {
        self.handle_resize_at(width, height, Instant::now())
    }

    /// Handle a resize event at a specific time (for testing).
    pub fn handle_resize_at(&mut self, width: u16, height: u16, now: Instant) -> CoalesceAction {
        self.event_count += 1;

        // Calculate dt
        let dt = self.last_event.map(|t| duration_since_or_zero(now, t));
        let dt_ms = dt.map(|d| d.as_secs_f64() * 1000.0).unwrap_or(0.0);

        // Track event time for rate calculation
        // Clear stale events that artificially inflate the window duration
        if dt_ms > 1000.0 {
            self.event_times.clear();
        }

        self.event_times.push_back(now);
        while self.event_times.len() > self.config.rate_window_size {
            self.event_times.pop_front();
        }

        // Update regime based on event rate
        self.update_regime(now);

        self.last_event = Some(now);

        // If no pending, and this matches current size, no action needed
        if self.pending_size.is_none() && (width, height) == self.last_applied {
            self.log_decision(now, "skip_same_size", false, Some(dt_ms), None);
            return CoalesceAction::None;
        }

        // Whether work was already waiting BEFORE this event arrived: a
        // deadline apply counts as forced only if it breached the SLA of
        // previously-pending work, not this fresh event (bd-1za0z).
        let had_pending_before_event = self.pending_size.is_some();

        // Update pending size (latest wins)
        self.pending_size = Some((width, height));

        // Track events in current coalesce window (bd-1rz0.7)
        self.events_in_window += 1;

        // Mark window start if this is first event
        if self.window_start.is_none() {
            self.window_start = Some(now);
        }

        // Check hard deadline. `forced` must mean "work was ALREADY waiting
        // and sat past its SLA deadline" — an isolated resize arriving after
        // a quiet gap is applied instantly, but it is not a deadline breach
        // and must not inflate the SLA forced count (bd-1za0z).
        let time_since_render = duration_since_or_zero(now, self.last_render);
        if time_since_render >= Duration::from_millis(self.config.hard_deadline_ms) {
            return self.apply_pending_at(now, had_pending_before_event);
        }

        // If enough time has passed since the last event, apply now.
        // In heuristic mode, only apply immediately in Steady; burst applies via tick.
        let time_ok = match dt {
            Some(d) => d >= Duration::from_millis(self.current_delay_ms()),
            None => false, // First event must be coalesced to establish steady state timing
        };

        if time_ok && (self.bocpd.is_some() || self.regime == Regime::Steady) {
            return self.apply_pending_at(now, false);
        }

        self.log_decision(now, "coalesce", false, Some(dt_ms), None);

        // Fire decision hook for coalesce events (bd-1rz0.7)
        if let Some(ref hooks) = self.telemetry_hooks
            && let Some(entry) = self.logs.last()
        {
            hooks.fire_decision(entry);
        }

        CoalesceAction::ShowPlaceholder
    }

    /// Tick the coalescer (call each frame).
    ///
    /// Returns the action to take.
    pub fn tick(&mut self) -> CoalesceAction {
        self.tick_at(Instant::now())
    }

    /// Tick at a specific time (for testing).
    pub fn tick_at(&mut self, now: Instant) -> CoalesceAction {
        // Heuristic cooldown exit: HEURISTIC MODE ONLY. In BOCPD mode the
        // posterior owns regime changes (BocpdPosteriorSteady on the next
        // event); letting this rate-based exit fire too produced
        // contradictory transitions — and with a low-rate custom BocpdConfig
        // it was the only between-events Burst exit, fighting the posterior
        // and flapping the regime (bd-1za0z).
        if self.regime == Regime::Burst && self.bocpd.is_none() {
            let rate = self.calculate_event_rate(now);
            if rate >= self.config.burst_exit_rate {
                self.cooldown_remaining = self.config.cooldown_frames.max(1);
            } else if self.cooldown_remaining > 0 {
                self.cooldown_remaining -= 1;
                if self.cooldown_remaining == 0 {
                    self.record_regime_transition(
                        now,
                        Regime::Steady,
                        TransitionReasonCode::HeuristicExitBurstCooldown,
                        (1.0 - (rate / self.config.burst_exit_rate)).clamp(0.0, 1.0),
                        rate,
                        None,
                    );
                }
            }
        }

        if self.pending_size.is_none() {
            return CoalesceAction::None;
        }

        if self.window_start.is_none() {
            return CoalesceAction::None;
        }

        // Check hard deadline
        let time_since_render = duration_since_or_zero(now, self.last_render);
        if time_since_render >= Duration::from_millis(self.config.hard_deadline_ms) {
            return self.apply_pending_at(now, true);
        }

        let delay_ms = self.current_delay_ms();

        // Check if enough time has passed since last event
        if let Some(last_event) = self.last_event {
            let since_last_event = duration_since_or_zero(now, last_event);
            if since_last_event >= Duration::from_millis(delay_ms) {
                return self.apply_pending_at(now, false);
            }
        }

        CoalesceAction::None
    }

    /// Time until the pending resize should be applied.
    pub fn time_until_apply(&self, now: Instant) -> Option<Duration> {
        let _pending = self.pending_size?;

        // 1. Check hard deadline relative to last_render
        let time_since_render = duration_since_or_zero(now, self.last_render);
        let hard_deadline = Duration::from_millis(self.config.hard_deadline_ms);
        let hard_deadline_remaining = hard_deadline.saturating_sub(time_since_render);

        // 2. Check delay since last event
        let delay_remaining = if let Some(last_event) = self.last_event {
            let since_last_event = duration_since_or_zero(now, last_event);
            let delay = Duration::from_millis(self.current_delay_ms());
            delay.saturating_sub(since_last_event)
        } else {
            Duration::ZERO
        };

        // We apply when BOTH conditions are met (or hard deadline reached)
        // Wait, the logic in handle_resize/tick is:
        // IF (time_since_render >= hard_deadline) OR (since_last_event >= delay)
        // So time_until_apply should be the MIN of these two.
        Some(hard_deadline_remaining.min(delay_remaining))
    }

    /// Check if there's a pending resize.
    #[inline]
    pub fn has_pending(&self) -> bool {
        self.pending_size.is_some()
    }

    /// Get the current regime.
    #[inline]
    pub fn regime(&self) -> Regime {
        self.regime
    }

    /// Check if BOCPD-based regime detection is enabled.
    #[inline]
    pub fn bocpd_enabled(&self) -> bool {
        self.bocpd.is_some()
    }

    /// Get the BOCPD detector for inspection (if enabled).
    ///
    /// Returns `None` if BOCPD is not enabled.
    #[inline]
    pub fn bocpd(&self) -> Option<&BocpdDetector> {
        self.bocpd.as_ref()
    }

    /// Get the current P(burst) from BOCPD (if enabled).
    ///
    /// Returns the posterior probability that the system is in burst regime.
    /// Returns `None` if BOCPD is not enabled.
    #[inline]
    pub fn bocpd_p_burst(&self) -> Option<f64> {
        self.bocpd.as_ref().map(|b| b.p_burst())
    }

    /// Get the recommended delay from BOCPD (if enabled).
    ///
    /// Returns the recommended coalesce delay in milliseconds based on the
    /// current posterior distribution. Returns `None` if BOCPD is not enabled.
    #[inline]
    pub fn bocpd_recommended_delay(&self) -> Option<u64> {
        self.bocpd
            .as_ref()
            .map(|b| b.recommended_delay(self.config.steady_delay_ms, self.config.burst_delay_ms))
    }

    /// Get the current event rate (events/second).
    pub fn event_rate(&self) -> f64 {
        self.calculate_event_rate(Instant::now())
    }

    /// Get the last applied size.
    #[inline]
    pub fn last_applied(&self) -> (u16, u16) {
        self.last_applied
    }

    /// Get decision logs (if logging enabled).
    pub fn logs(&self) -> &[DecisionLog] {
        &self.logs
    }

    /// Get regime transition evidence logs.
    pub fn transition_logs(&self) -> &[RegimeTransitionLog] {
        &self.transition_logs
    }

    /// Record a coalesce cycle time, retaining only the most recent window.
    ///
    /// Unbounded retention is a slow leak in long-lived sessions (one entry
    /// per applied resize); percentiles over the recent window are also the
    /// more meaningful telemetry.
    fn record_cycle_time(&mut self, coalesce_ms: f64) {
        self.cycle_times.push(coalesce_ms);
        let len = self.cycle_times.len();
        if len > MAX_CYCLE_TIME_SAMPLES {
            self.cycle_times.drain(..len - MAX_CYCLE_TIME_SAMPLES);
        }
    }

    /// Append a regime-transition evidence entry, retaining only the most
    /// recent window (one entry per regime flip, never truncated by the
    /// runtime otherwise).
    fn push_transition_log(&mut self, log: RegimeTransitionLog) {
        self.transition_logs.push(log);
        let len = self.transition_logs.len();
        if len > MAX_TRANSITION_LOGS {
            self.transition_logs.drain(..len - MAX_TRANSITION_LOGS);
        }
    }

    /// Clear decision logs.
    pub fn clear_logs(&mut self) {
        self.logs.clear();
        self.transition_logs.clear();
        self.pending_transition_evidence = None;
        self.log_start = None;
        self.config_logged = false;
    }

    /// Get statistics about the coalescer.
    pub fn stats(&self) -> CoalescerStats {
        CoalescerStats {
            event_count: self.event_count,
            regime: self.regime,
            event_rate: self.event_rate(),
            has_pending: self.pending_size.is_some(),
            last_applied: self.last_applied,
        }
    }

    /// Export decision logs as JSONL (one entry per line).
    #[must_use]
    pub fn decision_logs_jsonl(&self) -> String {
        let (cols, rows) = self.last_applied;
        let run_id = self.evidence_run_id.as_str();
        let screen_mode = self.evidence_screen_mode;
        self.logs
            .iter()
            .map(|entry| entry.to_jsonl(run_id, screen_mode, cols, rows))
            .collect::<Vec<_>>()
            .join("\n")
    }

    /// Compute a deterministic checksum of decision logs.
    #[must_use]
    pub fn decision_checksum(&self) -> u64 {
        let mut hash = FNV_OFFSET_BASIS;
        for entry in &self.logs {
            fnv_hash_bytes(&mut hash, &entry.event_idx.to_le_bytes());
            fnv_hash_bytes(&mut hash, &entry.elapsed_ms.to_bits().to_le_bytes());
            fnv_hash_bytes(&mut hash, &entry.dt_ms.to_bits().to_le_bytes());
            fnv_hash_bytes(&mut hash, &entry.event_rate.to_bits().to_le_bytes());
            fnv_hash_bytes(
                &mut hash,
                &[match entry.regime {
                    Regime::Steady => 0u8,
                    Regime::Burst => 1u8,
                }],
            );
            fnv_hash_bytes(&mut hash, entry.action.as_bytes());
            fnv_hash_bytes(&mut hash, &[0u8]); // separator

            fnv_hash_bytes(&mut hash, &[entry.pending_size.is_some() as u8]);
            if let Some((w, h)) = entry.pending_size {
                fnv_hash_bytes(&mut hash, &w.to_le_bytes());
                fnv_hash_bytes(&mut hash, &h.to_le_bytes());
            }

            fnv_hash_bytes(&mut hash, &[entry.applied_size.is_some() as u8]);
            if let Some((w, h)) = entry.applied_size {
                fnv_hash_bytes(&mut hash, &w.to_le_bytes());
                fnv_hash_bytes(&mut hash, &h.to_le_bytes());
            }

            fnv_hash_bytes(
                &mut hash,
                &entry.time_since_render_ms.to_bits().to_le_bytes(),
            );
            fnv_hash_bytes(&mut hash, &[entry.coalesce_ms.is_some() as u8]);
            if let Some(ms) = entry.coalesce_ms {
                fnv_hash_bytes(&mut hash, &ms.to_bits().to_le_bytes());
            }
            fnv_hash_bytes(&mut hash, &[entry.forced as u8]);
            fnv_hash_bytes(&mut hash, &[entry.transition_reason_code.is_some() as u8]);
            if let Some(reason_code) = entry.transition_reason_code {
                fnv_hash_bytes(&mut hash, reason_code.as_str().as_bytes());
            }
            fnv_hash_bytes(&mut hash, &[entry.transition_confidence.is_some() as u8]);
            if let Some(confidence) = entry.transition_confidence {
                fnv_hash_bytes(&mut hash, &confidence.to_bits().to_le_bytes());
            }
        }
        hash
    }

    /// Compute checksum as hex string.
    #[must_use]
    pub fn decision_checksum_hex(&self) -> String {
        format!("{:016x}", self.decision_checksum())
    }

    /// Compute a summary of the decision log.
    #[must_use]
    #[allow(clippy::field_reassign_with_default)]
    pub fn decision_summary(&self) -> DecisionSummary {
        let mut summary = DecisionSummary::default();
        summary.decision_count = self.logs.len();
        summary.last_applied = self.last_applied;
        summary.regime = self.regime;

        for entry in &self.logs {
            match entry.action {
                "apply" | "apply_forced" | "apply_immediate" => {
                    summary.apply_count += 1;
                    if entry.forced {
                        summary.forced_apply_count += 1;
                    }
                }
                "coalesce" => summary.coalesce_count += 1,
                "skip_same_size" => summary.skip_count += 1,
                _ => {}
            }
        }

        summary.checksum = self.decision_checksum();
        summary
    }

    /// Export config + decision logs + summary as JSONL.
    #[must_use]
    pub fn evidence_to_jsonl(&self) -> String {
        let mut lines = Vec::with_capacity(self.logs.len() + self.transition_logs.len() + 2);
        let (cols, rows) = self.last_applied;
        let run_id = self.evidence_run_id.as_str();
        let screen_mode = self.evidence_screen_mode;
        let summary_event_idx = self
            .logs
            .last()
            .map(|entry| entry.event_idx)
            .or_else(|| self.transition_logs.last().map(|entry| entry.event_idx))
            .unwrap_or(0);
        lines.push(self.config.to_jsonl(run_id, screen_mode, cols, rows, 0));
        lines.extend(
            self.logs
                .iter()
                .map(|entry| entry.to_jsonl(run_id, screen_mode, cols, rows)),
        );
        lines.extend(
            self.transition_logs
                .iter()
                .map(|entry| entry.to_jsonl(run_id, screen_mode, cols, rows)),
        );
        lines.push(self.decision_summary().to_jsonl(
            run_id,
            screen_mode,
            cols,
            rows,
            summary_event_idx,
        ));
        lines.join("\n")
    }

    // --- Internal methods ---

    fn apply_pending_at(&mut self, now: Instant, forced: bool) -> CoalesceAction {
        let Some((width, height)) = self.pending_size.take() else {
            return CoalesceAction::None;
        };

        let coalesce_time = self
            .window_start
            .map(|s| duration_since_or_zero(now, s))
            .unwrap_or(Duration::ZERO);
        let coalesce_ms = coalesce_time.as_secs_f64() * 1000.0;

        // Track cycle time for percentile calculation (bd-1rz0.7)
        self.record_cycle_time(coalesce_ms);

        self.window_start = None;
        self.last_applied = (width, height);
        self.last_render = now;

        // Reset events in window counter
        self.events_in_window = 0;

        self.log_decision(
            now,
            if forced { "apply_forced" } else { "apply" },
            forced,
            None,
            Some(coalesce_ms),
        );

        // Fire telemetry hooks (bd-1rz0.7)
        if let Some(ref hooks) = self.telemetry_hooks
            && let Some(entry) = self.logs.last()
        {
            hooks.fire_resize_applied(entry);
        }

        CoalesceAction::ApplyResize {
            width,
            height,
            coalesce_time,
            forced_by_deadline: forced,
        }
    }

    #[inline]
    fn current_delay_ms(&self) -> u64 {
        if let Some(ref bocpd) = self.bocpd {
            bocpd.recommended_delay(self.config.steady_delay_ms, self.config.burst_delay_ms)
        } else {
            match self.regime {
                Regime::Steady => self.config.steady_delay_ms,
                Regime::Burst => self.config.burst_delay_ms,
            }
        }
    }

    fn update_regime(&mut self, now: Instant) {
        // Use BOCPD for regime detection when enabled
        if self.bocpd.is_some() {
            let transition = {
                let mut pending = None;
                if let Some(bocpd) = self.bocpd.as_mut() {
                    // Update BOCPD with the event timestamp (it calculates inter-arrival internally)
                    bocpd.observe_event(now);

                    let p_burst = bocpd.p_burst();
                    // Map BOCPD regime to coalescer regime.
                    let proposed = match bocpd.regime() {
                        BocpdRegime::Steady => Regime::Steady,
                        BocpdRegime::Burst => Regime::Burst,
                        BocpdRegime::Transitional => {
                            // During transition, maintain current regime to avoid thrashing
                            self.regime
                        }
                    };
                    if proposed != self.regime {
                        let (reason_code, confidence) = if proposed == Regime::Burst {
                            (
                                TransitionReasonCode::BocpdPosteriorBurst,
                                p_burst.clamp(0.0, 1.0),
                            )
                        } else {
                            (
                                TransitionReasonCode::BocpdPosteriorSteady,
                                (1.0 - p_burst).clamp(0.0, 1.0),
                            )
                        };
                        pending = Some((proposed, reason_code, confidence, p_burst));
                    }
                }
                pending
            };

            if let Some((proposed, reason_code, confidence, p_burst)) = transition {
                let rate = self.calculate_event_rate(now);
                self.record_regime_transition(
                    now,
                    proposed,
                    reason_code,
                    confidence,
                    rate,
                    Some(p_burst),
                );
            }
        } else {
            // Fall back to heuristic rate-based detection
            let rate = self.calculate_event_rate(now);

            match self.regime {
                Regime::Steady => {
                    if rate >= self.config.burst_enter_rate {
                        self.cooldown_remaining = self.config.cooldown_frames.max(1);
                        let confidence = (rate / self.config.burst_enter_rate).clamp(0.0, 1.0);
                        self.record_regime_transition(
                            now,
                            Regime::Burst,
                            TransitionReasonCode::HeuristicEnterBurstRate,
                            confidence,
                            rate,
                            None,
                        );
                    }
                }
                Regime::Burst => {
                    if rate >= self.config.burst_exit_rate {
                        self.cooldown_remaining = self.config.cooldown_frames.max(1);
                    }
                }
            }
        }
    }

    fn record_regime_transition(
        &mut self,
        now: Instant,
        to_regime: Regime,
        reason_code: TransitionReasonCode,
        confidence: f64,
        event_rate: f64,
        p_burst: Option<f64>,
    ) {
        let from_regime = self.regime;
        if from_regime == to_regime {
            return;
        }
        self.regime = to_regime;
        self.regime_transitions += 1;
        self.pending_transition_evidence = Some(PendingTransitionEvidence {
            reason_code,
            confidence,
        });
        self.push_transition_log(RegimeTransitionLog {
            timestamp: now,
            event_idx: self.event_count,
            from_regime,
            to_regime,
            reason_code,
            confidence,
            event_rate,
            p_burst,
            cooldown_remaining: self.cooldown_remaining,
        });
        if let Some(ref hooks) = self.telemetry_hooks {
            hooks.fire_regime_change(from_regime, to_regime);
        }

        if let Some(ref sink) = self.evidence_sink {
            let (cols, rows) = self.last_applied;
            let run_id = self.evidence_run_id.as_str();
            let screen_mode = self.evidence_screen_mode;
            if !self.config_logged {
                let _ = sink.write_jsonl(&self.config.to_jsonl(run_id, screen_mode, cols, rows, 0));
                self.config_logged = true;
            }
            if let Some(entry) = self.transition_logs.last() {
                let _ = sink.write_jsonl(&entry.to_jsonl(run_id, screen_mode, cols, rows));
            }
        }
    }

    fn calculate_event_rate(&self, now: Instant) -> f64 {
        if self.event_times.len() < 2 {
            return 0.0;
        }

        let first = *self
            .event_times
            .front()
            .expect("event_times has >=2 elements per length guard");
        let window_duration = match now.checked_duration_since(first) {
            Some(duration) => duration,
            None => return 0.0,
        };

        // Enforce a minimum duration of 1ms to prevent divide-by-zero or instability
        // and to correctly reflect high rates for near-instantaneous bursts.
        let duration_secs = window_duration.as_secs_f64().max(0.001);

        // Number of intervals is len - 1
        ((self.event_times.len() - 1) as f64) / duration_secs
    }

    fn log_decision(
        &mut self,
        now: Instant,
        action: &'static str,
        forced: bool,
        dt_ms_override: Option<f64>,
        coalesce_ms: Option<f64>,
    ) {
        if !self.config.enable_logging {
            return;
        }

        if self.log_start.is_none() {
            self.log_start = Some(now);
        }

        let elapsed_ms = self
            .log_start
            .map(|t| duration_since_or_zero(now, t).as_secs_f64() * 1000.0)
            .unwrap_or(0.0);

        let dt_ms = dt_ms_override
            .or_else(|| {
                self.last_event
                    .map(|t| duration_since_or_zero(now, t).as_secs_f64() * 1000.0)
            })
            .unwrap_or(0.0);

        let time_since_render_ms =
            duration_since_or_zero(now, self.last_render).as_secs_f64() * 1000.0;

        let applied_size =
            if action == "apply" || action == "apply_forced" || action == "apply_immediate" {
                Some(self.last_applied)
            } else {
                None
            };
        let (transition_reason_code, transition_confidence) =
            match self.pending_transition_evidence.take() {
                Some(ev) => (Some(ev.reason_code), Some(ev.confidence)),
                None => (None, None),
            };

        self.logs.push(DecisionLog {
            timestamp: now,
            elapsed_ms,
            event_idx: self.event_count,
            dt_ms,
            event_rate: self.calculate_event_rate(now),
            regime: self.regime,
            action,
            pending_size: self.pending_size,
            applied_size,
            time_since_render_ms,
            coalesce_ms,
            forced,
            transition_reason_code,
            transition_confidence,
        });

        if let Some(ref sink) = self.evidence_sink {
            let (cols, rows) = self.last_applied;
            let run_id = self.evidence_run_id.as_str();
            let screen_mode = self.evidence_screen_mode;
            if !self.config_logged {
                let _ = sink.write_jsonl(&self.config.to_jsonl(run_id, screen_mode, cols, rows, 0));
                self.config_logged = true;
            }
            if let Some(entry) = self.logs.last() {
                let _ = sink.write_jsonl(&entry.to_jsonl(run_id, screen_mode, cols, rows));
            }
            if let Some(ref bocpd) = self.bocpd
                && let Some(jsonl) = bocpd.decision_log_jsonl(
                    self.config.steady_delay_ms,
                    self.config.burst_delay_ms,
                    forced,
                )
            {
                let _ = sink.write_jsonl(&jsonl);
            }
        }
    }
}

/// Statistics about the coalescer state.
#[derive(Debug, Clone)]
pub struct CoalescerStats {
    /// Total events processed.
    pub event_count: u64,
    /// Current regime.
    pub regime: Regime,
    /// Current event rate (events/sec).
    pub event_rate: f64,
    /// Whether there's a pending resize.
    pub has_pending: bool,
    /// Last applied size.
    pub last_applied: (u16, u16),
}

/// Summary of decision logs.
#[derive(Debug, Clone, Default)]
pub struct DecisionSummary {
    /// Total number of decisions logged.
    pub decision_count: usize,
    /// Total apply decisions.
    pub apply_count: usize,
    /// Applies forced by deadline.
    pub forced_apply_count: usize,
    /// Total coalesce decisions.
    pub coalesce_count: usize,
    /// Total skip decisions.
    pub skip_count: usize,
    /// Final regime at summary time.
    pub regime: Regime,
    /// Last applied size.
    pub last_applied: (u16, u16),
    /// Checksum for the decision log.
    pub checksum: u64,
}

impl DecisionSummary {
    /// Checksum as hex string.
    #[must_use]
    pub fn checksum_hex(&self) -> String {
        format!("{:016x}", self.checksum)
    }

    /// Serialize summary to JSONL format.
    #[must_use]
    pub fn to_jsonl(
        &self,
        run_id: &str,
        screen_mode: ScreenMode,
        cols: u16,
        rows: u16,
        event_idx: u64,
    ) -> String {
        let prefix = evidence_prefix(run_id, screen_mode, cols, rows, event_idx);
        format!(
            r#"{{{prefix},"event":"summary","decisions":{},"applies":{},"forced_applies":{},"coalesces":{},"skips":{},"regime":"{}","last_w":{},"last_h":{},"checksum":"{}"}}"#,
            self.decision_count,
            self.apply_count,
            self.forced_apply_count,
            self.coalesce_count,
            self.skip_count,
            self.regime.as_str(),
            self.last_applied.0,
            self.last_applied.1,
            self.checksum_hex()
        )
    }
}

// =============================================================================
// Telemetry Hooks (bd-1rz0.7)
// =============================================================================

/// Callback type for resize applied events.
pub type OnResizeApplied = Box<dyn Fn(&DecisionLog) + Send + Sync>;
/// Callback type for regime change events.
pub type OnRegimeChange = Box<dyn Fn(Regime, Regime) + Send + Sync>;
/// Callback type for coalesce decision events.
pub type OnCoalesceDecision = Box<dyn Fn(&DecisionLog) + Send + Sync>;

/// Telemetry hooks for observing resize coalescer events.
///
/// # Example
///
/// ```ignore
/// use ftui_runtime::resize_coalescer::{ResizeCoalescer, TelemetryHooks, CoalescerConfig};
///
/// let hooks = TelemetryHooks::new()
///     .on_resize_applied(|entry| println!("Applied: {}x{}", entry.applied_size.unwrap().0, entry.applied_size.unwrap().1))
///     .on_regime_change(|from, to| println!("Regime: {:?} -> {:?}", from, to));
///
/// let mut coalescer = ResizeCoalescer::new(CoalescerConfig::default(), (80, 24))
///     .with_telemetry_hooks(hooks);
/// ```
pub struct TelemetryHooks {
    /// Called when a resize is applied.
    on_resize_applied: Option<OnResizeApplied>,
    /// Called when regime changes (Steady <-> Burst).
    on_regime_change: Option<OnRegimeChange>,
    /// Called for every decision (coalesce, apply, skip).
    on_decision: Option<OnCoalesceDecision>,
    /// Enable tracing events (requires `tracing` feature).
    emit_tracing: bool,
}

impl Default for TelemetryHooks {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Debug for TelemetryHooks {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TelemetryHooks")
            .field("on_resize_applied", &self.on_resize_applied.is_some())
            .field("on_regime_change", &self.on_regime_change.is_some())
            .field("on_decision", &self.on_decision.is_some())
            .field("emit_tracing", &self.emit_tracing)
            .finish()
    }
}

impl TelemetryHooks {
    /// Create a new empty hooks instance.
    #[must_use]
    pub fn new() -> Self {
        Self {
            on_resize_applied: None,
            on_regime_change: None,
            on_decision: None,
            emit_tracing: false,
        }
    }

    /// Set callback for resize applied events.
    #[must_use]
    pub fn on_resize_applied<F>(mut self, callback: F) -> Self
    where
        F: Fn(&DecisionLog) + Send + Sync + 'static,
    {
        self.on_resize_applied = Some(Box::new(callback));
        self
    }

    /// Set callback for regime change events.
    #[must_use]
    pub fn on_regime_change<F>(mut self, callback: F) -> Self
    where
        F: Fn(Regime, Regime) + Send + Sync + 'static,
    {
        self.on_regime_change = Some(Box::new(callback));
        self
    }

    /// Set callback for all decision events.
    #[must_use]
    pub fn on_decision<F>(mut self, callback: F) -> Self
    where
        F: Fn(&DecisionLog) + Send + Sync + 'static,
    {
        self.on_decision = Some(Box::new(callback));
        self
    }

    /// Enable tracing event emission for OpenTelemetry integration.
    ///
    /// When enabled, decision events are emitted as `tracing::event!` calls
    /// with target `ftui.decision.resize` and all evidence ledger fields.
    #[must_use]
    pub fn with_tracing(mut self, enabled: bool) -> Self {
        self.emit_tracing = enabled;
        self
    }

    /// Check if a resize-applied hook is registered.
    pub fn has_resize_applied(&self) -> bool {
        self.on_resize_applied.is_some()
    }

    /// Check if a regime-change hook is registered.
    pub fn has_regime_change(&self) -> bool {
        self.on_regime_change.is_some()
    }

    /// Check if a decision hook is registered.
    pub fn has_decision(&self) -> bool {
        self.on_decision.is_some()
    }

    /// Invoke the resize applied callback if set.
    fn fire_resize_applied(&self, entry: &DecisionLog) {
        if let Some(ref cb) = self.on_resize_applied {
            cb(entry);
        }
        if self.emit_tracing {
            Self::emit_resize_tracing(entry);
        }
    }

    /// Invoke the regime change callback if set.
    fn fire_regime_change(&self, from: Regime, to: Regime) {
        if let Some(ref cb) = self.on_regime_change {
            cb(from, to);
        }
        if self.emit_tracing {
            tracing::debug!(
                target: "ftui.decision.resize",
                from_regime = %from.as_str(),
                to_regime = %to.as_str(),
                "regime_change"
            );
        }
    }

    /// Invoke the decision callback if set.
    fn fire_decision(&self, entry: &DecisionLog) {
        if let Some(ref cb) = self.on_decision {
            cb(entry);
        }
    }

    /// Emit a tracing event for resize decisions.
    fn emit_resize_tracing(entry: &DecisionLog) {
        let (pending_w, pending_h) = entry.pending_size.unwrap_or((0, 0));
        let (applied_w, applied_h) = entry.applied_size.unwrap_or((0, 0));
        let coalesce_ms = entry.coalesce_ms.unwrap_or(0.0);

        tracing::info!(
            target: "ftui.decision.resize",
            event_idx = entry.event_idx,
            elapsed_ms = entry.elapsed_ms,
            dt_ms = entry.dt_ms,
            event_rate = entry.event_rate,
            regime = %entry.regime.as_str(),
            action = entry.action,
            pending_w = pending_w,
            pending_h = pending_h,
            applied_w = applied_w,
            applied_h = applied_h,
            time_since_render_ms = entry.time_since_render_ms,
            coalesce_ms = coalesce_ms,
            forced = entry.forced,
            "resize_decision"
        );
    }
}

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

    #[test]
    fn retained_diagnostics_are_bounded() {
        let mut c = ResizeCoalescer::new(test_config(), (80, 24));
        for i in 0..(MAX_CYCLE_TIME_SAMPLES + 500) {
            c.record_cycle_time(i as f64);
        }
        assert_eq!(c.cycle_times.len(), MAX_CYCLE_TIME_SAMPLES);
        // The retained window is the most recent samples.
        assert_eq!(c.cycle_times[0], 500.0);
        // Percentiles still work over the bounded window.
        assert!(c.cycle_time_percentiles().is_some());

        let now = Instant::now();
        for i in 0..(MAX_TRANSITION_LOGS + 100) {
            c.push_transition_log(RegimeTransitionLog {
                timestamp: now,
                event_idx: i as u64,
                from_regime: Regime::Steady,
                to_regime: Regime::Burst,
                reason_code: TransitionReasonCode::HeuristicEnterBurstRate,
                confidence: 1.0,
                event_rate: 0.0,
                p_burst: None,
                cooldown_remaining: 0,
            });
        }
        assert_eq!(c.transition_logs.len(), MAX_TRANSITION_LOGS);
        assert_eq!(c.transition_logs[0].event_idx, 100);
    }

    fn test_config() -> CoalescerConfig {
        CoalescerConfig {
            steady_delay_ms: 16,
            burst_delay_ms: 40,
            hard_deadline_ms: 100,
            burst_enter_rate: 10.0,
            burst_exit_rate: 5.0,
            cooldown_frames: 3,
            rate_window_size: 8,
            enable_logging: true,
            enable_bocpd: false,
            bocpd_config: None,
        }
    }

    #[derive(Debug, Clone, Copy)]
    struct SimulationMetrics {
        event_count: u64,
        apply_count: u64,
        forced_count: u64,
        mean_coalesce_ms: f64,
        max_coalesce_ms: f64,
        decision_checksum: u64,
        final_regime: Regime,
    }

    impl SimulationMetrics {
        fn to_jsonl(self, pattern: &str, mode: &str) -> String {
            let pattern = json_escape(pattern);
            let mode = json_escape(mode);
            let apply_ratio = if self.event_count == 0 {
                0.0
            } else {
                self.apply_count as f64 / self.event_count as f64
            };

            format!(
                r#"{{"event":"simulation_summary","pattern":"{pattern}","mode":"{mode}","events":{},"applies":{},"forced":{},"apply_ratio":{:.4},"mean_coalesce_ms":{:.3},"max_coalesce_ms":{:.3},"final_regime":"{}","checksum":"{:016x}"}}"#,
                self.event_count,
                self.apply_count,
                self.forced_count,
                apply_ratio,
                self.mean_coalesce_ms,
                self.max_coalesce_ms,
                self.final_regime.as_str(),
                self.decision_checksum
            )
        }
    }

    #[derive(Debug, Clone, Copy)]
    struct SimulationComparison {
        apply_delta: i64,
        mean_coalesce_delta_ms: f64,
    }

    impl SimulationComparison {
        fn from_metrics(heuristic: SimulationMetrics, bocpd: SimulationMetrics) -> Self {
            let heuristic_apply = i64::try_from(heuristic.apply_count).unwrap_or(i64::MAX);
            let bocpd_apply = i64::try_from(bocpd.apply_count).unwrap_or(i64::MAX);
            let apply_delta = heuristic_apply.saturating_sub(bocpd_apply);
            let mean_coalesce_delta_ms = heuristic.mean_coalesce_ms - bocpd.mean_coalesce_ms;
            Self {
                apply_delta,
                mean_coalesce_delta_ms,
            }
        }

        fn to_jsonl(self, pattern: &str) -> String {
            let pattern = json_escape(pattern);
            format!(
                r#"{{"event":"simulation_compare","pattern":"{pattern}","apply_delta":{},"mean_coalesce_delta_ms":{:.3}}}"#,
                self.apply_delta, self.mean_coalesce_delta_ms
            )
        }
    }

    fn as_u64(value: usize) -> u64 {
        u64::try_from(value).unwrap_or(u64::MAX)
    }

    fn build_schedule(base: Instant, events: &[(u16, u16, u64)]) -> Vec<(Instant, u16, u16)> {
        let mut schedule = Vec::with_capacity(events.len());
        let mut elapsed_ms = 0u64;
        for (w, h, delay_ms) in events {
            elapsed_ms = elapsed_ms.saturating_add(*delay_ms);
            schedule.push((base + Duration::from_millis(elapsed_ms), *w, *h));
        }
        schedule
    }

    fn run_simulation(
        events: &[(u16, u16, u64)],
        config: CoalescerConfig,
        tick_ms: u64,
    ) -> SimulationMetrics {
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let base = Instant::now();
        let schedule = build_schedule(base, events);
        let last_event_ms = schedule
            .last()
            .map(|(time, _, _)| {
                u64::try_from(duration_since_or_zero(*time, base).as_millis()).unwrap_or(u64::MAX)
            })
            .unwrap_or(0);
        let end_ms = last_event_ms
            .saturating_add(c.config.hard_deadline_ms)
            .saturating_add(tick_ms);

        let mut next_idx = 0usize;
        let mut now_ms = 0u64;
        while now_ms <= end_ms {
            let now = base + Duration::from_millis(now_ms);

            while next_idx < schedule.len() && schedule[next_idx].0 <= now {
                let (event_time, w, h) = schedule[next_idx];
                let _ = c.handle_resize_at(w, h, event_time);
                next_idx += 1;
            }

            let _ = c.tick_at(now);
            now_ms = now_ms.saturating_add(tick_ms);
        }

        let mut coalesce_values = Vec::new();
        let mut apply_count = 0usize;
        let mut forced_count = 0usize;
        for entry in c.logs() {
            if matches!(entry.action, "apply" | "apply_forced" | "apply_immediate") {
                apply_count += 1;
                if entry.forced {
                    forced_count += 1;
                }
                if let Some(ms) = entry.coalesce_ms {
                    coalesce_values.push(ms);
                }
            }
        }

        let max_coalesce_ms = coalesce_values
            .iter()
            .copied()
            .fold(0.0_f64, |acc, value| acc.max(value));
        let mean_coalesce_ms = if coalesce_values.is_empty() {
            0.0
        } else {
            let sum = coalesce_values.iter().sum::<f64>();
            sum / as_u64(coalesce_values.len()) as f64
        };

        SimulationMetrics {
            event_count: as_u64(events.len()),
            apply_count: as_u64(apply_count),
            forced_count: as_u64(forced_count),
            mean_coalesce_ms,
            max_coalesce_ms,
            decision_checksum: c.decision_checksum(),
            final_regime: c.regime(),
        }
    }

    fn steady_pattern() -> Vec<(u16, u16, u64)> {
        let mut events = Vec::new();
        for i in 0..8u16 {
            let width = 90 + i;
            let height = 30 + (i % 3);
            events.push((width, height, 300));
        }
        events
    }

    fn burst_pattern() -> Vec<(u16, u16, u64)> {
        let mut events = Vec::new();
        for i in 0..30u16 {
            let width = 100 + i;
            let height = 25 + (i % 5);
            events.push((width, height, 10));
        }
        events
    }

    fn oscillatory_pattern() -> Vec<(u16, u16, u64)> {
        let mut events = Vec::new();
        let sizes = [(120, 40), (140, 28), (130, 36), (150, 32)];
        let delays = [40u64, 200u64, 60u64, 180u64];
        for i in 0..16usize {
            let (w, h) = sizes[i % sizes.len()];
            let delay = delays[i % delays.len()];
            events.push((w + (i as u16 % 3), h, delay));
        }
        events
    }

    #[test]
    fn new_coalescer_starts_in_steady() {
        let c = ResizeCoalescer::new(CoalescerConfig::default(), (80, 24));
        assert_eq!(c.regime(), Regime::Steady);
        assert!(!c.has_pending());
    }

    #[test]
    fn same_size_returns_none() {
        let mut c = ResizeCoalescer::new(test_config(), (80, 24));
        let action = c.handle_resize(80, 24);
        assert_eq!(action, CoalesceAction::None);
    }

    #[test]
    fn different_size_shows_placeholder() {
        let mut c = ResizeCoalescer::new(test_config(), (80, 24));
        let action = c.handle_resize(100, 40);
        assert_eq!(action, CoalesceAction::ShowPlaceholder);
        assert!(c.has_pending());
    }

    #[test]
    fn latest_wins_semantics() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // Rapid sequence of resizes
        c.handle_resize_at(90, 30, base);
        c.handle_resize_at(100, 40, base + Duration::from_millis(5));
        c.handle_resize_at(110, 50, base + Duration::from_millis(10));

        // Wait for coalesce delay
        let action = c.tick_at(base + Duration::from_millis(60));

        let (width, height) = if let CoalesceAction::ApplyResize { width, height, .. } = action {
            (width, height)
        } else {
            assert!(
                matches!(action, CoalesceAction::ApplyResize { .. }),
                "Expected ApplyResize, got {action:?}"
            );
            return;
        };
        assert_eq!((width, height), (110, 50), "Should apply latest size");
    }

    #[test]
    fn hard_deadline_forces_apply() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // First resize
        c.handle_resize_at(100, 40, base);

        // Wait past hard deadline
        let action = c.tick_at(base + Duration::from_millis(150));

        let forced_by_deadline = if let CoalesceAction::ApplyResize {
            forced_by_deadline, ..
        } = action
        {
            forced_by_deadline
        } else {
            assert!(
                matches!(action, CoalesceAction::ApplyResize { .. }),
                "Expected ApplyResize, got {action:?}"
            );
            return;
        };
        assert!(forced_by_deadline, "Should be forced by deadline");
    }

    /// CONTRACT (bd-1za0z): an isolated resize arriving after a quiet gap is
    /// applied instantly but NOT flagged forced — nothing was waiting on the
    /// deadline, so it is not an SLA breach.
    #[test]
    fn isolated_resize_after_quiet_gap_is_not_forced() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();
        let action = c.handle_resize_at(120, 40, base + Duration::from_millis(500));

        match action {
            CoalesceAction::ApplyResize {
                forced_by_deadline, ..
            } => {
                assert!(
                    !forced_by_deadline,
                    "quiet-gap instant apply must not count as forced"
                );
            }
            other => panic!("Expected ApplyResize, got {other:?}"),
        }
    }

    /// CONTRACT (bd-1za0z): when a resize arrives while an earlier resize is
    /// STILL pending past the hard deadline, that IS an SLA breach — forced
    /// stays true on the event path too.
    #[test]
    fn resize_arriving_while_pending_past_deadline_stays_forced() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();
        // First event coalesces (no dt yet), leaving work waiting.
        c.handle_resize_at(90, 30, base);

        // Second event arrives after the hard deadline elapsed with the
        // first still unapplied.
        let action = c.handle_resize_at(120, 40, base + Duration::from_millis(500));

        match action {
            CoalesceAction::ApplyResize {
                forced_by_deadline, ..
            } => {
                assert!(
                    forced_by_deadline,
                    "deadline breach of WAITING work must stay forced"
                );
            }
            other => panic!("Expected ApplyResize, got {other:?}"),
        }
    }

    #[test]
    fn burst_mode_detection() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // Rapid events to trigger burst mode
        for i in 0..15 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }

        assert_eq!(c.regime(), Regime::Burst);
    }

    #[test]
    fn steady_mode_fast_response() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // Single resize
        c.handle_resize_at(100, 40, base);

        // In steady mode, should apply after steady_delay
        let action = c.tick_at(base + Duration::from_millis(20));

        assert!(matches!(action, CoalesceAction::ApplyResize { .. }));
    }

    #[test]
    fn record_external_apply_updates_state_and_logs() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);
        c.record_external_apply(120, 50, base + Duration::from_millis(5));

        assert!(!c.has_pending());
        assert_eq!(c.last_applied(), (120, 50));

        let summary = c.decision_summary();
        assert_eq!(summary.apply_count, 1);
        assert_eq!(summary.last_applied, (120, 50));
        assert!(
            c.logs()
                .iter()
                .any(|entry| entry.action == "apply_immediate"),
            "record_external_apply should emit apply_immediate decision"
        );
    }

    #[test]
    fn coalesce_time_tracked() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        c.handle_resize_at(100, 40, base);
        let action = c.tick_at(base + Duration::from_millis(50));

        let coalesce_time = if let CoalesceAction::ApplyResize { coalesce_time, .. } = action {
            coalesce_time
        } else {
            assert!(
                matches!(action, CoalesceAction::ApplyResize { .. }),
                "Expected ApplyResize"
            );
            return;
        };
        assert!(coalesce_time >= Duration::from_millis(40));
        assert!(coalesce_time <= Duration::from_millis(60));
    }

    #[test]
    fn event_rate_calculation() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // 10 events over 1 second = 10 events/sec
        for i in 0..10 {
            c.handle_resize_at(80 + i, 24, base + Duration::from_millis(i as u64 * 100));
        }

        let rate = c.calculate_event_rate(base + Duration::from_millis(1000));
        assert!(rate > 8.0 && rate < 12.0, "Rate should be ~10 events/sec");
    }

    #[test]
    fn rapid_burst_triggers_high_rate() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
        let base = Instant::now();

        // Simulate 8 events arriving at the EXACT same time (or < 1ms)
        for _ in 0..8 {
            c.handle_resize_at(80, 24, base);
        }

        let rate = c.calculate_event_rate(base);
        // 8 events / 0.001s = 8000 events/sec
        assert!(
            rate >= 1000.0,
            "Rate should be high for instantaneous burst, got {}",
            rate
        );
    }

    #[test]
    fn cooldown_prevents_immediate_exit() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // Enter burst mode
        for i in 0..15 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }
        assert_eq!(c.regime(), Regime::Burst);

        // Rate should drop but cooldown prevents immediate exit
        c.tick_at(base + Duration::from_millis(500));
        c.tick_at(base + Duration::from_millis(600));

        // Cooldown holds while the rate window still reports the storm.
        assert_eq!(c.regime(), Regime::Burst);

        // The event-rate window clears after a >1s quiet gap; the cooldown
        // then decrements per tick and Burst exits.
        for i in 0..6 {
            c.tick_at(base + Duration::from_millis(1300 + 100 * i));
        }

        // Should have exited burst (this pins the cooldown exit path —
        // with cooldown_frames armed to at least 1, Burst always has an
        // exit once the rate decays).
        assert_eq!(c.stats().regime, Regime::Steady);
    }

    #[test]
    fn cooldown_frames_zero_cannot_dead_end_burst() {
        // Regression: cooldown_frames = 0 armed cooldown_remaining = 0 on
        // Burst entry; the only Burst->Steady exit required
        // cooldown_remaining > 0, so the regime pinned at Burst forever and
        // the load governor latched Stressed for the session. Arming now
        // clamps to >= 1 frame.
        let mut c = ResizeCoalescer::new(
            CoalescerConfig {
                cooldown_frames: 0,
                ..Default::default()
            },
            (80, 24),
        );
        let base = Instant::now();
        // Storm: enter Burst.
        for i in 0..10 {
            let _ =
                c.handle_resize_at(100 + i, 40, base + Duration::from_millis(10 * u64::from(i)));
        }
        assert_eq!(c.stats().regime, Regime::Burst);
        // Storm over: quiet ticks must eventually exit Burst.
        for i in 1..=20 {
            c.tick_at(base + Duration::from_millis(100 + 200 * i));
        }
        assert_eq!(
            c.stats().regime,
            Regime::Steady,
            "Burst must always have an exit path"
        );
    }

    #[test]
    fn logging_captures_decisions() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);
        c.tick_at(base + Duration::from_millis(50));

        assert!(!c.logs().is_empty());
        assert_eq!(c.logs()[0].action, "coalesce");
    }

    #[test]
    fn logging_jsonl_format() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        c.handle_resize_at(100, 40, Instant::now());
        c.tick_at(Instant::now() + Duration::from_millis(50));

        let (cols, rows) = c.last_applied();
        let jsonl = c.logs()[0].to_jsonl("resize-test", ScreenMode::AltScreen, cols, rows);

        assert!(jsonl.contains("\"event\":\"decision\""));
        assert!(jsonl.contains("\"action\":\"coalesce\""));
        assert!(jsonl.contains("\"regime\":\"steady\""));
        assert!(jsonl.contains("\"pending_w\":100"));
        assert!(jsonl.contains("\"pending_h\":40"));
    }

    #[test]
    fn apply_logs_coalesce_ms() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);
        let action = c.tick_at(base + Duration::from_millis(50));
        assert!(matches!(action, CoalesceAction::ApplyResize { .. }));

        let last = c.logs().last().expect("Expected a decision log entry");
        assert!(last.coalesce_ms.is_some());
        assert!(last.coalesce_ms.unwrap() >= 0.0);
    }

    #[test]
    fn decision_checksum_is_stable() {
        let mut config = test_config();
        config.enable_logging = true;

        let base = Instant::now();
        let mut c1 = ResizeCoalescer::new(config.clone(), (80, 24)).with_last_render(base);
        let mut c2 = ResizeCoalescer::new(config, (80, 24)).with_last_render(base);

        for c in [&mut c1, &mut c2] {
            c.handle_resize_at(90, 30, base);
            c.handle_resize_at(100, 40, base + Duration::from_millis(10));
            let _ = c.tick_at(base + Duration::from_millis(80));
        }

        assert_eq!(c1.decision_checksum(), c2.decision_checksum());
    }

    #[test]
    fn evidence_jsonl_includes_summary() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        c.handle_resize_at(100, 40, Instant::now());
        c.tick_at(Instant::now() + Duration::from_millis(50));

        let jsonl = c.evidence_to_jsonl();

        assert!(jsonl.contains("\"event\":\"config\""));
        assert!(jsonl.contains("\"event\":\"summary\""));
    }

    #[test]
    fn evidence_jsonl_parses_and_has_required_fields() {
        use serde_json::Value;

        let mut config = test_config();
        config.enable_logging = true;
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24))
            .with_last_render(base)
            .with_evidence_run_id("resize-test")
            .with_screen_mode(ScreenMode::AltScreen);

        c.handle_resize_at(90, 30, base);
        c.handle_resize_at(100, 40, base + Duration::from_millis(10));
        let _ = c.tick_at(base + Duration::from_millis(120));

        let jsonl = c.evidence_to_jsonl();
        let mut saw_config = false;
        let mut saw_decision = false;
        let mut saw_summary = false;

        for line in jsonl.lines() {
            let value: Value = serde_json::from_str(line).expect("valid JSONL evidence");
            let event = value
                .get("event")
                .and_then(Value::as_str)
                .expect("event field");
            assert_eq!(value["schema_version"], EVIDENCE_SCHEMA_VERSION);
            assert_eq!(value["run_id"], "resize-test");
            assert!(
                value["event_idx"].is_number(),
                "event_idx should be numeric"
            );
            assert_eq!(value["screen_mode"], "altscreen");
            assert!(value["cols"].is_number(), "cols should be numeric");
            assert!(value["rows"].is_number(), "rows should be numeric");
            match event {
                "config" => {
                    for key in [
                        "steady_delay_ms",
                        "burst_delay_ms",
                        "hard_deadline_ms",
                        "burst_enter_rate",
                        "burst_exit_rate",
                        "cooldown_frames",
                        "rate_window_size",
                        "logging_enabled",
                    ] {
                        assert!(value.get(key).is_some(), "missing config field {key}");
                    }
                    saw_config = true;
                }
                "decision" => {
                    for key in [
                        "idx",
                        "elapsed_ms",
                        "dt_ms",
                        "event_rate",
                        "regime",
                        "action",
                        "pending_w",
                        "pending_h",
                        "applied_w",
                        "applied_h",
                        "time_since_render_ms",
                        "coalesce_ms",
                        "forced",
                        "transition_reason_code",
                        "transition_confidence",
                    ] {
                        assert!(value.get(key).is_some(), "missing decision field {key}");
                    }
                    saw_decision = true;
                }
                "summary" => {
                    for key in [
                        "decisions",
                        "applies",
                        "forced_applies",
                        "coalesces",
                        "skips",
                        "regime",
                        "last_w",
                        "last_h",
                        "checksum",
                    ] {
                        assert!(value.get(key).is_some(), "missing summary field {key}");
                    }
                    saw_summary = true;
                }
                _ => {}
            }
        }

        assert!(saw_config, "config evidence missing");
        assert!(saw_decision, "decision evidence missing");
        assert!(saw_summary, "summary evidence missing");
    }

    #[test]
    fn evidence_jsonl_is_deterministic_for_fixed_schedule() {
        let mut config = test_config();
        config.enable_logging = true;
        let base = Instant::now();

        let run = || {
            let mut c = ResizeCoalescer::new(config.clone(), (80, 24))
                .with_last_render(base)
                .with_evidence_run_id("resize-test")
                .with_screen_mode(ScreenMode::AltScreen);
            c.handle_resize_at(90, 30, base);
            c.handle_resize_at(100, 40, base + Duration::from_millis(10));
            let _ = c.tick_at(base + Duration::from_millis(120));
            c.evidence_to_jsonl()
        };

        let first = run();
        let second = run();
        assert_eq!(first, second);
    }

    #[test]
    fn bocpd_logging_inherits_coalescer_logging() {
        let mut config = test_config();
        config.enable_bocpd = true;
        config.bocpd_config = Some(BocpdConfig::default());

        let c = ResizeCoalescer::new(config, (80, 24));
        let bocpd = c.bocpd().expect("BOCPD should be enabled");
        assert!(bocpd.config().enable_logging);
    }

    #[test]
    fn stats_reflect_state() {
        let mut c = ResizeCoalescer::new(test_config(), (80, 24));
        let base = Instant::now();

        c.handle_resize_at(100, 40, base);
        let action = c.tick_at(base + Duration::from_millis(5));
        assert_eq!(action, CoalesceAction::None);

        let stats = c.stats();
        assert_eq!(stats.event_count, 1);
        assert!(stats.has_pending);
        assert_eq!(stats.last_applied, (80, 24));

        let action = c.tick_at(base + Duration::from_millis(50));
        assert_eq!(
            action,
            CoalesceAction::ApplyResize {
                width: 100,
                height: 40,
                coalesce_time: Duration::from_millis(50),
                forced_by_deadline: false,
            }
        );

        let stats = c.stats();
        assert!(!stats.has_pending);
        assert_eq!(stats.last_applied, (100, 40));
    }

    #[test]
    fn time_until_apply_calculation() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);

        let time_left = c.time_until_apply(base + Duration::from_millis(5));
        assert!(time_left.is_some());
        let time_left = time_left.unwrap();
        assert!(time_left.as_millis() > 0);
        assert!(time_left.as_millis() < config.steady_delay_ms as u128);
    }

    #[test]
    fn deterministic_behavior() {
        let config = test_config();

        // Run twice with same inputs
        let results: Vec<_> = (0..2)
            .map(|_| {
                let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                let base = Instant::now();

                for i in 0..5 {
                    c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 20));
                }

                c.tick_at(base + Duration::from_millis(200))
            })
            .collect();

        assert_eq!(results[0], results[1], "Results must be deterministic");
    }

    #[test]
    fn transition_reason_codes_and_evidence_fields_are_logged() {
        let mut config = test_config();
        config.enable_logging = true;
        config.hard_deadline_ms = 5_000;
        config.burst_delay_ms = 50;
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24)).with_last_render(base);

        for i in 0..12 {
            c.handle_resize_at(90 + i, 30, base + Duration::from_millis(i as u64 * 10));
        }

        let transition = c
            .transition_logs()
            .first()
            .expect("rapid events should trigger a transition");
        assert_eq!(transition.from_regime, Regime::Steady);
        assert_eq!(transition.to_regime, Regime::Burst);
        assert_eq!(
            transition.reason_code,
            TransitionReasonCode::HeuristicEnterBurstRate
        );
        assert!(
            (0.0..=1.0).contains(&transition.confidence),
            "transition confidence should be normalized"
        );
        assert!(
            transition.event_rate >= 0.0,
            "event-rate evidence should be included"
        );

        let decision_with_transition = c
            .logs()
            .iter()
            .find(|entry| entry.transition_reason_code.is_some())
            .expect("transition decisions should include reason code/evidence");
        assert_eq!(
            decision_with_transition.transition_reason_code,
            Some(TransitionReasonCode::HeuristicEnterBurstRate)
        );
        assert!(decision_with_transition.transition_confidence.is_some());

        let jsonl = c.evidence_to_jsonl();
        assert!(jsonl.contains("\"event\":\"regime_transition\""));
        assert!(jsonl.contains("\"reason_code\":\"heuristic_enter_burst_rate\""));
        assert!(jsonl.contains("\"transition_reason_code\":"));
    }

    #[test]
    fn regime_transition_sequence_is_deterministic_for_fixed_schedule() {
        let config = CoalescerConfig {
            burst_enter_rate: 5.0,
            burst_exit_rate: 2.0,
            cooldown_frames: 3,
            rate_window_size: 4,
            steady_delay_ms: 10,
            burst_delay_ms: 50,
            hard_deadline_ms: 5_000,
            enable_logging: true,
            enable_bocpd: false,
            bocpd_config: None,
        };
        let base = Instant::now();

        let run = || {
            let mut c = ResizeCoalescer::new(config.clone(), (80, 24)).with_last_render(base);

            // Enter burst with rapid events.
            for i in 0..8u64 {
                let t = base + Duration::from_millis(30 * i);
                c.handle_resize_at(80 + i as u16, 24 + i as u16, t);
            }

            // Apply pending and flush rate window with slow events.
            let mut t = base + Duration::from_millis(280);
            let _ = c.tick_at(t);
            for i in 0..5u64 {
                t += Duration::from_secs(1);
                c.handle_resize_at(100 + i as u16, 30 + i as u16, t);
                let _ = c.tick_at(t + Duration::from_millis(60));
            }

            // Drain cooldown without triggering apply.
            t += Duration::from_millis(70);
            c.handle_resize_at(120, 35, t);
            for step in 1..=config.cooldown_frames {
                let _ = c.tick_at(t + Duration::from_millis(step as u64 * 5));
            }

            c.transition_logs()
                .iter()
                .map(|entry| {
                    (
                        entry.from_regime,
                        entry.to_regime,
                        entry.reason_code,
                        entry.event_idx,
                        entry.cooldown_remaining,
                    )
                })
                .collect::<Vec<_>>()
        };

        let first = run();
        let second = run();
        assert_eq!(first, second);
        assert!(
            first.iter().any(|(_, to, reason, _, _)| {
                *to == Regime::Burst && *reason == TransitionReasonCode::HeuristicEnterBurstRate
            }),
            "expected steady->burst transition with heuristic reason"
        );
        assert!(
            first.iter().any(|(_, to, reason, _, _)| {
                *to == Regime::Steady && *reason == TransitionReasonCode::HeuristicExitBurstCooldown
            }),
            "expected burst->steady transition with cooldown reason"
        );
    }

    #[test]
    fn bounded_oscillation_and_converges_to_steady() {
        let config = CoalescerConfig {
            burst_enter_rate: 5.0,
            burst_exit_rate: 2.0,
            cooldown_frames: 3,
            rate_window_size: 4,
            steady_delay_ms: 10,
            burst_delay_ms: 50,
            hard_deadline_ms: 5_000,
            enable_logging: true,
            enable_bocpd: false,
            bocpd_config: None,
        };
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24)).with_last_render(base);
        let mut t = base;

        // Alternate burst pulses with limited cooldown opportunities.
        for cycle in 0..30u64 {
            for pulse in 0..6u64 {
                t += Duration::from_millis(30);
                c.handle_resize_at(80 + ((cycle + pulse) % 40) as u16, 24 + pulse as u16, t);
            }
            t += Duration::from_millis(70);
            let _ = c.tick_at(t);

            t += Duration::from_secs(1);
            c.handle_resize_at(120 + (cycle % 20) as u16, 30 + (cycle % 5) as u16, t);
            let _ = c.tick_at(t + Duration::from_millis(60));
        }

        let transitions_before_convergence = c.regime_transition_count();
        assert!(
            transitions_before_convergence <= 4,
            "oscillation should stay bounded, transitions={}",
            transitions_before_convergence
        );

        // Final quiet period: explicitly drain cooldown and verify convergence.
        t += Duration::from_secs(1);
        c.handle_resize_at(160, 40, t);
        for step in 1..=config.cooldown_frames {
            let _ = c.tick_at(t + Duration::from_millis(step as u64 * 5));
        }
        assert_eq!(c.regime(), Regime::Steady);
        let last_transition = c
            .transition_logs()
            .last()
            .expect("expected at least one transition");
        assert_eq!(last_transition.to_regime, Regime::Steady);
        assert_eq!(
            last_transition.reason_code,
            TransitionReasonCode::HeuristicExitBurstCooldown
        );
    }

    #[test]
    fn simulation_bocpd_vs_heuristic_metrics() {
        let tick_ms = 5;
        // Keep heuristic thresholds high so burst classification is conservative,
        // while BOCPD uses a responsive posterior to detect bursty streams.
        let mut heuristic_config = test_config();
        heuristic_config.burst_enter_rate = 60.0;
        heuristic_config.burst_exit_rate = 30.0;
        let mut bocpd_cfg = BocpdConfig::responsive();
        bocpd_cfg.burst_prior = 0.35;
        bocpd_cfg.steady_threshold = 0.2;
        bocpd_cfg.burst_threshold = 0.6;
        let bocpd_config = heuristic_config.clone().with_bocpd_config(bocpd_cfg);
        let patterns = vec![
            ("steady", steady_pattern()),
            ("burst", burst_pattern()),
            ("oscillatory", oscillatory_pattern()),
        ];

        for (pattern, events) in patterns {
            let heuristic = run_simulation(&events, heuristic_config.clone(), tick_ms);
            let bocpd = run_simulation(&events, bocpd_config.clone(), tick_ms);

            let heuristic_jsonl = heuristic.to_jsonl(pattern, "heuristic");
            let bocpd_jsonl = bocpd.to_jsonl(pattern, "bocpd");
            let comparison = SimulationComparison::from_metrics(heuristic, bocpd);
            let comparison_jsonl = comparison.to_jsonl(pattern);

            eprintln!("{heuristic_jsonl}");
            eprintln!("{bocpd_jsonl}");
            eprintln!("{comparison_jsonl}");

            assert!(heuristic_jsonl.contains("\"event\":\"simulation_summary\""));
            assert!(bocpd_jsonl.contains("\"event\":\"simulation_summary\""));
            assert!(comparison_jsonl.contains("\"event\":\"simulation_compare\""));

            #[allow(clippy::cast_precision_loss)]
            let max_allowed = test_config().hard_deadline_ms as f64 + 1.0;
            assert!(
                heuristic.max_coalesce_ms <= max_allowed,
                "heuristic latency bounded for {pattern}"
            );
            assert!(
                bocpd.max_coalesce_ms <= max_allowed,
                "bocpd latency bounded for {pattern}"
            );

            if pattern == "burst" {
                let event_count = as_u64(events.len());
                assert!(
                    heuristic.apply_count < event_count,
                    "heuristic should coalesce under burst pattern"
                );
                assert!(
                    bocpd.apply_count < event_count,
                    "bocpd should coalesce under burst pattern"
                );
                assert!(
                    comparison.apply_delta >= 0,
                    "BOCPD should not increase renders in burst (apply_delta={})",
                    comparison.apply_delta
                );
            }
        }
    }

    #[test]
    fn never_drops_final_size() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();

        // Many rapid resizes that may trigger some applies due to hard deadline
        let mut intermediate_applies = Vec::new();
        for i in 0..100 {
            let action = c.handle_resize_at(
                80 + (i % 50),
                24 + (i % 30),
                base + Duration::from_millis(i as u64 * 5),
            );
            if let CoalesceAction::ApplyResize { width, height, .. } = action {
                intermediate_applies.push((width, height));
            }
        }

        // The final size - may apply immediately if deadline is hit
        let final_action = c.handle_resize_at(200, 100, base + Duration::from_millis(600));

        let applied_size = if let CoalesceAction::ApplyResize { width, height, .. } = final_action {
            Some((width, height))
        } else {
            // If not applied immediately, tick until it is
            let mut result = None;
            for tick in 0..100 {
                let action = c.tick_at(base + Duration::from_millis(700 + tick * 20));
                if let CoalesceAction::ApplyResize { width, height, .. } = action {
                    result = Some((width, height));
                    break;
                }
            }
            result
        };

        assert_eq!(
            applied_size,
            Some((200, 100)),
            "Must apply final size 200x100"
        );
    }

    #[test]
    fn bounded_latency_invariant() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);

        // Simulate time passing without any new events
        let mut applied_at = None;
        for ms in 0..200 {
            let now = base + Duration::from_millis(ms);
            let action = c.tick_at(now);
            if matches!(action, CoalesceAction::ApplyResize { .. }) {
                applied_at = Some(ms);
                break;
            }
        }

        assert!(applied_at.is_some(), "Must apply within reasonable time");
        assert!(
            applied_at.unwrap() <= config.hard_deadline_ms,
            "Must apply within hard deadline"
        );
    }

    // =========================================================================
    // Property tests (bd-1rz0.8)
    // =========================================================================

    mod property {
        use super::*;
        use proptest::prelude::*;

        /// Strategy for generating resize dimensions.
        fn dimension() -> impl Strategy<Value = u16> {
            1u16..500
        }

        /// Strategy for generating resize event sequences.
        fn resize_sequence(max_len: usize) -> impl Strategy<Value = Vec<(u16, u16, u64)>> {
            proptest::collection::vec((dimension(), dimension(), 0u64..200), 0..max_len)
        }

        proptest! {
            /// Property: identical sequences yield identical final results.
            ///
            /// The coalescer must be deterministic - same inputs → same outputs.
            #[test]
            fn determinism_across_sequences(
                events in resize_sequence(50),
                tick_offset in 100u64..500
            ) {
                let config = CoalescerConfig::default();

                let results: Vec<_> = (0..2)
                    .map(|_| {
                        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                        let base = Instant::now();

                        for (i, (w, h, delay)) in events.iter().enumerate() {
                            let offset = events[..i].iter().map(|(_, _, d)| *d).sum::<u64>() + delay;
                            c.handle_resize_at(*w, *h, base + Duration::from_millis(offset));
                        }

                        // Tick to trigger apply
                        let total_time = events.iter().map(|(_, _, d)| d).sum::<u64>() + tick_offset;
                        c.tick_at(base + Duration::from_millis(total_time))
                    })
                    .collect();

                prop_assert_eq!(results[0], results[1], "Results must be deterministic");
            }

            /// Property: the latest resize is never lost (latest-wins semantics).
            ///
            /// When all coalescing completes, the applied size must match the
            /// final requested size.
            #[test]
            fn latest_wins_never_drops(
                events in resize_sequence(20),
                final_w in dimension(),
                final_h in dimension()
            ) {
                if events.is_empty() {
                    // No events to test
                    return Ok(());
                }

                let config = CoalescerConfig::default();
                let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                let base = Instant::now();

                // Feed all events
                let mut offset = 0u64;
                for (w, h, delay) in &events {
                    offset += delay;
                    c.handle_resize_at(*w, *h, base + Duration::from_millis(offset));
                }

                // Add final event
                offset += 50;
                c.handle_resize_at(final_w, final_h, base + Duration::from_millis(offset));

                // Tick until we get an apply
                let mut result = None;
                for tick in 0..200 {
                    let action = c.tick_at(base + Duration::from_millis(offset + 10 + tick * 20));
                    if let CoalesceAction::ApplyResize { width, height, .. } = action {
                        result = Some((width, height));
                        break;
                    }
                }

                // The final applied size must match the latest requested size
                if let Some((applied_w, applied_h)) = result {
                    prop_assert_eq!(
                        (applied_w, applied_h),
                        (final_w, final_h),
                        "Must apply the final size {} x {}",
                        final_w,
                        final_h
                    );
                }
            }

            /// Property: bounded latency is always maintained.
            ///
            /// A pending resize must be applied within hard_deadline_ms.
            #[test]
            fn bounded_latency_maintained(
                w in dimension(),
                h in dimension()
            ) {
                let config = CoalescerConfig::default();
                // Use (0,0) so no generated size (1..500) hits the skip_same_size path
                let mut c = ResizeCoalescer::new(config.clone(), (0, 0));
                let base = Instant::now();

                c.handle_resize_at(w, h, base);

                // Tick forward until applied
                let mut applied_at = None;
                for ms in 0..=config.hard_deadline_ms + 50 {
                    let action = c.tick_at(base + Duration::from_millis(ms));
                    if matches!(action, CoalesceAction::ApplyResize { .. }) {
                        applied_at = Some(ms);
                        break;
                    }
                }

                prop_assert!(applied_at.is_some(), "Resize must be applied");
                prop_assert!(
                    applied_at.unwrap() <= config.hard_deadline_ms,
                    "Must apply within hard deadline ({}ms), took {}ms",
                    config.hard_deadline_ms,
                    applied_at.unwrap()
                );
            }

            /// Property: applied sizes are never corrupted.
            ///
            /// When a resize is applied, the dimensions must exactly match
            /// what was requested (no off-by-one, no swapped axes).
            #[test]
            fn no_size_corruption(
                w in dimension(),
                h in dimension()
            ) {
                let config = CoalescerConfig::default();
                // Use (0,0) so no generated size (1..500) hits the skip_same_size path
                let mut c = ResizeCoalescer::new(config.clone(), (0, 0));
                let base = Instant::now();

                c.handle_resize_at(w, h, base);

                // Tick until applied
                let mut result = None;
                for ms in 0..200 {
                    let action = c.tick_at(base + Duration::from_millis(ms));
                    if let CoalesceAction::ApplyResize { width, height, .. } = action {
                        result = Some((width, height));
                        break;
                    }
                }

                prop_assert!(result.is_some());
                let (applied_w, applied_h) = result.unwrap();
                prop_assert_eq!(applied_w, w, "Width must not be corrupted");
                prop_assert_eq!(applied_h, h, "Height must not be corrupted");
            }

            /// Property: regime transitions are monotonic with event rate.
            ///
            /// Higher event rates should more reliably trigger burst mode.
            #[test]
            fn regime_follows_event_rate(
                event_count in 1usize..30
            ) {
                let config = CoalescerConfig {
                    burst_enter_rate: 10.0,
                    burst_exit_rate: 5.0,
                    ..CoalescerConfig::default()
                };
                let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                let base = Instant::now();

                // Very fast events (>10/sec) should trigger burst mode
                for i in 0..event_count {
                    c.handle_resize_at(
                        80 + i as u16,
                        24,
                        base + Duration::from_millis(i as u64 * 50), // 20 events/sec
                    );
                }

                // With enough fast events, should enter burst
                if event_count >= 10 {
                    prop_assert_eq!(
                        c.regime(),
                        Regime::Burst,
                        "Many rapid events should trigger burst mode"
                    );
                }
            }

            /// Property: event count invariant - coalescer tracks all incoming events.
            ///
            /// The `event_count` field tracks ALL resize events for rate calculation
            /// and telemetry, including same-size events that are skipped.
            #[test]
            fn event_count_invariant(
                events in resize_sequence(100)
            ) {
                let config = CoalescerConfig::default();
                let mut c = ResizeCoalescer::new(config, (80, 24));
                let base = Instant::now();

                for (w, h, delay) in &events {
                    c.handle_resize_at(*w, *h, base + Duration::from_millis(*delay));
                }

                let stats = c.stats();
                // Event count should equal total incoming events (for rate calculation).
                prop_assert_eq!(
                    stats.event_count,
                    events.len() as u64,
                    "Event count should match total incoming events"
                );
            }

            // =========================================================================
            // BOCPD Property Tests (bd-3e1t.2.5)
            // =========================================================================

            /// Property: BOCPD determinism - identical sequences yield identical results.
            #[test]
            fn bocpd_determinism_across_sequences(
                events in resize_sequence(30),
                tick_offset in 100u64..400
            ) {
                let config = CoalescerConfig::default().with_bocpd();

                let results: Vec<_> = (0..2)
                    .map(|_| {
                        let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                        let base = Instant::now();

                        for (i, (w, h, delay)) in events.iter().enumerate() {
                            let offset = events[..i].iter().map(|(_, _, d)| *d).sum::<u64>() + delay;
                            c.handle_resize_at(*w, *h, base + Duration::from_millis(offset));
                        }

                        let total_time = events.iter().map(|(_, _, d)| d).sum::<u64>() + tick_offset;
                        let action = c.tick_at(base + Duration::from_millis(total_time));
                        (action, c.regime(), c.bocpd_p_burst())
                    })
                    .collect();

                prop_assert_eq!(results[0], results[1], "BOCPD results must be deterministic");
            }

            /// Property: BOCPD latest-wins - final resize is always applied.
            #[test]
            fn bocpd_latest_wins_never_drops(
                events in resize_sequence(15),
                final_w in dimension(),
                final_h in dimension()
            ) {
                if events.is_empty() {
                    return Ok(());
                }

                let config = CoalescerConfig::default().with_bocpd();
                let mut c = ResizeCoalescer::new(config, (80, 24));
                let base = Instant::now();

                let mut offset = 0u64;
                for (w, h, delay) in &events {
                    offset += delay;
                    c.handle_resize_at(*w, *h, base + Duration::from_millis(offset));
                }

                offset += 50;
                c.handle_resize_at(final_w, final_h, base + Duration::from_millis(offset));

                let mut final_applied = None;
                for tick in 0..200 {
                    let action = c.tick_at(base + Duration::from_millis(offset + 10 + tick * 20));
                    if let CoalesceAction::ApplyResize { width, height, .. } = action {
                        final_applied = Some((width, height));
                    }
                    if !c.has_pending() && final_applied.is_some() {
                        break;
                    }
                }

                if let Some((applied_w, applied_h)) = final_applied {
                    prop_assert_eq!(
                        (applied_w, applied_h),
                        (final_w, final_h),
                        "BOCPD must apply the final size"
                    );
                }
            }

            /// Property: BOCPD bounded latency - hard deadline is always met.
            #[test]
            fn bocpd_bounded_latency_maintained(
                w in dimension(),
                h in dimension()
            ) {
                let config = CoalescerConfig::default().with_bocpd();
                let mut c = ResizeCoalescer::new(config.clone(), (0, 0));
                let base = Instant::now();

                c.handle_resize_at(w, h, base);

                let mut applied_at = None;
                for ms in 0..=config.hard_deadline_ms + 50 {
                    let action = c.tick_at(base + Duration::from_millis(ms));
                    if matches!(action, CoalesceAction::ApplyResize { .. }) {
                        applied_at = Some(ms);
                        break;
                    }
                }

                prop_assert!(applied_at.is_some(), "BOCPD resize must be applied");
                prop_assert!(
                    applied_at.unwrap() <= config.hard_deadline_ms,
                    "BOCPD must apply within hard deadline ({}ms), took {}ms",
                    config.hard_deadline_ms,
                    applied_at.unwrap()
                );
            }

            /// Property: BOCPD posterior always valid (normalized, bounded).
            #[test]
            fn bocpd_posterior_always_valid(
                events in resize_sequence(50)
            ) {
                if events.is_empty() {
                    return Ok(());
                }

                let config = CoalescerConfig::default().with_bocpd();
                let mut c = ResizeCoalescer::new(config, (80, 24));
                let base = Instant::now();

                for (w, h, delay) in &events {
                    c.handle_resize_at(*w, *h, base + Duration::from_millis(*delay));

                    // Check posterior validity after each event
                    if let Some(bocpd) = c.bocpd() {
                        let sum: f64 = bocpd.run_length_posterior().iter().sum();
                        prop_assert!(
                            (sum - 1.0).abs() < 1e-8,
                            "Posterior must sum to 1, got {}",
                            sum
                        );
                    }

                    let p_burst = c.bocpd_p_burst().unwrap();
                    prop_assert!(
                        (0.0..=1.0).contains(&p_burst),
                        "P(burst) must be in [0,1], got {}",
                        p_burst
                    );
                }
            }
        }
    }

    // =========================================================================
    // Telemetry Hooks Tests (bd-1rz0.7)
    // =========================================================================

    #[test]
    fn telemetry_hooks_fire_on_resize_applied() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicU32, Ordering};

        let applied_count = Arc::new(AtomicU32::new(0));
        let applied_count_clone = applied_count.clone();

        let hooks = TelemetryHooks::new().on_resize_applied(move |_entry| {
            applied_count_clone.fetch_add(1, Ordering::SeqCst);
        });

        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24)).with_telemetry_hooks(hooks);

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);
        c.tick_at(base + Duration::from_millis(50));

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

    #[test]
    fn telemetry_hooks_fire_on_regime_change() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicU32, Ordering};

        let regime_changes = Arc::new(AtomicU32::new(0));
        let regime_changes_clone = regime_changes.clone();

        let hooks = TelemetryHooks::new().on_regime_change(move |_from, _to| {
            regime_changes_clone.fetch_add(1, Ordering::SeqCst);
        });

        let config = test_config();
        let mut c = ResizeCoalescer::new(config, (80, 24)).with_telemetry_hooks(hooks);

        let base = Instant::now();

        // Rapid events to trigger burst mode
        for i in 0..15 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }

        // Should have triggered at least one regime change (Steady -> Burst)
        assert!(regime_changes.load(Ordering::SeqCst) >= 1);
    }

    #[test]
    fn regime_transition_count_tracks_changes() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        assert_eq!(c.regime_transition_count(), 0);

        let base = Instant::now();

        // Rapid events to trigger burst mode
        for i in 0..15 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }

        // Should have transitioned to Burst at least once
        assert!(c.regime_transition_count() >= 1);
    }

    #[test]
    fn cycle_time_percentiles_calculated() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        // Initially no percentiles
        assert!(c.cycle_time_percentiles().is_none());

        let base = Instant::now();

        // Generate multiple applies to get cycle times
        for i in 0..5 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 100));
            c.tick_at(base + Duration::from_millis(i as u64 * 100 + 50));
        }

        // Now should have percentiles
        let percentiles = c.cycle_time_percentiles();
        assert!(percentiles.is_some());

        let p = percentiles.unwrap();
        assert!(p.count >= 1);
        assert!(p.mean_ms >= 0.0);
        assert!(p.p50_ms >= 0.0);
        assert!(p.p95_ms >= p.p50_ms);
        assert!(p.p99_ms >= p.p95_ms);
    }

    #[test]
    fn cycle_time_percentiles_jsonl_format() {
        let percentiles = CycleTimePercentiles {
            p50_ms: 10.5,
            p95_ms: 25.3,
            p99_ms: 42.1,
            count: 100,
            mean_ms: 15.2,
        };

        let jsonl = percentiles.to_jsonl();
        assert!(jsonl.contains("\"event\":\"cycle_time_percentiles\""));
        assert!(jsonl.contains("\"p50_ms\":10.500"));
        assert!(jsonl.contains("\"p95_ms\":25.300"));
        assert!(jsonl.contains("\"p99_ms\":42.100"));
        assert!(jsonl.contains("\"mean_ms\":15.200"));
        assert!(jsonl.contains("\"count\":100"));
    }

    // =========================================================================
    // BOCPD Integration Tests (bd-3e1t.2.2)
    // =========================================================================

    #[test]
    fn bocpd_disabled_by_default() {
        let c = ResizeCoalescer::new(CoalescerConfig::default(), (80, 24));
        assert!(!c.bocpd_enabled());
        assert!(c.bocpd().is_none());
        assert!(c.bocpd_p_burst().is_none());
    }

    #[test]
    fn bocpd_enabled_with_config() {
        let config = CoalescerConfig::default().with_bocpd();
        let c = ResizeCoalescer::new(config, (80, 24));
        assert!(c.bocpd_enabled());
        assert!(c.bocpd().is_some());
    }

    #[test]
    fn bocpd_posterior_normalized() {
        let config = CoalescerConfig::default().with_bocpd();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();

        // Feed events with various inter-arrival times
        for i in 0..20 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 50));
        }

        // Check posterior is valid probability
        let p_burst = c.bocpd_p_burst().expect("BOCPD should be enabled");
        assert!(
            (0.0..=1.0).contains(&p_burst),
            "P(burst) must be in [0,1], got {}",
            p_burst
        );

        // Check BOCPD internal posterior is normalized
        if let Some(bocpd) = c.bocpd() {
            let sum: f64 = bocpd.run_length_posterior().iter().sum();
            assert!(
                (sum - 1.0).abs() < 1e-9,
                "Posterior must sum to 1, got {}",
                sum
            );
        }
    }

    #[test]
    fn bocpd_detects_burst_from_rapid_events() {
        use crate::bocpd::BocpdConfig;

        // Configure BOCPD with clear burst detection
        let bocpd_config = BocpdConfig {
            mu_steady_ms: 200.0,
            mu_burst_ms: 20.0,
            burst_threshold: 0.6,
            steady_threshold: 0.4,
            ..BocpdConfig::default()
        };

        let config = CoalescerConfig::default().with_bocpd_config(bocpd_config);
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();

        // Feed rapid events (10ms intervals = burst-like)
        for i in 0..30 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }

        // Should have high P(burst) and be in Burst regime
        let p_burst = c.bocpd_p_burst().expect("BOCPD should be enabled");
        assert!(
            p_burst > 0.5,
            "Rapid events should yield high P(burst), got {}",
            p_burst
        );
        assert_eq!(
            c.regime(),
            Regime::Burst,
            "Regime should be Burst with rapid events"
        );
    }

    #[test]
    fn bocpd_detects_steady_from_slow_events() {
        use crate::bocpd::BocpdConfig;

        // Configure BOCPD with clear steady detection
        let bocpd_config = BocpdConfig {
            mu_steady_ms: 200.0,
            mu_burst_ms: 20.0,
            burst_threshold: 0.7,
            steady_threshold: 0.3,
            ..BocpdConfig::default()
        };

        let config = CoalescerConfig::default().with_bocpd_config(bocpd_config);
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();

        // Feed slow events (300ms intervals = steady-like)
        for i in 0..10 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 300));
        }

        // Should have low P(burst) and be in Steady regime
        let p_burst = c.bocpd_p_burst().expect("BOCPD should be enabled");
        assert!(
            p_burst < 0.5,
            "Slow events should yield low P(burst), got {}",
            p_burst
        );
        assert_eq!(
            c.regime(),
            Regime::Steady,
            "Regime should be Steady with slow events"
        );
    }

    #[test]
    fn bocpd_recommended_delay_varies_with_regime() {
        let config = CoalescerConfig::default().with_bocpd();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();

        // Initial delay (before any events)
        c.handle_resize_at(85, 30, base);
        let delay_initial = c.bocpd_recommended_delay().expect("BOCPD enabled");

        // Feed burst-like events
        for i in 1..30 {
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 10));
        }
        let delay_burst = c.bocpd_recommended_delay().expect("BOCPD enabled");

        // Recommended delay should be positive
        assert!(delay_initial > 0, "Initial delay should be positive");
        assert!(delay_burst > 0, "Burst delay should be positive");
    }

    #[test]
    fn bocpd_update_is_deterministic() {
        let config = CoalescerConfig::default().with_bocpd();

        let base = Instant::now();

        // Run twice with identical inputs
        let results: Vec<_> = (0..2)
            .map(|_| {
                let mut c = ResizeCoalescer::new(config.clone(), (80, 24));
                for i in 0..20 {
                    c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(i as u64 * 25));
                }
                (c.regime(), c.bocpd_p_burst())
            })
            .collect();

        assert_eq!(
            results[0], results[1],
            "BOCPD results must be deterministic"
        );
    }

    #[test]
    fn bocpd_memory_bounded() {
        use crate::bocpd::BocpdConfig;

        // Use a small max_run_length to test memory bounds
        let bocpd_config = BocpdConfig {
            max_run_length: 50,
            ..BocpdConfig::default()
        };

        let config = CoalescerConfig::default().with_bocpd_config(bocpd_config);
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();

        // Feed many events
        for i in 0u64..200 {
            c.handle_resize_at(
                80 + (i as u16 % 100),
                24 + (i as u16 % 50),
                base + Duration::from_millis(i * 20),
            );
        }

        // Check posterior length is bounded
        if let Some(bocpd) = c.bocpd() {
            let posterior_len = bocpd.run_length_posterior().len();
            assert!(
                posterior_len <= 51, // max_run_length + 1
                "Posterior length should be bounded, got {}",
                posterior_len
            );
        }
    }

    #[test]
    fn bocpd_stable_under_mixed_traffic() {
        let config = CoalescerConfig::default().with_bocpd();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        let base = Instant::now();
        let mut offset = 0u64;

        // Steady period
        for i in 0..5 {
            offset += 200;
            c.handle_resize_at(80 + i, 24 + i, base + Duration::from_millis(offset));
        }

        // Burst period
        for i in 0..15 {
            offset += 15;
            c.handle_resize_at(90 + i, 30 + i, base + Duration::from_millis(offset));
        }

        // Steady period again
        for i in 0..5 {
            offset += 250;
            c.handle_resize_at(100 + i, 40 + i, base + Duration::from_millis(offset));
        }

        // Posterior should still be valid
        let p_burst = c.bocpd_p_burst().expect("BOCPD enabled");
        assert!(
            (0.0..=1.0).contains(&p_burst),
            "P(burst) must remain valid after mixed traffic"
        );

        if let Some(bocpd) = c.bocpd() {
            let sum: f64 = bocpd.run_length_posterior().iter().sum();
            assert!((sum - 1.0).abs() < 1e-9, "Posterior must remain normalized");
        }
    }

    // =========================================================================
    // Evidence JSONL field validation tests (bd-plwf)
    // =========================================================================

    #[test]
    fn evidence_decision_jsonl_contains_all_required_fields() {
        let log = DecisionLog {
            timestamp: Instant::now(),
            elapsed_ms: 16.5,
            event_idx: 1,
            dt_ms: 16.0,
            event_rate: 62.5,
            regime: Regime::Steady,
            action: "apply",
            pending_size: Some((100, 40)),
            applied_size: Some((100, 40)),
            time_since_render_ms: 16.2,
            coalesce_ms: Some(16.0),
            forced: false,
            transition_reason_code: None,
            transition_confidence: None,
        };

        let jsonl = log.to_jsonl("test-run-1", ScreenMode::AltScreen, 100, 40);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("Decision JSONL must be valid JSON");

        // Schema fields
        assert_eq!(
            parsed["schema_version"].as_str().unwrap(),
            EVIDENCE_SCHEMA_VERSION
        );
        assert_eq!(parsed["run_id"].as_str().unwrap(), "test-run-1");
        assert_eq!(parsed["event_idx"].as_u64().unwrap(), 1);
        assert_eq!(parsed["screen_mode"].as_str().unwrap(), "altscreen");
        assert_eq!(parsed["cols"].as_u64().unwrap(), 100);
        assert_eq!(parsed["rows"].as_u64().unwrap(), 40);

        // Event-specific fields
        assert_eq!(parsed["event"].as_str().unwrap(), "decision");
        assert!(parsed["elapsed_ms"].as_f64().is_some());
        assert!(parsed["dt_ms"].as_f64().is_some());
        assert!(parsed["event_rate"].as_f64().is_some());
        assert_eq!(parsed["regime"].as_str().unwrap(), "steady");
        assert_eq!(parsed["action"].as_str().unwrap(), "apply");
        assert_eq!(parsed["pending_w"].as_u64().unwrap(), 100);
        assert_eq!(parsed["pending_h"].as_u64().unwrap(), 40);
        assert_eq!(parsed["applied_w"].as_u64().unwrap(), 100);
        assert_eq!(parsed["applied_h"].as_u64().unwrap(), 40);
        assert!(parsed["time_since_render_ms"].as_f64().is_some());
        assert!(parsed["coalesce_ms"].as_f64().is_some());
        assert!(!parsed["forced"].as_bool().unwrap());
        assert!(parsed["transition_reason_code"].is_null());
        assert!(parsed["transition_confidence"].is_null());
    }

    #[test]
    fn evidence_decision_jsonl_null_fields_when_no_pending() {
        let log = DecisionLog {
            timestamp: Instant::now(),
            elapsed_ms: 0.0,
            event_idx: 0,
            dt_ms: 0.0,
            event_rate: 0.0,
            regime: Regime::Steady,
            action: "skip_same_size",
            pending_size: None,
            applied_size: None,
            time_since_render_ms: 0.0,
            coalesce_ms: None,
            forced: false,
            transition_reason_code: None,
            transition_confidence: None,
        };

        let jsonl = log.to_jsonl("test-run-2", ScreenMode::AltScreen, 80, 24);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("Decision JSONL must be valid JSON");

        assert!(parsed["pending_w"].is_null());
        assert!(parsed["pending_h"].is_null());
        assert!(parsed["applied_w"].is_null());
        assert!(parsed["applied_h"].is_null());
        assert!(parsed["coalesce_ms"].is_null());
        assert!(parsed["transition_reason_code"].is_null());
        assert!(parsed["transition_confidence"].is_null());
    }

    #[test]
    fn evidence_config_jsonl_contains_all_fields() {
        let config = test_config();
        let jsonl = config.to_jsonl("cfg-run", ScreenMode::AltScreen, 80, 24, 0);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("Config JSONL must be valid JSON");

        assert_eq!(parsed["event"].as_str().unwrap(), "config");
        assert_eq!(
            parsed["schema_version"].as_str().unwrap(),
            EVIDENCE_SCHEMA_VERSION
        );
        assert_eq!(parsed["steady_delay_ms"].as_u64().unwrap(), 16);
        assert_eq!(parsed["burst_delay_ms"].as_u64().unwrap(), 40);
        assert_eq!(parsed["hard_deadline_ms"].as_u64().unwrap(), 100);
        assert!(parsed["burst_enter_rate"].as_f64().is_some());
        assert!(parsed["burst_exit_rate"].as_f64().is_some());
        assert_eq!(parsed["cooldown_frames"].as_u64().unwrap(), 3);
        assert_eq!(parsed["rate_window_size"].as_u64().unwrap(), 8);
    }

    #[test]
    fn evidence_inline_screen_mode_string() {
        let log = DecisionLog {
            timestamp: Instant::now(),
            elapsed_ms: 0.0,
            event_idx: 0,
            dt_ms: 0.0,
            event_rate: 0.0,
            regime: Regime::Burst,
            action: "coalesce",
            pending_size: Some((120, 40)),
            applied_size: None,
            time_since_render_ms: 5.0,
            coalesce_ms: None,
            forced: false,
            transition_reason_code: None,
            transition_confidence: None,
        };

        let jsonl = log.to_jsonl("inline-run", ScreenMode::Inline { ui_height: 12 }, 120, 40);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("JSONL must be valid JSON");

        assert_eq!(parsed["screen_mode"].as_str().unwrap(), "inline");
        assert_eq!(parsed["regime"].as_str().unwrap(), "burst");
    }

    #[test]
    fn resize_scheduling_steady_applies_within_steady_delay() {
        let config = CoalescerConfig {
            steady_delay_ms: 20,
            burst_delay_ms: 50,
            hard_deadline_ms: 200,
            enable_logging: true,
            ..test_config()
        };
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        // Single resize in steady mode
        let action = c.handle_resize_at(100, 40, base);
        // First resize may or may not apply immediately depending on implementation
        match action {
            CoalesceAction::ApplyResize { width, height, .. } => {
                assert_eq!(width, 100);
                assert_eq!(height, 40);
            }
            CoalesceAction::None | CoalesceAction::ShowPlaceholder => {
                // Tick past steady delay to get the apply
                let later = base + Duration::from_millis(25);
                let action = c.tick_at(later);
                if let CoalesceAction::ApplyResize { width, height, .. } = action {
                    assert_eq!(width, 100);
                    assert_eq!(height, 40);
                }
            }
        }

        // Verify final applied size
        assert_eq!(c.last_applied(), (100, 40));
    }

    #[test]
    fn resize_scheduling_burst_regime_coalesces_rapid_events() {
        let config = CoalescerConfig {
            steady_delay_ms: 16,
            burst_delay_ms: 40,
            hard_deadline_ms: 100,
            burst_enter_rate: 10.0,
            enable_logging: true,
            ..test_config()
        };
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let mut apply_count = 0u32;

        // Rapid resize events (20 events at 50ms intervals = 20 Hz)
        for i in 0..20 {
            let t = base + Duration::from_millis(i * 50);
            let action = c.handle_resize_at(80 + (i as u16), 24, t);
            if matches!(action, CoalesceAction::ApplyResize { .. }) {
                apply_count += 1;
            }
            // Tick between events
            let tick_t = t + Duration::from_millis(10);
            let tick_action = c.tick_at(tick_t);
            if matches!(tick_action, CoalesceAction::ApplyResize { .. }) {
                apply_count += 1;
            }
        }

        // Should have coalesced: fewer applies than events
        assert!(
            apply_count < 20,
            "Expected coalescing: {apply_count} applies for 20 events"
        );
        // But should still have rendered at least once
        assert!(apply_count > 0, "Should have at least one apply");
    }

    #[test]
    fn evidence_summary_jsonl_includes_checksum() {
        let config = CoalescerConfig {
            enable_logging: true,
            ..test_config()
        };
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24));

        // Generate some events
        c.handle_resize_at(100, 40, base + Duration::from_millis(10));
        c.tick_at(base + Duration::from_millis(30));

        let all_lines = c.evidence_to_jsonl();
        let summary_line = all_lines.lines().last().expect("Should have summary line");
        let parsed: serde_json::Value =
            serde_json::from_str(summary_line).expect("Summary JSONL line must be valid JSON");

        assert_eq!(parsed["event"].as_str().unwrap(), "summary");
        assert!(parsed["decisions"].as_u64().is_some());
        assert!(parsed["applies"].as_u64().is_some());
        assert!(parsed["forced_applies"].as_u64().is_some());
        assert!(parsed["coalesces"].as_u64().is_some());
        assert!(parsed["skips"].as_u64().is_some());
        assert!(parsed["regime"].as_str().is_some());
        assert!(parsed["checksum"].as_str().is_some());
    }

    // =========================================================================
    // DecisionEvidence tests (bd-dionl)
    // =========================================================================

    #[test]
    fn decision_evidence_favor_apply_steady() {
        let ev = DecisionEvidence::favor_apply(Regime::Steady, 80.0, 2.0);
        // Steady regime contrib = 1.0, timing = min(80/50, 2) = 1.6, rate<5 = 0.5
        assert!(ev.log_bayes_factor > 0.0, "Should favor apply");
        assert_eq!(ev.regime_contribution, 1.0);
        assert!((ev.timing_contribution - 1.6).abs() < 0.01);
        assert_eq!(ev.rate_contribution, 0.5);
        assert!(ev.is_strong());
        assert!(ev.is_decisive());
    }

    #[test]
    fn decision_evidence_favor_apply_burst_regime() {
        let ev = DecisionEvidence::favor_apply(Regime::Burst, 10.0, 20.0);
        // Burst regime contrib = -0.5, timing = min(10/50, 2) = 0.2, rate>=5 = -0.3
        assert_eq!(ev.regime_contribution, -0.5);
        assert!((ev.timing_contribution - 0.2).abs() < 0.01);
        assert_eq!(ev.rate_contribution, -0.3);
        // LBF = -0.5 + 0.2 + (-0.3) = -0.6, so not strong
        assert!(!ev.is_strong());
    }

    #[test]
    fn decision_evidence_favor_coalesce_burst() {
        let ev = DecisionEvidence::favor_coalesce(Regime::Burst, 5.0, 15.0);
        // Burst regime contrib = 1.0, timing = min(20/5, 2) = 2.0, rate>10 = 0.5
        // LBF = -(1.0 + 2.0 + 0.5) = -3.5
        assert!(ev.log_bayes_factor < 0.0, "Should favor coalesce");
        assert_eq!(ev.regime_contribution, 1.0);
        assert!((ev.timing_contribution - 2.0).abs() < 0.01);
        assert_eq!(ev.rate_contribution, 0.5);
        assert!(ev.is_strong());
        assert!(ev.is_decisive());
    }

    #[test]
    fn decision_evidence_favor_coalesce_steady_regime() {
        let ev = DecisionEvidence::favor_coalesce(Regime::Steady, 100.0, 3.0);
        // Steady regime contrib = -0.5, timing = min(20/100, 2) = 0.2, rate<=10 = -0.3
        assert_eq!(ev.regime_contribution, -0.5);
        assert!((ev.timing_contribution - 0.2).abs() < 0.01);
        assert_eq!(ev.rate_contribution, -0.3);
    }

    #[test]
    fn decision_evidence_forced_deadline() {
        let ev = DecisionEvidence::forced_deadline(100.0);
        assert!(ev.log_bayes_factor.is_infinite());
        assert_eq!(ev.regime_contribution, 0.0);
        assert!((ev.timing_contribution - 100.0).abs() < 0.01);
        assert_eq!(ev.rate_contribution, 0.0);
        assert!(ev.is_strong());
        assert!(ev.is_decisive());
        assert!(ev.explanation.contains("100.0ms"));
    }

    #[test]
    fn decision_evidence_is_strong_boundary() {
        // Exactly 1.0 is NOT strong (need >1.0)
        let ev = DecisionEvidence {
            log_bayes_factor: 1.0,
            regime_contribution: 0.0,
            timing_contribution: 0.0,
            rate_contribution: 0.0,
            explanation: String::new(),
        };
        assert!(!ev.is_strong());

        let ev2 = DecisionEvidence {
            log_bayes_factor: 1.001,
            ..ev.clone()
        };
        assert!(ev2.is_strong());

        // Negative values also count
        let ev3 = DecisionEvidence {
            log_bayes_factor: -1.5,
            ..ev
        };
        assert!(ev3.is_strong());
    }

    #[test]
    fn decision_evidence_is_decisive_boundary() {
        let ev = DecisionEvidence {
            log_bayes_factor: 2.0,
            regime_contribution: 0.0,
            timing_contribution: 0.0,
            rate_contribution: 0.0,
            explanation: String::new(),
        };
        assert!(!ev.is_decisive()); // need >2.0

        let ev2 = DecisionEvidence {
            log_bayes_factor: 2.001,
            ..ev
        };
        assert!(ev2.is_decisive());
    }

    #[test]
    fn decision_evidence_to_jsonl_valid() {
        let ev = DecisionEvidence::favor_apply(Regime::Steady, 50.0, 3.0);
        let jsonl = ev.to_jsonl("test-run", ScreenMode::AltScreen, 80, 24, 5);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("DecisionEvidence JSONL must be valid JSON");

        assert_eq!(parsed["event"].as_str().unwrap(), "decision_evidence");
        assert!(parsed["log_bayes_factor"].as_f64().is_some());
        assert!(parsed["regime_contribution"].as_f64().is_some());
        assert!(parsed["timing_contribution"].as_f64().is_some());
        assert!(parsed["rate_contribution"].as_f64().is_some());
        assert!(parsed["explanation"].as_str().is_some());
    }

    #[test]
    fn decision_evidence_to_jsonl_infinity() {
        let ev = DecisionEvidence::forced_deadline(100.0);
        let jsonl = ev.to_jsonl("test-run", ScreenMode::AltScreen, 80, 24, 0);
        // Infinity serialized as "inf" string
        assert!(jsonl.contains("\"inf\""));
    }

    // =========================================================================
    // Edge case tests (bd-dionl)
    // =========================================================================

    #[test]
    fn hard_deadline_zero_applies_immediately() {
        let config = CoalescerConfig {
            hard_deadline_ms: 0,
            enable_logging: true,
            ..test_config()
        };
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let base = Instant::now();

        // Any resize should apply immediately since deadline is 0
        let action = c.handle_resize_at(100, 40, base);
        assert!(
            matches!(action, CoalesceAction::ApplyResize { .. }),
            "hard_deadline_ms=0 should force immediate apply, got {action:?}"
        );
    }

    #[test]
    fn rate_window_size_zero_returns_zero_rate() {
        let config = CoalescerConfig {
            rate_window_size: 0,
            enable_logging: true,
            ..test_config()
        };
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let base = Instant::now();

        for i in 0..5 {
            c.handle_resize_at(80 + i, 24, base + Duration::from_millis(i as u64 * 10));
        }
        // With window=0, only 1 event kept, so < 2 elements → rate=0
        let rate = c.calculate_event_rate(base + Duration::from_millis(50));
        assert_eq!(rate, 0.0, "rate_window_size=0 should yield 0 rate");
    }

    #[test]
    fn tick_no_pending_returns_none() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let base = Instant::now();

        // No resize events, tick should return None
        let action = c.tick_at(base);
        assert_eq!(action, CoalesceAction::None);
        let action = c.tick_at(base + Duration::from_millis(500));
        assert_eq!(action, CoalesceAction::None);
    }

    #[test]
    fn time_until_apply_none_when_no_pending() {
        let c = ResizeCoalescer::new(test_config(), (80, 24));
        assert!(c.time_until_apply(Instant::now()).is_none());
    }

    #[test]
    fn time_until_apply_zero_when_past_delay() {
        let config = test_config();
        let mut c = ResizeCoalescer::new(config, (80, 24));
        let base = Instant::now();

        c.handle_resize_at(100, 40, base);
        // Well past the steady_delay_ms of 16ms
        let result = c.time_until_apply(base + Duration::from_millis(500));
        assert_eq!(result, Some(Duration::ZERO));
    }

    // =========================================================================
    // json_escape tests (bd-dionl)
    // =========================================================================

    #[test]
    fn json_escape_special_characters() {
        assert_eq!(json_escape("hello"), "hello");
        assert_eq!(json_escape("a\"b"), "a\\\"b");
        assert_eq!(json_escape("a\\b"), "a\\\\b");
        assert_eq!(json_escape("a\nb"), "a\\nb");
        assert_eq!(json_escape("a\rb"), "a\\rb");
        assert_eq!(json_escape("a\tb"), "a\\tb");
    }

    #[test]
    fn json_escape_control_characters() {
        // Control char \x01 should be escaped as \u0001
        let input = "a\x01b";
        let escaped = json_escape(input);
        assert_eq!(escaped, "a\\u0001b");
    }

    #[test]
    fn json_escape_empty_string() {
        assert_eq!(json_escape(""), "");
    }

    // =========================================================================
    // clear_logs and decision_logs_jsonl tests (bd-dionl)
    // =========================================================================

    #[test]
    fn clear_logs_resets_state() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        c.handle_resize_at(100, 40, Instant::now());
        c.tick_at(Instant::now() + Duration::from_millis(50));

        assert!(!c.logs().is_empty());

        c.clear_logs();
        assert!(c.logs().is_empty());

        // After clearing, new logs should work
        c.handle_resize_at(120, 50, Instant::now());
        c.tick_at(Instant::now() + Duration::from_millis(50));
        assert!(!c.logs().is_empty());
    }

    #[test]
    fn decision_logs_jsonl_each_line_valid() {
        let mut config = test_config();
        config.enable_logging = true;
        let base = Instant::now();
        let mut c = ResizeCoalescer::new(config, (80, 24))
            .with_evidence_run_id("jsonl-test")
            .with_screen_mode(ScreenMode::AltScreen);

        c.handle_resize_at(100, 40, base);
        c.handle_resize_at(110, 50, base + Duration::from_millis(5));
        c.tick_at(base + Duration::from_millis(50));

        let jsonl = c.decision_logs_jsonl();
        assert!(!jsonl.is_empty());
        for line in jsonl.lines() {
            let _: serde_json::Value =
                serde_json::from_str(line).expect("Each JSONL line must be valid JSON");
        }
    }

    // =========================================================================
    // TelemetryHooks tests (bd-dionl)
    // =========================================================================

    #[test]
    fn telemetry_hooks_has_methods() {
        let hooks = TelemetryHooks::new();
        assert!(!hooks.has_resize_applied());
        assert!(!hooks.has_regime_change());
        assert!(!hooks.has_decision());

        let hooks = hooks.on_resize_applied(|_| {});
        assert!(hooks.has_resize_applied());
        assert!(!hooks.has_regime_change());
        assert!(!hooks.has_decision());
    }

    #[test]
    fn telemetry_hooks_with_tracing() {
        let hooks = TelemetryHooks::new().with_tracing(true);
        let debug_str = format!("{:?}", hooks);
        assert!(debug_str.contains("emit_tracing: true"));
    }

    #[test]
    fn telemetry_hooks_default_equals_new() {
        let h1 = TelemetryHooks::default();
        let h2 = TelemetryHooks::new();
        assert!(!h1.has_resize_applied());
        assert!(!h2.has_resize_applied());
    }

    #[test]
    fn telemetry_hooks_on_decision_fires() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicU32, Ordering};

        let count = Arc::new(AtomicU32::new(0));
        let count_clone = count.clone();

        let hooks = TelemetryHooks::new().on_decision(move |_entry| {
            count_clone.fetch_add(1, Ordering::SeqCst);
        });

        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24)).with_telemetry_hooks(hooks);

        let base = Instant::now();
        c.handle_resize_at(100, 40, base);

        // The coalesce decision should fire the on_decision hook
        assert!(count.load(Ordering::SeqCst) >= 1);
    }

    // =========================================================================
    // Regime::as_str tests (bd-dionl)
    // =========================================================================

    #[test]
    fn regime_as_str_values() {
        assert_eq!(Regime::Steady.as_str(), "steady");
        assert_eq!(Regime::Burst.as_str(), "burst");
    }

    #[test]
    fn regime_default_is_steady() {
        assert_eq!(Regime::default(), Regime::Steady);
    }

    // =========================================================================
    // DecisionSummary tests (bd-dionl)
    // =========================================================================

    #[test]
    fn decision_summary_checksum_hex_format() {
        let summary = DecisionSummary {
            checksum: 0x0123456789ABCDEF,
            ..DecisionSummary::default()
        };
        assert_eq!(summary.checksum_hex(), "0123456789abcdef");
    }

    #[test]
    fn decision_summary_default_values() {
        let summary = DecisionSummary::default();
        assert_eq!(summary.decision_count, 0);
        assert_eq!(summary.apply_count, 0);
        assert_eq!(summary.forced_apply_count, 0);
        assert_eq!(summary.coalesce_count, 0);
        assert_eq!(summary.skip_count, 0);
        assert_eq!(summary.regime, Regime::Steady);
        assert_eq!(summary.last_applied, (0, 0));
        assert_eq!(summary.checksum, 0);
    }

    #[test]
    fn decision_summary_to_jsonl_valid() {
        let summary = DecisionSummary {
            decision_count: 5,
            apply_count: 2,
            forced_apply_count: 1,
            coalesce_count: 2,
            skip_count: 1,
            regime: Regime::Burst,
            last_applied: (120, 40),
            checksum: 0xDEADBEEF,
        };
        let jsonl = summary.to_jsonl("run-1", ScreenMode::AltScreen, 120, 40, 5);
        let parsed: serde_json::Value =
            serde_json::from_str(&jsonl).expect("Summary JSONL must be valid JSON");

        assert_eq!(parsed["event"].as_str().unwrap(), "summary");
        assert_eq!(parsed["decisions"].as_u64().unwrap(), 5);
        assert_eq!(parsed["applies"].as_u64().unwrap(), 2);
        assert_eq!(parsed["forced_applies"].as_u64().unwrap(), 1);
        assert_eq!(parsed["coalesces"].as_u64().unwrap(), 2);
        assert_eq!(parsed["skips"].as_u64().unwrap(), 1);
        assert_eq!(parsed["regime"].as_str().unwrap(), "burst");
        assert_eq!(parsed["last_w"].as_u64().unwrap(), 120);
        assert_eq!(parsed["last_h"].as_u64().unwrap(), 40);
        assert!(parsed["checksum"].as_str().unwrap().contains("deadbeef"));
    }

    // =========================================================================
    // screen_mode_str tests (bd-dionl)
    // =========================================================================

    #[test]
    fn screen_mode_str_all_variants() {
        assert_eq!(screen_mode_str(ScreenMode::AltScreen), "altscreen");
        assert_eq!(
            screen_mode_str(ScreenMode::Inline { ui_height: 10 }),
            "inline"
        );
        assert_eq!(
            screen_mode_str(ScreenMode::InlineAuto {
                min_height: 5,
                max_height: 20,
            }),
            "inline_auto"
        );
    }

    // =========================================================================
    // Config builder chaining tests (bd-dionl)
    // =========================================================================

    #[test]
    fn config_with_logging_chaining() {
        let config = CoalescerConfig::default().with_logging(true);
        assert!(config.enable_logging);

        let config2 = config.with_logging(false);
        assert!(!config2.enable_logging);
    }

    #[test]
    fn config_with_bocpd_chaining() {
        let config = CoalescerConfig::default().with_bocpd();
        assert!(config.enable_bocpd);
        assert!(config.bocpd_config.is_some());
    }

    #[test]
    fn config_with_bocpd_config_chaining() {
        let bocpd_cfg = BocpdConfig {
            max_run_length: 42,
            ..BocpdConfig::default()
        };
        let config = CoalescerConfig::default().with_bocpd_config(bocpd_cfg);
        assert!(config.enable_bocpd);
        assert_eq!(config.bocpd_config.as_ref().unwrap().max_run_length, 42);
    }

    // =========================================================================
    // Coalescer builder chaining tests (bd-dionl)
    // =========================================================================

    #[test]
    fn coalescer_with_evidence_run_id() {
        let c = ResizeCoalescer::new(test_config(), (80, 24)).with_evidence_run_id("custom-run-id");
        // Verify via evidence JSONL output
        let jsonl = c.evidence_to_jsonl();
        assert!(jsonl.contains("custom-run-id"));
    }

    #[test]
    fn coalescer_with_screen_mode() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24))
            .with_screen_mode(ScreenMode::Inline { ui_height: 8 });

        c.handle_resize(100, 40);
        let jsonl = c.evidence_to_jsonl();
        assert!(jsonl.contains("\"screen_mode\":\"inline\""));
    }

    #[test]
    fn coalescer_set_evidence_sink_clears_config_logged() {
        let mut config = test_config();
        config.enable_logging = true;
        let mut c = ResizeCoalescer::new(config, (80, 24));

        // Generate some events so config is logged
        c.handle_resize(100, 40);

        // Setting evidence sink to None should reset config_logged
        c.set_evidence_sink(None);

        // Evidence JSONL should still have config line since it rebuilds from config
        let jsonl = c.evidence_to_jsonl();
        assert!(jsonl.contains("\"event\":\"config\""));
    }

    // =========================================================================
    // duration_since_or_zero tests (bd-dionl)
    // =========================================================================

    #[test]
    fn duration_since_or_zero_normal() {
        let earlier = Instant::now();
        std::thread::sleep(Duration::from_millis(1));
        let now = Instant::now();
        let result = duration_since_or_zero(now, earlier);
        assert!(result >= Duration::from_millis(1));
    }

    #[test]
    fn duration_since_or_zero_same_instant() {
        let now = Instant::now();
        let result = duration_since_or_zero(now, now);
        assert_eq!(result, Duration::ZERO);
    }

    // =========================================================================
    // ResizeAppliedEvent and RegimeChangeEvent struct tests (bd-dionl)
    // =========================================================================

    #[test]
    fn resize_applied_event_fields() {
        let event = ResizeAppliedEvent {
            new_size: (100, 40),
            old_size: (80, 24),
            elapsed: Duration::from_millis(42),
            forced: true,
        };
        assert_eq!(event.new_size, (100, 40));
        assert_eq!(event.old_size, (80, 24));
        assert_eq!(event.elapsed, Duration::from_millis(42));
        assert!(event.forced);
    }

    #[test]
    fn regime_change_event_fields() {
        let event = RegimeChangeEvent {
            from: Regime::Steady,
            to: Regime::Burst,
            event_idx: 42,
            reason_code: TransitionReasonCode::HeuristicEnterBurstRate,
            confidence: 0.91,
        };
        assert_eq!(event.from, Regime::Steady);
        assert_eq!(event.to, Regime::Burst);
        assert_eq!(event.event_idx, 42);
        assert_eq!(
            event.reason_code,
            TransitionReasonCode::HeuristicEnterBurstRate
        );
        assert!((event.confidence - 0.91).abs() < f64::EPSILON);
    }

    // =========================================================================
    // CoalesceAction equality edge cases (bd-dionl)
    // =========================================================================

    #[test]
    fn coalesce_action_show_placeholder_eq() {
        assert_eq!(
            CoalesceAction::ShowPlaceholder,
            CoalesceAction::ShowPlaceholder
        );
        assert_ne!(CoalesceAction::ShowPlaceholder, CoalesceAction::None);
    }

    #[test]
    fn coalesce_action_apply_resize_eq() {
        let a = CoalesceAction::ApplyResize {
            width: 100,
            height: 40,
            coalesce_time: Duration::from_millis(16),
            forced_by_deadline: false,
        };
        let b = CoalesceAction::ApplyResize {
            width: 100,
            height: 40,
            coalesce_time: Duration::from_millis(16),
            forced_by_deadline: false,
        };
        assert_eq!(a, b);

        let c = CoalesceAction::ApplyResize {
            width: 100,
            height: 40,
            coalesce_time: Duration::from_millis(16),
            forced_by_deadline: true,
        };
        assert_ne!(a, c);
    }

    // =========================================================================
    // FNV hash consistency (bd-dionl)
    // =========================================================================

    #[test]
    fn fnv_hash_deterministic() {
        let mut h1 = FNV_OFFSET_BASIS;
        fnv_hash_bytes(&mut h1, b"hello world");

        let mut h2 = FNV_OFFSET_BASIS;
        fnv_hash_bytes(&mut h2, b"hello world");

        assert_eq!(h1, h2);
    }

    #[test]
    fn fnv_hash_different_inputs_different_hashes() {
        let mut h1 = FNV_OFFSET_BASIS;
        fnv_hash_bytes(&mut h1, b"hello");

        let mut h2 = FNV_OFFSET_BASIS;
        fnv_hash_bytes(&mut h2, b"world");

        assert_ne!(h1, h2);
    }

    #[test]
    fn fnv_hash_empty_input_returns_basis() {
        let mut hash = FNV_OFFSET_BASIS;
        fnv_hash_bytes(&mut hash, b"");
        assert_eq!(hash, FNV_OFFSET_BASIS);
    }
}