//! The metric scenario event loop.

use std::sync::{Arc, RwLock};
use std::time::{Duration, SystemTime};

use tokio_util::sync::CancellationToken;

use crate::config::ScenarioConfig;
use crate::encoder::create_encoder;
use crate::generator::create_generator;
use crate::model::metric::{Labels, MetricEvent, ValidatedMetricName};
use crate::schedule::core_loop::{
    self, CloseEmitFn, CloseSignal, GateContext, TickContext, TickOutput, TickResult, WriteCommand,
};
use crate::schedule::gate_bus::GateBus;
use crate::schedule::is_in_spike;
use crate::schedule::stats::ScenarioStats;
use crate::schedule::ParsedSchedule;
use crate::sink::{create_sink, Sink, SinkConfig};
use crate::SondaError;

/// Run a scenario to completion, emitting encoded metric events at the configured rate.
pub async fn run(config: &ScenarioConfig) -> Result<(), SondaError> {
    let mut sink = create_sink(&config.sink, None).await?;
    let cancel = CancellationToken::new();
    run_with_sink(config, &mut sink, &cancel, None).await
}

/// Run a scenario to completion, writing encoded events into the provided sink.
pub async fn run_with_sink(
    config: &ScenarioConfig,
    sink: &mut Box<dyn Sink>,
    cancel: &CancellationToken,
    stats: Option<Arc<RwLock<ScenarioStats>>>,
) -> Result<(), SondaError> {
    run_with_sink_gated(config, sink, cancel, stats, None, None).await
}

pub async fn run_with_sink_gated(
    config: &ScenarioConfig,
    sink: &mut Box<dyn Sink>,
    cancel: &CancellationToken,
    stats: Option<Arc<RwLock<ScenarioStats>>>,
    upstream_bus: Option<Arc<GateBus>>,
    gate_ctx: Option<GateContext>,
) -> Result<(), SondaError> {
    let schedule = ParsedSchedule::from_base_config(&config.base)?;

    let generator = create_generator(&config.generator, config.rate)?;
    let generator =
        crate::generator::wrap_with_jitter(generator, config.base.jitter, config.base.jitter_seed);
    let encoder: Arc<dyn crate::encoder::Encoder> = Arc::from(create_encoder(&config.encoder)?);

    let labels: Arc<Labels> = {
        let inner = if let Some(ref label_map) = config.labels {
            let pairs: Vec<(&str, &str)> = label_map
                .iter()
                .map(|(k, v)| (k.as_str(), v.as_str()))
                .collect();
            Labels::from_pairs(&pairs)?
        } else {
            Labels::from_pairs(&[])?
        };
        Arc::new(inner)
    };

    let name = ValidatedMetricName::new(&config.name)?;

    let mut buf: Vec<u8> = Vec::with_capacity(256);

    let upstream_bus_for_tick = upstream_bus.clone();
    let mut tick_fn = |ctx: &TickContext<'_>,
                       output: &mut TickOutput,
                       events_buf: &mut Vec<MetricEvent>|
     -> Result<TickResult, SondaError> {
        let wall_now = ctx.wall_clock;

        let value = generator.value(ctx.tick);
        if let Some(ref bus) = upstream_bus_for_tick {
            bus.tick(value);
        }

        let needs_dynamic = !ctx.dynamic_labels.is_empty();
        let tick_labels: Arc<Labels> = if ctx.spike_windows.is_empty() && !needs_dynamic {
            Arc::clone(&labels)
        } else {
            let mut mutated: Option<Labels> = None;
            if needs_dynamic {
                let tl = mutated.get_or_insert_with(|| (*labels).clone());
                for dl in ctx.dynamic_labels {
                    tl.insert(dl.key.clone(), dl.label_value_for_tick(ctx.tick));
                }
            }
            for sw in ctx.spike_windows {
                if is_in_spike(ctx.elapsed, sw) {
                    let tl = mutated.get_or_insert_with(|| (*labels).clone());
                    tl.insert(sw.label.clone(), sw.label_value_for_tick(ctx.tick));
                }
            }
            match mutated {
                Some(tl) => Arc::new(tl),
                None => Arc::clone(&labels),
            }
        };

        let event = MetricEvent::from_parts(name.clone(), value, tick_labels, wall_now);

        buf.clear();
        encoder.encode_metric(&event, &mut buf)?;
        let bytes_written = buf.len() as u64;
        output
            .writes
            .push(WriteCommand::Bytes(std::mem::take(&mut buf)));

        events_buf.push(event);

        Ok(TickResult {
            bytes_written,
            delivered: true,
        })
    };

    match gate_ctx {
        None => {
            core_loop::run_schedule_loop(&schedule, config.rate, cancel, stats, sink, &mut tick_fn)
                .await
        }
        Some(mut ctx) => {
            ctx.close_emit = build_close_emit(
                config,
                stats.as_ref(),
                ctx.delay.as_ref(),
                Arc::clone(&encoder),
            );
            ctx.holds_on_close = ctx.delay.as_ref().and_then(|d| d.close_snap_to).is_some();
            core_loop::gated_loop(
                &schedule,
                config.rate,
                cancel,
                stats,
                ctx,
                sink,
                &mut tick_fn,
            )
            .await
        }
    }
}

/// `RemoteWrite` sinks default to a stale-NaN marker; other sinks emit only
/// when the user sets `delay.close.snap_to`.
///
/// Resolves the close-emit policy (StaleMarker vs SnapTo) using the shared
/// scenario encoder. Returns `None` when no emission is wanted.
fn build_close_emit(
    config: &ScenarioConfig,
    stats: Option<&Arc<RwLock<ScenarioStats>>>,
    delay: Option<&crate::compiler::DelayClause>,
    encoder: Arc<dyn crate::encoder::Encoder>,
) -> Option<CloseEmitFn> {
    let stats = stats?.clone();

    let user_explicitly_disabled = matches!(delay.and_then(|d| d.close_stale_marker), Some(false));
    let user_snap = delay.and_then(|d| d.close_snap_to);

    let is_remote_write = is_remote_write_sink(&config.sink);
    let should_build = if is_remote_write {
        user_snap.is_some() || !user_explicitly_disabled
    } else {
        user_snap.is_some()
    };
    if !should_build {
        return None;
    }

    let signal = match user_snap {
        Some(v) => CloseSignal::SnapTo(v),
        None => CloseSignal::StaleMarker,
    };

    match stats.write() {
        Ok(mut st) => st.enable_close_series_tracking(),
        Err(p) => p.into_inner().enable_close_series_tracking(),
    }
    Some(make_close_emitter(stats, encoder, signal))
}

fn make_close_emitter(
    stats: Arc<RwLock<ScenarioStats>>,
    encoder: Arc<dyn crate::encoder::Encoder>,
    signal: CloseSignal,
) -> CloseEmitFn {
    let value = match signal {
        #[cfg(feature = "remote-write")]
        CloseSignal::StaleMarker => crate::encoder::remote_write::PROMETHEUS_STALE_NAN,
        #[cfg(not(feature = "remote-write"))]
        CloseSignal::StaleMarker => f64::from_bits(0x7ff0000000000002),
        CloseSignal::SnapTo(v) => v,
    };

    let mut buf: Vec<u8> = Vec::with_capacity(256);
    Box::new(move |output: &mut TickOutput| -> Result<(), SondaError> {
        let (series, watermark) = match stats.write() {
            Ok(mut st) => st.drain_close_emit_series(),
            Err(p) => p.into_inner().drain_close_emit_series(),
        };

        // Derived from the last active emission, not real `now`: a
        // `start_time`-shifted scenario's watermark sits in the shifted
        // window, and the +1ms keeps the marker strictly after it.
        let close_ts = match watermark {
            Some(max_recent) => max_recent
                .checked_add(Duration::from_millis(1))
                .unwrap_or(max_recent),
            None => SystemTime::now(),
        };

        for (name, labels) in series {
            buf.clear();
            let marker = MetricEvent::from_parts(name, value, labels, close_ts);
            encoder.encode_metric(&marker, &mut buf)?;
            output
                .writes
                .push(WriteCommand::Bytes(std::mem::take(&mut buf)));
        }
        Ok(())
    })
}

#[cfg(feature = "remote-write")]
fn is_remote_write_sink(sink: &SinkConfig) -> bool {
    matches!(sink, SinkConfig::RemoteWrite { .. })
}

#[cfg(not(feature = "remote-write"))]
fn is_remote_write_sink(_sink: &SinkConfig) -> bool {
    false
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use async_trait::async_trait;
    use tokio_util::sync::CancellationToken;

    use crate::config::{BaseScheduleConfig, GapConfig, ScenarioConfig};
    use crate::encoder::EncoderConfig;
    use crate::generator::GeneratorConfig;
    use crate::sink::{Sink, SinkConfig};
    use crate::SondaError;

    type SharedBuf = std::sync::Arc<Mutex<Vec<u8>>>;

    struct ProbeSink(SharedBuf);

    #[async_trait]
    impl Sink for ProbeSink {
        async fn write(&mut self, data: &[u8]) -> Result<(), SondaError> {
            self.0
                .lock()
                .expect("probe sink mutex poisoned")
                .extend_from_slice(data);
            Ok(())
        }
        async fn flush(&mut self) -> Result<(), SondaError> {
            Ok(())
        }
    }

    fn probe_sink() -> (Box<dyn Sink>, SharedBuf) {
        let buf: SharedBuf = std::sync::Arc::new(Mutex::new(Vec::new()));
        let sink: Box<dyn Sink> = Box::new(ProbeSink(SharedBuf::clone(&buf)));
        (sink, buf)
    }

    /// Build a minimal ScenarioConfig suitable for a short integration run.
    fn make_config(rate: f64, duration: &str, gaps: Option<GapConfig>) -> ScenarioConfig {
        ScenarioConfig {
            base: BaseScheduleConfig {
                name: "up".to_string(),
                rate,
                duration: Some(duration.to_string()),
                gaps,
                bursts: None,
                cardinality_spikes: None,
                dynamic_labels: None,
                labels: None,
                sink: SinkConfig::Stdout, // not used — tests use run_with_sink directly
                phase_offset: None,
                clock_group: None,
                clock_group_is_auto: None,
                start_time: None,
                jitter: None,
                jitter_seed: None,
                on_sink_error: crate::OnSinkError::Warn,
            },
            generator: GeneratorConfig::Constant { value: 1.0 },
            encoder: EncoderConfig::PrometheusText { precision: None },
            metric_type: None,
            help: None,
        }
    }

    // ---- run: basic correctness ----------------------------------------------

    /// run() with a short duration should complete without error.
    #[tokio::test]
    async fn run_completes_without_error_for_short_duration() {
        let config = make_config(100.0, "100ms", None);
        let result = super::run(&config).await;
        assert!(
            result.is_ok(),
            "run must succeed for valid config: {result:?}"
        );
    }

    // ---- Integration: ~rate events emitted over duration --------------------

    /// At rate=100 for 1 second we expect approximately 100 newline-terminated events.
    /// We allow a ±20% window to accommodate scheduling jitter.
    #[tokio::test]
    async fn integration_rate_100_duration_1s_emits_approximately_100_events() {
        let config = make_config(100.0, "1s", None);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        assert!(
            (80..=120).contains(&newlines),
            "expected ~100 events (80–120), got {newlines}"
        );
    }

    /// Each emitted line is valid UTF-8 and starts with the metric name.
    #[tokio::test]
    async fn integration_output_lines_start_with_metric_name() {
        let config = make_config(50.0, "200ms", None);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("output must be valid UTF-8");
        for line in output.lines() {
            assert!(
                line.starts_with("up"),
                "each line must start with metric name 'up', got: {line:?}"
            );
        }
    }

    /// Each emitted Prometheus line ends with a newline.
    #[tokio::test]
    async fn integration_output_ends_with_newline() {
        let config = make_config(50.0, "200ms", None);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        assert!(captured.ends_with(b"\n"), "output must end with a newline");
    }

    // ---- Integration: gap suppresses events ----------------------------------

    /// With rate=100 for 5s and a gap_every=3s gap_for=1s, we expect fewer than
    /// 500 events because the gap suppresses approximately 1 second of output per
    /// 3-second cycle (~100 events lost from the first gap, plus ~100 from the
    /// second). We use 380 as a conservative upper bound below 500.
    #[tokio::test]
    async fn integration_gap_suppresses_events() {
        let config = make_config(
            100.0,
            "5s",
            Some(GapConfig {
                every: "3s".to_string(),
                r#for: "1s".to_string(),
            }),
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        assert!(
            newlines < 500,
            "gap must suppress events: expected < 500, got {newlines}"
        );
        // Also confirm events were actually emitted (not zero).
        assert!(
            newlines > 0,
            "some events must be emitted outside of gaps, got {newlines}"
        );
    }

    // ---- run: invalid config is rejected -------------------------------------

    /// A config with an unparseable duration returns Err.
    #[tokio::test]
    async fn run_with_invalid_duration_returns_err() {
        let mut config = make_config(100.0, "bad_duration", None);
        // Manually set an invalid duration string.
        config.duration = Some("not_a_duration".to_string());
        let result = super::run(&config).await;
        assert!(result.is_err(), "invalid duration must return Err");
    }

    /// A config with an invalid gap duration returns Err.
    #[tokio::test]
    async fn run_with_invalid_gap_every_returns_err() {
        let mut config = make_config(100.0, "1s", None);
        config.gaps = Some(GapConfig {
            every: "bad".to_string(),
            r#for: "1s".to_string(),
        });
        let result = super::run(&config).await;
        assert!(result.is_err(), "invalid gap.every must return Err");
    }

    // ---- run: labels appear in output ---------------------------------------

    /// When labels are configured they appear in the encoded output.
    #[tokio::test]
    async fn integration_labels_appear_in_output() {
        let mut config = make_config(50.0, "100ms", None);
        let mut label_map = std::collections::HashMap::new();
        label_map.insert("host".to_string(), "server1".to_string());
        config.labels = Some(label_map);

        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("output must be valid UTF-8");
        assert!(
            output.contains("host=\"server1\""),
            "label must appear in output, got:\n{output}"
        );
    }

    // ---- Integration: burst increases event rate ----------------------------

    /// Helper that builds a ScenarioConfig with an optional BurstConfig.
    fn make_config_with_burst(
        rate: f64,
        duration: &str,
        gaps: Option<crate::config::GapConfig>,
        bursts: Option<crate::config::BurstConfig>,
    ) -> crate::config::ScenarioConfig {
        crate::config::ScenarioConfig {
            base: crate::config::BaseScheduleConfig {
                name: "up".to_string(),
                rate,
                duration: Some(duration.to_string()),
                gaps,
                bursts,
                cardinality_spikes: None,
                dynamic_labels: None,
                labels: None,
                sink: crate::sink::SinkConfig::Stdout,
                phase_offset: None,
                clock_group: None,
                clock_group_is_auto: None,
                start_time: None,
                jitter: None,
                jitter_seed: None,
                on_sink_error: crate::OnSinkError::Warn,
            },
            generator: crate::generator::GeneratorConfig::Constant { value: 1.0 },
            encoder: crate::encoder::EncoderConfig::PrometheusText { precision: None },
            metric_type: None,
            help: None,
        }
    }

    /// With rate=10 and burst_multiplier=5 for the entire 1s run (burst_every=10s,
    /// burst_for=5s so the burst covers the full 1s window), we should get
    /// significantly more than 10 events — approximately 50.
    ///
    /// The burst occupies [0, burst_for) of each burst_every cycle.
    /// With burst_every=10s and burst_for=9s, the first 1s of the run is always
    /// inside a burst (cycle_pos=0..1 < 9), so effective_interval = base/5.
    #[tokio::test]
    async fn integration_burst_increases_event_count() {
        let config = make_config_with_burst(
            10.0,
            "1s",
            None,
            Some(crate::config::BurstConfig {
                every: "10s".to_string(),
                r#for: "9s".to_string(),
                multiplier: 5.0,
            }),
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        // Without burst: ~10 events. With 5x burst for entire 1s: ~50 events.
        // We allow a wide range to accommodate scheduling jitter.
        assert!(
            newlines > 15,
            "burst must increase event count above base rate: expected >15, got {newlines}"
        );
        assert!(
            newlines < 100,
            "event count must be sane (not runaway): expected <100, got {newlines}"
        );
    }

    /// Rate=100 for 2s with burst_every=10s, burst_for=1s, multiplier=5.
    /// The first 1s of the run is in a burst (rate=500), the second 1s is not (rate=100).
    /// Total expected events: ~500 + ~100 = ~600.
    /// We use a range of 400–800 to accommodate scheduling jitter.
    #[tokio::test]
    async fn integration_burst_then_normal_produces_mixed_rate() {
        let config = make_config_with_burst(
            100.0,
            "2s",
            None,
            Some(crate::config::BurstConfig {
                every: "10s".to_string(),
                r#for: "1s".to_string(),
                multiplier: 5.0,
            }),
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        // Without any burst: ~200 events. With burst for first 1s: ~600.
        assert!(
            newlines > 200,
            "burst phase must produce more than base rate alone: expected >200, got {newlines}"
        );
        assert!(
            newlines <= 900,
            "total event count must be in expected range, got {newlines}"
        );
    }

    // ---- Integration: gap wins over burst -----------------------------------

    /// When a gap and burst overlap, the gap must win — no events are emitted
    /// during the gap window regardless of burst state.
    ///
    /// We configure a 3s run where:
    /// - Gap: every=5s, for=2s → gap occupies seconds [3, 5) of each cycle.
    ///   For a 3s run starting at t=0, no gap is active (gap starts at 3s and
    ///   the run ends at 3s). So we set the gap at [0,2) to make it overlap
    ///   with the burst.
    ///
    /// Actually, the gap occupies the END of each cycle (tail), and the burst
    /// occupies the START. To force overlap, we use a very short every (3s)
    /// so that the gap tail (last 2s of a 3s cycle = seconds [1, 3)) overlaps
    /// with a burst that starts at the beginning of the next cycle.
    ///
    /// Simpler approach: use duration=1s with gap_every=2s gap_for=1s (entire
    /// run is in a gap since the gap occupies seconds [1,2) of each 2s cycle,
    /// but actually for a 1s run, cycle_pos ∈ [0,1) which is NOT in the gap
    /// [every-for, every) = [1, 2)). Let's use a different approach.
    ///
    /// Clearest approach: run for 3s, gap_every=3s, gap_for=2s.
    /// Gap occupies [1s, 3s) in each cycle. With burst_every=3s, burst_for=2.5s,
    /// burst occupies [0, 2.5s) of each 3s cycle.
    /// Overlap at [1s, 2.5s): gap wins → no events during overlap.
    ///
    /// During [0, 1s): no gap, burst active → ~100*5=500 events in that 1s.
    /// During [1s, 3s): gap wins → 0 events.
    /// Total: only ~500 events (much less than 100*3=300 without gap/burst,
    /// and much less than 500*2+100*1 with burst only).
    #[tokio::test]
    async fn integration_gap_wins_over_burst_suppresses_events() {
        // Gap occupies [every-for, every) = [3-2, 3) = [1s, 3s) per cycle.
        // Burst occupies [0, burst_for) = [0, 2.5s) per cycle.
        // Overlap: [1s, 2.5s) — gap wins here.
        // Only [0, 1s) has burst active with no gap.
        let config = make_config_with_burst(
            100.0,
            "3s",
            Some(GapConfig {
                every: "3s".to_string(),
                r#for: "2s".to_string(),
            }),
            Some(crate::config::BurstConfig {
                every: "3s".to_string(),
                r#for: "2500ms".to_string(), // 2.5s burst overlaps with gap starting at 1s
                multiplier: 5.0,
            }),
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();

        // During gap: 0 events. During [0,1s) with burst (5x): ~500 events.
        // But the run exits after 3s total, and the gap sleeps through most of it.
        // We expect far fewer than 300 baseline events (no gap/burst at 100/s for 3s).
        // The gap eats 2 out of every 3 seconds → baseline would be ~100 events.
        // With burst for the first 1s → ~500 events in that 1s alone.
        // The key assertion is that the gap suppressed events: significantly less
        // than what 5x burst for 3s would produce (1500), and also that the
        // gap didn't allow events during its window.
        assert!(
            newlines < 1000,
            "gap must suppress many events that burst would have produced: expected <1000, got {newlines}"
        );
        // Some events must have been produced during the non-gap window.
        assert!(
            newlines > 0,
            "some events must be emitted outside of gaps, got {newlines}"
        );
    }

    /// Simpler gap-wins-over-burst test: run for 100ms with no gap and burst
    /// to establish a baseline, then run with both gap and burst where the
    /// gap covers the entire duration — expect zero events.
    #[tokio::test]
    async fn integration_gap_covering_full_window_produces_zero_events_even_with_burst() {
        // Gap: every=1s, for=500ms → gap occupies [500ms, 1000ms) in each 1s cycle.
        // Run for 200ms starting at cycle_pos approaching 500ms.
        //
        // Instead use: gap every=1s, for=900ms → gap occupies [100ms, 1000ms).
        // We need to start IN the gap. Since cycle_pos = elapsed % every,
        // we'd need to offset the start, which we can't do.
        //
        // Better: gap every=500ms, for=400ms → gap at [100ms, 500ms) per cycle.
        // Start at 0ms: not in gap. At 100ms: in gap. Sleep 400ms. Resume at 500ms.
        // At 500ms: not in gap again. Very few events are emitted.
        //
        // Simplest verifiable test: confirm gap suppresses events.
        // Run rate=1000 for 500ms with gap_every=1s gap_for=900ms.
        // Gap occupies [100ms, 1000ms) per 1s cycle.
        // At t=0: not in gap → emit events for ~100ms at 1000/s → ~100 events.
        // At t=100ms: in gap → sleep until t=1000ms → but run ends at 500ms.
        // After sleep wake at t=1000ms > 500ms → exit immediately.
        // Total: only ~100 events despite rate=1000 for 500ms baseline of ~500.
        let config_no_gap_burst = make_config_with_burst(
            1000.0,
            "500ms",
            None,
            Some(crate::config::BurstConfig {
                every: "1s".to_string(),
                r#for: "900ms".to_string(),
                multiplier: 5.0,
            }),
        );
        let (mut sink_no_gap, buf_no_gap) = probe_sink();
        super::run_with_sink(
            &config_no_gap_burst,
            &mut sink_no_gap,
            &CancellationToken::new(),
            None,
        )
        .await
        .expect("run must succeed");
        let events_burst_only = buf_no_gap
            .lock()
            .unwrap()
            .iter()
            .filter(|&&b| b == b'\n')
            .count();

        // With the burst for 900ms of each 1s cycle, over 500ms we'd expect ~4500 events.
        // This shows burst is working.
        let config_gap_and_burst = make_config_with_burst(
            1000.0,
            "500ms",
            Some(GapConfig {
                every: "1s".to_string(),
                r#for: "900ms".to_string(),
            }),
            Some(crate::config::BurstConfig {
                every: "1s".to_string(),
                r#for: "900ms".to_string(),
                multiplier: 5.0,
            }),
        );
        let (mut sink_gap_burst, buf_gap_burst) = probe_sink();
        super::run_with_sink(
            &config_gap_and_burst,
            &mut sink_gap_burst,
            &CancellationToken::new(),
            None,
        )
        .await
        .expect("run must succeed");
        let events_gap_and_burst = buf_gap_burst
            .lock()
            .unwrap()
            .iter()
            .filter(|&&b| b == b'\n')
            .count();

        // The gap must suppress the burst: far fewer events when gap is active.
        // Gap occupies [100ms, 1000ms) so burst during that window is suppressed.
        // Only events in [0, 100ms) should fire.
        assert!(
            events_gap_and_burst < events_burst_only,
            "gap must suppress burst events: gap+burst={events_gap_and_burst} must be < burst-only={events_burst_only}"
        );
    }

    // ---- Shutdown flag with burst scenario -----------------------------------

    // ---- Integration: stats buffer receives metric events ---------------------

    #[tokio::test]
    async fn runner_pushes_metric_events_to_stats_current_values() {
        use std::sync::{Arc, RwLock};

        let config = make_config(50.0, "200ms", None);
        let (mut sink, _buf) = probe_sink();
        let stats = Arc::new(RwLock::new(crate::schedule::stats::ScenarioStats::default()));

        super::run_with_sink(
            &config,
            &mut sink,
            &CancellationToken::new(),
            Some(Arc::clone(&stats)),
        )
        .await
        .expect("run must succeed");

        let st = stats.read().expect("lock must not be poisoned");
        assert!(
            !st.current_values.is_empty(),
            "runner must push events into stats.current_values"
        );
    }

    /// When no stats arc is provided (None), the runner does not panic and
    /// still produces output normally.
    #[tokio::test]
    async fn runner_without_stats_does_not_push_metrics() {
        let config = make_config(50.0, "100ms", None);
        let (mut sink, buf) = probe_sink();

        // Pass None for stats — should work fine without buffering.
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        assert!(
            newlines > 0,
            "runner without stats must still produce output"
        );
    }

    #[tokio::test]
    async fn runner_stats_current_values_have_correct_metric_name() {
        use std::sync::{Arc, RwLock};

        let config = make_config(50.0, "100ms", None);
        let (mut sink, _buf) = probe_sink();
        let stats = Arc::new(RwLock::new(crate::schedule::stats::ScenarioStats::default()));

        super::run_with_sink(
            &config,
            &mut sink,
            &CancellationToken::new(),
            Some(Arc::clone(&stats)),
        )
        .await
        .expect("run must succeed");

        let st = stats.read().expect("lock must not be poisoned");
        for event in st.current_values.values() {
            assert_eq!(&*event.name, "up");
        }
    }

    /// Cancelling the token stops the runner even during a burst window.
    #[tokio::test]
    async fn cancel_stops_run_during_burst() {
        let config = make_config_with_burst(
            1000.0,
            "60s",
            None,
            Some(crate::config::BurstConfig {
                every: "10s".to_string(),
                r#for: "9s".to_string(),
                multiplier: 5.0,
            }),
        );

        let cancel = CancellationToken::new();
        let cancel_for_thread = cancel.clone();

        let handle = std::thread::spawn(move || {
            std::thread::sleep(std::time::Duration::from_millis(200));
            cancel_for_thread.cancel();
        });

        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &cancel, None)
            .await
            .expect("run must succeed");
        handle.join().expect("thread must complete");

        // The run stopped after ~200ms due to shutdown flag.
        // At rate=1000 with 5x burst: ~1000 events in 200ms.
        // We just assert it stopped without hanging (the test would time out otherwise)
        // and produced some output.
        let captured = buf.lock().unwrap();
        let newlines = captured.iter().filter(|&&b| b == b'\n').count();
        assert!(
            newlines > 0,
            "some events must have been emitted before shutdown"
        );
    }

    // ---- Integration: cardinality spikes inject dynamic labels ----------------

    /// Helper that builds a ScenarioConfig with a cardinality spike.
    fn make_config_with_spike(
        rate: f64,
        duration: &str,
        spike: crate::config::CardinalitySpikeConfig,
    ) -> crate::config::ScenarioConfig {
        crate::config::ScenarioConfig {
            base: crate::config::BaseScheduleConfig {
                name: "up".to_string(),
                rate,
                duration: Some(duration.to_string()),
                gaps: None,
                bursts: None,
                cardinality_spikes: Some(vec![spike]),
                dynamic_labels: None,
                labels: None,
                sink: crate::sink::SinkConfig::Stdout,
                phase_offset: None,
                clock_group: None,
                clock_group_is_auto: None,
                start_time: None,
                jitter: None,
                jitter_seed: None,
                on_sink_error: crate::OnSinkError::Warn,
            },
            generator: crate::generator::GeneratorConfig::Constant { value: 1.0 },
            encoder: crate::encoder::EncoderConfig::PrometheusText { precision: None },
            metric_type: None,
            help: None,
        }
    }

    /// When the entire run is inside a spike window, every output line must
    /// contain the spike label key.
    #[tokio::test]
    async fn integration_spike_labels_appear_during_spike_window() {
        let spike = crate::config::CardinalitySpikeConfig {
            label: "pod_name".to_string(),
            every: "10s".to_string(),
            r#for: "9s".to_string(),
            cardinality: 5,
            strategy: crate::config::SpikeStrategy::Counter,
            prefix: Some("pod-".to_string()),
            seed: None,
        };
        // Run for 500ms inside a spike window (spike occupies 0..9s of each 10s cycle)
        let config = make_config_with_spike(50.0, "500ms", spike);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("output must be valid UTF-8");
        for line in output.lines() {
            assert!(
                line.contains("pod_name="),
                "every line during spike must contain pod_name label, got: {line:?}"
            );
        }
    }

    /// When no spike windows are configured, output does not contain spike labels.
    #[tokio::test]
    async fn integration_no_spike_config_produces_no_spike_labels() {
        let config = make_config(50.0, "200ms", None);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("output must be valid UTF-8");
        for line in output.lines() {
            assert!(
                !line.contains("pod_name="),
                "without spike config, pod_name must not appear: {line:?}"
            );
        }
    }

    /// Counter strategy produces unique values bounded by cardinality.
    #[tokio::test]
    async fn integration_spike_counter_strategy_produces_bounded_values() {
        let spike = crate::config::CardinalitySpikeConfig {
            label: "pod_name".to_string(),
            every: "10s".to_string(),
            r#for: "9s".to_string(),
            cardinality: 3,
            strategy: crate::config::SpikeStrategy::Counter,
            prefix: Some("pod-".to_string()),
            seed: None,
        };
        let config = make_config_with_spike(50.0, "500ms", spike);
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("output must be valid UTF-8");
        let mut seen_values = std::collections::HashSet::new();
        for line in output.lines() {
            // Extract value of pod_name from Prometheus output like: pod_name="pod-0"
            if let Some(start) = line.find("pod_name=\"") {
                let rest = &line[start + 10..];
                if let Some(end) = rest.find('"') {
                    seen_values.insert(rest[..end].to_string());
                }
            }
        }
        // With cardinality=3, we should see at most 3 unique values
        assert!(
            seen_values.len() <= 3,
            "counter strategy with cardinality=3 should produce at most 3 unique values, got {}: {:?}",
            seen_values.len(),
            seen_values
        );
        assert!(
            !seen_values.is_empty(),
            "must have produced at least one spike label value"
        );
    }

    /// Stats correctly reports `in_cardinality_spike` during a spike window.
    #[tokio::test]
    async fn integration_spike_stats_reports_in_cardinality_spike() {
        use std::sync::{Arc, RwLock};

        let spike = crate::config::CardinalitySpikeConfig {
            label: "pod_name".to_string(),
            every: "10s".to_string(),
            r#for: "9s".to_string(),
            cardinality: 5,
            strategy: crate::config::SpikeStrategy::Counter,
            prefix: Some("pod-".to_string()),
            seed: None,
        };
        let config = make_config_with_spike(50.0, "200ms", spike);
        let (mut sink, _buf) = probe_sink();
        let stats = Arc::new(RwLock::new(crate::schedule::stats::ScenarioStats::default()));

        super::run_with_sink(
            &config,
            &mut sink,
            &CancellationToken::new(),
            Some(Arc::clone(&stats)),
        )
        .await
        .expect("run must succeed");

        // The entire 200ms run is inside the spike window (0..9s of 10s cycle).
        // The final stats snapshot should show in_cardinality_spike = true.
        let st = stats.read().expect("lock must not be poisoned");
        assert!(
            st.in_cardinality_spike,
            "stats must report in_cardinality_spike=true during spike window"
        );
    }

    // ---- Arc sharing: name and labels are reference-counted, not deep-cloned ---

    /// All metric events buffered in stats must share the same Arc<str> name
    #[tokio::test]
    async fn current_values_holds_one_entry_for_single_series_scenario() {
        use std::sync::{Arc, RwLock};

        let config = make_config(200.0, "100ms", None);
        let (mut sink, _buf) = probe_sink();
        let stats = Arc::new(RwLock::new(crate::schedule::stats::ScenarioStats::default()));

        super::run_with_sink(
            &config,
            &mut sink,
            &CancellationToken::new(),
            Some(Arc::clone(&stats)),
        )
        .await
        .expect("run must succeed");

        let st = stats.read().expect("lock must not be poisoned");
        assert_eq!(
            st.current_values.len(),
            1,
            "a single-series scenario must collapse to one current value"
        );
        let persisted = st
            .current_values
            .values()
            .next()
            .expect("runner must push at least one event");
        assert_eq!(&*persisted.name, "up");
    }

    /// Invalid metric name in config is caught before the hot loop, not during.
    #[tokio::test]
    async fn invalid_metric_name_returns_config_error_before_loop() {
        let config = ScenarioConfig {
            base: BaseScheduleConfig {
                name: "123-invalid".to_string(),
                rate: 10.0,
                duration: Some("100ms".to_string()),
                gaps: None,
                bursts: None,
                cardinality_spikes: None,
                dynamic_labels: None,
                labels: None,
                sink: SinkConfig::Stdout,
                phase_offset: None,
                clock_group: None,
                clock_group_is_auto: None,
                start_time: None,
                jitter: None,
                jitter_seed: None,
                on_sink_error: crate::OnSinkError::Warn,
            },
            generator: GeneratorConfig::Constant { value: 1.0 },
            encoder: EncoderConfig::PrometheusText { precision: None },
            metric_type: None,
            help: None,
        };
        let (mut sink, _buf) = probe_sink();
        let result =
            super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None).await;
        assert!(
            matches!(result, Err(crate::SondaError::Config(ref e)) if e.to_string().contains("123-invalid")),
            "expected Config error for invalid name, got: {result:?}"
        );
    }

    // ---- Integration: dynamic labels appear in metric output --------------------

    /// Helper that builds a ScenarioConfig with dynamic_labels.
    fn make_config_with_dynamic_labels(
        rate: f64,
        duration: &str,
        dynamic_labels: Vec<crate::config::DynamicLabelConfig>,
    ) -> ScenarioConfig {
        ScenarioConfig {
            base: BaseScheduleConfig {
                name: "up".to_string(),
                rate,
                duration: Some(duration.to_string()),
                gaps: None,
                bursts: None,
                cardinality_spikes: None,
                dynamic_labels: Some(dynamic_labels),
                labels: None,
                sink: SinkConfig::Stdout,
                phase_offset: None,
                clock_group: None,
                clock_group_is_auto: None,
                start_time: None,
                jitter: None,
                jitter_seed: None,
                on_sink_error: crate::OnSinkError::Warn,
            },
            generator: GeneratorConfig::Constant { value: 1.0 },
            encoder: EncoderConfig::PrometheusText { precision: None },
            metric_type: None,
            help: None,
        }
    }

    /// Dynamic labels with counter strategy appear in every Prometheus line.
    #[tokio::test]
    async fn dynamic_labels_counter_appear_in_metric_output() {
        let config = make_config_with_dynamic_labels(
            10.0,
            "1s",
            vec![crate::config::DynamicLabelConfig {
                key: "hostname".to_string(),
                strategy: crate::config::DynamicLabelStrategy::Counter {
                    prefix: Some("host-".to_string()),
                    cardinality: 5,
                },
            }],
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("valid UTF-8");
        let lines: Vec<&str> = output.lines().collect();
        assert!(
            !lines.is_empty(),
            "runner must produce at least one line of output"
        );

        for line in &lines {
            assert!(
                line.contains("hostname=\"host-"),
                "every metric line must contain dynamic label hostname; line: {line}"
            );
        }
    }

    /// Dynamic labels with values list cycle through the values.
    #[tokio::test]
    async fn dynamic_labels_values_list_cycle_in_metric_output() {
        let config = make_config_with_dynamic_labels(
            10.0,
            "1s",
            vec![crate::config::DynamicLabelConfig {
                key: "region".to_string(),
                strategy: crate::config::DynamicLabelStrategy::ValuesList {
                    values: vec!["alpha".to_string(), "beta".to_string(), "gamma".to_string()],
                },
            }],
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("valid UTF-8");
        let lines: Vec<&str> = output.lines().collect();
        assert!(!lines.is_empty());

        // All lines must contain the label key
        for line in &lines {
            assert!(
                line.contains("region=\""),
                "every metric line must contain dynamic label region; line: {line}"
            );
        }

        // Check that multiple distinct values appear across the output
        let has_alpha = lines.iter().any(|l| l.contains("region=\"alpha\""));
        let has_beta = lines.iter().any(|l| l.contains("region=\"beta\""));
        assert!(
            has_alpha || has_beta,
            "at least one distinct dynamic label value should appear in output"
        );
    }

    /// Cardinality ceiling is respected: only cardinality distinct values appear.
    #[tokio::test]
    async fn dynamic_labels_counter_respects_cardinality_ceiling_in_output() {
        let config = make_config_with_dynamic_labels(
            50.0,
            "1s",
            vec![crate::config::DynamicLabelConfig {
                key: "hostname".to_string(),
                strategy: crate::config::DynamicLabelStrategy::Counter {
                    prefix: Some("host-".to_string()),
                    cardinality: 3,
                },
            }],
        );
        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("valid UTF-8");
        let mut distinct_values = std::collections::HashSet::new();
        for line in output.lines() {
            // Extract the hostname="..." value
            if let Some(start) = line.find("hostname=\"") {
                let rest = &line[start + 10..];
                if let Some(end) = rest.find('"') {
                    distinct_values.insert(rest[..end].to_string());
                }
            }
        }
        assert_eq!(
            distinct_values.len(),
            3,
            "with cardinality=3, exactly 3 distinct values must appear, got {:?}",
            distinct_values
        );
    }

    /// Dynamic labels and static labels coexist: both appear in output.
    #[tokio::test]
    async fn dynamic_labels_and_static_labels_coexist_in_output() {
        let mut config = make_config_with_dynamic_labels(
            10.0,
            "1s",
            vec![crate::config::DynamicLabelConfig {
                key: "hostname".to_string(),
                strategy: crate::config::DynamicLabelStrategy::Counter {
                    prefix: Some("host-".to_string()),
                    cardinality: 5,
                },
            }],
        );
        let mut label_map = std::collections::HashMap::new();
        label_map.insert("env".to_string(), "prod".to_string());
        config.labels = Some(label_map);

        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("valid UTF-8");
        for line in output.lines() {
            assert!(
                line.contains("env=\"prod\""),
                "static label must appear; line: {line}"
            );
            assert!(
                line.contains("hostname=\"host-"),
                "dynamic label must appear; line: {line}"
            );
        }
    }

    /// Dynamic label wins on key collision with static label.
    #[tokio::test]
    async fn dynamic_label_wins_on_key_collision_with_static() {
        let mut config = make_config_with_dynamic_labels(
            10.0,
            "500ms",
            vec![crate::config::DynamicLabelConfig {
                key: "hostname".to_string(),
                strategy: crate::config::DynamicLabelStrategy::Counter {
                    prefix: Some("dynamic-".to_string()),
                    cardinality: 3,
                },
            }],
        );
        let mut label_map = std::collections::HashMap::new();
        label_map.insert("hostname".to_string(), "static-value".to_string());
        config.labels = Some(label_map);

        let (mut sink, buf) = probe_sink();
        super::run_with_sink(&config, &mut sink, &CancellationToken::new(), None)
            .await
            .expect("run must succeed");

        let captured = buf.lock().unwrap();
        let output = std::str::from_utf8(&captured).expect("valid UTF-8");
        for line in output.lines() {
            assert!(
                line.contains("hostname=\"dynamic-"),
                "dynamic label must overwrite static label; line: {line}"
            );
            assert!(
                !line.contains("hostname=\"static-value\""),
                "static value must not appear when dynamic label overrides it; line: {line}"
            );
        }
    }

    async fn drain_emit_to_memory(
        emit: &mut crate::schedule::core_loop::CloseEmitFn,
        dest: &mut crate::sink::memory::MemorySink,
    ) {
        use crate::schedule::core_loop::{TickOutput, WriteCommand};
        use crate::sink::Sink as SinkTrait;
        let mut output = TickOutput::default();
        emit(&mut output).expect("close-emit must succeed");
        for cmd in output.writes.drain(..) {
            match cmd {
                WriteCommand::Bytes(buf) => {
                    SinkTrait::write(dest, &buf).await.expect("memory write ok")
                }
                WriteCommand::LogEvent { event, bytes } => {
                    SinkTrait::write_log_event(dest, &event, &bytes)
                        .await
                        .expect("memory log write ok")
                }
            }
        }
    }

    #[tokio::test]
    async fn close_emitter_is_idempotent_after_draining_series() {
        use crate::encoder::create_encoder;
        use crate::model::metric::{Labels, MetricEvent, ValidatedMetricName};
        use crate::schedule::core_loop::CloseSignal;
        use crate::schedule::stats::ScenarioStats;
        use crate::sink::memory::MemorySink;
        use std::sync::{Arc, RwLock};
        use std::time::SystemTime;

        let stats = Arc::new(RwLock::new(ScenarioStats::default()));
        {
            let mut st = stats.write().unwrap();
            st.enable_close_series_tracking();
            let name = ValidatedMetricName::new("up").unwrap();
            let labels = Arc::new(Labels::from_pairs(&[("host", "a")]).unwrap());
            st.push_metric(MetricEvent::from_parts(
                name,
                1.0,
                labels,
                SystemTime::now(),
            ));
        }

        let encoder: Arc<dyn crate::encoder::Encoder> =
            Arc::from(create_encoder(&EncoderConfig::PrometheusText { precision: None }).unwrap());
        let mut emit = super::make_close_emitter(stats.clone(), encoder, CloseSignal::SnapTo(0.0));

        let mut first = MemorySink::new();
        drain_emit_to_memory(&mut emit, &mut first).await;
        assert!(
            !first.buffer.is_empty(),
            "first invocation must emit the tracked series"
        );
        {
            let st = stats.read().unwrap();
            assert!(
                st.close_emit_series.is_empty(),
                "series set must be drained after the first invocation"
            );
            assert!(
                st.last_emit_ts.is_none(),
                "watermark must reset after the first invocation"
            );
        }

        let mut second = MemorySink::new();
        drain_emit_to_memory(&mut emit, &mut second).await;
        assert!(
            second.buffer.is_empty(),
            "second invocation with no new series must emit nothing"
        );
    }

    #[tokio::test]
    async fn close_emit_timestamp_strictly_after_recent_active_emissions() {
        use crate::encoder::create_encoder;
        use crate::model::metric::{Labels, MetricEvent, ValidatedMetricName};
        use crate::schedule::core_loop::CloseSignal;
        use crate::schedule::stats::ScenarioStats;
        use crate::sink::memory::MemorySink;
        use std::sync::{Arc, RwLock};
        use std::time::{Duration, UNIX_EPOCH};

        let watermark_ts = UNIX_EPOCH + Duration::from_millis(1_700_000_000_000);

        let stats = Arc::new(RwLock::new(ScenarioStats::default()));
        {
            let mut st = stats.write().unwrap();
            st.enable_close_series_tracking();
            let name = ValidatedMetricName::new("up").unwrap();
            let labels_a = Arc::new(Labels::from_pairs(&[("host", "a")]).unwrap());
            let labels_b = Arc::new(Labels::from_pairs(&[("host", "b")]).unwrap());
            st.push_metric(MetricEvent::from_parts(
                name.clone(),
                1.0,
                labels_a,
                watermark_ts,
            ));
            st.push_metric(MetricEvent::from_parts(name, 1.0, labels_b, watermark_ts));
        }

        let encoder: Arc<dyn crate::encoder::Encoder> =
            Arc::from(create_encoder(&EncoderConfig::PrometheusText { precision: None }).unwrap());
        let mut emit = super::make_close_emitter(stats.clone(), encoder, CloseSignal::SnapTo(0.0));

        let mut sink = MemorySink::new();
        drain_emit_to_memory(&mut emit, &mut sink).await;

        let output = std::str::from_utf8(&sink.buffer).expect("output must be valid UTF-8");
        let lines: Vec<&str> = output.lines().filter(|l| !l.is_empty()).collect();
        assert_eq!(
            lines.len(),
            2,
            "expected one marker line per distinct (name, labels) series, got: {output}"
        );

        let watermark_ms = watermark_ts.duration_since(UNIX_EPOCH).unwrap().as_millis() as i128;

        for line in &lines {
            let ts_str = line
                .rsplit(' ')
                .next()
                .expect("Prometheus text line must end with a timestamp token");
            let close_ms: i128 = ts_str
                .parse()
                .unwrap_or_else(|_| panic!("close ts must parse as i128, line: {line}"));

            assert!(
                close_ms > watermark_ms,
                "close ts {close_ms} must be strictly greater than watermark ts {watermark_ms}; line: {line}"
            );
        }
    }

    #[tokio::test]
    async fn close_emit_emits_every_distinct_series_regardless_of_count() {
        use crate::encoder::create_encoder;
        use crate::model::metric::{Labels, MetricEvent, ValidatedMetricName};
        use crate::schedule::core_loop::CloseSignal;
        use crate::schedule::stats::ScenarioStats;
        use crate::sink::memory::MemorySink;
        use std::sync::{Arc, RwLock};
        use std::time::SystemTime;

        let series_count: usize = 300;
        let stats = Arc::new(RwLock::new(ScenarioStats::default()));
        {
            let mut st = stats.write().unwrap();
            st.enable_close_series_tracking();
            let name = ValidatedMetricName::new("up").unwrap();
            for i in 0..series_count {
                let labels =
                    Arc::new(Labels::from_pairs(&[("host", format!("h{i}").as_str())]).unwrap());
                st.push_metric(MetricEvent::from_parts(
                    name.clone(),
                    1.0,
                    labels,
                    SystemTime::now(),
                ));
            }
        }

        let encoder: Arc<dyn crate::encoder::Encoder> =
            Arc::from(create_encoder(&EncoderConfig::PrometheusText { precision: None }).unwrap());
        let mut emit = super::make_close_emitter(stats.clone(), encoder, CloseSignal::SnapTo(0.0));

        let mut sink = MemorySink::new();
        drain_emit_to_memory(&mut emit, &mut sink).await;

        let output = std::str::from_utf8(&sink.buffer).expect("output must be valid UTF-8");
        let lines = output.lines().filter(|l| !l.is_empty()).count();
        assert_eq!(lines, series_count);
    }

    #[tokio::test]
    async fn close_emit_dedups_repeated_series_to_one_marker() {
        use crate::encoder::create_encoder;
        use crate::model::metric::{Labels, MetricEvent, ValidatedMetricName};
        use crate::schedule::core_loop::CloseSignal;
        use crate::schedule::stats::ScenarioStats;
        use crate::sink::memory::MemorySink;
        use std::sync::{Arc, RwLock};
        use std::time::SystemTime;

        let stats = Arc::new(RwLock::new(ScenarioStats::default()));
        {
            let mut st = stats.write().unwrap();
            st.enable_close_series_tracking();
            let name = ValidatedMetricName::new("up").unwrap();
            for _ in 0..50 {
                let labels = Arc::new(Labels::from_pairs(&[("host", "a")]).unwrap());
                st.push_metric(MetricEvent::from_parts(
                    name.clone(),
                    1.0,
                    labels,
                    SystemTime::now(),
                ));
            }
        }

        let encoder: Arc<dyn crate::encoder::Encoder> =
            Arc::from(create_encoder(&EncoderConfig::PrometheusText { precision: None }).unwrap());
        let mut emit = super::make_close_emitter(stats.clone(), encoder, CloseSignal::SnapTo(0.0));

        let mut sink = MemorySink::new();
        drain_emit_to_memory(&mut emit, &mut sink).await;

        let output = std::str::from_utf8(&sink.buffer).expect("output must be valid UTF-8");
        let lines = output.lines().filter(|l| !l.is_empty()).count();
        assert_eq!(
            lines, 1,
            "the same (name, labels) series pushed many times must emit exactly one marker"
        );
    }

    #[test]
    fn build_close_emit_enables_series_tracking_on_stats() {
        use crate::schedule::stats::ScenarioStats;
        use std::sync::{Arc, RwLock};

        let mut config = make_config(10.0, "100ms", None);
        config.encoder = EncoderConfig::PrometheusText { precision: None };
        let stats = Arc::new(RwLock::new(ScenarioStats::default()));

        let delay = crate::compiler::DelayClause {
            open: None,
            close: None,
            close_stale_marker: None,
            close_snap_to: Some(0.0),
        };
        let encoder: Arc<dyn crate::encoder::Encoder> =
            Arc::from(crate::encoder::create_encoder(&config.encoder).unwrap());
        let emitter = super::build_close_emit(&config, Some(&stats), Some(&delay), encoder);
        assert!(emitter.is_some(), "snap_to must yield a close-emitter");
        assert!(
            stats.read().unwrap().track_close_series,
            "build_close_emit must enable series tracking when it returns Some"
        );
    }
}
