use crate::checkpoint::merge_checkpoint;
use crate::heartbeat::host_name;
use crate::model::Checkpoint;
use crate::redaction::{RedactionAudit, SecretRedactor};
use crate::sources::shared::truncate_chars;
use crate::tee::{
    backend_sinks_json, filter_batch_for_backend, BackendSinkCell, ReplayFloor,
    SharedRouteResolver, StatusRegistry,
};
use crate::{
    DispatchState, Metrics, SinkMessage, WATCHER_BACKEND_MIXED, WATCHER_BACKEND_NATIVE,
    WATCHER_BACKEND_POLL,
};
use anyhow::Context;
use chrono::{DateTime, Utc};
use moraine_clickhouse::{is_oversized_json_each_row_insert_error, ClickHouseClient};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, Notify, RwLock};
use tokio::task::JoinHandle;
use tracing::{info, warn};

const CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES: usize = 10 * 1024 * 1024;
const OVERSIZED_ROW_FRAGMENT_CHARS: usize = 20_000;
const SINK_JSON_OBJECT_TOO_LARGE: &str = "sink_json_object_too_large";
const MAX_SINK_ERROR_TEXT_CHARS: usize = 1_000;

/// What a sink task is writing for. The flush/checkpoint/retry machinery is
/// identical for both; the role decides intake filtering, heartbeats, and
/// health transitions.
pub(crate) enum SinkRole {
    /// The always-on local sink: receives everything and owns heartbeats.
    Default {
        dispatch: Arc<Mutex<DispatchState>>,
        /// Mirror sink status cells, surfaced through the heartbeat.
        backends: StatusRegistry,
        /// Whether `ingest_heartbeats` carries the `backend_sinks` column
        /// (migration 017). Pre-017 schemas reject unknown columns at insert
        /// time, so the field is attached only when the column exists.
        backend_sinks_column: bool,
        /// Whether `ingest_heartbeats` carries the `redactions_total` column
        /// (migration 022). Redaction still runs without this audit surface.
        redactions_column: bool,
        redactions: Arc<RedactionAudit>,
    },
    /// A mirror sink for one named backend: filters intake down to the
    /// sessions routed to it and reports flush health on its status cell.
    Backend {
        cell: Arc<BackendSinkCell>,
        resolver: SharedRouteResolver,
        /// Signalled on the lagging/unreachable -> ok transition so the
        /// backend's supervisor schedules a catch-up replay pass.
        replay_notify: Arc<Notify>,
        /// Checkpoint floor handed to that pass, captured at the same
        /// transition (see `note_flush_outcome` for why it must be taken
        /// here, inside the sink task, and not when the supervisor wakes).
        replay_floor: ReplayFloor,
        redactor: Arc<SecretRedactor>,
        redactions: Arc<RedactionAudit>,
    },
}

/// Backend-role health bookkeeping after a flush attempt; no-op for the
/// default sink. Recovery requires a drained queue so one successful flush
/// mid-backlog does not trigger a replay storm.
async fn note_flush_outcome(
    role: &SinkRole,
    checkpoints: &Arc<RwLock<HashMap<String, Checkpoint>>>,
    flush_ok: bool,
    queue_drained: bool,
) {
    let SinkRole::Backend {
        cell,
        replay_notify,
        replay_floor,
        ..
    } = role
    else {
        return;
    };

    if !flush_ok {
        if cell.mark_flush_failure() {
            warn!(
                backend = cell.name(),
                "backend flush failing; marked unreachable (live mirroring pauses; \
                 catch-up replay runs once it recovers)"
            );
        }
        return;
    }

    if queue_drained && cell.mark_recovered() {
        // Capture the replay floor BEFORE signalling the supervisor. This
        // sink task is the checkpoint map's only writer and has not yet
        // accepted any post-recovery batch, so the snapshot cannot contain
        // a live-forwarded checkpoint that jumps past the outage's dropped
        // span — the gap the scheduled pass exists to close. Snapshotting
        // when the supervisor wakes instead would race the next live flush.
        let floor = checkpoints.read().await.clone();
        *replay_floor.lock().expect("replay floor mutex poisoned") = Some(floor);
        info!(
            backend = cell.name(),
            dropped_batches = cell.dropped_batches(),
            "backend drained after lagging/unreachable; scheduling catch-up replay"
        );
        replay_notify.notify_one();
    }
}

fn watcher_backend_label(value: u64) -> &'static str {
    match value {
        WATCHER_BACKEND_NATIVE => "native",
        WATCHER_BACKEND_POLL => "poll",
        WATCHER_BACKEND_MIXED => "mixed",
        _ => "unknown",
    }
}

fn saturating_u64_to_u32(value: u64) -> u32 {
    value.min(u32::MAX as u64) as u32
}

fn duration_from_config_seconds(seconds: f64, minimum_seconds: f64, field_name: &str) -> Duration {
    if !seconds.is_finite() {
        warn!("non-finite config value for `{field_name}` ({seconds}); using {minimum_seconds}");
        return Duration::from_secs_f64(minimum_seconds);
    }

    let sanitized_seconds = seconds.max(minimum_seconds);
    Duration::try_from_secs_f64(sanitized_seconds).unwrap_or_else(|_| {
        warn!(
            "out-of-range config value for `{field_name}` ({sanitized_seconds}); using {minimum_seconds}"
        );
        Duration::from_secs_f64(minimum_seconds)
    })
}

fn append_to_visible_percentile(sorted_latencies_ms: &[u64], quantile: f64) -> u64 {
    debug_assert!(!sorted_latencies_ms.is_empty());
    let rank = ((sorted_latencies_ms.len() as f64) * quantile).ceil() as usize;
    sorted_latencies_ms[rank.saturating_sub(1).min(sorted_latencies_ms.len() - 1)]
}

fn compute_append_to_visible_stats(
    raw_rows: &[Value],
    visible_at: DateTime<Utc>,
) -> Option<(u32, u32)> {
    let mut latencies_ms: Vec<u64> = raw_rows
        .iter()
        .filter_map(|row| row.get("record_ts").and_then(Value::as_str))
        .filter_map(|record_ts| DateTime::parse_from_rfc3339(record_ts).ok())
        .map(|record_ts| {
            visible_at
                .signed_duration_since(record_ts.with_timezone(&Utc))
                .num_milliseconds()
                .max(0) as u64
        })
        .collect();

    if latencies_ms.is_empty() {
        return None;
    }

    latencies_ms.sort_unstable();
    let p50 = append_to_visible_percentile(&latencies_ms, 0.50);
    let p95 = append_to_visible_percentile(&latencies_ms, 0.95);
    Some((saturating_u64_to_u32(p50), saturating_u64_to_u32(p95)))
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn spawn_sink_task(
    config: moraine_config::AppConfig,
    clickhouse: ClickHouseClient,
    checkpoints: Arc<RwLock<HashMap<String, Checkpoint>>>,
    metrics: Arc<Metrics>,
    mut rx: mpsc::Receiver<SinkMessage>,
    checkpoint_cursor_columns: bool,
    role: SinkRole,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut raw_rows = Vec::<Value>::new();
        let mut event_rows = Vec::<Value>::new();
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();
        let mut checkpoint_updates = HashMap::<String, Checkpoint>::new();
        let mut pending_batch_bytes = 0usize;

        // Mirror sinks share one ingest_checkpoints table per team backend,
        // so their rows are scoped per host (migration 018; guaranteed
        // present by the schema handshake). The default backend stays
        // single-writer and keeps writing host-less rows, which also keeps
        // it working before `moraine db migrate` adds the column locally.
        let checkpoint_host = match &role {
            SinkRole::Backend { .. } => host_name(),
            SinkRole::Default { .. } => String::new(),
        };

        let flush_interval = duration_from_config_seconds(
            config.ingest.flush_interval_seconds,
            0.05,
            "ingest.flush_interval_seconds",
        );
        let heartbeat_interval = duration_from_config_seconds(
            config.ingest.heartbeat_interval_seconds,
            1.0,
            "ingest.heartbeat_interval_seconds",
        );
        let retry_backoff = duration_from_config_seconds(
            config.ingest.flush_interval_seconds * 2.0,
            0.25,
            "ingest.flush_interval_seconds * 2.0",
        );

        let mut flush_tick = tokio::time::interval(flush_interval);
        let mut heartbeat_tick = tokio::time::interval(heartbeat_interval);
        let mut throttling_flush_retries = false;

        loop {
            if throttling_flush_retries
                && has_pending_data(
                    &raw_rows,
                    &event_rows,
                    &link_rows,
                    &tool_rows,
                    &error_rows,
                    &checkpoint_updates,
                )
            {
                let flush_ok = flush_pending(
                    &clickhouse,
                    &checkpoints,
                    &metrics,
                    &mut raw_rows,
                    &mut event_rows,
                    &mut link_rows,
                    &mut tool_rows,
                    &mut error_rows,
                    &mut checkpoint_updates,
                    checkpoint_cursor_columns,
                    &checkpoint_host,
                )
                .await;
                note_flush_outcome(&role, &checkpoints, flush_ok, rx.is_empty()).await;
                if flush_ok {
                    pending_batch_bytes = 0;
                    throttling_flush_retries = false;
                    info!("flush retry succeeded; resuming sink intake");
                } else {
                    tokio::select! {
                        _ = tokio::time::sleep(retry_backoff) => {}
                        _ = heartbeat_tick.tick() => {
                            emit_heartbeat(&clickhouse, &metrics, &role).await;
                        }
                    }
                }
                continue;
            }

            tokio::select! {
                maybe_msg = rx.recv() => {
                    match maybe_msg {
                        Some(SinkMessage::Batch(batch)) => {
                            // Mirror sinks only buffer the sessions routed to
                            // them; the default sink takes batches whole.
                            let batch = match &role {
                                SinkRole::Backend {
                                    cell,
                                    resolver,
                                    redactor,
                                    redactions,
                                    ..
                                } => {
                                    let mut resolver =
                                        resolver.lock().expect("route resolver mutex poisoned");
                                    let mut batch =
                                        filter_batch_for_backend(&batch, cell.name(), &mut resolver);
                                    let report = redactor.redact_batch(&mut batch);
                                    redactions.record(&report);
                                    batch
                                }
                                SinkRole::Default { .. } => batch,
                            };
                            pending_batch_bytes =
                                pending_batch_bytes.saturating_add(batch.approx_bytes());
                            raw_rows.extend(batch.raw_rows);
                            event_rows.extend(batch.event_rows);
                            link_rows.extend(batch.link_rows);
                            tool_rows.extend(batch.tool_rows);
                            error_rows.extend(batch.error_rows);
                            if let Some(cp) = batch.checkpoint {
                                merge_checkpoint(&mut checkpoint_updates, cp);
                            }

                            let total_rows = raw_rows.len() + event_rows.len() + link_rows.len() + tool_rows.len() + error_rows.len();
                            if total_rows >= config.ingest.batch_size
                                || pending_batch_bytes >= config.ingest.max_batch_bytes.max(1)
                            {
                                let flush_ok = flush_pending(
                                    &clickhouse,
                                    &checkpoints,
                                    &metrics,
                                    &mut raw_rows,
                                    &mut event_rows,
                                    &mut link_rows,
                                    &mut tool_rows,
                                    &mut error_rows,
                                    &mut checkpoint_updates,
                                    checkpoint_cursor_columns,
                                    &checkpoint_host,
                                ).await;
                                note_flush_outcome(&role, &checkpoints, flush_ok, rx.is_empty()).await;
                                if !flush_ok {
                                    if !throttling_flush_retries {
                                        warn!(
                                            "flush failed; pausing sink intake and retrying pending rows every {} ms",
                                            retry_backoff.as_millis()
                                        );
                                    }
                                    throttling_flush_retries = true;
                                } else {
                                    pending_batch_bytes = 0;
                                }
                            }
                        }
                        None => break,
                    }
                }
                _ = flush_tick.tick() => {
                    if has_pending_data(&raw_rows, &event_rows, &link_rows, &tool_rows, &error_rows, &checkpoint_updates) {
                        let flush_ok = flush_pending(
                            &clickhouse,
                            &checkpoints,
                            &metrics,
                            &mut raw_rows,
                            &mut event_rows,
                            &mut link_rows,
                            &mut tool_rows,
                            &mut error_rows,
                            &mut checkpoint_updates,
                            checkpoint_cursor_columns,
                            &checkpoint_host,
                        ).await;
                        note_flush_outcome(&role, &checkpoints, flush_ok, rx.is_empty()).await;
                        if !flush_ok {
                            if !throttling_flush_retries {
                                warn!(
                                    "flush failed; pausing sink intake and retrying pending rows every {} ms",
                                    retry_backoff.as_millis()
                                );
                            }
                            throttling_flush_retries = true;
                        } else {
                            pending_batch_bytes = 0;
                        }
                    }
                }
                _ = heartbeat_tick.tick() => {
                    emit_heartbeat(&clickhouse, &metrics, &role).await;
                }
            }
        }

        if has_pending_data(
            &raw_rows,
            &event_rows,
            &link_rows,
            &tool_rows,
            &error_rows,
            &checkpoint_updates,
        ) {
            let flush_ok = flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                checkpoint_cursor_columns,
                &checkpoint_host,
            )
            .await;
            note_flush_outcome(&role, &checkpoints, flush_ok, true).await;
        }
    })
}

fn has_pending_data(
    raw_rows: &[Value],
    event_rows: &[Value],
    link_rows: &[Value],
    tool_rows: &[Value],
    error_rows: &[Value],
    checkpoint_updates: &HashMap<String, Checkpoint>,
) -> bool {
    !(raw_rows.is_empty()
        && event_rows.is_empty()
        && link_rows.is_empty()
        && tool_rows.is_empty()
        && error_rows.is_empty()
        && checkpoint_updates.is_empty())
}

fn quarantine_oversized_rows(
    table: &str,
    rows: &mut Vec<Value>,
    error_rows: &mut Vec<Value>,
) -> anyhow::Result<usize> {
    let mut kept = Vec::with_capacity(rows.len());
    let mut quarantined = 0usize;

    for row in std::mem::take(rows) {
        let row_bytes = serde_json::to_vec(&row)
            .with_context(|| format!("failed to encode {table} row for oversized-row guard"))?
            .len();
        if row_bytes > CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES {
            error_rows.push(oversized_row_error(table, &row, row_bytes));
            quarantined = quarantined.saturating_add(1);
        } else {
            kept.push(row);
        }
    }

    *rows = kept;
    Ok(quarantined)
}

fn oversized_row_error(table: &str, row: &Value, row_bytes: usize) -> Value {
    let source = ingest_error_source_from_row(row);
    json!({
        "source_name": source.source_name,
        "harness": source.harness,
        "source_file": source.source_file,
        "source_inode": source.source_inode,
        "source_generation": source.source_generation,
        "source_line_no": source.source_line_no,
        "source_offset": source.source_offset,
        "error_kind": "oversized_json_row",
        "error_text": format!(
            "skipped {table} row: serialized JSONEachRow object is {row_bytes} bytes, above ClickHouse limit of {CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES} bytes"
        ),
        "raw_fragment": oversized_row_fragment(row),
    })
}

fn parse_source_ref(row: &Value) -> Option<(String, u64, u64)> {
    let source_ref = row.get("source_ref").and_then(Value::as_str)?;
    let mut parts = source_ref.rsplitn(3, ':');
    let line_no = parts.next()?.parse::<u64>().ok()?;
    let source_generation = parts.next()?.parse::<u64>().ok()?;
    let source_file = parts.next()?.to_string();
    Some((source_file, source_generation, line_no))
}

fn row_str(row: &Value, key: &str) -> Option<String> {
    row.get(key).and_then(Value::as_str).map(ToOwned::to_owned)
}

fn row_u64(row: &Value, key: &str) -> Option<u64> {
    row.get(key).and_then(Value::as_u64)
}

fn oversized_row_fragment(row: &Value) -> String {
    for key in [
        "raw_json",
        "payload_json",
        "output_json",
        "input_json",
        "output_text",
        "text_content",
        "metadata_json",
    ] {
        if let Some(value) = row
            .get(key)
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty())
        {
            return truncate_chars(value, OVERSIZED_ROW_FRAGMENT_CHARS);
        }
    }

    serde_json::to_string(row)
        .map(|value| truncate_chars(&value, OVERSIZED_ROW_FRAGMENT_CHARS))
        .unwrap_or_default()
}

#[derive(Clone, Copy, Debug)]
enum SinkStage {
    RawEvents,
    Events,
    EventLinks,
    ToolIo,
    IngestErrors,
    IngestCheckpoints,
}

impl SinkStage {
    fn table(self) -> &'static str {
        match self {
            Self::RawEvents => "raw_events",
            Self::Events => "events",
            Self::EventLinks => "event_links",
            Self::ToolIo => "tool_io",
            Self::IngestErrors => "ingest_errors",
            Self::IngestCheckpoints => "ingest_checkpoints",
        }
    }
}

struct SinkFlushFailure {
    stage: SinkStage,
    error: anyhow::Error,
}

impl SinkFlushFailure {
    fn new(stage: SinkStage, error: anyhow::Error) -> Self {
        Self { stage, error }
    }
}

#[derive(Default)]
struct IngestErrorSource {
    source_name: String,
    harness: String,
    source_file: String,
    source_inode: u64,
    source_generation: u64,
    source_line_no: u64,
    source_offset: u64,
}

#[derive(Clone, Copy)]
struct PendingSinkData<'a> {
    raw_rows: &'a [Value],
    event_rows: &'a [Value],
    link_rows: &'a [Value],
    tool_rows: &'a [Value],
    error_rows: &'a [Value],
    checkpoint_updates: &'a HashMap<String, Checkpoint>,
}

impl PendingSinkData<'_> {
    fn source_for(self, stage: SinkStage) -> (IngestErrorSource, &'static str) {
        let stage_rows = match stage {
            SinkStage::RawEvents => self.raw_rows,
            SinkStage::Events => self.event_rows,
            SinkStage::EventLinks => self.link_rows,
            SinkStage::ToolIo => self.tool_rows,
            SinkStage::IngestErrors => self.error_rows,
            SinkStage::IngestCheckpoints => &[],
        };

        if let [row] = stage_rows {
            return (ingest_error_source_from_row(row), "row");
        }

        if let Some(source) = self
            .checkpoint_updates
            .values()
            .next()
            .map(ingest_error_source_from_checkpoint)
        {
            return (source, "checkpoint");
        }

        (IngestErrorSource::default(), "batch")
    }

    fn stage_len(self, stage: SinkStage) -> usize {
        match stage {
            SinkStage::RawEvents => self.raw_rows.len(),
            SinkStage::Events => self.event_rows.len(),
            SinkStage::EventLinks => self.link_rows.len(),
            SinkStage::ToolIo => self.tool_rows.len(),
            SinkStage::IngestErrors => self.error_rows.len(),
            SinkStage::IngestCheckpoints => self.checkpoint_updates.len(),
        }
    }

    fn total_len(self) -> usize {
        self.raw_rows.len()
            + self.event_rows.len()
            + self.link_rows.len()
            + self.tool_rows.len()
            + self.error_rows.len()
            + self.checkpoint_updates.len()
    }
}

fn ingest_error_source_from_row(row: &Value) -> IngestErrorSource {
    let source_ref = parse_source_ref(row);
    IngestErrorSource {
        source_name: row_str(row, "source_name").unwrap_or_default(),
        harness: row_str(row, "harness").unwrap_or_default(),
        source_file: row_str(row, "source_file")
            .or_else(|| source_ref.as_ref().map(|parts| parts.0.clone()))
            .unwrap_or_default(),
        source_inode: row_u64(row, "source_inode").unwrap_or_default(),
        source_generation: row_u64(row, "source_generation")
            .or_else(|| source_ref.as_ref().map(|parts| parts.1))
            .unwrap_or_default(),
        source_line_no: row_u64(row, "source_line_no")
            .or_else(|| source_ref.as_ref().map(|parts| parts.2))
            .unwrap_or_default(),
        source_offset: row_u64(row, "source_offset").unwrap_or_default(),
    }
}

fn ingest_error_source_from_checkpoint(checkpoint: &Checkpoint) -> IngestErrorSource {
    IngestErrorSource {
        source_name: checkpoint.source_name.clone(),
        source_file: checkpoint.source_file.clone(),
        source_inode: checkpoint.source_inode,
        source_generation: checkpoint.source_generation as u64,
        source_line_no: checkpoint.last_line_no,
        source_offset: checkpoint.last_offset,
        ..Default::default()
    }
}

fn non_retryable_sink_error_row(
    stage: SinkStage,
    error_text: &str,
    pending: PendingSinkData<'_>,
) -> Value {
    let (source, source_scope) = pending.source_for(stage);
    let stage_pending_rows = pending.stage_len(stage);
    let total_pending_rows = pending.total_len();

    json!({
        "source_name": source.source_name,
        "harness": source.harness,
        "source_file": source.source_file,
        "source_inode": source.source_inode,
        "source_generation": source.source_generation,
        "source_line_no": source.source_line_no,
        "source_offset": source.source_offset,
        "error_kind": SINK_JSON_OBJECT_TOO_LARGE,
        "error_text": truncate_chars(error_text, MAX_SINK_ERROR_TEXT_CHARS),
        "raw_fragment": format!(
            "source_scope={source_scope}; sink_stage={}; stage_pending_rows={stage_pending_rows}; total_pending_rows={total_pending_rows}; checkpoint_updates={}; original_payload_omitted=true",
            stage.table(),
            pending.checkpoint_updates.len()
        ),
    })
}

async fn record_non_retryable_sink_error(
    clickhouse: &ClickHouseClient,
    metrics: &Arc<Metrics>,
    stage: SinkStage,
    error_text: &str,
    pending: PendingSinkData<'_>,
) -> anyhow::Result<()> {
    let row = non_retryable_sink_error_row(stage, error_text, pending);

    clickhouse
        .insert_json_rows("ingest_errors", &[row])
        .await
        .with_context(|| {
            format!(
                "failed to record non-retryable sink error for {}",
                stage.table()
            )
        })?;
    metrics.err_rows_written.fetch_add(1, Ordering::Relaxed);
    Ok(())
}

fn clear_pending_sink_data(
    raw_rows: &mut Vec<Value>,
    event_rows: &mut Vec<Value>,
    link_rows: &mut Vec<Value>,
    tool_rows: &mut Vec<Value>,
    error_rows: &mut Vec<Value>,
    checkpoint_updates: &mut HashMap<String, Checkpoint>,
) {
    raw_rows.clear();
    event_rows.clear();
    link_rows.clear();
    tool_rows.clear();
    error_rows.clear();
    checkpoint_updates.clear();
}

async fn commit_checkpoint_updates(
    clickhouse: &ClickHouseClient,
    checkpoints: &Arc<RwLock<HashMap<String, Checkpoint>>>,
    checkpoint_rows: &[Value],
    checkpoint_updates: &HashMap<String, Checkpoint>,
) -> anyhow::Result<()> {
    if checkpoint_rows.is_empty() {
        return Ok(());
    }

    clickhouse
        .insert_json_rows("ingest_checkpoints", checkpoint_rows)
        .await
        .context("failed to insert ingest checkpoint rows")?;

    // Offset-monotone merge, not a blind insert: a backend map has two
    // producers (live router batches and catch-up replay), so an out-of-order
    // flush must never regress what the map already committed.
    let mut state = checkpoints.write().await;
    for cp in checkpoint_updates.values() {
        merge_checkpoint(&mut state, cp.clone());
    }
    Ok(())
}

/// Default-role only: mirror sinks write rows into databases this host does
/// not own, so the ingest heartbeat (including per-backend mirror status)
/// always lands in the default backend.
async fn emit_heartbeat(clickhouse: &ClickHouseClient, metrics: &Arc<Metrics>, role: &SinkRole) {
    let SinkRole::Default {
        dispatch,
        backends,
        backend_sinks_column,
        redactions_column,
        redactions,
    } = role
    else {
        return;
    };

    let files_active = {
        let state = dispatch.lock().expect("dispatch mutex poisoned");
        state.inflight.len() as u32
    };
    let files_watched = metrics.watcher_registrations.load(Ordering::Relaxed) as u32;
    let last_error = {
        metrics
            .last_error
            .lock()
            .expect("metrics last_error mutex poisoned")
            .clone()
    };
    let watcher_backend =
        watcher_backend_label(metrics.watcher_backend_state.load(Ordering::Relaxed));

    let mut heartbeat = json!({
        "host": host_name(),
        "service_version": env!("CARGO_PKG_VERSION"),
        "queue_depth": metrics.queue_depth.load(Ordering::Relaxed),
        "files_active": files_active,
        "files_watched": files_watched,
        "rows_raw_written": metrics.raw_rows_written.load(Ordering::Relaxed),
        "rows_events_written": metrics.event_rows_written.load(Ordering::Relaxed),
        "rows_errors_written": metrics.err_rows_written.load(Ordering::Relaxed),
        "flush_latency_ms": saturating_u64_to_u32(metrics.last_flush_ms.load(Ordering::Relaxed)),
        "append_to_visible_p50_ms": saturating_u64_to_u32(metrics.append_to_visible_p50_ms.load(Ordering::Relaxed)),
        "append_to_visible_p95_ms": saturating_u64_to_u32(metrics.append_to_visible_p95_ms.load(Ordering::Relaxed)),
        "watcher_backend": watcher_backend,
        "watcher_error_count": metrics.watcher_error_count.load(Ordering::Relaxed),
        "watcher_reset_count": metrics.watcher_reset_count.load(Ordering::Relaxed),
        "watcher_last_reset_unix_ms": metrics.watcher_last_reset_unix_ms.load(Ordering::Relaxed),
        "last_error": last_error,
    });

    // Attached only when at least one mirror sink exists AND the column is
    // present (pre-017 schemas reject unknown columns at insert time).
    if *backend_sinks_column {
        if let Some(statuses) = backend_sinks_json(backends) {
            if let Some(obj) = heartbeat.as_object_mut() {
                obj.insert("backend_sinks".to_string(), json!(statuses));
            }
        }
    }

    if *redactions_column {
        let counts = redactions.snapshot();
        if !counts.is_empty() {
            if let Some(obj) = heartbeat.as_object_mut() {
                let counts_json = serde_json::to_string(&counts).unwrap_or_default();
                obj.insert("redactions_total".to_string(), Value::String(counts_json));
            }
        }
    }

    if let Err(exc) = clickhouse
        .insert_json_rows("ingest_heartbeats", &[heartbeat])
        .await
    {
        warn!("heartbeat insert failed: {exc}");
    }
}

#[allow(clippy::too_many_arguments)]
async fn flush_pending(
    clickhouse: &ClickHouseClient,
    checkpoints: &Arc<RwLock<HashMap<String, Checkpoint>>>,
    metrics: &Arc<Metrics>,
    raw_rows: &mut Vec<Value>,
    event_rows: &mut Vec<Value>,
    link_rows: &mut Vec<Value>,
    tool_rows: &mut Vec<Value>,
    error_rows: &mut Vec<Value>,
    checkpoint_updates: &mut HashMap<String, Checkpoint>,
    checkpoint_cursor_columns: bool,
    checkpoint_host: &str,
) -> bool {
    let started = Instant::now();

    let checkpoint_rows: Vec<Value> = checkpoint_updates
        .values()
        .map(|cp| {
            let mut row = json!({
                "source_name": cp.source_name,
                "source_file": cp.source_file,
                "source_inode": cp.source_inode,
                "source_generation": cp.source_generation,
                "last_offset": cp.last_offset,
                "last_line_no": cp.last_line_no,
                "status": cp.status,
            });
            // Older schemas (pre-015) reject unknown columns at insert time,
            // so the SQLite cursor fields are attached only when present.
            if checkpoint_cursor_columns {
                if let Some(obj) = row.as_object_mut() {
                    obj.insert("cursor_json".to_string(), json!(cp.cursor_json));
                    obj.insert(
                        "source_fingerprint".to_string(),
                        json!(cp.source_fingerprint),
                    );
                    obj.insert(
                        "schema_fingerprint".to_string(),
                        json!(cp.schema_fingerprint),
                    );
                }
            }
            // Backend-role only (migration 018): scopes rows in a shared
            // team table to this host. Empty for the default sink.
            if !checkpoint_host.is_empty() {
                if let Some(obj) = row.as_object_mut() {
                    obj.insert("host".to_string(), json!(checkpoint_host));
                }
            }
            row
        })
        .collect();

    let quarantined = match (|| -> anyhow::Result<usize> {
        Ok(
            quarantine_oversized_rows("raw_events", raw_rows, error_rows)?
                + quarantine_oversized_rows("events", event_rows, error_rows)?
                + quarantine_oversized_rows("event_links", link_rows, error_rows)?
                + quarantine_oversized_rows("tool_io", tool_rows, error_rows)?,
        )
    })() {
        Ok(quarantined) => quarantined,
        Err(exc) => {
            metrics.flush_failures.fetch_add(1, Ordering::Relaxed);
            *metrics
                .last_error
                .lock()
                .expect("metrics last_error mutex poisoned") = exc.to_string();
            warn!("flush failed during oversized-row guard: {exc}");
            return false;
        }
    };
    if quarantined > 0 {
        warn!(
            rows = quarantined,
            max_bytes = CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES,
            "quarantined oversized rows before ClickHouse insert"
        );
    }

    let flush_result = async {
        if !raw_rows.is_empty() {
            clickhouse
                .insert_json_rows("raw_events", raw_rows)
                .await
                .map_err(|error| SinkFlushFailure::new(SinkStage::RawEvents, error))?;
            metrics
                .raw_rows_written
                .fetch_add(raw_rows.len() as u64, Ordering::Relaxed);
            if let Some((p50_ms, p95_ms)) = compute_append_to_visible_stats(raw_rows, Utc::now()) {
                metrics
                    .append_to_visible_p50_ms
                    .store(p50_ms as u64, Ordering::Relaxed);
                metrics
                    .append_to_visible_p95_ms
                    .store(p95_ms as u64, Ordering::Relaxed);
            }
            raw_rows.clear();
        }

        if !event_rows.is_empty() {
            clickhouse
                .insert_json_rows("events", event_rows)
                .await
                .map_err(|error| SinkFlushFailure::new(SinkStage::Events, error))?;
            metrics
                .event_rows_written
                .fetch_add(event_rows.len() as u64, Ordering::Relaxed);
            event_rows.clear();
        }

        if !link_rows.is_empty() {
            clickhouse
                .insert_json_rows("event_links", link_rows)
                .await
                .map_err(|error| SinkFlushFailure::new(SinkStage::EventLinks, error))?;
            link_rows.clear();
        }

        if !tool_rows.is_empty() {
            clickhouse
                .insert_json_rows("tool_io", tool_rows)
                .await
                .map_err(|error| SinkFlushFailure::new(SinkStage::ToolIo, error))?;
            tool_rows.clear();
        }

        if !error_rows.is_empty() {
            clickhouse
                .insert_json_rows("ingest_errors", error_rows)
                .await
                .map_err(|error| SinkFlushFailure::new(SinkStage::IngestErrors, error))?;
            metrics
                .err_rows_written
                .fetch_add(error_rows.len() as u64, Ordering::Relaxed);
            error_rows.clear();
        }

        if !checkpoint_rows.is_empty() {
            commit_checkpoint_updates(
                clickhouse,
                checkpoints,
                &checkpoint_rows,
                checkpoint_updates,
            )
            .await
            .map_err(|error| SinkFlushFailure::new(SinkStage::IngestCheckpoints, error))?;
            checkpoint_updates.clear();
        }

        metrics
            .last_flush_ms
            .store(started.elapsed().as_millis() as u64, Ordering::Relaxed);
        Ok::<(), SinkFlushFailure>(())
    }
    .await;

    match flush_result {
        Ok(()) => true,
        Err(failure) => {
            metrics.flush_failures.fetch_add(1, Ordering::Relaxed);
            let error_text = failure.error.to_string();
            if is_oversized_json_each_row_insert_error(&failure.error) {
                let last_error = format!(
                    "non-retryable sink flush failure at {}: {error_text}",
                    failure.stage.table()
                );
                *metrics
                    .last_error
                    .lock()
                    .expect("metrics last_error mutex poisoned") = last_error;
                warn!(
                    sink_stage = failure.stage.table(),
                    "non-retryable oversized JSONEachRow insert failure; recording compact ingest error, committing checkpoint, and dropping pending batch: {}",
                    failure.error
                );
                if let Err(error) = record_non_retryable_sink_error(
                    clickhouse,
                    metrics,
                    failure.stage,
                    &error_text,
                    PendingSinkData {
                        raw_rows,
                        event_rows,
                        link_rows,
                        tool_rows,
                        error_rows,
                        checkpoint_updates,
                    },
                )
                .await
                {
                    let last_error = format!(
                        "failed to record non-retryable sink error at {}: {error}",
                        failure.stage.table()
                    );
                    *metrics
                        .last_error
                        .lock()
                        .expect("metrics last_error mutex poisoned") = last_error.clone();
                    warn!(
                        sink_stage = failure.stage.table(),
                        original_error = %failure.error,
                        "{last_error}"
                    );
                    return false;
                }
                if let Err(error) = commit_checkpoint_updates(
                    clickhouse,
                    checkpoints,
                    &checkpoint_rows,
                    checkpoint_updates,
                )
                .await
                {
                    let last_error = format!(
                        "failed to checkpoint after non-retryable sink failure at {}: {error}",
                        failure.stage.table()
                    );
                    *metrics
                        .last_error
                        .lock()
                        .expect("metrics last_error mutex poisoned") = last_error.clone();
                    warn!(
                        sink_stage = failure.stage.table(),
                        original_error = %failure.error,
                        "{last_error}"
                    );
                    return false;
                }
                metrics
                    .last_flush_ms
                    .store(started.elapsed().as_millis() as u64, Ordering::Relaxed);
                clear_pending_sink_data(
                    raw_rows,
                    event_rows,
                    link_rows,
                    tool_rows,
                    error_rows,
                    checkpoint_updates,
                );
                return true;
            }

            *metrics
                .last_error
                .lock()
                .expect("metrics last_error mutex poisoned") = error_text;
            warn!(
                sink_stage = failure.stage.table(),
                "flush failed: {}", failure.error
            );
            false
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::checkpoint::checkpoint_key;
    use crate::model::RowBatch;
    use axum::{
        extract::{Query, State},
        http::StatusCode,
        routing::post,
        Router,
    };
    use chrono::{DateTime, Utc};
    use serde_json::json;
    use tokio::time::timeout;

    #[derive(Clone, Default)]
    struct MockClickHouseState {
        calls_by_table: Arc<Mutex<HashMap<String, usize>>>,
        rows_by_table: Arc<Mutex<HashMap<String, Vec<Value>>>>,
        fail_once_by_table: Arc<Mutex<HashMap<String, usize>>>,
        fail_always_by_table: Arc<Mutex<HashMap<String, String>>>,
    }

    impl MockClickHouseState {
        fn with_single_failure(table: &str) -> Self {
            let state = Self::default();
            state
                .fail_once_by_table
                .lock()
                .expect("mock fail_once mutex poisoned")
                .insert(table.to_string(), 1);
            state
        }

        fn with_permanent_failure(table: &str, body: &str) -> Self {
            let state = Self::default();
            state.fail_permanently(table, body);
            state
        }

        fn fail_permanently(&self, table: &str, body: &str) {
            self.fail_always_by_table
                .lock()
                .expect("mock fail_always mutex poisoned")
                .insert(table.to_string(), body.to_string());
        }

        fn call_count(&self, table: &str) -> usize {
            *self
                .calls_by_table
                .lock()
                .expect("mock calls mutex poisoned")
                .get(table)
                .unwrap_or(&0)
        }

        fn rows(&self, table: &str) -> Vec<Value> {
            self.rows_by_table
                .lock()
                .expect("mock rows mutex poisoned")
                .get(table)
                .cloned()
                .unwrap_or_default()
        }
    }

    fn inserted_table_name(query: &str) -> Option<&'static str> {
        if query.contains("`raw_events`") {
            Some("raw_events")
        } else if query.contains("`event_links`") {
            Some("event_links")
        } else if query.contains("`tool_io`") {
            Some("tool_io")
        } else if query.contains("`ingest_errors`") {
            Some("ingest_errors")
        } else if query.contains("`ingest_checkpoints`") {
            Some("ingest_checkpoints")
        } else if query.contains("`ingest_heartbeats`") {
            Some("ingest_heartbeats")
        } else if query.contains("`events`") {
            Some("events")
        } else {
            None
        }
    }

    async fn mock_clickhouse_handler(
        State(state): State<MockClickHouseState>,
        Query(params): Query<HashMap<String, String>>,
        body: String,
    ) -> (StatusCode, String) {
        let query = params.get("query").cloned().unwrap_or_default();
        let Some(table) = inserted_table_name(&query) else {
            return (
                StatusCode::BAD_REQUEST,
                format!("unexpected query payload: {query}"),
            );
        };

        {
            let mut calls = state
                .calls_by_table
                .lock()
                .expect("mock calls mutex poisoned");
            *calls.entry(table.to_string()).or_insert(0) += 1;
        }
        {
            let mut fail_once = state
                .fail_once_by_table
                .lock()
                .expect("mock fail_once mutex poisoned");
            if let Some(remaining) = fail_once.get_mut(table) {
                if *remaining > 0 {
                    *remaining -= 1;
                    return (
                        StatusCode::INTERNAL_SERVER_ERROR,
                        format!("intentional failure for {table}"),
                    );
                }
            }
        }
        {
            let fail_always = state
                .fail_always_by_table
                .lock()
                .expect("mock fail_always mutex poisoned");
            if let Some(body) = fail_always.get(table) {
                return (StatusCode::BAD_REQUEST, body.clone());
            }
        }

        {
            let mut rows = state
                .rows_by_table
                .lock()
                .expect("mock rows mutex poisoned");
            let entry = rows.entry(table.to_string()).or_default();
            for line in body.lines().filter(|line| !line.trim().is_empty()) {
                if let Ok(row) = serde_json::from_str::<Value>(line) {
                    entry.push(row);
                }
            }
        }

        (StatusCode::OK, String::new())
    }

    async fn spawn_mock_clickhouse(
        fail_once_table: &str,
    ) -> (ClickHouseClient, MockClickHouseState) {
        let state = MockClickHouseState::with_single_failure(fail_once_table);
        spawn_mock_clickhouse_with_state(state).await
    }

    async fn spawn_mock_clickhouse_with_state(
        state: MockClickHouseState,
    ) -> (ClickHouseClient, MockClickHouseState) {
        let app = Router::new()
            .route("/", post(mock_clickhouse_handler))
            .with_state(state.clone());

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind mock clickhouse listener");
        let addr = listener.local_addr().expect("mock listener addr");

        tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });

        let mut config = moraine_config::AppConfig::default();
        config.clickhouse.url = format!("http://{}", addr);
        config.clickhouse.timeout_seconds = 1.0;
        let clickhouse = ClickHouseClient::new(config.clickhouse)
            .expect("mock clickhouse client should initialize");

        (clickhouse, state)
    }

    fn sample_checkpoint() -> Checkpoint {
        Checkpoint {
            source_name: "source-a".to_string(),
            source_file: "/tmp/source-a.jsonl".to_string(),
            source_inode: 42,
            source_generation: 1,
            last_offset: 100,
            last_line_no: 3,
            status: "active".to_string(),
            ..Default::default()
        }
    }

    fn single_row_batch(id: u64) -> SinkMessage {
        let mut batch = RowBatch::default();
        batch.raw_rows.push(json!({ "id": id }));
        SinkMessage::Batch(batch)
    }

    fn test_redactor() -> Arc<SecretRedactor> {
        Arc::new(
            SecretRedactor::new(&moraine_config::RedactionConfig::default())
                .expect("test redactor compiles"),
        )
    }

    fn test_redaction_audit() -> Arc<RedactionAudit> {
        Arc::new(RedactionAudit::default())
    }

    fn default_role() -> SinkRole {
        SinkRole::Default {
            dispatch: Arc::new(Mutex::new(DispatchState::default())),
            backends: Arc::new(Mutex::new(std::collections::BTreeMap::new())),
            backend_sinks_column: false,
            redactions_column: false,
            redactions: test_redaction_audit(),
        }
    }

    const CLICKHOUSE_OVERSIZED_JSON_OBJECT_ERROR: &str =
        "Code: 117. DB::Exception: Size of JSON object at position 104890103 is extremely large. \
         Expected not greater than 10485760 bytes, but current is 104890103 bytes per row. \
         While executing ParallelParsingBlockInputFormat.";

    #[tokio::test]
    async fn failed_flush_throttles_sink_consumption() {
        let mut config = moraine_config::AppConfig::default();
        config.clickhouse.url = "http://127.0.0.1:1".to_string();
        config.clickhouse.timeout_seconds = 1.0;
        config.ingest.batch_size = 1;
        config.ingest.flush_interval_seconds = 0.05;
        config.ingest.heartbeat_interval_seconds = 60.0;

        let clickhouse = ClickHouseClient::new(config.clickhouse.clone())
            .expect("clickhouse client should initialize");
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());
        let (tx, rx) = mpsc::channel(1);

        let handle = spawn_sink_task(
            config,
            clickhouse,
            checkpoints,
            metrics,
            rx,
            true,
            default_role(),
        );

        tx.send(single_row_batch(1))
            .await
            .expect("first send should succeed");
        tx.send(single_row_batch(2))
            .await
            .expect("second send should succeed");

        let third_send = timeout(Duration::from_millis(350), tx.send(single_row_batch(3))).await;
        assert!(
            third_send.is_err(),
            "third send should block while sink retries failed flushes"
        );

        handle.abort();
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_retries_only_unfinished_tables() {
        let (clickhouse, mock_state) = spawn_mock_clickhouse("events").await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let mut raw_rows = vec![json!({
            "record_ts": "2026-02-17T00:00:01.000Z",
            "event_uid": "evt-1"
        })];
        let mut event_rows = vec![json!({"event_uid": "evt-1"})];
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();
        let mut checkpoint_updates = HashMap::new();
        let checkpoint = sample_checkpoint();
        checkpoint_updates.insert(
            checkpoint_key(&checkpoint.source_name, &checkpoint.source_file),
            checkpoint.clone(),
        );

        let first_attempt = flush_pending(
            &clickhouse,
            &checkpoints,
            &metrics,
            &mut raw_rows,
            &mut event_rows,
            &mut link_rows,
            &mut tool_rows,
            &mut error_rows,
            &mut checkpoint_updates,
            true,
            "",
        )
        .await;
        assert!(!first_attempt, "first flush should fail at events stage");

        assert!(raw_rows.is_empty(), "raw rows should not be retried");
        assert_eq!(
            event_rows.len(),
            1,
            "event rows remain pending after failure"
        );
        assert_eq!(
            checkpoint_updates.len(),
            1,
            "checkpoint update must remain pending until checkpoint flush succeeds"
        );
        assert_eq!(metrics.raw_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.event_rows_written.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        let second_attempt = flush_pending(
            &clickhouse,
            &checkpoints,
            &metrics,
            &mut raw_rows,
            &mut event_rows,
            &mut link_rows,
            &mut tool_rows,
            &mut error_rows,
            &mut checkpoint_updates,
            true,
            "",
        )
        .await;
        assert!(
            second_attempt,
            "second flush should complete remaining stages"
        );

        assert!(event_rows.is_empty());
        assert!(checkpoint_updates.is_empty());
        assert_eq!(metrics.raw_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.event_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        assert_eq!(mock_state.call_count("raw_events"), 1);
        assert_eq!(mock_state.call_count("events"), 2);
        assert_eq!(mock_state.call_count("ingest_checkpoints"), 1);

        let state = checkpoints.read().await;
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        assert!(
            state.contains_key(&checkpoint_key_value),
            "checkpoint cache should advance after checkpoint stage succeeds"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_records_non_retryable_sink_oversized_json_and_clears_batch() {
        let (clickhouse, mock_state) =
            spawn_mock_clickhouse_with_state(MockClickHouseState::with_permanent_failure(
                "raw_events",
                CLICKHOUSE_OVERSIZED_JSON_OBJECT_ERROR,
            ))
            .await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let checkpoint = sample_checkpoint();
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value.clone(), checkpoint)]);
        let mut raw_rows = vec![json!({
            "source_name": "source-a",
            "harness": "codex",
            "source_file": "/tmp/source-a.jsonl",
            "source_inode": 42u64,
            "source_generation": 1u32,
            "source_line_no": 7u64,
            "source_offset": 2048u64,
            "record_ts": "2026-02-17T00:00:01.000Z",
            "top_type": "response_item",
            "session_id": "session-a",
            "raw_json": "payload-marker-that-must-not-be-copied",
            "raw_json_hash": 1u64,
            "event_uid": "evt-fallback",
        })];
        let mut event_rows = vec![json!({"event_uid": "evt-fallback"})];
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "non-retryable oversized JSONEachRow failures should clear the stuck batch"
        );

        assert!(raw_rows.is_empty());
        assert!(event_rows.is_empty());
        assert!(checkpoint_updates.is_empty());
        assert_eq!(mock_state.call_count("raw_events"), 1);
        assert_eq!(
            mock_state.call_count("events"),
            0,
            "the failed stage stops the normal flush sequence"
        );
        assert_eq!(mock_state.call_count("ingest_errors"), 1);
        assert_eq!(
            mock_state.call_count("ingest_checkpoints"),
            1,
            "the skipped batch checkpoint should commit after the compact error row"
        );

        let persisted_errors = mock_state.rows("ingest_errors");
        assert_eq!(persisted_errors.len(), 1);
        let error = &persisted_errors[0];
        assert_eq!(
            error.get("error_kind").and_then(Value::as_str),
            Some(SINK_JSON_OBJECT_TOO_LARGE)
        );
        assert_eq!(
            error.get("source_file").and_then(Value::as_str),
            Some("/tmp/source-a.jsonl")
        );
        assert_eq!(error.get("source_line_no").and_then(Value::as_u64), Some(7));
        assert_eq!(
            error.get("source_offset").and_then(Value::as_u64),
            Some(2048)
        );
        assert!(error
            .get("error_text")
            .and_then(Value::as_str)
            .is_some_and(|text| text.contains("Size of JSON object")));
        let raw_fragment = error
            .get("raw_fragment")
            .and_then(Value::as_str)
            .expect("fallback error should carry a compact fragment");
        assert!(raw_fragment.contains("sink_stage=raw_events"));
        assert!(raw_fragment.contains("source_scope=row"));
        assert!(raw_fragment.contains("original_payload_omitted=true"));
        assert!(
            !raw_fragment.contains("payload-marker"),
            "fallback error must not copy the original row payload"
        );
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        let last_error = metrics
            .last_error
            .lock()
            .expect("metrics last_error mutex poisoned")
            .clone();
        assert!(last_error.contains("non-retryable sink flush failure"));
        assert!(last_error.contains("raw_events"));

        let state = checkpoints.read().await;
        assert!(
            state.contains_key(&checkpoint_key_value),
            "the fallback should advance the checkpoint after recording the compact sink error"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_keeps_non_retryable_batch_when_error_record_fails() {
        let mock_state = MockClickHouseState::default();
        mock_state.fail_permanently("raw_events", CLICKHOUSE_OVERSIZED_JSON_OBJECT_ERROR);
        mock_state.fail_permanently("ingest_errors", "intentional ingest_errors failure");
        let (clickhouse, mock_state) = spawn_mock_clickhouse_with_state(mock_state).await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let checkpoint = sample_checkpoint();
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value, checkpoint)]);
        let mut raw_rows = vec![json!({
            "source_name": "source-a",
            "source_file": "/tmp/source-a.jsonl",
            "source_line_no": 7u64,
            "source_offset": 2048u64,
            "record_ts": "2026-02-17T00:00:01.000Z",
            "top_type": "response_item",
            "session_id": "session-a",
            "raw_json": "{}",
            "raw_json_hash": 1u64,
            "event_uid": "evt-error-record-fails",
        })];
        let mut event_rows = vec![json!({"event_uid": "evt-error-record-fails"})];
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            !flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "fallback should remain retryable when the compact error row cannot be written"
        );

        assert_eq!(raw_rows.len(), 1);
        assert_eq!(event_rows.len(), 1);
        assert_eq!(checkpoint_updates.len(), 1);
        assert_eq!(mock_state.call_count("ingest_errors"), 1);
        assert_eq!(mock_state.call_count("ingest_checkpoints"), 0);
        assert!(mock_state.rows("ingest_errors").is_empty());
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        let last_error = metrics
            .last_error
            .lock()
            .expect("metrics last_error mutex poisoned")
            .clone();
        assert!(last_error.contains("failed to record non-retryable sink error"));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_checkpoints_after_later_stage_non_retryable_fallback() {
        let (clickhouse, mock_state) =
            spawn_mock_clickhouse_with_state(MockClickHouseState::with_permanent_failure(
                "events",
                CLICKHOUSE_OVERSIZED_JSON_OBJECT_ERROR,
            ))
            .await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let checkpoint = sample_checkpoint();
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value.clone(), checkpoint)]);
        let mut raw_rows = vec![json!({
            "record_ts": "2026-02-17T00:00:01.000Z",
            "event_uid": "evt-later-stage",
            "raw_json": "{}",
        })];
        let mut event_rows = vec![json!({
            "event_uid": "evt-later-stage",
            "source_name": "source-a",
            "source_file": "/tmp/source-a.jsonl",
            "source_line_no": 17u64,
            "source_offset": 4096u64,
        })];
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "non-retryable fallback after a partial commit should finish by checkpointing"
        );

        assert!(raw_rows.is_empty());
        assert!(event_rows.is_empty());
        assert!(checkpoint_updates.is_empty());
        assert_eq!(mock_state.rows("raw_events").len(), 1);
        assert!(mock_state.rows("events").is_empty());
        assert_eq!(mock_state.rows("ingest_errors").len(), 1);
        assert_eq!(mock_state.rows("ingest_checkpoints").len(), 1);
        assert_eq!(metrics.raw_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.event_rows_written.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        let error = &mock_state.rows("ingest_errors")[0];
        assert_eq!(
            error.get("source_line_no").and_then(Value::as_u64),
            Some(17)
        );
        assert!(error
            .get("raw_fragment")
            .and_then(Value::as_str)
            .is_some_and(|fragment| fragment.contains("source_scope=row")
                && fragment.contains("sink_stage=events")));

        let state = checkpoints.read().await;
        assert!(
            state.contains_key(&checkpoint_key_value),
            "checkpoint cache should advance after later-stage fallback"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_uses_checkpoint_source_for_multi_row_non_retryable_fallback() {
        let (clickhouse, mock_state) =
            spawn_mock_clickhouse_with_state(MockClickHouseState::with_permanent_failure(
                "raw_events",
                CLICKHOUSE_OVERSIZED_JSON_OBJECT_ERROR,
            ))
            .await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let mut checkpoint = sample_checkpoint();
        checkpoint.last_line_no = 99;
        checkpoint.last_offset = 8192;
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value, checkpoint)]);
        let mut raw_rows = vec![
            json!({
                "source_name": "source-a",
                "source_file": "/tmp/source-a.jsonl",
                "source_line_no": 7u64,
                "source_offset": 2048u64,
                "record_ts": "2026-02-17T00:00:01.000Z",
                "event_uid": "evt-first",
                "raw_json": "{}",
            }),
            json!({
                "source_name": "source-a",
                "source_file": "/tmp/source-a.jsonl",
                "source_line_no": 8u64,
                "source_offset": 4096u64,
                "record_ts": "2026-02-17T00:00:02.000Z",
                "event_uid": "evt-second",
                "raw_json": "{}",
            }),
        ];
        let mut event_rows = Vec::<Value>::new();
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "multi-row non-retryable fallback should be batch-handled"
        );

        let persisted_errors = mock_state.rows("ingest_errors");
        assert_eq!(persisted_errors.len(), 1);
        let error = &persisted_errors[0];
        assert_eq!(
            error.get("source_line_no").and_then(Value::as_u64),
            Some(99)
        );
        assert_eq!(
            error.get("source_offset").and_then(Value::as_u64),
            Some(8192)
        );
        assert!(error
            .get("raw_fragment")
            .and_then(Value::as_str)
            .is_some_and(|fragment| fragment.contains("source_scope=checkpoint")
                && fragment.contains("stage_pending_rows=2")));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_retries_other_code_117_failures() {
        let (clickhouse, mock_state) =
            spawn_mock_clickhouse_with_state(MockClickHouseState::with_permanent_failure(
                "raw_events",
                "Code: 117. DB::Exception: Unknown field found while parsing JSONEachRow: \
                 unexpected_column",
            ))
            .await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let checkpoint = sample_checkpoint();
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value, checkpoint)]);
        let mut raw_rows = vec![json!({
            "source_name": "source-a",
            "harness": "codex",
            "source_file": "/tmp/source-a.jsonl",
            "source_inode": 42u64,
            "source_generation": 1u32,
            "source_line_no": 7u64,
            "source_offset": 2048u64,
            "record_ts": "2026-02-17T00:00:01.000Z",
            "top_type": "response_item",
            "session_id": "session-a",
            "raw_json": "{}",
            "raw_json_hash": 1u64,
            "event_uid": "evt-retry",
        })];
        let mut event_rows = vec![json!({"event_uid": "evt-retry"})];
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            !flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "other ClickHouse Code 117 failures should remain retryable"
        );

        assert_eq!(raw_rows.len(), 1);
        assert_eq!(event_rows.len(), 1);
        assert_eq!(checkpoint_updates.len(), 1);
        assert!(mock_state.rows("ingest_errors").is_empty());
        assert_eq!(mock_state.call_count("ingest_errors"), 0);
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.flush_failures.load(Ordering::Relaxed), 1);

        let last_error = metrics
            .last_error
            .lock()
            .expect("metrics last_error mutex poisoned")
            .clone();
        assert!(last_error.contains("Unknown field"));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_quarantines_oversized_raw_rows() {
        let (clickhouse, mock_state) = spawn_mock_clickhouse("unused-table").await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let checkpoint = sample_checkpoint();
        let checkpoint_key_value = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let mut checkpoint_updates = HashMap::from([(checkpoint_key_value.clone(), checkpoint)]);
        let mut raw_rows = vec![json!({
            "source_name": "source-a",
            "harness": "codex",
            "source_file": "/tmp/source-a.jsonl",
            "source_inode": 42u64,
            "source_generation": 1u32,
            "source_line_no": 7u64,
            "source_offset": 2048u64,
            "record_ts": "2026-02-17T00:00:01.000Z",
            "top_type": "response_item",
            "session_id": "session-a",
            "raw_json": "x".repeat(CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES),
            "raw_json_hash": 1u64,
            "event_uid": "evt-oversized",
        })];
        let mut event_rows = Vec::<Value>::new();
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        assert!(
            flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "oversized rows should be quarantined without failing the flush"
        );

        assert!(
            mock_state.rows("raw_events").is_empty(),
            "oversized raw row must not be sent to raw_events"
        );
        let error_rows = mock_state.rows("ingest_errors");
        assert_eq!(error_rows.len(), 1, "one compact ingest error is written");
        let error = &error_rows[0];
        assert_eq!(
            error.get("error_kind").and_then(Value::as_str),
            Some("oversized_json_row")
        );
        assert_eq!(
            error.get("source_file").and_then(Value::as_str),
            Some("/tmp/source-a.jsonl")
        );
        assert_eq!(error.get("source_line_no").and_then(Value::as_u64), Some(7));
        assert_eq!(
            error.get("source_offset").and_then(Value::as_u64),
            Some(2048)
        );
        assert_eq!(
            error
                .get("raw_fragment")
                .and_then(Value::as_str)
                .map(str::len),
            Some(OVERSIZED_ROW_FRAGMENT_CHARS)
        );
        assert_eq!(mock_state.rows("ingest_checkpoints").len(), 1);
        assert!(checkpoint_updates.is_empty());
        assert_eq!(metrics.raw_rows_written.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 1);

        let state = checkpoints.read().await;
        assert!(
            state.contains_key(&checkpoint_key_value),
            "checkpoint cache should advance after quarantining the row"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_quarantines_oversized_tool_rows() {
        let (clickhouse, mock_state) = spawn_mock_clickhouse("unused-table").await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());

        let mut raw_rows = Vec::<Value>::new();
        let mut event_rows = Vec::<Value>::new();
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = vec![json!({
            "event_uid": "evt-tool",
            "session_id": "session-a",
            "harness": "codex",
            "source_name": "source-a",
            "tool_call_id": "call-1",
            "tool_name": "shell",
            "tool_phase": "response",
            "output_json": "x".repeat(CLICKHOUSE_JSON_EACH_ROW_OBJECT_MAX_BYTES),
            "source_ref": "/tmp/source-a.jsonl:3:17",
            "event_version": 1u64,
        })];
        let mut error_rows = Vec::<Value>::new();
        let mut checkpoint_updates = HashMap::new();

        assert!(
            flush_pending(
                &clickhouse,
                &checkpoints,
                &metrics,
                &mut raw_rows,
                &mut event_rows,
                &mut link_rows,
                &mut tool_rows,
                &mut error_rows,
                &mut checkpoint_updates,
                true,
                "",
            )
            .await,
            "oversized tool rows should be quarantined without failing the flush"
        );

        assert!(
            mock_state.rows("tool_io").is_empty(),
            "oversized tool row must not be sent to tool_io"
        );
        let error_rows = mock_state.rows("ingest_errors");
        assert_eq!(error_rows.len(), 1, "one compact ingest error is written");
        let error = &error_rows[0];
        assert_eq!(
            error.get("source_file").and_then(Value::as_str),
            Some("/tmp/source-a.jsonl")
        );
        assert_eq!(
            error.get("source_generation").and_then(Value::as_u64),
            Some(3)
        );
        assert_eq!(
            error.get("source_line_no").and_then(Value::as_u64),
            Some(17)
        );
        assert_eq!(error.get("source_inode").and_then(Value::as_u64), Some(0));
        assert_eq!(metrics.err_rows_written.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn backend_sink_filters_intake_to_routed_sessions() {
        let (clickhouse, mock_state) = spawn_mock_clickhouse("unused-table").await;

        let mut route_config = moraine_config::AppConfig::default();
        route_config.backends.insert(
            "team-ch".to_string(),
            moraine_config::ClickHouseConfig::default(),
        );
        route_config.routes.push(moraine_config::RouteConfig {
            dir: "/work/team/**".to_string(),
            backend: "team-ch".to_string(),
            mode: moraine_config::ROUTE_MODE_MIRROR.to_string(),
        });
        let resolver: SharedRouteResolver = Arc::new(Mutex::new(crate::tee::RouteResolver::new(
            Arc::new(route_config),
        )));

        let mut config = moraine_config::AppConfig::default();
        config.ingest.heartbeat_interval_seconds = 60.0;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());
        let redactions = test_redaction_audit();
        let (tx, rx) = mpsc::channel::<SinkMessage>(8);

        let handle = spawn_sink_task(
            config,
            clickhouse,
            checkpoints,
            metrics,
            rx,
            true,
            SinkRole::Backend {
                cell: Arc::new(BackendSinkCell::new("team-ch")),
                resolver,
                replay_notify: Arc::new(Notify::new()),
                replay_floor: Arc::new(Mutex::new(None)),
                redactor: test_redactor(),
                redactions: redactions.clone(),
            },
        );

        let raw_secret = "ghp_abcdefghijklmnopqrstuvwxyzABCDE12345";
        let mut batch = RowBatch::default();
        batch.push_raw_row(json!({
            "session_id": "team-session",
            "cwd": "/work/team/project",
            "record_ts": "2026-02-17T00:00:01.000Z",
            "raw_json": format!("{{\"token\":\"{raw_secret}\"}}"),
            "raw_json_hash": 1u64,
        }));
        batch.push_raw_row(json!({
            "session_id": "other-session",
            "cwd": "/home/other",
            "record_ts": "2026-02-17T00:00:01.000Z",
            "raw_json": format!("{{\"token\":\"{raw_secret}\"}}"),
            "raw_json_hash": 1u64,
        }));
        batch.checkpoint = Some(sample_checkpoint());
        tx.send(SinkMessage::Batch(batch))
            .await
            .expect("send mixed batch");
        drop(tx); // close the channel so the final flush runs deterministically

        timeout(Duration::from_secs(5), handle)
            .await
            .expect("backend sink should finish")
            .expect("backend sink task should not panic");

        let raw_rows = mock_state.rows("raw_events");
        assert_eq!(raw_rows.len(), 1, "only the routed session's rows flush");
        assert_eq!(
            raw_rows[0].get("session_id").and_then(Value::as_str),
            Some("team-session")
        );
        assert_eq!(
            raw_rows[0].get("raw_json").and_then(Value::as_str),
            Some("{\"token\":\"[REDACTED:github-token]\"}"),
            "backend sink backstop redacts the remote copy after route filtering"
        );
        assert_eq!(
            redactions.snapshot().get("github-token"),
            Some(&1),
            "backend redactions are counted without secret values"
        );
        let checkpoint_rows = mock_state.rows("ingest_checkpoints");
        assert_eq!(
            checkpoint_rows.len(),
            1,
            "the checkpoint lands in the backend's own database even though \
             part of the batch was filtered away"
        );
        assert_eq!(
            checkpoint_rows[0].get("host").and_then(Value::as_str),
            Some(host_name().as_str()),
            "backend checkpoint rows are host-scoped: team backends share one \
             ingest_checkpoints table across members"
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn flush_pending_map_merge_is_offset_monotone() {
        let (clickhouse, mock_state) = spawn_mock_clickhouse("unused-table").await;
        let checkpoints = Arc::new(RwLock::new(HashMap::new()));
        let metrics = Arc::new(Metrics::default());
        let mut raw_rows = Vec::<Value>::new();
        let mut event_rows = Vec::<Value>::new();
        let mut link_rows = Vec::<Value>::new();
        let mut tool_rows = Vec::<Value>::new();
        let mut error_rows = Vec::<Value>::new();

        let ahead = sample_checkpoint(); // offset 100
        let key = checkpoint_key(&ahead.source_name, &ahead.source_file);
        let mut behind = sample_checkpoint();
        behind.last_offset = 50;

        for cp in [ahead, behind] {
            let mut checkpoint_updates = HashMap::new();
            checkpoint_updates.insert(key.clone(), cp);
            assert!(
                flush_pending(
                    &clickhouse,
                    &checkpoints,
                    &metrics,
                    &mut raw_rows,
                    &mut event_rows,
                    &mut link_rows,
                    &mut tool_rows,
                    &mut error_rows,
                    &mut checkpoint_updates,
                    true,
                    "",
                )
                .await
            );
        }

        // A backend map has two producers (live router batches and replay),
        // so an out-of-order flush must never rewind the committed offset —
        // the replay pass would otherwise re-trust a stale floor.
        let state = checkpoints.read().await;
        assert_eq!(
            state.get(&key).map(|cp| cp.last_offset),
            Some(100),
            "a lower-offset flush must not regress the committed map"
        );
        assert!(
            mock_state.rows("ingest_checkpoints").len() == 2,
            "both checkpoint rows still flush; only the map merge is monotone"
        );
        let default_role_rows = mock_state.rows("ingest_checkpoints");
        assert!(
            default_role_rows
                .iter()
                .all(|row| row.get("host").is_none()),
            "an empty checkpoint host (default role) must omit the column"
        );
    }

    #[tokio::test]
    async fn recovery_captures_the_replay_floor_before_post_recovery_flushes() {
        let cell = Arc::new(BackendSinkCell::new("team-ch"));
        cell.set_status(crate::tee::BackendSinkStatus::Lagging);
        let replay_notify = Arc::new(Notify::new());
        let replay_floor: ReplayFloor = Arc::new(Mutex::new(None));
        let role = SinkRole::Backend {
            cell,
            resolver: Arc::new(Mutex::new(crate::tee::RouteResolver::new(Arc::new(
                moraine_config::AppConfig::default(),
            )))),
            replay_notify: replay_notify.clone(),
            replay_floor: replay_floor.clone(),
            redactor: test_redactor(),
            redactions: test_redaction_audit(),
        };

        let checkpoint = sample_checkpoint(); // offset 100
        let key = checkpoint_key(&checkpoint.source_name, &checkpoint.source_file);
        let checkpoints = Arc::new(RwLock::new(HashMap::from([(
            key.clone(),
            checkpoint.clone(),
        )])));

        note_flush_outcome(&role, &checkpoints, true, true).await;

        // A live batch flushed after recovery jumps the map past the outage
        // gap; the floor captured at the transition must not move with it.
        {
            let mut state = checkpoints.write().await;
            let entry = state.get_mut(&key).expect("checkpoint present");
            entry.last_offset = 900;
        }

        let floor = replay_floor
            .lock()
            .expect("replay floor mutex poisoned")
            .take()
            .expect("recovery must capture a replay floor");
        assert_eq!(
            floor.get(&key).map(|cp| cp.last_offset),
            Some(100),
            "the floor reflects what the backend had flushed at recovery"
        );
        assert!(
            timeout(Duration::from_millis(100), replay_notify.notified())
                .await
                .is_ok(),
            "recovery must schedule a replay pass"
        );
    }

    #[test]
    fn compute_append_to_visible_stats_uses_real_record_timestamps() {
        let visible_at = DateTime::parse_from_rfc3339("2026-02-17T00:00:10.000Z")
            .expect("valid timestamp")
            .with_timezone(&Utc);
        let raw_rows = vec![
            json!({"record_ts": "2026-02-17T00:00:00.000Z"}),
            json!({"record_ts": "2026-02-17T00:00:05.000Z"}),
            json!({"record_ts": "2026-02-17T00:00:09.000Z"}),
        ];

        let (p50, p95) = compute_append_to_visible_stats(&raw_rows, visible_at)
            .expect("expected percentile stats");

        assert_eq!(p50, 5_000);
        assert_eq!(p95, 10_000);
    }

    #[test]
    fn compute_append_to_visible_stats_returns_none_for_unparseable_rows() {
        let visible_at = DateTime::parse_from_rfc3339("2026-02-17T00:00:10.000Z")
            .expect("valid timestamp")
            .with_timezone(&Utc);
        let raw_rows = vec![
            json!({"record_ts": "not-a-timestamp"}),
            json!({"record_ts": ""}),
            json!({}),
        ];

        let stats = compute_append_to_visible_stats(&raw_rows, visible_at);
        assert!(stats.is_none());
    }

    #[test]
    fn compute_append_to_visible_stats_clamps_future_timestamps_to_zero() {
        let visible_at = DateTime::parse_from_rfc3339("2026-02-17T00:00:10.000Z")
            .expect("valid timestamp")
            .with_timezone(&Utc);
        let raw_rows = vec![json!({"record_ts": "2026-02-17T00:00:20.000Z"})];

        let (p50, p95) = compute_append_to_visible_stats(&raw_rows, visible_at)
            .expect("expected percentile stats");

        assert_eq!(p50, 0);
        assert_eq!(p95, 0);
    }

    #[test]
    fn duration_from_config_seconds_clamps_to_minimum() {
        let duration = duration_from_config_seconds(0.001, 0.05, "ingest.flush_interval_seconds");
        assert_eq!(duration, Duration::from_millis(50));
    }

    #[test]
    fn duration_from_config_seconds_handles_non_finite_values() {
        let nan = duration_from_config_seconds(f64::NAN, 0.05, "ingest.flush_interval_seconds");
        let pos_inf =
            duration_from_config_seconds(f64::INFINITY, 0.05, "ingest.flush_interval_seconds");
        let neg_inf =
            duration_from_config_seconds(f64::NEG_INFINITY, 0.05, "ingest.flush_interval_seconds");

        assert_eq!(nan, Duration::from_millis(50));
        assert_eq!(pos_inf, Duration::from_millis(50));
        assert_eq!(neg_inf, Duration::from_millis(50));
    }
}
