//! Neuro-symbolic data dictionary: parsing `stats` / `frequency` CSVs into
//! records, deterministically generating dictionary entries, and merging them
//! with the LLM-produced Label + Description + Content Type.
//!
//! Fields on these structs are `pub(super)` so sibling submodules (e.g.
//! `formatters`) can read them; all types themselves remain crate-private.

use foldhash::{HashMap, HashMapExt, HashSet};
use indexmap::IndexMap;
use serde::{Deserialize, Serialize};

use super::{CliError, CliResult, extract_json_from_output};

/// Curated, documented vocabulary of semantic Content Type tokens. Each token is
/// intended to map cleanly to a `fake-rs` faker for a future `synthesize` command.
///
/// Primitive numeric/boolean types (`integer`, `decimal`, `boolean`) are
/// deliberately excluded — they are redundant with the dictionary's deterministic
/// `type` column. `synthesize` falls back to `type` + `min`/`max` for plain
/// numeric fields whose `content_type` is `unknown`.
///
/// `time` (time-of-day, e.g. `HH:MM:SS`) and `duration` (elapsed time) ARE
/// included because qsv's stats reports them as `String`, so the deterministic
/// `type` column doesn't cover them; without these tokens `synthesize` would
/// fall through to lorem text for fields that are clearly temporal. They map
/// to `fake::faker::time::en::Time` and `Duration` respectively.
///
/// `date` and `datetime` are stamped DETERMINISTICALLY by
/// `generate_code_based_dictionary` from the stats `Type` column (`Date` /
/// `DateTime`); the LLM never classifies them. They carry an optional `:<fmt>`
/// suffix holding an LLM-inferred chrono strftime format (e.g.
/// `"datetime:%m/%d/%Y %I:%M:%S %p"`) describing the column's on-disk format.
/// The suffix is syntactically validated by `normalize_datetime_token`,
/// semantically validated against real frequency samples by
/// `validate_date_formats`, and consumed by
/// `synthesize::faker_map::parse_date_format`. A bare `date`/`datetime` falls
/// back to `synthesize`'s hardcoded `%Y-%m-%d` / RFC3339 output.
///
/// `downgrade_all_midnight_datetime_columns` reclassifies a `datetime` column
/// as `date` (stripping the time specifiers from `<fmt>`) when every
/// frequency-sampled value falls on midnight — `stats` reports `DateTime`
/// whenever any single value carries a time, which over-reports columns that
/// are plainly dates stored with a zero time-of-day.
///
/// `duration` accepts an optional `:N` suffix carrying an LLM-inferred
/// upper bound in seconds (e.g. `"duration:3600"` for an hour cap). The
/// suffix is normalized by `normalize_duration_token` and consumed by
/// `synthesize::faker_map::parse_duration_cap`. Bare `"duration"` falls
/// back to a 24-hour default at generation time.
///
/// `unique_id` marks fields where every row has a distinct non-null value
/// (i.e. stats `cardinality == rowcount`, no nulls — primary keys, surrogate
/// keys, sequence numbers). It is set DETERMINISTICALLY by
/// `generate_code_based_dictionary` via a structural check on the field's
/// frequency table (single row whose `count == cardinality`), independent of
/// the literal sentinel text, and overrides any token the LLM produced for
/// that field. The vocabulary entry is still exposed to the LLM so its
/// classification stays consistent, but the deterministic check is
/// authoritative — and LLM-supplied `unique_id` is rejected by
/// `parse_llm_dictionary_response` and `combine_dictionary_entries` to keep
/// the contract one-way.
///
/// NOTE on spellings:
/// - `license_plate` uses the AMERICAN spelling in the vocab (the user-facing surface), but maps to
///   fake-rs's `automotive::LicencePlate` (British) at the faker layer in `synthesize::faker_map`.
///   The vocab spelling is what the LLM sees and what users author by hand.
/// - `ipv6_address` mirrors the existing `ip_address` (which is IPv4 today); keeping the explicit
///   version in the token avoids ambiguity for the LLM.
pub(crate) const CONTENT_TYPE_VOCAB: &[&str] = &[
    // person / identity
    "first_name",
    "last_name",
    "full_name",
    "username",
    "password",
    "email",
    "phone",
    // address / location
    "street_address",
    "street_name",
    "building_number",
    "secondary_address",
    "city",
    "state",
    "state_abbr",
    "zip_code",
    "country",
    "country_code",
    "latitude",
    "longitude",
    "time_zone",
    // company / job
    "company_name",
    "industry",
    "job_title",
    "profession",
    // identifiers / technical
    "unique_id",
    "uuid",
    "credit_card",
    "currency_code",
    "isbn",
    "ip_address",
    "ipv6_address",
    "mac_address",
    "url",
    "user_agent",
    "file_name",
    "file_path",
    "mime_type",
    "color_hex",
    "license_plate",
    // temporal (`date`/`datetime` carry an optional LLM-inferred chrono
    // strftime `:<fmt>` suffix and are stamped deterministically from the
    // stats `Type` column - see `normalize_datetime_token`)
    "date",
    "datetime",
    "time",
    "duration",
    // generic / fallback
    "category",
    "lorem_word",
    "lorem_sentence",
    "lorem_paragraph",
    "free_text",
    "unknown",
];

/// Render `CONTENT_TYPE_VOCAB` as a comma-separated string for prompt injection.
pub(super) fn content_type_vocab_list() -> String {
    CONTENT_TYPE_VOCAB.join(", ")
}

/// Closed set of analytical column `role` tokens. Unlike `content_type` (which
/// describes the *kind of value* for synthesis/PII) and `concept` (which gives a
/// *cross-dataset semantic identity* for joining), `role` describes how an agent
/// should USE the column in a query: `dimension` (group/filter by it), `measure`
/// (aggregate it), `identifier` (a key naming an entity), or `timestamp` (a
/// date/time axis). `identifier` and `timestamp` are stamped deterministically
/// (from `is_unique_id` and the stats `Date`/`DateTime` type); `dimension` vs
/// `measure` is the LLM's call, falling back to a type-based default.
pub(crate) const ROLE_VOCAB: &[&str] = &["dimension", "measure", "identifier", "timestamp"];

/// Catalog-wide, namespaced `concept` vocabulary. A concept is the column's
/// real-world semantic identity; two columns in *different* datasets that carry
/// the SAME concept ID denote the same thing and are join-compatible. This is the
/// mechanism by which an agent correlates datasets across a catalog (decision:
/// "shared concept vocabulary only" — no explicit cross-dataset foreign keys).
///
/// Concept is distinct from `content_type`: `content_type` drives `synthesize`
/// (fake-rs fakers) and PII handling; `concept` drives catalog join discovery and
/// is hierarchical/namespaced. Many concepts are SEEDED deterministically from
/// `content_type` (see `concept_from_content_type`) so a `--infer-content-type`
/// run gets concepts for free; the LLM fills/refines the rest from this list.
///
/// Namespaces:
///   * `geo.*`   — spatial identity (join keys for places)
///   * `time.*`  — temporal axes
///   * `id.*`    — entity keys (`id.surrogate_key` is deterministic, tied to `is_unique_id`)
///   * `org.*`   — organizations
///   * `pii.*`   — sensitive personal data (drives the PII quality flag; not a join target)
///   * `measure.*`, `category.*` — quantities / categoricals (not join targets)
///   * domain prefixes (e.g. `nyc.*`) — catalog-specific shared keys
pub(crate) const CONCEPT_VOCAB: &[&str] = &[
    // spatial
    "geo.zip_code",
    "geo.city",
    "geo.county",
    "geo.county_fips",
    "geo.state",
    "geo.state_fips",
    "geo.country",
    "geo.latitude",
    "geo.longitude",
    "geo.coordinate_pair",
    "geo.street_address",
    "geo.census_tract",
    "geo.crs_stateplane_x",
    "geo.crs_stateplane_y",
    // temporal
    "time.event_timestamp",
    "time.created_at",
    "time.closed_at",
    "time.updated_at",
    "time.due_at",
    "time.date",
    "time.duration",
    // identifiers
    "id.surrogate_key",
    "id.natural_key",
    "id.foreign_key",
    "id.uuid",
    // organizations
    "org.agency",
    "org.company",
    "org.industry",
    // sensitive personal data
    "pii.email",
    "pii.phone",
    "pii.full_name",
    "pii.address",
    // quantities / categoricals
    "measure.count",
    "measure.amount",
    "measure.ratio",
    "category.status",
    "category.type",
    "category.channel",
    // NYC-domain extension (illustrative shared catalog keys)
    "nyc.bbl",
    "nyc.borough",
    "nyc.community_board",
    "nyc.complaint_type",
    // fallback
    "unknown",
];

/// Controlled vocabulary for the dataset-level temporal cadence token
/// (`x-qsv.cadence`). Unlike `CONCEPT_VOCAB`, this is NOT injected into any LLM
/// prompt: cadence is computed deterministically from cached stats
/// (`formatters::infer_cadence`) — a machine token, never prose (the PR #4321
/// discipline). The const exists so producers (`infer_cadence`) and consumers
/// (`viz smart`'s trend-bucket floor, issue #4216) share one authoritative
/// token list; a unit test pins `infer_cadence`'s outputs to it. The sibling
/// vocabulary in `qsv profile` is dcat-us-v3.yaml's accrualPeriodicity slugs.
pub(super) const CADENCE_VOCAB: &[&str] = &["daily", "weekly", "monthly", "quarterly", "annual"];

/// Render `CONCEPT_VOCAB` as a comma-separated string for prompt injection.
pub(super) fn concept_vocab_list() -> String {
    CONCEPT_VOCAB.join(", ")
}

/// Render `ROLE_VOCAB` as a comma-separated string for prompt injection.
pub(super) fn role_vocab_list() -> String {
    ROLE_VOCAB.join(", ")
}

/// Concept namespaces that denote a shared real-world entity an agent can join
/// ON across datasets. `pii.*`, `measure.*`, `category.*`, and `unknown` are
/// excluded: they are sensitive, additive, or not stable identities.
const LINKABLE_CONCEPT_PREFIXES: &[&str] = &["geo.", "time.", "id.", "org.", "nyc."];

/// True when `concept` names a cross-dataset-joinable identity (drives the
/// schema-table `Join?` hint and the per-column join line).
pub(super) fn is_linkable_concept(concept: &str) -> bool {
    !concept.is_empty()
        && concept != "unknown"
        && LINKABLE_CONCEPT_PREFIXES
            .iter()
            .any(|p| concept.starts_with(p))
}

/// Deterministic seed mapping from a (base) `content_type` token to a catalog
/// `concept`, so a `--infer-content-type` run gets concepts without extra LLM
/// work. Returns `None` for tokens with no obvious 1:1 concept (the LLM fills
/// those). Caller passes the bare token (no `:<fmt>`/`:N` suffix).
pub(super) fn concept_from_content_type(content_type_base: &str) -> Option<&'static str> {
    Some(match content_type_base {
        "unique_id" => "id.surrogate_key",
        "uuid" => "id.uuid",
        "zip_code" => "geo.zip_code",
        "city" => "geo.city",
        "state" | "state_abbr" => "geo.state",
        "country" | "country_code" => "geo.country",
        "latitude" => "geo.latitude",
        "longitude" => "geo.longitude",
        "street_address" | "street_name" => "geo.street_address",
        "email" => "pii.email",
        "phone" => "pii.phone",
        "first_name" | "last_name" | "full_name" => "pii.full_name",
        "company_name" => "org.company",
        "industry" => "org.industry",
        "date" => "time.date",
        "datetime" => "time.event_timestamp",
        _ => return None,
    })
}

/// Temporal `content_type` base tokens whose `role` must be `timestamp` (a clock
/// time / calendar date is never a quantity to aggregate). `duration` is
/// deliberately excluded — `duration:N` is a magnitude you sum/average, so it is
/// measure-admissible. See issue #4177.
fn is_temporal_content_type(base: &str) -> bool {
    matches!(base, "date" | "datetime" | "time")
}

/// id-family `content_type` base tokens where a stray `role: measure` should be
/// corrected to `identifier` rather than `dimension` (these name an entity, they
/// are not categorical group-by attributes). See issue #4177.
fn is_identifier_content_type(base: &str) -> bool {
    matches!(
        base,
        "unique_id"
            | "uuid"
            | "isbn"
            | "credit_card"
            | "ip_address"
            | "ipv6_address"
            | "mac_address"
            | "license_plate"
    )
}

/// The `role` a `content_type` base token reliably implies, or `None` when it does
/// not constrain role at all. `unknown`, `duration`, and any out-of-vocabulary token
/// return `None` — numeric measures legitimately carry `content_type: unknown` and
/// `duration:N` is a magnitude you aggregate, so role/concept are left as inferred.
/// Anchors both the `role` coercion (#4177) and the `concept` reconciliation (#4176).
fn role_from_content_type(base: &str) -> Option<&'static str> {
    if is_temporal_content_type(base) {
        Some("timestamp")
    } else if is_identifier_content_type(base) {
        Some("identifier")
    } else if base != "unknown" && base != "duration" && CONTENT_TYPE_VOCAB.contains(&base) {
        Some("dimension")
    } else {
        None
    }
}

/// The `role` a catalog `concept`'s namespace implies (the segment before the first
/// `.`), or `None` for `unknown` and any namespace with no role mapping. Used to
/// detect a `concept` that contradicts a reliable `content_type` signal (#4176).
fn role_from_concept_namespace(concept: &str) -> Option<&'static str> {
    Some(match concept.split('.').next().unwrap_or_default() {
        "measure" => "measure",
        "time" => "timestamp",
        "id" => "identifier",
        "geo" | "org" | "pii" | "category" | "nyc" => "dimension",
        _ => return None,
    })
}

/// Merge an LLM-supplied `concept` onto the (possibly deterministically seeded)
/// `current` value. `id.surrogate_key` is deterministic (tied to `is_unique_id`)
/// and never overridden. Otherwise the LLM fills an empty slot, and on the
/// refine pass (`allow_override`) may replace a non-authoritative seed.
fn merge_concept(current: &str, llm: &str, allow_override: bool) -> String {
    if current == "id.surrogate_key" {
        return current.to_string();
    }
    if !llm.is_empty() && llm != "id.surrogate_key" && (current.is_empty() || allow_override) {
        return llm.to_string();
    }
    current.to_string()
}

/// Merge an LLM-supplied `role` onto the (possibly deterministically seeded)
/// `current` value. `identifier` and `timestamp` are stamped deterministically
/// and never overridden; `dimension`/`measure` come from the LLM (or the
/// type-based fallback applied in `combine_dictionary_entries`).
fn merge_role(current: &str, llm: &str, allow_override: bool) -> String {
    if current == "identifier" || current == "timestamp" {
        return current.to_string();
    }
    if ROLE_VOCAB.contains(&llm) && (current.is_empty() || allow_override) {
        return llm.to_string();
    }
    current.to_string()
}

/// Normalize the `duration` content type, optionally with an LLM-inferred
/// per-field upper-bound suffix.
///
/// Accepts:
///   * `"duration"` → returns `Some("duration")` (synthesize falls back to its default 24-hour cap)
///   * `"duration:N"` where N is a positive integer → returns `Some("duration:N")` (whitespace
///     around N is tolerated)
///   * `"duration:<malformed>"` (non-numeric / zero) → returns `Some("duration")` so a bad suffix
///     degrades gracefully to the unbounded form rather than dropping the classification entirely
///
/// Returns `None` for anything that isn't a duration token, so the caller can
/// fall back to the regular `CONTENT_TYPE_VOCAB` membership check.
///
/// Caller is responsible for lowercasing / outer-trimming the input first.
pub(super) fn normalize_duration_token(raw: &str) -> Option<String> {
    if raw == "duration" {
        return Some("duration".to_string());
    }
    let suffix = raw.strip_prefix("duration:")?;
    match suffix.trim().parse::<u64>() {
        Ok(n) if n > 0 => Some(format!("duration:{n}")),
        _ => Some("duration".to_string()),
    }
}

/// True if `fmt` is a syntactically valid chrono strftime format string, i.e.
/// it contains no unrecognized `%` specifiers. A plain literal with no `%` is
/// considered valid here; semantic mismatches are caught later by
/// `validate_date_formats` parsing a real sample value.
pub(crate) fn is_valid_strftime(fmt: &str) -> bool {
    use chrono::format::{Item, StrftimeItems};
    StrftimeItems::new(fmt).all(|item| !matches!(item, Item::Error))
}

/// Normalize a `date` / `datetime` content type, optionally carrying an
/// LLM-inferred chrono strftime format suffix.
///
/// Accepts:
///   * `"date"` / `"datetime"` → returns the bare token
///   * `"date:<fmt>"` / `"datetime:<fmt>"` → returns the token verbatim when `<fmt>` is a
///     syntactically valid chrono strftime string; an empty or malformed `<fmt>` degrades to the
///     bare token (mirroring how `normalize_duration_token` degrades a bad `:N`)
///
/// Returns `None` for anything that is not a date/datetime token, so the caller
/// can fall back to the regular `CONTENT_TYPE_VOCAB` membership check.
///
/// Unlike `normalize_duration_token`, the input must NOT be fully lowercased by
/// the caller: chrono strftime specifiers are case-sensitive (`%m` month vs
/// `%M` minute, `%p` AM/PM). Only the token prefix (before the first `:`) is
/// case-folded here; the `<fmt>` suffix is preserved verbatim. Splitting on the
/// first `:` keeps colons inside the format (e.g. `%H:%M:%S`) intact.
pub(super) fn normalize_datetime_token(raw: &str) -> Option<String> {
    let raw = raw.trim();
    let (head, fmt) = match raw.split_once(':') {
        Some((h, f)) => (h, Some(f)),
        None => (raw, None),
    };
    let token = match head.to_ascii_lowercase().as_str() {
        "date" => "date",
        "datetime" => "datetime",
        _ => return None,
    };
    match fmt {
        Some(fmt) if !fmt.is_empty() && is_valid_strftime(fmt) => Some(format!("{token}:{fmt}")),
        _ => Some(token.to_string()),
    }
}

/// Returns the bare-token prefix of a `content_type`, i.e. the part before the
/// first `:` (`"datetime:%F"` → `"datetime"`, `"duration:3600"` → `"duration"`).
/// Non-suffixed tokens are returned unchanged.
pub(super) fn content_type_base(token: &str) -> &str {
    token.split(':').next().unwrap_or(token)
}

/// Extract the strftime format suffix from a `date`/`datetime` `content_type`,
/// when present and syntactically valid (e.g. `"date:%m/%d/%Y"` →
/// `Some("%m/%d/%Y")`). Bare tokens and non-date content types yield `None`.
pub(crate) fn content_type_date_format(content_type: &str) -> Option<&str> {
    let fmt = content_type
        .strip_prefix("datetime:")
        .or_else(|| content_type.strip_prefix("date:"))?;
    (!fmt.is_empty() && is_valid_strftime(fmt)).then_some(fmt)
}

/// Reformat an (RFC 3339-normalized) date string `value` using the strftime
/// format carried by `content_type`, when present. Returns `value` unchanged
/// when there is no inferred format, the value is empty, or it cannot be parsed.
///
/// This mirrors the markdown dictionary template's `datefmt` filter so the
/// JSON/TOON/JSONSchema outputs present date Min/Max in the same inferred
/// format as the markdown dictionary, instead of qsv's normalized RFC 3339.
pub(super) fn format_date_value<'a>(
    content_type: &str,
    value: &'a str,
) -> std::borrow::Cow<'a, str> {
    use std::borrow::Cow;
    if value.is_empty() {
        return Cow::Borrowed(value);
    }
    let Some(fmt) = content_type_date_format(content_type) else {
        return Cow::Borrowed(value);
    };
    // Disambiguate the parse using the inferred format's own field order: when
    // `%d` precedes `%m` the column is day-first, so an ambiguous value like
    // "01/02/2020" under `date:%d/%m/%Y` parses as 1 Feb (not 2 Jan) and
    // round-trips unchanged instead of being silently swapped. Min/Max are
    // RFC3339 (preference-invariant), so this only matters for raw Examples.
    match qsv_dateparser::parse_with_preference(
        value,
        strftime_dmy_preference(fmt).unwrap_or(false),
    ) {
        Ok(dt) => Cow::Owned(dt.format(fmt).to_string()),
        Err(_) => Cow::Borrowed(value),
    }
}

/// The DMY-vs-MDY preference a strftime format expresses, as understood by
/// `qsv_dateparser::parse_with_preference`:
///
/// - `Some(true)`  — day-first: the numeric day field precedes the numeric month.
/// - `Some(false)` — month-first: the numeric month precedes the numeric day, in a layout the
///   preference bit can actually act on.
/// - `None`        — the format expresses no *actionable* preference, so callers should fall back
///   to their own default (e.g. `QSV_PREFER_DMY`).
///
/// `None` covers two cases: formats with no numeric day and/or month (`%H:%M:%S`,
/// `%b %d, %Y`), and **year-leading** layouts (`%Y-%m-%d`, `%F`, `%y-%m-%d`).
/// The latter matters because `qsv_dateparser` only consults the preference bit for
/// slash-separated `d/m` values (its `slash_mdy_family`, regex `^\d{1,2}/\d{1,2}`);
/// year-leading values are read unambiguously regardless, so claiming "month-first"
/// there would override a caller's default for no reason.
///
/// Parses the format with chrono's `StrftimeItems` so every day/month padding
/// variant (`%d`, `%e`, `%-d`, `%0d`, `%_d`, `%m`, `%-m`, `%0m`, `%_m`, …) and every
/// compound specifier (`%F` → `%Y-%m-%d`, `%D` → `%m/%d/%y`) is recognized.
pub(crate) fn strftime_dmy_preference(fmt: &str) -> Option<bool> {
    use chrono::format::{Item, Numeric, StrftimeItems};
    let mut day_idx = None;
    let mut month_idx = None;
    let mut year_idx = None;
    for (i, item) in StrftimeItems::new(fmt).enumerate() {
        match item {
            Item::Numeric(Numeric::Day, _) if day_idx.is_none() => day_idx = Some(i),
            Item::Numeric(Numeric::Month, _) if month_idx.is_none() => month_idx = Some(i),
            Item::Numeric(Numeric::Year | Numeric::YearMod100 | Numeric::IsoYear, _)
                if year_idx.is_none() =>
            {
                year_idx = Some(i);
            },
            _ => {},
        }
    }
    let (d, m) = (day_idx?, month_idx?);
    if d < m {
        return Some(true);
    }
    // month-first, but a leading year makes the value unambiguous to the parser.
    // Keep this check gated on `m < d` so `%Y/%d/%m` stays day-first.
    if year_idx.is_some_and(|y| y < m) {
        return None;
    }
    Some(false)
}

/// Reformat the value part of each `"value [count]"` example line to the
/// `content_type`'s inferred date format, so the Examples column reads
/// consistently with the date-formatted Min/Max. Passthrough when
/// `content_type` carries no date format and on the `<ALL_UNIQUE>` sentinel.
/// The trailing ` [count]` suffix is preserved verbatim, and values that
/// cannot be parsed (frequency aggregation buckets like `Other…`/`(NULL)…`,
/// truncated `value…` entries) are left unchanged by `format_date_value`.
pub(super) fn format_date_examples(content_type: &str, examples: &str) -> String {
    if examples == "<ALL_UNIQUE>" || content_type_date_format(content_type).is_none() {
        return examples.to_string();
    }
    examples
        .lines()
        .map(|line| {
            // Split off a trailing " [count]" suffix (rfind so values that
            // themselves contain "[" aren't truncated mid-value), reformat the
            // value, then rejoin value + suffix.
            if let Some(idx) = line.rfind(" [")
                && line.ends_with(']')
            {
                let (value, suffix) = line.split_at(idx);
                format!("{}{suffix}", format_date_value(content_type, value))
            } else {
                format_date_value(content_type, line).into_owned()
            }
        })
        .collect::<Vec<_>>()
        .join("\n")
}

/// LLM-inferred fields for a single dictionary column, keyed by field name in the
/// map returned by `parse_llm_dictionary_response`. `content_type` stays empty
/// unless `--infer-content-type` is set.
#[derive(Debug, Clone, Default)]
pub(super) struct LlmDictField {
    pub(super) label:        String,
    pub(super) description:  String,
    pub(super) content_type: String,
    /// Catalog concept ID (e.g. `geo.zip_code`); empty unless concept inference
    /// is on. Validated against `CONCEPT_VOCAB` in `parse_llm_dictionary_response`.
    pub(super) concept:      String,
    /// Analytical role (`dimension`/`measure`/`identifier`/`timestamp`); empty
    /// unless concept inference is on. Validated against `ROLE_VOCAB`.
    pub(super) role:         String,
    /// RAW null-sentinel tokens proposed by the LLM (e.g. `NULL`, `N/A`, `-999`).
    /// Deliberately unclassified here: the model proposes, `classify_null_values`
    /// (which alone can see the column's type and observed values) decides which
    /// are verifiable. Empty unless the dictionary prompt asked for them.
    pub(super) null_values:  Vec<String>,
    /// RAW `[min, max]` gauge scale proposed by the LLM for a numeric measure.
    /// `None` unless the dictionary prompt asked for it (under `--infer-content-type`)
    /// and the model returned a well-formed 2-number array with `min < max`. The
    /// role/stats VERIFICATION happens later in `combine_dictionary_entries`, not
    /// here — this struct only carries the model's raw proposal.
    pub(super) gauge_range:  Option<[f64; 2]>,
}

pub(crate) struct StatsRecord {
    pub(crate) field:       String,
    pub(crate) r#type:      String,
    pub(crate) cardinality: u64,
    pub(crate) nullcount:   u64,
    pub(crate) min:         String, // Empty string if not available
    pub(crate) max:         String, // Empty string if not available
    pub(crate) addl_cols:   IndexMap<String, String>, // Additional columns (preserves CSV order)
}

pub(crate) struct FrequencyRecord {
    pub(crate) field:      String,
    pub(crate) value:      String,
    pub(crate) count:      u64,
    pub(crate) percentage: f64,
    pub(crate) rank:       f64,
}

/// One weighted frequency sample for a field, carrying the qsv-computed
/// `percentage` and `rank` that the flat `examples` string discards. Populated
/// for the same top-N values shown in `examples`; consumed by the `SemanticMd`
/// formatter to render richer Frequency tables. `value` is already display-
/// formatted (bucket "…" suffix + `truncate_str` truncation) to match `examples`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct FreqDetail {
    pub(super) value:      String,
    pub(super) count:      u64,
    pub(super) percentage: f64,
    /// Dense rank from `frequency`; `0.0` for aggregation buckets (`Other…`/`(NULL)…`).
    pub(super) rank:       f64,
}

/// One row in the generated data dictionary. `label`, `description` and
/// `content_type` start empty and are filled by the LLM pass (`content_type`
/// only when `--infer-content-type` is set); all other fields are populated
/// deterministically from `StatsRecord` + `FrequencyRecord`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct DictionaryEntry {
    pub(super) name:            String,
    pub(super) r#type:          String,
    pub(super) label:           String,
    pub(super) description:     String,
    pub(super) content_type:    String, /* Curated semantic token; empty unless
                                         * --infer-content-type */
    pub(super) min:             String, // Empty string if not available
    pub(super) max:             String, // Empty string if not available
    pub(super) cardinality:     u64,
    pub(super) enumeration:     String, // Empty if not enumerable, otherwise one value per line
    pub(super) null_count:      u64,
    pub(super) addl_cols:       IndexMap<String, String>, // Preserves column order
    pub(super) examples:        String,                   /* Format: "val1 [cnt1]\nval2 [cnt2]…"
                                                           * or
                                                           * "<ALL_UNIQUE>" */
    /// `examples` without the `--truncate-str` truncation (bucket "…" suffixes and the
    /// `<ALL_UNIQUE>` sentinel are identical in both). Truncation exists to bound the LLM
    /// prompt, so it still governs `examples` — which feeds the prompt and the CSV/JSON/
    /// semantic-md dictionaries — while the machine-facing JSON Schema output emits these
    /// exact values. `#[serde(default)]` keeps older cached dictionaries (written before this
    /// field existed) deserializable; an empty value falls back to `examples`.
    #[serde(default)]
    pub(super) examples_full:   String,
    /// Structured counterpart to `examples`, retaining per-value percentage and rank.
    /// `#[serde(default)]` keeps older cached dictionaries (written before this field
    /// existed) deserializable.
    #[serde(default)]
    pub(super) freq_details:    Vec<FreqDetail>,
    /// Structural "every row carries a distinct non-null value" flag (`cardinality ==
    /// rowcount`, no nulls), computed deterministically at generation time. Distinct
    /// from the overloaded `examples == "<ALL_UNIQUE>"` sentinel (which is also set for
    /// constant-value and `HIGH_CARDINALITY` columns at 100% frequency). Consumed by the
    /// `SemanticMd` formatter for primary-key inference. `#[serde(default)]` for cache
    /// backward-compatibility.
    #[serde(default)]
    pub(super) is_unique_id:    bool,
    /// Catalog-wide semantic identity used for cross-dataset join discovery
    /// (e.g. `geo.zip_code`). Deterministically seeded from `content_type` and
    /// refined by the LLM; empty unless content-type/concept inference is on.
    /// `#[serde(default)]` for cache backward-compatibility.
    #[serde(default)]
    pub(super) concept:         String,
    /// Analytical role: `dimension`, `measure`, `identifier`, or `timestamp`.
    /// `identifier`/`timestamp` are deterministic; the rest are LLM-filled with a
    /// type-based fallback. `#[serde(default)]` for cache backward-compatibility.
    #[serde(default)]
    pub(super) role:            String,
    /// Null sentinels the LLM proposed AND that qsv independently confirmed are
    /// present as literal values in this `String` column. The confirmation is one
    /// of PRESENCE, not of meaning: qsv verifies the literal occurs here; that it
    /// stands for "missing" remains the LLM's judgment. Spellings are the ones
    /// OBSERVED in the data (every casing), not the model's echo of them.
    ///
    /// Strictly wider than what `qsv denull` reports: denull confirms only columns
    /// that would promote to a numeric type once blanked, so a purely categorical
    /// column (`status` = ok/pending/NULL) is confirmed here and ignored there.
    /// `#[serde(default)]` for cache backward-compatibility.
    #[serde(default)]
    pub(super) null_values:     Vec<String>,
    /// Null sentinels the LLM proposed that qsv CANNOT confirm by scanning —
    /// numeric/date placeholders (`-999`, `9999`, `9999-12-31`) that parse as
    /// valid values of the column's type, or tokens never observed in the data.
    /// These are unfalsifiable by construction: `-999` is a legal integer and no
    /// scan can distinguish it from a real reading. Emitted with
    /// `confirm_required: true` so no consumer can auto-apply a guess.
    /// `#[serde(default)]` for cache backward-compatibility.
    #[serde(default)]
    pub(super) null_candidates: Vec<String>,
    /// Optional `[min, max]` canonical scale for a KPI gauge on a numeric MEASURE
    /// (e.g. `[0, 100]` for a percent, `[0, 5]` for a rating). The LLM proposes it
    /// only under `--infer-content-type`; `combine_dictionary_entries` then keeps it
    /// ONLY when the column's finalized `role` is `measure` AND the observed
    /// `[min, max]` lies inside the proposed range — an unfalsifiable or
    /// out-of-scale range is dropped, mirroring the propose-then-verify discipline
    /// of `null_values`. Consumed by `viz smart --dictionary` to draw a gauge tile
    /// (`x-qsv.gauge_range`). `#[serde(default)]` for cache backward-compatibility.
    #[serde(default)]
    pub(super) gauge_range:     Option<[f64; 2]>,
}

/// Parse the `stats` CSV into structured records, returning the records plus
/// the ordered list of additional (non-standard) column names in CSV order.
pub(crate) fn parse_stats_csv(stats_csv: &str) -> CliResult<(Vec<StatsRecord>, Vec<String>)> {
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(true)
        .from_reader(stats_csv.as_bytes());

    let headers = rdr.headers()?.clone();

    let std_cols: HashSet<&str> = ["field", "type", "cardinality", "nullcount", "min", "max"]
        .iter()
        .copied()
        .collect();

    let field_idx = headers
        .iter()
        .position(|h| h == "field")
        .ok_or_else(|| CliError::Other("Stats CSV missing 'field' column".to_string()))?;

    let type_idx = headers
        .iter()
        .position(|h| h == "type")
        .ok_or_else(|| CliError::Other("Stats CSV missing 'type' column".to_string()))?;

    let cardinality_idx = headers.iter().position(|h| h == "cardinality");
    let nullcount_idx = headers
        .iter()
        .position(|h| h == "nullcount")
        .ok_or_else(|| CliError::Other("Stats CSV missing 'nullcount' column".to_string()))?;
    let min_idx = headers.iter().position(|h| h == "min");
    let max_idx = headers.iter().position(|h| h == "max");

    let addl_col_indices: Vec<(usize, String)> = headers
        .iter()
        .enumerate()
        .filter_map(|(idx, header)| {
            if std_cols.contains(header) {
                None
            } else {
                Some((idx, header.to_string()))
            }
        })
        .collect();

    let mut records = Vec::new();

    for result in rdr.records() {
        let record = result?;
        let field = record
            .get(field_idx)
            .ok_or_else(|| CliError::Other("Stats CSV record missing field value".to_string()))?
            .to_string();

        let r#type = record
            .get(type_idx)
            .ok_or_else(|| CliError::Other("Stats CSV record missing type value".to_string()))?
            .to_string();

        let cardinality = cardinality_idx
            .and_then(|idx| record.get(idx))
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(0);

        let nullcount = record
            .get(nullcount_idx)
            .ok_or_else(|| CliError::Other("Stats CSV record missing nullcount value".to_string()))?
            .parse::<u64>()
            .map_err(|e| CliError::Other(format!("Failed to parse nullcount: {e}")))?;

        let min = min_idx
            .and_then(|idx| record.get(idx))
            .map(std::string::ToString::to_string)
            .unwrap_or_default();

        let max = max_idx
            .and_then(|idx| record.get(idx))
            .map(std::string::ToString::to_string)
            .unwrap_or_default();

        let mut addl_cols = IndexMap::new();
        for (idx, col_name) in &addl_col_indices {
            let value = record
                .get(*idx)
                .map(std::string::ToString::to_string)
                .unwrap_or_default();
            addl_cols.insert(col_name.clone(), value);
        }

        records.push(StatsRecord {
            field,
            r#type,
            cardinality,
            nullcount,
            min,
            max,
            addl_cols,
        });
    }

    let ordered_col_names: Vec<String> =
        addl_col_indices.into_iter().map(|(_, name)| name).collect();

    Ok((records, ordered_col_names))
}

/// Parse the `frequency` CSV into structured records.
pub(crate) fn parse_frequency_csv(frequency_csv: &str) -> CliResult<Vec<FrequencyRecord>> {
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(true)
        .from_reader(frequency_csv.as_bytes());

    let headers = rdr.headers()?.clone();

    let field_idx = headers
        .iter()
        .position(|h| h == "field")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'field' column".to_string()))?;

    let value_idx = headers
        .iter()
        .position(|h| h == "value")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'value' column".to_string()))?;

    let count_idx = headers
        .iter()
        .position(|h| h == "count")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'count' column".to_string()))?;

    let percentage_idx = headers
        .iter()
        .position(|h| h == "percentage")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'percentage' column".to_string()))?;

    let rank_idx = headers
        .iter()
        .position(|h| h == "rank")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'rank' column".to_string()))?;

    let mut records = Vec::new();

    for result in rdr.records() {
        let record = result?;
        let field = record
            .get(field_idx)
            .ok_or_else(|| CliError::Other("Frequency CSV record missing field value".to_string()))?
            .to_string();

        let value = record
            .get(value_idx)
            .ok_or_else(|| CliError::Other("Frequency CSV record missing value".to_string()))?
            .to_string();

        let count = record
            .get(count_idx)
            .ok_or_else(|| CliError::Other("Frequency CSV record missing count".to_string()))
            .and_then(|s| {
                if s.is_empty() {
                    Ok(0)
                } else {
                    s.parse::<u64>().map_err(|e| {
                        CliError::Other(format!("Failed to parse count in frequency CSV: {e}"))
                    })
                }
            })?;

        let percentage = record
            .get(percentage_idx)
            .ok_or_else(|| CliError::Other("Frequency CSV record missing percentage".to_string()))
            .and_then(|s| {
                if s.is_empty() {
                    Ok(0.0)
                } else {
                    s.parse::<f64>().map_err(|e| {
                        CliError::Other(format!("Failed to parse percentage in frequency CSV: {e}"))
                    })
                }
            })?;

        let rank = record
            .get(rank_idx)
            .ok_or_else(|| CliError::Other("Frequency CSV record missing rank".to_string()))
            .and_then(|s| {
                if s.is_empty() {
                    Ok(0.0)
                } else {
                    s.parse::<f64>().map_err(|e| {
                        CliError::Other(format!("Failed to parse rank in frequency CSV: {e}"))
                    })
                }
            })?;

        records.push(FrequencyRecord {
            field,
            value,
            count,
            percentage,
            rank,
        });
    }

    Ok(records)
}

/// Generate dictionary entries deterministically from `stats` + `frequency`
/// data. Label and Description are left empty for the LLM pass to fill.
///
/// When `infer_content_type` is set, `content_type` is also deterministically
/// pre-set to `"unique_id"` for fields where `cardinality == rowcount` with no
/// nulls. The detection is structural (single frequency row whose
/// `count == cardinality`), not text-matching, so it works regardless of the
/// `frequency --all-unique-text` setting and won't be confused by fields whose
/// values literally contain the string `<ALL_UNIQUE>`. All other fields'
/// `content_type` stays empty and is filled by the LLM pass.
pub(super) fn generate_code_based_dictionary(
    stats_records: &[StatsRecord],
    frequency_records: &[FrequencyRecord],
    enum_threshold: usize,
    num_examples: u16,
    truncate_str: usize,
    addl_cols: &[String],
    infer_content_type: bool,
) -> Vec<DictionaryEntry> {
    let mut frequency_by_field: HashMap<String, Vec<&FrequencyRecord>> = HashMap::new();
    for freq_record in frequency_records {
        frequency_by_field
            .entry(freq_record.field.clone())
            .or_default()
            .push(freq_record);
    }

    let mut dictionary_entries = Vec::new();

    for stats_record in stats_records {
        let field_name = &stats_record.field;
        let field_frequencies = frequency_by_field
            .get(field_name)
            .cloned()
            .unwrap_or_default();

        let enumeration = if stats_record.cardinality <= enum_threshold as u64 {
            // Check for rank=0 "Other" bucket or <ALL_UNIQUE> sentinel
            let has_other = field_frequencies
                .iter()
                .any(|f| f.rank == 0.0 && !f.value.contains("<ALL_UNIQUE>"));
            if has_other {
                String::new()
            } else {
                let mut enum_values: Vec<String> = field_frequencies
                    .iter()
                    .filter(|f| !f.value.contains("<ALL_UNIQUE>"))
                    .map(|f| f.value.clone())
                    .collect();
                enum_values.sort();
                enum_values.join("\n")
            }
        } else {
            String::new()
        };

        let (examples, examples_full, freq_details) = if field_frequencies
            .iter()
            .any(|f| (f.percentage - 100.0).abs() < 0.0001)
        {
            (
                "<ALL_UNIQUE>".to_string(),
                "<ALL_UNIQUE>".to_string(),
                Vec::new(),
            )
        } else {
            let mut sorted_freqs = field_frequencies.clone();
            sorted_freqs.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.value.cmp(&b.value)));

            let mut top_n: Vec<String> = Vec::new();
            let mut top_n_full: Vec<String> = Vec::new();
            let mut details: Vec<FreqDetail> = Vec::new();
            for f in sorted_freqs.iter().take(num_examples as usize) {
                // For frequency bucket entries (rank == 0.0), strip the redundant
                // "(n)" count and append "…" to disambiguate from literal values with
                // the same name (e.g. bucket "Other… [4,091]" vs literal "Other [2,006]")
                let raw_value = if f.rank == 0.0 {
                    let base = if let Some(pos) = f.value.rfind(" (") {
                        &f.value[..pos]
                    } else {
                        &f.value
                    };
                    format!("{base}…")
                } else {
                    f.value.clone()
                };

                let v = if truncate_str > 0 && raw_value.chars().count() > truncate_str {
                    let mut s = raw_value.chars().take(truncate_str).collect::<String>();
                    s.push('…');
                    s
                } else {
                    raw_value.clone()
                };
                top_n.push(format!("{} [{}]", v, f.count));
                // untruncated counterpart — the bucket "…" suffix above is already applied, so
                // only the `truncate_str` step is skipped. Consumed by the JSON Schema output
                // (machine-facing, and what `viz --dict-info` renders); the prompt and the
                // CSV/JSON/semantic-md dictionaries keep the truncated `examples`.
                top_n_full.push(format!("{raw_value} [{}]", f.count));
                details.push(FreqDetail {
                    value:      v,
                    count:      f.count,
                    percentage: f.percentage,
                    rank:       f.rank,
                });
            }

            (top_n.join("\n"), top_n_full.join("\n"), details)
        };

        let mut entry_addl_cols = IndexMap::new();
        for col_name in addl_cols {
            if let Some(value) = stats_record.addl_cols.get(col_name) {
                entry_addl_cols.insert(col_name.clone(), value.clone());
            }
        }

        // Deterministically classify fields where every row carries a distinct
        // non-null value as `unique_id`. qsv's frequency command emits an
        // ALL_UNIQUE sentinel row exactly when `cardinality == rowcount` (a
        // single row with `count == rowcount` and `percentage == 100.0`). We
        // detect that structurally — `len() == 1` + `count == cardinality` —
        // rather than matching the literal `<ALL_UNIQUE>` text so that:
        //   - a real field whose values literally contain the string "<ALL_UNIQUE>" isn't
        //     mislabeled (constants have `cardinality == 1`; mixed fields produce more than one
        //     frequency row);
        //   - a custom `frequency --all-unique-text` sentinel is still detected correctly (the text
        //     doesn't matter, only the structural cardinality==count invariant does);
        //   - HIGH_CARDINALITY sentinel rows (also single-row, percentage 100.0, count == rowcount)
        //     are excluded because for them `cardinality < rowcount == count`.
        // Also requires `cardinality > 1` and `nullcount == 0` to enforce
        // the semantic contract (every row has a distinct non-null value), and that the
        // single row covers the whole column (`percentage == 100.0`). The percentage
        // guard rejects truncated/custom frequency data (e.g. `--limit 1 --no-other` or a
        // `file:` frequency CSV) where the lone emitted top row's `count` coincidentally
        // equals the column cardinality even though the column is not unique.
        // Pre-set value takes precedence over whatever the LLM returns (see
        // `combine_dictionary_entries`). Only populate when `--infer-content-type`
        // is on; otherwise the `content_type` column is suppressed entirely.
        let is_all_unique = stats_record.cardinality > 1
            && stats_record.nullcount == 0
            && field_frequencies.len() == 1
            && field_frequencies[0].count == stats_record.cardinality
            && (field_frequencies[0].percentage - 100.0).abs() < 0.0001;
        // Deterministically stamp the bare `content_type` token. `unique_id`
        // keeps priority over `date`/`datetime` (a unique timestamp key is
        // still a key). `date`/`datetime` are derived from the stats `Type`
        // column; the LLM later supplies only the optional strftime `:<fmt>`
        // suffix (merged in `combine_dictionary_entries`).
        let content_type = if !infer_content_type {
            String::new()
        } else if is_all_unique {
            "unique_id".to_string()
        } else {
            match stats_record.r#type.as_str() {
                "Date" => "date".to_string(),
                "DateTime" => "datetime".to_string(),
                _ => String::new(),
            }
        };

        // Deterministically seed `concept` + `role`, gated on the same flag as
        // `content_type`. `concept` seeds from the content_type token (e.g.
        // `zip_code` → `geo.zip_code`); `role` from the structural unique-id flag
        // (`identifier`) and the stats Date/DateTime type (`timestamp`). The LLM
        // fills/refines empty or non-authoritative values in
        // `combine_dictionary_entries`.
        let (concept, role) = if infer_content_type {
            let concept = concept_from_content_type(content_type_base(&content_type))
                .map(ToString::to_string)
                .unwrap_or_default();
            let role = if is_all_unique {
                "identifier".to_string()
            } else {
                match stats_record.r#type.as_str() {
                    "Date" | "DateTime" => "timestamp".to_string(),
                    _ => String::new(),
                }
            };
            (concept, role)
        } else {
            (String::new(), String::new())
        };

        dictionary_entries.push(DictionaryEntry {
            name: stats_record.field.clone(),
            r#type: stats_record.r#type.clone(),
            label: String::new(),       // Filled by LLM
            description: String::new(), // Filled by LLM
            content_type,               /* Pre-set to "unique_id" for ALL_UNIQUE fields;
                                         * otherwise filled by LLM */
            min: stats_record.min.clone(),
            max: stats_record.max.clone(),
            cardinality: stats_record.cardinality,
            enumeration,
            null_count: stats_record.nullcount,
            addl_cols: entry_addl_cols,
            examples,
            examples_full,
            freq_details,
            is_unique_id: is_all_unique,
            concept,
            role,
            // Filled by the LLM pass via `classify_null_values`; there is no
            // deterministic seed, since proposing a sentinel requires judgment.
            null_values: Vec::new(),
            null_candidates: Vec::new(),
            // Proposed by the LLM pass (measure gauges) and verified against the
            // observed min/max in `combine_dictionary_entries`; no deterministic seed.
            gauge_range: None,
        });
    }

    dictionary_entries
}

/// Merge a single LLM-supplied `content_type` onto the (possibly
/// deterministically pre-stamped) `current` value, returning the value the
/// entry should hold. Encodes the whole `content_type` merge contract:
///
/// * A code-stamped `date`/`datetime` column (`current` base is `date`/`datetime`) only accepts an
///   LLM value that agrees on the base token AND carries a strftime `:<fmt>` suffix — the LLM may
///   add or correct the format but never reclassify the column or wipe a known format back to bare.
/// * `unique_id` is deterministic — the LLM can never override it.
/// * The LLM may never introduce a `date`/`datetime` token onto a column the stats did not type as
///   such (only the deterministic stamp may), and never supply `unique_id`.
/// * Otherwise an LLM token is written into an empty slot; when `allow_override` is set (the refine
///   pass) it may also replace an existing non-`unique_id`, non-date token.
fn merge_content_type(current: &str, llm: &str, allow_override: bool) -> String {
    let current_base = content_type_base(current);
    if current_base == "date" || current_base == "datetime" {
        if llm.contains(':') && content_type_base(llm) == current_base {
            return llm.to_string();
        }
        return current.to_string();
    }
    if current == "unique_id" {
        return current.to_string();
    }
    if !llm.is_empty()
        && llm != "unique_id"
        && !matches!(content_type_base(llm), "date" | "datetime")
        && (current.is_empty() || allow_override)
    {
        return llm.to_string();
    }
    current.to_string()
}

/// Merge code-generated dictionary entries with the LLM-generated fields (Label,
/// Description and, when `--infer-content-type` is set, Content Type) keyed by
/// field name.
///
/// `content_type` merge contract — see `merge_content_type`:
/// - Code-derived `content_type` always wins. `generate_code_based_dictionary` stamps `"unique_id"`
///   on `<ALL_UNIQUE>` fields and bare `"date"`/`"datetime"` on `Date`/`DateTime`-typed fields.
/// - For a code-stamped `date`/`datetime` field the LLM may only contribute the strftime `:<fmt>`
///   suffix; it can neither reclassify the column nor introduce a `date`/`datetime` token on a
///   non-date column.
/// - LLM-supplied `"unique_id"` is refused (defense in depth — the parser also strips it).
///
/// When `infer_content_type` is set, this is the single point that guarantees
/// every entry has a non-empty `content_type`: any field the LLM classified
/// with an invalid token, omitted the `content_type` key for, or left out of
/// its response entirely falls back to `"unknown"`.
/// Split one field's raw null-sentinel proposals into a VERIFIABLE bucket and an
/// UNVERIFIABLE one. The model proposes; this function - which alone sees the
/// column's type and the values actually present in it - decides. The model cannot
/// promote its own guess into the trusted bucket, mirroring the deterministic-only
/// contract that makes `parse_llm_dictionary_response` refuse an LLM `unique_id`.
///
/// "Trusted" means only that the literal is PRESENT in the column. Whether it
/// actually denotes a missing value is the model's call, and nothing downstream
/// acts on it automatically.
///
/// `null_text` is `frequency`'s label for the already-empty cells it aggregates into
/// the null row (default `(NULL)`). That label is DISPLAY, not data - yet it is right
/// there in the Frequency Distribution the model reads, and models do propose it
/// (observed on real data, on a file `denull --apply` had just cleaned). A proposal
/// that merely echoes the label is dropped outright rather than demoted, because it
/// describes cells qsv ALREADY counts as null. It is dropped only when the token is
/// not among `observed`: a column that genuinely holds the literal text `(NULL)` as a
/// value still confirms it normally.
///
/// A proposal is trusted only when BOTH hold:
///   * `qsv_type` is `String` - the only type whose sentinels are detectable at all. A `-999` in an
///     `Integer` column parses as a valid integer; no scan can tell it from a real reading (and a
///     well-depth column legitimately reaches `-140`), so it can never be confirmed.
///   * the token appears in `observed`, the column's real values.
///
/// `observed` must be the FULL set of raw frequency values for the column (see
/// `usable_samples_by_field`), never the display `examples`/`freq_details` subset:
/// those are capped at `--num-examples` and are truncated for display, so a
/// sentinel at rank 6 - or a long one rendered as `NOT APPLICAB…` - would be
/// falsely demoted, or worse, a truncated prefix confirmed.
///
/// Matching is ASCII-case-insensitive (models don't reliably echo casing), but the
/// emitted spellings are the ones found IN THE DATA, never the model's echo - and
/// every observed casing is emitted, because a consumer must know every spelling it
/// would have to blank. Everything else falls through to `null_candidates`, carrying
/// the model's raw token for a human to confirm.
fn classify_null_values(
    qsv_type: &str,
    observed: &[&str],
    proposed: &[String],
    null_text: &str,
) -> (Vec<String>, Vec<String>) {
    let is_string_col = qsv_type == "String";

    let mut confirmed: Vec<String> = Vec::new();
    let mut candidates: Vec<String> = Vec::new();

    for token in proposed {
        let token = token.trim();
        if token.is_empty() {
            continue;
        }
        let matches: Vec<&str> = if is_string_col {
            observed
                .iter()
                .filter(|v| v.eq_ignore_ascii_case(token))
                .copied()
                .collect()
        } else {
            Vec::new()
        };
        // The model echoed frequency's null-row LABEL, and no real value matches it.
        // Those cells are already null; reporting the label as a sentinel would be
        // noise on data that has none. Checked after `matches` so a column genuinely
        // holding the literal text still confirms it.
        if matches.is_empty() && token.eq_ignore_ascii_case(null_text) {
            continue;
        }
        // A non-numeric token cannot occur in a numeric column: if a single cell held the
        // text "NULL", `stats` could not have typed the column Integer/Float. Such a
        // proposal is IMPOSSIBLE, not merely unconfirmable, so drop it instead of asking a
        // human to confirm a cell that cannot exist. A numeric placeholder (-999) parses
        // fine and still rides through as a candidate - that is the whole point.
        if (qsv_type == "Integer" || qsv_type == "Float") && token.parse::<f64>().is_err() {
            continue;
        }
        if matches.is_empty() {
            if !candidates.iter().any(|c| c == token) {
                candidates.push(token.to_string());
            }
        } else {
            for m in matches {
                if !confirmed.iter().any(|c| c == m) {
                    confirmed.push(m.to_string());
                }
            }
        }
    }
    (confirmed, candidates)
}

/// The raw null-sentinel proposals to classify, per field.
///
/// For a `--two-pass` run, `refine` wins only where it actually re-states a field's
/// sentinels: the refine prompt never asks for them, so an empty refine list means
/// "not re-stated", NOT "retracted". Taking it unconditionally would wipe the first
/// pass's sentinels on every two-pass run - the same invariant that guards
/// Label/Description in `combine_dictionary_entries_with_baseline`.
pub(super) fn null_value_proposals(
    baseline: &HashMap<String, LlmDictField>,
    refine: Option<&HashMap<String, LlmDictField>>,
) -> HashMap<String, Vec<String>> {
    let mut proposals: HashMap<String, Vec<String>> = baseline
        .iter()
        .filter(|(_, f)| !f.null_values.is_empty())
        .map(|(name, f)| (name.clone(), f.null_values.clone()))
        .collect();
    if let Some(refine) = refine {
        for (name, f) in refine {
            if !f.null_values.is_empty() {
                proposals.insert(name.clone(), f.null_values.clone());
            }
        }
    }
    proposals
}

/// Stamp each entry's `null_values` / `null_candidates` from the LLM's raw proposals,
/// classified against the column's real values.
///
/// Deliberately a separate pass rather than part of `combine_dictionary_entries`:
/// classification needs the frequency records (for the full observed-value set) and
/// `null_text`, neither of which the combine step has - and building the sample map
/// once here is cheaper than rebuilding it per entry.
///
/// A no-op when the LLM proposed nothing, which is the case whenever the dictionary
/// prompt did not ask (i.e. without `--infer-null-values`).
pub(super) fn apply_null_value_classification(
    entries: &mut [DictionaryEntry],
    proposals: &HashMap<String, Vec<String>>,
    frequency_records: &[FrequencyRecord],
    null_text: &str,
) {
    if proposals.is_empty() {
        return;
    }
    // Excludes the rank-0 "Other" bucket, the `<ALL_UNIQUE>` sentinel and the emitted
    // null row, and yields RAW (untruncated) values - exactly the confirmation set.
    let samples_by_field = usable_samples_by_field(frequency_records, entries, null_text);
    for entry in entries.iter_mut() {
        let Some(proposed) = proposals.get(&entry.name) else {
            continue;
        };
        let observed: &[&str] = samples_by_field
            .get(entry.name.as_str())
            .map_or(&[], Vec::as_slice);
        let (confirmed, candidates) =
            classify_null_values(&entry.r#type, observed, proposed, null_text);
        entry.null_values = confirmed;
        entry.null_candidates = candidates;
    }
}

pub(super) fn combine_dictionary_entries(
    mut code_entries: Vec<DictionaryEntry>,
    llm_fields: &HashMap<String, LlmDictField>,
    infer_content_type: bool,
) -> Vec<DictionaryEntry> {
    for entry in &mut code_entries {
        if let Some(llm) = llm_fields.get(&entry.name) {
            entry.label = llm.label.clone();
            entry.description = llm.description.clone();
            entry.content_type = merge_content_type(&entry.content_type, &llm.content_type, false);
            entry.concept = merge_concept(&entry.concept, &llm.concept, false);
            entry.role = merge_role(&entry.role, &llm.role, false);
            entry.gauge_range = llm.gauge_range;
        }
        if infer_content_type {
            if entry.content_type.is_empty() {
                entry.content_type = "unknown".to_string();
            }
            coerce_role_concept(entry);
            // AFTER role is finalized: keep the proposed gauge scale only if the
            // column is a measure and its observed range fits inside it.
            verify_gauge_range(entry);
        }
    }
    code_entries
}

fn coerce_role_concept(entry: &mut DictionaryEntry) {
    // Backfill the deterministic concept mapping from the now-merged `content_type`
    // before defaulting to "unknown". This recovers the documented mapping when a
    // model (or an older cached response, or a custom prompt) returns a valid
    // `content_type` like `zip_code` but supplies no concept — either omitting it
    // (empty) or emitting the literal `"unknown"` (a valid vocab token). The seed in
    // `generate_code_based_dictionary` only sees the pre-LLM (deterministic)
    // content_type, so this is the merge path's chance to apply it.
    if entry.concept.is_empty() || entry.concept == "unknown" {
        entry.concept = concept_from_content_type(content_type_base(&entry.content_type))
            .map_or_else(|| "unknown".to_string(), ToString::to_string);
    }
    if entry.role.is_empty() {
        entry.role = if entry.r#type == "Integer" || entry.r#type == "Float" {
            "measure".to_string()
        } else {
            "dimension".to_string()
        };
    }

    let base = content_type_base(&entry.content_type);

    // issue #4177: `content_type` constrains `role`. The LLM infers `content_type`,
    // `role`, and `concept` independently, so it can emit contradictions like a
    // `time` column tagged `role: measure` (a clock time is not a quantity to
    // aggregate). Downstream consumers (e.g. `viz smart`'s `derive_semantics`) route
    // on `role`, so the contradiction becomes a visible defect. Reconcile here — the
    // single coercion choke point shared by both merge paths:
    //   - a temporal content_type is always a `timestamp` (overrides any role);
    //   - any other recognized (non-numeric) content_type must not be a `measure`, so a stray
    //     `measure` becomes `identifier` (id-family) or `dimension`.
    // `unknown`/`duration` yield no constraint (measure-admissible), so a numeric
    // measure is left untouched.
    match role_from_content_type(base) {
        Some("timestamp") => entry.role = "timestamp".to_string(),
        Some(required) if entry.role == "measure" => entry.role = required.to_string(),
        _ => {},
    }

    // issue #4176: `concept` must not contradict a reliable `content_type`. describegpt
    // infers concept independently, so it can emit e.g. `concept: measure.ratio` on a
    // column whose `content_type` (`category`) and `role` (`dimension`) both say
    // categorical. `viz`'s `derive_semantics` prefers concept over role, so the lone
    // dissenting concept silently misroutes (and can drop) the column — charting it
    // WORSE than with no dictionary at all. Reconcile at two granularities:
    //   1. When the content_type pins a deterministic concept (`concept_from_content_type`),
    //      require an EXACT match by default — most mappings are exact (a `city` is `geo.city`, an
    //      `email` is `pii.email`), so any other concept is a real contradiction: `geo.city` tagged
    //      `geo.country` corrupts join identity (both are linkable), and `geo.city` on a `latitude`
    //      column would suppress `Route::MapCoord`. A few content_types map COARSELY to a generic
    //      placeholder that has legitimate more-specific siblings; for those,
    //      `admissible_refinements` lists the EXACT set the LLM may substitute. A namespace-wide
    //      check was too broad — e.g. `company_name` → `org.company` admits `org.agency` (a
    //      government "Agency Name" column has no `agency_name` content_type) but NOT
    //      `org.industry` (a classification, not an org name); `date` admits the `time.*` event
    //      points but not `time.duration` (a span). Any concept outside the admissible set is reset
    //      to the deterministic value.
    //   2. Otherwise, when the content_type still constrains the role, a concept whose namespace
    //      implies a different role is the dissenter → drop to "unknown".
    // `content_type: unknown` pins neither, so a specific concept like `geo.census_tract`
    // on an untyped numeric column is deliberately left to win — that override is why
    // `route_from_concept` exists.
    //
    // Coarse content_types → the exact concepts a model may substitute for the deterministic
    // seed. A content_type absent here is EXACT (only its deterministic concept is admissible).
    fn admissible_refinements(base: &str) -> &'static [&'static str] {
        match base {
            // a date/datetime is a point-in-time event, so any temporal event concept fits,
            // but NOT `time.duration` (a span) or `time.date` vs `time.*` distinctions.
            "date" | "datetime" => &[
                "time.date",
                "time.event_timestamp",
                "time.created_at",
                "time.closed_at",
                "time.updated_at",
                "time.due_at",
            ],
            // a uuid is a key; it may be a surrogate/natural/foreign key (id.surrogate_key is
            // reserved+deterministic and handled upstream, so it is not offered here).
            "uuid" => &["id.uuid", "id.natural_key", "id.foreign_key"],
            // an org-name column may be a company or a (government) agency, never an industry.
            "company_name" => &["org.company", "org.agency"],
            _ => &[],
        }
    }
    if let Some(deterministic) = concept_from_content_type(base) {
        let admissible = admissible_refinements(base);
        let contradicts = if admissible.is_empty() {
            entry.concept != deterministic
        } else {
            !admissible.contains(&entry.concept.as_str())
        };
        if contradicts {
            entry.concept = deterministic.to_string();
        }
    } else if let Some(ct_role) = role_from_content_type(base)
        && let Some(concept_role) = role_from_concept_namespace(&entry.concept)
        && ct_role != concept_role
    {
        entry.concept = "unknown".to_string();
    }
}

/// Verify (and otherwise drop) a proposed `gauge_range`. A KPI gauge is only
/// meaningful, and only non-misleading, when:
///
///   1. the column's qsv `type` is numeric (`Integer`/`Float`) — a gauge scale on a String/Date
///      column is nonsense even if the LLM mis-tagged it `role: measure` and its lexicographic
///      min/max happen to parse as floats, and
///   2. the column's FINALIZED `role` is `measure` (gauges are for quantities you aggregate — not
///      dimensions, identifiers, or timestamps), and
///   3. the column's OBSERVED `[min, max]` lies inside the proposed `[lo, hi]`.
///
/// Requirement 3 is the guardrail against a misleading gauge: `viz` renders the
/// measure's value against this scale, so a range that does not contain the data
/// would draw a needle pinned past the end. The shape (two finite numbers, `lo <
/// hi`) was already validated in `parse_llm_dictionary_response`; this is where the
/// SEMANTIC check happens, because only here are the merged role, the column type,
/// and the observed min/max all known. Anything that fails is reset to `None` — an
/// unverifiable gauge is simply not emitted (byte-identical to a run that never
/// proposed one).
///
/// Must be called AFTER `coerce_role_concept`, so `role` is final.
fn verify_gauge_range(entry: &mut DictionaryEntry) {
    let Some([lo, hi]) = entry.gauge_range else {
        return;
    };
    let keep = if matches!(entry.r#type.as_str(), "Integer" | "Float")
        && entry.role == "measure"
        && let Ok(obs_min) = entry.min.parse::<f64>()
        && let Ok(obs_max) = entry.max.parse::<f64>()
    {
        obs_min.is_finite() && obs_max.is_finite() && obs_min >= lo && obs_max <= hi
    } else {
        false
    };
    if !keep {
        entry.gauge_range = None;
    }
}

/// Two-pass-aware merge: seed `code_entries` with the BASELINE LLM Label / Description /
/// Content Type (from the first pass) and then overlay the REFINE pass's LLM fields on top.
/// If the refine pass omits a field, the baseline Label / Description / Content Type are
/// preserved — this is the critical correctness invariant for `--two-pass`. Without it, a
/// refine response that returns a subset of fields would silently wipe the first-pass
/// human-friendly fields back to code-derived defaults.
///
/// The `content_type` merge contract from `combine_dictionary_entries` is preserved via
/// `merge_content_type`, applied once per pass. The baseline pass may only fill an empty
/// slot (`allow_override = false`); the refine pass may additionally upgrade an existing
/// non-`unique_id`, non-date token (`allow_override = true`). Neither pass can override the
/// deterministic `"unique_id"` stamp, reclassify a `date`/`datetime` column, or introduce a
/// `date`/`datetime` token onto a non-date column — only the strftime `:<fmt>` suffix may
/// be added or corrected on an already-date-typed column.
pub(super) fn combine_dictionary_entries_with_baseline(
    mut code_entries: Vec<DictionaryEntry>,
    baseline_llm_fields: &HashMap<String, LlmDictField>,
    refine_llm_fields: &HashMap<String, LlmDictField>,
    infer_content_type: bool,
) -> Vec<DictionaryEntry> {
    for entry in &mut code_entries {
        // Stage 1: apply baseline (first-pass) LLM values, mirroring
        // `combine_dictionary_entries`'s behavior for the single-pass case.
        if let Some(baseline) = baseline_llm_fields.get(&entry.name) {
            entry.label = baseline.label.clone();
            entry.description = baseline.description.clone();
            entry.content_type =
                merge_content_type(&entry.content_type, &baseline.content_type, false);
            entry.concept = merge_concept(&entry.concept, &baseline.concept, false);
            entry.role = merge_role(&entry.role, &baseline.role, false);
            entry.gauge_range = baseline.gauge_range;
        }
        // Stage 2: overlay refine-pass LLM values where present. Omitted fields keep their
        // baseline values from stage 1 — this is the whole point of the baseline merge.
        if let Some(refine) = refine_llm_fields.get(&entry.name) {
            if !refine.label.is_empty() {
                entry.label = refine.label.clone();
            }
            if !refine.description.is_empty() {
                entry.description = refine.description.clone();
            }
            entry.content_type =
                merge_content_type(&entry.content_type, &refine.content_type, true);
            entry.concept = merge_concept(&entry.concept, &refine.concept, true);
            entry.role = merge_role(&entry.role, &refine.role, true);
            // A refine pass that re-proposes a gauge scale overrides the baseline; one
            // that omits it keeps the baseline value (mirrors the label/description rule).
            if refine.gauge_range.is_some() {
                entry.gauge_range = refine.gauge_range;
            }
        }
        // Stage 3: same final "unknown"/fallback coercion as `combine_dictionary_entries` so
        // the two-pass output matches single-pass invariants.
        if infer_content_type {
            if entry.content_type.is_empty() {
                entry.content_type = "unknown".to_string();
            }
            coerce_role_concept(entry);
            verify_gauge_range(entry);
        }
    }
    code_entries
}

/// True if `sample` parses cleanly with the chrono strftime `fmt`. For
/// `datetime`, an offset-naive parse is tried first, then an offset-aware one
/// (`%z` / `%:z` formats are rejected by `NaiveDateTime`).
fn sample_parses_with_format(sample: &str, fmt: &str, is_datetime: bool) -> bool {
    use chrono::{DateTime, NaiveDate, NaiveDateTime};
    if is_datetime {
        NaiveDateTime::parse_from_str(sample, fmt).is_ok()
            || DateTime::parse_from_str(sample, fmt).is_ok()
    } else {
        NaiveDate::parse_from_str(sample, fmt).is_ok()
    }
}

/// Group usable raw frequency values by field name — every value except the
/// rank-0 rows (the "Other" bucket and the `<ALL_UNIQUE>` / `<HIGH_CARDINALITY>`
/// sentinels, all of which `frequency` stamps with rank `0.0`) and the emitted
/// null row.
///
/// The all-unique sentinel is excluded STRUCTURALLY, by its rank, never by matching
/// its literal text: `frequency --all-unique-text` is configurable, so a text match
/// would both miss a customized sentinel and — the reason this matters — wrongly drop
/// a real datum that merely contains the default `<ALL_UNIQUE>` string, silently
/// removing it from the observed-value set that `classify_null_values` confirms
/// against. (Contrast `generate_code_based_dictionary`, where a text check legitimately
/// disambiguates *between* two different rank-0 rows.)
///
/// The null row is identified by BOTH signals `frequency` guarantees for it,
/// never by either one alone:
///   * its `value` equals the configured `--null-text` (`null_text`, default `(NULL)`), and
///   * its `count` equals the column's null count (carried on `DictionaryEntry.null_count`).
///
/// Value alone is too broad — `frequency`'s null text is configurable, and a
/// real datum that merely reads as the null label would be dropped. Count
/// alone is also too broad — a real value that merely shares the column's null
/// count would be dropped, which can keep a bad date suffix or wrongly
/// downgrade a `datetime` column. Requiring both confines the exclusion to the
/// genuine null sentinel, even when `frequency --pct-nulls` gives that row a
/// real (non-zero) rank that a rank check alone would let through.
///
/// Borrows the sample strings from `frequency_records`.
fn usable_samples_by_field<'a>(
    frequency_records: &'a [FrequencyRecord],
    entries: &[DictionaryEntry],
    null_text: &str,
) -> HashMap<&'a str, Vec<&'a str>> {
    let null_count_by_field: HashMap<&str, u64> = entries
        .iter()
        .filter(|e| e.null_count > 0)
        .map(|e| (e.name.as_str(), e.null_count))
        .collect();

    let mut samples_by_field: HashMap<&str, Vec<&str>> = HashMap::new();
    for rec in frequency_records {
        if rec.rank == 0.0 {
            continue;
        }
        // the emitted null row: `frequency` writes it with `--null-text` as
        // its value AND a count equal to the column's null count
        if rec.value == null_text && null_count_by_field.get(rec.field.as_str()) == Some(&rec.count)
        {
            continue;
        }
        samples_by_field
            .entry(rec.field.as_str())
            .or_default()
            .push(rec.value.as_str());
    }
    samples_by_field
}

/// Semantically validate the LLM-inferred strftime `:<fmt>` suffix on every
/// `date`/`datetime` entry against the real on-disk sample values.
///
/// `normalize_datetime_token` only checks the suffix is *syntactically* valid
/// chrono strftime; a well-formed format can still mismatch the data (the LLM
/// guesses `%m/%d/%Y` for `%d/%m/%Y` data). For each date/datetime entry that
/// carries a suffix, this attempts to parse EVERY usable raw value from the
/// frequency records — skipping the rank-0 "Other" bucket and the
/// `<ALL_UNIQUE>` sentinel — with the inferred format. An ambiguous format can
/// parse an early sample (`01/02/2020`) yet be disproven by a later one
/// (`13/02/2020`), so all samples must parse; if any fails, the suffix is
/// stripped back to the bare token.
///
/// Entries with no usable sample (an `ALL_UNIQUE` date column, or one whose only
/// frequency row is the "Other" bucket) keep their suffix unchanged — there is
/// nothing to validate them against.
pub(super) fn validate_date_formats(
    entries: &mut [DictionaryEntry],
    frequency_records: &[FrequencyRecord],
    null_text: &str,
) {
    let samples_by_field = usable_samples_by_field(frequency_records, entries, null_text);

    for entry in entries {
        let base = content_type_base(&entry.content_type);
        if base != "date" && base != "datetime" {
            continue;
        }
        let Some(fmt) = entry
            .content_type
            .strip_prefix("datetime:")
            .or_else(|| entry.content_type.strip_prefix("date:"))
        else {
            continue; // bare token, nothing to validate
        };
        let Some(samples) = samples_by_field.get(entry.name.as_str()) else {
            continue; // no usable sample (e.g. ALL_UNIQUE) - accept as-is
        };
        // Every usable sample must parse: an ambiguous format can match the
        // first sample yet be disproven by a later one. If any fails, the
        // inferred format is wrong for this column - strip back to bare token.
        if !samples
            .iter()
            .all(|s| sample_parses_with_format(s, fmt, base == "datetime"))
        {
            entry.content_type = base.to_string();
        }
    }
}

/// Produce the date-only portion of a chrono strftime format by truncating at
/// the first time-related specifier (`%H`, `%I`, `%M`, `%S`, `%p`, `%z`, …).
///
/// Used when an all-midnight `datetime` column is reclassified as `date`: the
/// on-disk format `%m/%d/%Y %I:%M:%S %p` becomes the date-only `%m/%d/%Y`.
///
/// Returns `None` when there is no clean date-only prefix — a datetime-compound
/// specifier (`%c`, `%+`, `%s`) or a time-first layout — so the caller falls
/// back to a bare `date` token.
fn strip_time_from_format(fmt: &str) -> Option<String> {
    let bytes = fmt.as_bytes();
    let mut i = 0;
    let mut cut: Option<usize> = None;
    while i < bytes.len() {
        if bytes[i] != b'%' {
            i += 1;
            continue;
        }
        let pct = i;
        i += 1; // past '%'
        if i >= bytes.len() {
            break;
        }
        if bytes[i] == b'%' {
            i += 1; // literal "%%"
            continue;
        }
        // skip modifier chars (`-_0.:#` and digits) to reach the specifier letter
        while i < bytes.len() && matches!(bytes[i], b'-' | b'_' | b'0'..=b'9' | b'.' | b':' | b'#')
        {
            i += 1;
        }
        if i >= bytes.len() {
            break;
        }
        let letter = bytes[i] as char;
        i += 1;
        match letter {
            // datetime-compound: no clean date-only prefix
            'c' | '+' | 's' => return None,
            // time-related specifier: the date part ends at this `%`
            'H' | 'k' | 'I' | 'l' | 'P' | 'p' | 'M' | 'S' | 'f' | 'R' | 'T' | 'X' | 'r' | 'Z'
            | 'z' => {
                cut = Some(pct);
                break;
            },
            // date specifier (or a harmless literal like `%n`/`%t`) — keep scanning
            _ => {},
        }
    }
    let date_part = cut.map_or(fmt, |c| &fmt[..c]);
    // trim the trailing date/time separator literal (whitespace, the ISO `T`,
    // or punctuation like `-` `/` `.` `:` `,` `_`) so a format such as
    // `%Y-%m-%d-%H:%M:%S` yields `%Y-%m-%d`, not `%Y-%m-%d-`
    let trimmed = date_part.trim_end_matches(|c: char| {
        c.is_whitespace() || matches!(c, 'T' | '-' | '/' | '.' | ':' | ',' | '_')
    });
    (!trimmed.is_empty()).then(|| trimmed.to_string())
}

/// `Some(true)` if `sample` parses with the chrono strftime `fmt` AND its
/// time-of-day is exactly midnight; `Some(false)` if it parses with a
/// non-midnight time; `None` if it does not parse at all (so the caller can
/// ignore it rather than treat it as evidence either way).
fn sample_time_at_midnight(sample: &str, fmt: &str) -> Option<bool> {
    use chrono::{DateTime, NaiveDateTime, Timelike};
    let time = NaiveDateTime::parse_from_str(sample, fmt)
        .map(|dt| dt.time())
        .or_else(|_| DateTime::parse_from_str(sample, fmt).map(|dt| dt.time()))
        .ok()?;
    Some(time.num_seconds_from_midnight() == 0 && time.nanosecond() == 0)
}

/// Reclassify a `datetime` column as `date` when every usable frequency sample
/// parses with the inferred format AND falls exactly on midnight — i.e. the
/// column holds dates stored with a zero time-of-day. `qsv stats` types a
/// column `DateTime` whenever ANY value has a non-midnight time, which
/// over-reports columns whose dominant pattern (per the frequency
/// distribution) is plainly a date.
///
/// Runs after `validate_date_formats`, so a surviving `datetime:<fmt>` suffix
/// is already format-validated. On downgrade the time specifiers are stripped
/// from `<fmt>` via `strip_time_from_format`, yielding `date:<date-fmt>` (or a
/// bare `date` when the format has no clean date-only prefix).
///
/// Bare `datetime` (no inferred format) is left unchanged — there is no format
/// to test the samples against. A sample that does NOT parse with the format
/// blocks the downgrade (the column stays `datetime`): a non-parsing real
/// value could carry a non-midnight time, so ignoring it would risk a wrong
/// downgrade. The emitted null row never reaches here — `usable_samples_by_field`
/// excludes it by `--null-text` value plus null count, even when
/// `frequency --pct-nulls` gives it a real rank.
pub(super) fn downgrade_all_midnight_datetime_columns(
    entries: &mut [DictionaryEntry],
    frequency_records: &[FrequencyRecord],
    null_text: &str,
) {
    let samples_by_field = usable_samples_by_field(frequency_records, entries, null_text);

    for entry in entries {
        let Some(fmt) = entry.content_type.strip_prefix("datetime:") else {
            continue; // bare `datetime`, or not a datetime column
        };
        let Some(samples) = samples_by_field.get(entry.name.as_str()) else {
            continue; // no usable sample to judge by
        };
        // EVERY usable sample must parse with the format: collecting into an
        // `Option<Vec<_>>` yields `None` if any sample fails to parse, which
        // keeps the column `datetime` rather than silently dropping a value
        // that might carry a non-midnight time.
        let Some(midnight_flags) = samples
            .iter()
            .map(|s| sample_time_at_midnight(s, fmt))
            .collect::<Option<Vec<bool>>>()
        else {
            continue;
        };
        if !midnight_flags.is_empty() && midnight_flags.iter().all(|&midnight| midnight) {
            entry.content_type = strip_time_from_format(fmt)
                .map_or_else(|| "date".to_string(), |date_fmt| format!("date:{date_fmt}"));
        }
    }
}

/// Extract the `{field_name: {label, description[, content_type]}}` map from the
/// LLM's JSON response, restricted to the given `field_names`. When
/// `infer_content_type` is set, `content_type` is validated against
/// `CONTENT_TYPE_VOCAB`; a missing, empty, or out-of-vocabulary value is left empty
/// here — `combine_dictionary_entries` is the single point that coerces any
/// still-empty `content_type` to `"unknown"`. When the flag is unset, `content_type`
/// is always empty.
///
/// Most tokens are lowercased before the vocab lookup, but `date`/`datetime` are
/// matched case-preserving via `normalize_datetime_token` because their optional
/// `:<fmt>` chrono strftime suffix is case-sensitive.
///
/// `unique_id` is in the vocab but is REJECTED from LLM input here: it is set
/// deterministically based on the `<ALL_UNIQUE>` frequency sentinel and the LLM
/// has no way to verify that condition, so accepting it would let non-ALL_UNIQUE
/// fields be misclassified.
pub(super) fn parse_llm_dictionary_response(
    llm_response: &str,
    field_names: &[String],
    infer_content_type: bool,
) -> CliResult<HashMap<String, LlmDictField>> {
    let json_value = extract_json_from_output(llm_response)?;

    let mut result = HashMap::new();

    if let Some(obj) = json_value.as_object() {
        for field_name in field_names {
            if let Some(field_obj) = obj.get(field_name)
                && let Some(field_map) = field_obj.as_object()
            {
                let label = field_map
                    .get("label")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
                    .to_string();

                let description = field_map
                    .get("description")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
                    .to_string();

                let content_type = if infer_content_type {
                    // `CONTENT_TYPE_VOCAB` is all lowercase and LLMs don't reliably echo
                    // casing (e.g. "Email", "First_Name"), so most tokens are case-folded
                    // before the vocab lookup. A missing, empty, or out-of-vocabulary value
                    // is left empty here; `combine_dictionary_entries` coerces any
                    // still-empty content_type to "unknown" when the flag is set.
                    //
                    // `date`/`datetime` are handled FIRST and case-preserving: they may
                    // carry a `:<fmt>` chrono strftime suffix (e.g. "datetime:%m/%d/%Y
                    // %I:%M:%S %p") whose specifiers are case-sensitive (%m vs %M, %p), so
                    // `normalize_datetime_token` lowercases only the token prefix.
                    //
                    // `duration` is special: the LLM may append an upper-bound suffix (e.g.
                    // "duration:3600") that isn't in `CONTENT_TYPE_VOCAB` literally, so
                    // route it through `normalize_duration_token`.
                    //
                    // `unique_id` is REJECTED here even though it is in the vocab: it is set
                    // deterministically by `generate_code_based_dictionary` based on the
                    // `<ALL_UNIQUE>` frequency sentinel (`cardinality == rowcount`), and the
                    // LLM has no way to verify that condition. Accepting it from LLM output
                    // would let non-ALL_UNIQUE fields be misclassified as `unique_id`,
                    // breaking the deterministic-only contract documented on
                    // `CONTENT_TYPE_VOCAB`.
                    let raw_trimmed = field_map
                        .get("content_type")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .trim();
                    if let Some(normalized) = normalize_datetime_token(raw_trimmed) {
                        normalized
                    } else {
                        let raw = raw_trimmed.to_ascii_lowercase();
                        if let Some(normalized) = normalize_duration_token(&raw) {
                            normalized
                        } else if raw == "unique_id" {
                            String::new()
                        } else if CONTENT_TYPE_VOCAB.contains(&raw.as_str()) {
                            raw
                        } else {
                            String::new()
                        }
                    }
                } else {
                    String::new()
                };

                // `concept` and `role` ride the same `infer_content_type` gate as
                // `content_type`. `concept` is validated against `CONCEPT_VOCAB`
                // (case-folded); `id.surrogate_key` is deterministic and refused
                // from LLM input, mirroring the `unique_id` content_type rejection.
                // `role` is validated against `ROLE_VOCAB`. Out-of-vocab / missing
                // values are left empty; `combine_dictionary_entries` applies the
                // deterministic seed and the final fallback.
                let (concept, role) = if infer_content_type {
                    let concept_raw = field_map
                        .get("concept")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .trim()
                        .to_ascii_lowercase();
                    let concept = if concept_raw == "id.surrogate_key"
                        || !CONCEPT_VOCAB.contains(&concept_raw.as_str())
                    {
                        String::new()
                    } else {
                        concept_raw
                    };
                    let role_raw = field_map
                        .get("role")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .trim()
                        .to_ascii_lowercase();
                    let role = if ROLE_VOCAB.contains(&role_raw.as_str()) {
                        role_raw
                    } else {
                        String::new()
                    };
                    (concept, role)
                } else {
                    (String::new(), String::new())
                };

                // Parsed unconditionally rather than behind a flag: when the dictionary
                // prompt never asks for null sentinels the key is simply absent and this
                // yields an empty vec, so no gate is needed here. Gating lives at the
                // prompt (what we ask) and at emit (`build_x_qsv` only writes non-empty
                // lists), which keeps a no-flag dictionary byte-identical while still
                // honoring a custom --prompt-file that elects to supply them.
                //
                // Only JSON strings are accepted; a stray number/object/null in the array
                // is skipped rather than stringified, so `-999` must arrive as "-999".
                let null_values = field_map
                    .get("null_values")
                    .and_then(|v| v.as_array())
                    .map(|arr| {
                        arr.iter()
                            .filter_map(|v| v.as_str())
                            .map(str::trim)
                            .filter(|s| !s.is_empty())
                            .map(ToString::to_string)
                            .collect()
                    })
                    .unwrap_or_default();

                // `gauge_range` rides the same `infer_content_type` gate as role/concept
                // (a gauge only makes sense once the field is classed as a measure). We
                // accept only a well-formed `[min, max]` array of two FINITE numbers with
                // `min < max`; anything else (missing key, wrong arity, NaN/inf, min>=max)
                // yields `None`. This is purely SHAPE validation — whether the range is
                // appropriate (role == measure, observed data lies inside it) is decided
                // later in `combine_dictionary_entries`, the only place that sees the
                // finalized role and the column's observed min/max.
                let gauge_range = if infer_content_type {
                    field_map
                        .get("gauge_range")
                        .and_then(|v| v.as_array())
                        .and_then(|a| match a.as_slice() {
                            [lo, hi] => Some((lo.as_f64()?, hi.as_f64()?)),
                            _ => None,
                        })
                        .filter(|(lo, hi)| lo.is_finite() && hi.is_finite() && lo < hi)
                        .map(|(lo, hi)| [lo, hi])
                } else {
                    None
                };

                result.insert(
                    field_name.clone(),
                    LlmDictField {
                        label,
                        description,
                        content_type,
                        concept,
                        role,
                        null_values,
                        gauge_range,
                    },
                );
            }
        }
    }

    // A response that parsed but matched NOT ONE of the real column names is a red flag, not
    // an empty dictionary. The usual cause is malformed JSON: `extract_json_from_output`
    // fails on the whole object, falls through to its brace scan, and returns the first
    // NESTED object that happens to parse -- typically one field's `{label, description,
    // ...}` body. Every lookup below then misses, and without this warning the run silently
    // produces a dictionary with no LLM labels at all (single-pass) or silently discards the
    // baseline pass (`--two-pass`), having already paid for the inference.
    //
    // Guarded on non-empty `field_names` so a genuinely column-less input stays quiet.
    if result.is_empty() && !field_names.is_empty() {
        wwarn!(
            "LLM response yielded no usable fields (none of the {} column name(s) matched). The \
             response was likely malformed JSON; any grain/relationships it carried were dropped \
             too. Re-run to retry the inference.",
            field_names.len()
        );
    }

    Ok(result)
}

/// Upper bound on declared stages in a row-encoded `pipeline`, so a malformed
/// response cannot declare a thousand-band funnel. `viz` applies its own, tighter
/// display cap separately; this is only a structural-sanity guard.
const MAX_DECLARED_STAGES: usize = 64;

/// Extract the optional top-level `relationships` array from the LLM's dictionary
/// response, validated against `field_names`. Each surviving entry is a clean
/// JSON object ready to embed in the dictionary output and consume from
/// `synthesize` and `viz smart`.
///
/// Recognized kinds are `joint` / `ordered` / `correlated` / `pipeline`. An entry
/// is dropped when its kind is unknown, when it has fewer than two members, or
/// when any member (or an `ordered` anchor) names a column not in `field_names`.
/// Anything that isn't well-formed JSON yields an empty list — relationship
/// inference is best-effort and never fails the dictionary phase.
///
/// `pipeline` (issue #4222) declares a process whose stages narrow monotonically,
/// in either of two encodings. Stages-as-COLUMNS uses `members` alone, listed
/// upstream-first — the OPPOSITE direction from `ordered`, which is ascending.
/// Stages-as-ROW-VALUES additionally carries `stage_column`, an ordered `stages`
/// array of that column's values, and an optional `value_column` to sum. The
/// presence of `stage_column` is what discriminates the two.
///
/// **Every emitted entry carries `members`, unconditionally.** `synthesize`'s
/// `SynthRelationship::members` has no serde default (unlike `anchor`) and the
/// dictionary is deserialized in a single pass, so an entry lacking it does not
/// degrade — it hard-errors `synthesize --dictionary`. For the row encoding the
/// members are therefore SYNTHESIZED here from `stage_column`/`value_column`
/// rather than taken from the LLM.
pub(super) fn parse_llm_relationships(
    llm_response: &str,
    field_names: &[String],
) -> Vec<serde_json::Value> {
    let Ok(json_value) = extract_json_from_output(llm_response) else {
        return Vec::new();
    };
    let Some(raw) = json_value.get("relationships").and_then(|v| v.as_array()) else {
        return Vec::new();
    };
    let in_fields = |name: &str| field_names.iter().any(|f| f == name);
    let str_field = |obj: &serde_json::Map<String, serde_json::Value>, k: &str| {
        obj.get(k)
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(ToString::to_string)
    };

    let mut result = Vec::new();
    for entry in raw {
        let Some(obj) = entry.as_object() else {
            continue;
        };
        let kind = obj.get("kind").and_then(|v| v.as_str()).unwrap_or_default();
        if !matches!(kind, "joint" | "ordered" | "correlated" | "pipeline") {
            continue;
        }

        // A `pipeline` carrying `stage_column` is the ROW encoding: its stages are
        // values inside one column, which `members` cannot describe, so `members`
        // is synthesized from the column references instead.
        if kind == "pipeline"
            && let Some(stage_column) = str_field(obj, "stage_column")
        {
            if !in_fields(&stage_column) {
                continue;
            }
            let stages: Vec<String> = obj
                .get("stages")
                .and_then(|v| v.as_array())
                .map(|arr| {
                    arr.iter()
                        .filter_map(|s| s.as_str())
                        .map(str::trim)
                        .filter(|s| !s.is_empty())
                        .map(ToString::to_string)
                        .collect()
                })
                .unwrap_or_default();
            let mut seen_stages = std::collections::HashSet::new();
            if stages.len() < 2
                || stages.len() > MAX_DECLARED_STAGES
                || !stages.iter().all(|s| seen_stages.insert(s.as_str()))
            {
                continue;
            }
            // An unusable `value_column` drops the WHOLE entry rather than falling
            // back to counting rows: silently turning a declared measure funnel
            // into a row-count funnel changes what the chart claims.
            let value_column = match obj.get("value_column") {
                Some(v) if !v.is_null() => {
                    let Some(vc) = v.as_str().map(str::trim).filter(|s| !s.is_empty()) else {
                        continue;
                    };
                    if !in_fields(vc) || vc == stage_column {
                        continue;
                    }
                    Some(vc.to_string())
                },
                _ => None,
            };

            let mut members = vec![stage_column.clone()];
            members.extend(value_column.clone());

            let mut clean = serde_json::Map::new();
            clean.insert("kind".to_string(), serde_json::json!(kind));
            clean.insert("members".to_string(), serde_json::json!(members));
            clean.insert("stage_column".to_string(), serde_json::json!(stage_column));
            clean.insert("stages".to_string(), serde_json::json!(stages));
            if let Some(vc) = value_column {
                clean.insert("value_column".to_string(), serde_json::json!(vc));
            }
            result.push(serde_json::Value::Object(clean));
            continue;
        }

        let members: Vec<String> = obj
            .get("members")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|m| m.as_str())
                    .map(ToString::to_string)
                    .collect()
            })
            .unwrap_or_default();
        if members.len() < 2 || !members.iter().all(|m| in_fields(m)) {
            continue;
        }
        // A pipeline naming the same column at two ranks is malformed; the other
        // kinds are unordered sets, where a repeat is merely redundant.
        if kind == "pipeline" {
            let mut seen = std::collections::HashSet::new();
            if !members.iter().all(|m| seen.insert(m.as_str())) {
                continue;
            }
        }

        let mut clean = serde_json::Map::new();
        clean.insert("kind".to_string(), serde_json::json!(kind));
        clean.insert("members".to_string(), serde_json::json!(members));
        // `anchor` is meaningful only for `ordered` groups; keep it when the LLM
        // supplied one that names a real column.
        if kind == "ordered"
            && let Some(anchor) = obj.get("anchor").and_then(|v| v.as_str())
            && in_fields(anchor)
        {
            clean.insert("anchor".to_string(), serde_json::json!(anchor));
        }
        result.push(serde_json::Value::Object(clean));
    }
    result
}

/// Extract the optional top-level `grain` string from the LLM's dictionary
/// response — a one-sentence statement of what a single row represents (e.g.
/// "one row = one 311 service request"). Best-effort and dataset-level (it is
/// NOT a per-field value, so it rides alongside `parse_llm_relationships` rather
/// than through `LlmDictField`): malformed JSON or a missing/blank `grain` yields
/// `None`, never failing the dictionary phase.
pub(super) fn parse_llm_grain(llm_response: &str) -> Option<String> {
    let json_value = extract_json_from_output(llm_response).ok()?;
    let grain = json_value.get("grain")?.as_str()?.trim();
    (!grain.is_empty()).then(|| grain.to_string())
}

/// Extract the optional top-level `grain_unit` string from the LLM's dictionary
/// response — the bare entity noun behind the `grain` sentence ("311 service
/// request"), which `viz smart` renders VERBATIM as the subject of its
/// count-over-time chart titles.
///
/// This exists so viz never has to parse a natural-language sentence: the older
/// path split `grain` on the English literal `" one "`, which silently leaked
/// English into otherwise-localized dashboards (issue #4321) and could not work
/// at all for languages that use neither articles nor spaces. Under
/// `--language xx` the prompt asks for this field in the target language, so it
/// can be interpolated directly into a localized title template.
///
/// Best-effort, exactly like `parse_llm_grain`: malformed JSON or a
/// missing/blank `grain_unit` yields `None`, never failing the dictionary phase.
pub(super) fn parse_llm_grain_unit(llm_response: &str) -> Option<String> {
    let json_value = extract_json_from_output(llm_response).ok()?;
    let grain_unit = json_value.get("grain_unit")?.as_str()?.trim();
    (!grain_unit.is_empty()).then(|| grain_unit.to_string())
}

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

    fn blank_entry(name: &str) -> DictionaryEntry {
        DictionaryEntry {
            name:            name.to_string(),
            r#type:          "String".to_string(),
            label:           String::new(),
            description:     String::new(),
            content_type:    String::new(),
            min:             String::new(),
            max:             String::new(),
            cardinality:     0,
            enumeration:     String::new(),
            null_count:      0,
            addl_cols:       IndexMap::new(),
            examples:        String::new(),
            examples_full:   String::new(),
            freq_details:    Vec::new(),
            is_unique_id:    false,
            concept:         String::new(),
            role:            String::new(),
            null_values:     Vec::new(),
            null_candidates: Vec::new(),
            gauge_range:     None,
        }
    }

    /// Build an entry, apply the merge-path coercion, and return the resulting role.
    fn coerced_role(content_type: &str, role: &str, qsv_type: &str) -> String {
        let mut entry = blank_entry("col");
        entry.r#type = qsv_type.to_string();
        entry.content_type = content_type.to_string();
        entry.role = role.to_string();
        coerce_role_concept(&mut entry);
        entry.role
    }

    // issue #4177 — `content_type` constrains `role`. The matrix below makes the
    // precedence rule explicit across the four column families hiSandog called out:
    // temporal, identifier, categorical, and numeric.

    #[test]
    fn coerce_role_temporal_content_type_forces_timestamp() {
        // The reported repro: a `time` column the LLM tagged `role: measure`.
        assert_eq!(coerced_role("time", "measure", "String"), "timestamp");
        // A temporal content_type overrides ANY inferred role, and a `:fmt` suffix
        // is stripped before the check.
        for ct in [
            "date",
            "datetime",
            "time",
            "date:%Y-%m-%d",
            "datetime:%Y-%m-%dT%H:%M:%S",
        ] {
            for role in ["measure", "dimension", "timestamp", ""] {
                assert_eq!(
                    coerced_role(ct, role, "String"),
                    "timestamp",
                    "content_type={ct} role={role:?} should coerce to timestamp"
                );
            }
        }
    }

    #[test]
    fn coerce_role_identifier_content_type_forces_identifier() {
        // id-family content_types must never be a `measure`; a stray one becomes
        // `identifier` (an entity key), not `dimension`.
        for ct in [
            "uuid",
            "unique_id",
            "isbn",
            "credit_card",
            "ip_address",
            "ipv6_address",
            "mac_address",
            "license_plate",
        ] {
            assert_eq!(
                coerced_role(ct, "measure", "String"),
                "identifier",
                "content_type={ct} role=measure should coerce to identifier"
            );
        }
    }

    #[test]
    fn coerce_role_categorical_content_type_demotes_measure_to_dimension() {
        // A recognized non-numeric, non-id content_type tagged `measure` is a
        // contradiction — coerce it to `dimension`.
        for ct in ["category", "state", "country", "currency_code", "mime_type"] {
            assert_eq!(
                coerced_role(ct, "measure", "String"),
                "dimension",
                "content_type={ct} role=measure should coerce to dimension"
            );
            // A legitimate `dimension` role is left untouched.
            assert_eq!(coerced_role(ct, "dimension", "String"), "dimension");
        }
    }

    #[test]
    fn coerce_role_numeric_measure_is_not_over_coerced() {
        // Numeric measures legitimately carry `content_type: unknown` — role=measure
        // must survive, whether inferred by the LLM or filled from the qsv type.
        assert_eq!(coerced_role("unknown", "measure", "Integer"), "measure");
        assert_eq!(coerced_role("unknown", "measure", "Float"), "measure");
        assert_eq!(coerced_role("unknown", "", "Integer"), "measure");
        assert_eq!(coerced_role("unknown", "", "Float"), "measure");
        // `duration:N` is a magnitude you aggregate, so it is measure-admissible and
        // exempt from the temporal→timestamp and categorical→dimension coercions.
        assert_eq!(coerced_role("duration", "measure", "Integer"), "measure");
        assert_eq!(
            coerced_role("duration:3600", "measure", "Integer"),
            "measure"
        );
    }

    /// Build an entry, apply the merge-path coercion, and return `(role, concept)`.
    fn coerced_role_concept(
        content_type: &str,
        role: &str,
        concept: &str,
        qsv_type: &str,
    ) -> (String, String) {
        let mut entry = blank_entry("col");
        entry.r#type = qsv_type.to_string();
        entry.content_type = content_type.to_string();
        entry.role = role.to_string();
        entry.concept = concept.to_string();
        coerce_role_concept(&mut entry);
        (entry.role, entry.concept)
    }

    #[test]
    fn coerce_concept_contradicting_content_type_is_reset() {
        // issue #4176 repro: content_type=category and role=dimension both say
        // categorical, only concept=measure.ratio dissents. content_type breaks the
        // tie: the concept is reset (category has no deterministic concept -> unknown),
        // and role is left categorical.
        let (role, concept) =
            coerced_role_concept("category", "dimension", "measure.ratio", "String");
        assert_eq!(role, "dimension");
        assert_eq!(concept, "unknown");

        // A concept whose content_type HAS a deterministic mapping is backfilled to the
        // consistent value rather than dropped to "unknown".
        let (_, concept) =
            coerced_role_concept("zip_code", "dimension", "measure.amount", "String");
        assert_eq!(concept, "geo.zip_code");

        // Temporal content_type with a stray measure concept: role forced to timestamp,
        // concept reset (time has no deterministic concept -> unknown).
        let (role, concept) = coerced_role_concept("time", "measure", "measure.count", "String");
        assert_eq!(role, "timestamp");
        assert_eq!(concept, "unknown");
    }

    #[test]
    fn coerce_concept_override_on_unknown_content_type_is_preserved() {
        // The signal `route_from_concept` exists for: a specific concept on an untyped
        // numeric column (content_type=unknown) must WIN, so it is NOT reset even though
        // its namespace (geo -> dimension) disagrees with a role-defaulted `measure`.
        let (role, concept) =
            coerced_role_concept("unknown", "measure", "geo.census_tract", "Integer");
        assert_eq!(role, "measure");
        assert_eq!(concept, "geo.census_tract");

        // A concept that AGREES with a recognized content_type is untouched.
        let (_, concept) = coerced_role_concept("email", "dimension", "pii.email", "String");
        assert_eq!(concept, "pii.email");
    }

    #[test]
    fn coerce_concept_same_role_bucket_cross_namespace_is_reset() {
        // A content_type that pins a deterministic concept must win against a concept in a
        // DIFFERENT namespace even when both share a role bucket. `email` -> `pii.email`
        // and a supplied `geo.city` are both the `dimension` bucket, so a role-only check
        // would miss it and leave a PII field falsely tagged as linkable geography.
        let (_, concept) = coerced_role_concept("email", "dimension", "geo.city", "String");
        assert_eq!(concept, "pii.email");

        // Same namespace but a different leaf is an LLM refinement, left untouched.
        let (_, concept) =
            coerced_role_concept("first_name", "dimension", "pii.full_name", "String");
        assert_eq!(concept, "pii.full_name");

        // A deterministic-concept content_type with a stray measure concept is corrected
        // to the deterministic value (not merely dropped to "unknown").
        let (_, concept) = coerced_role_concept("city", "dimension", "measure.amount", "String");
        assert_eq!(concept, "geo.city");
    }

    #[test]
    fn coerce_concept_coordinate_content_type_requires_exact_match() {
        // roborev #3584: coordinate concepts route to MapCoord, while other geo.* route to
        // Dimension, so a same-namespace sibling on a latitude/longitude column suppresses
        // map rendering. Unlike coarse mappings, these enforce the EXACT concept.
        let (_, concept) = coerced_role_concept("latitude", "dimension", "geo.city", "String");
        assert_eq!(concept, "geo.latitude");
        let (_, concept) = coerced_role_concept("longitude", "dimension", "geo.zip_code", "String");
        assert_eq!(concept, "geo.longitude");
        // A swapped coordinate concept is corrected too.
        let (_, concept) = coerced_role_concept("latitude", "measure", "geo.longitude", "Float");
        assert_eq!(concept, "geo.latitude");
        // The correct coordinate concept is preserved.
        let (_, concept) = coerced_role_concept("longitude", "measure", "geo.longitude", "Float");
        assert_eq!(concept, "geo.longitude");
    }

    #[test]
    fn coerce_concept_exact_namespace_sibling_is_reset() {
        // roborev #3587: geo/pii/org content_types map EXACTLY, so a same-namespace sibling
        // is a real contradiction (wrong join-compatible identity). `city` pins `geo.city`,
        // so a supplied `geo.country` (both linkable geo.*) is corrected, not preserved.
        assert_eq!(
            coerced_role_concept("city", "dimension", "geo.country", "String").1,
            "geo.city"
        );
        assert_eq!(
            coerced_role_concept("state", "dimension", "geo.city", "String").1,
            "geo.state"
        );
        assert_eq!(
            coerced_role_concept("email", "dimension", "pii.phone", "String").1,
            "pii.email"
        );
    }

    #[test]
    fn coerce_concept_refinable_namespace_sibling_is_preserved() {
        // A few content_types map coarsely to a generic concept, so a more specific concept
        // from the admissible set is a legitimate LLM refinement and is kept.
        assert_eq!(
            coerced_role_concept("date", "timestamp", "time.created_at", "Date").1,
            "time.created_at"
        );
        assert_eq!(
            coerced_role_concept("datetime", "timestamp", "time.due_at", "DateTime").1,
            "time.due_at"
        );
        assert_eq!(
            coerced_role_concept("uuid", "identifier", "id.foreign_key", "String").1,
            "id.foreign_key"
        );
        // roborev #3590: `company_name` -> `org.company` is coarse (a government "Agency Name"
        // column has no `agency_name` content_type), so a supplied `org.agency` is preserved.
        assert_eq!(
            coerced_role_concept("company_name", "dimension", "org.agency", "String").1,
            "org.agency"
        );
        // `industry` -> `org.industry` is EXACT: a same-namespace sibling is reset.
        assert_eq!(
            coerced_role_concept("industry", "dimension", "org.company", "String").1,
            "org.industry"
        );
        // A CROSS-namespace concept on a refinable mapping is still reset.
        assert_eq!(
            coerced_role_concept("uuid", "identifier", "geo.city", "String").1,
            "id.uuid"
        );
        assert_eq!(
            coerced_role_concept("company_name", "dimension", "geo.city", "String").1,
            "org.company"
        );
    }

    #[test]
    fn coerce_concept_same_namespace_non_admissible_refinement_is_reset() {
        // roborev #3593: refinability is an EXACT admissible set, not a whole namespace.
        // `company_name` admits org.company/org.agency but NOT org.industry (an industry is a
        // classification, not an organization's name) — both are linkable, so the mistag
        // corrupts join identity.
        assert_eq!(
            coerced_role_concept("company_name", "dimension", "org.industry", "String").1,
            "org.company"
        );
        // A `date` is a point-in-time event, so `time.duration` (a span) is not admissible.
        assert_eq!(
            coerced_role_concept("date", "timestamp", "time.duration", "Date").1,
            "time.date"
        );
        // The reserved `id.surrogate_key` is not offered as a `uuid` refinement.
        assert_eq!(
            coerced_role_concept("uuid", "identifier", "id.surrogate_key", "String").1,
            "id.uuid"
        );
    }

    #[test]
    fn format_date_examples_reformats_values_keeps_counts() {
        // DateTime values with a time component, inferred date-only format:
        // values are reformatted, " [count]" suffixes preserved verbatim.
        let out = format_date_examples(
            "date:%m/%d/%Y",
            "01/24/2013 12:00:00 AM [5]\n01/07/2014 12:00:00 AM [3]",
        );
        assert_eq!(out, "01/24/2013 [5]\n01/07/2014 [3]");
    }

    #[test]
    fn format_date_examples_idempotent_when_already_formatted() {
        let out = format_date_examples("date:%m/%d/%Y", "01/24/2013 [5]");
        assert_eq!(out, "01/24/2013 [5]");
    }

    #[test]
    fn format_date_examples_respects_day_first_format() {
        // Ambiguous "01/02/2020" under a day-first inferred format must parse as
        // 1 Feb and round-trip unchanged, NOT be swapped to "02/01/2020" by a
        // hardcoded month-first preference.
        assert_eq!(
            format_date_examples("date:%d/%m/%Y", "01/02/2020 [5]"),
            "01/02/2020 [5]"
        );
        // Month-first format keeps month-first interpretation.
        assert_eq!(
            format_date_examples("date:%m/%d/%Y", "01/02/2020 [5]"),
            "01/02/2020 [5]"
        );
        // `format_date_value` collapses the tri-state with `.unwrap_or(false)`;
        // these asserts pin that collapsed behavior (see the tri-state test below
        // for what each format actually expresses).
        let prefer_dmy = |fmt: &str| strftime_dmy_preference(fmt).unwrap_or(false);
        assert!(prefer_dmy("%d/%m/%Y"));
        assert!(!prefer_dmy("%m/%d/%Y"));
        assert!(!prefer_dmy("%Y-%m-%d")); // ISO is month-before-day
        // Padding/variant modifiers must be recognized via StrftimeItems, not
        // a fixed token list: %0d/%0m, %_d/%_m, %-d/%-m, %e all map to the
        // Day/Month numeric fields regardless of padding.
        assert!(prefer_dmy("%0d/%0m/%Y"));
        assert!(prefer_dmy("%_d/%_m/%Y"));
        assert!(prefer_dmy("%-d/%-m/%Y"));
        assert!(prefer_dmy("%e/%m/%Y"));
        assert!(!prefer_dmy("%0m/%0d/%Y"));
        assert!(!prefer_dmy("%H:%M:%S")); // no date fields → default month-first
        // End-to-end through format_date_examples with a padded day-first fmt.
        assert_eq!(
            format_date_examples("date:%0d/%0m/%Y", "01/02/2020 [5]"),
            "01/02/2020 [5]"
        );
    }

    #[test]
    fn strftime_dmy_preference_is_tri_state() {
        // day before month → day-first, whatever the padding.
        assert_eq!(strftime_dmy_preference("%d/%m/%Y"), Some(true));
        assert_eq!(strftime_dmy_preference("%-d/%-m/%Y"), Some(true));
        // the year guard is gated on month-before-day, so a year-leading
        // day-first layout stays day-first.
        assert_eq!(strftime_dmy_preference("%Y/%d/%m"), Some(true));

        // month before day, no leading year → month-first (the case the
        // preference bit actually acts on).
        assert_eq!(strftime_dmy_preference("%m/%d/%Y"), Some(false));
        assert_eq!(strftime_dmy_preference("%0m/%0d/%Y"), Some(false));
        // %D expands to %m/%d/%y.
        assert_eq!(strftime_dmy_preference("%D"), Some(false));

        // year-leading layouts are unambiguous to qsv_dateparser, so they
        // express no actionable preference and must NOT override the caller's.
        assert_eq!(strftime_dmy_preference("%Y-%m-%d"), None);
        assert_eq!(strftime_dmy_preference("%y-%m-%d"), None);
        // %F expands to %Y-%m-%d.
        assert_eq!(strftime_dmy_preference("%F"), None);
        assert_eq!(strftime_dmy_preference("%Y-%m-%dT%H:%M:%S"), None);

        // no numeric day and/or month at all.
        assert_eq!(strftime_dmy_preference("%H:%M:%S"), None);
        assert_eq!(strftime_dmy_preference("%b %d, %Y"), None);
        assert_eq!(strftime_dmy_preference(""), None);
    }

    #[test]
    fn format_date_examples_passthrough_cases() {
        // No inferred format → unchanged.
        assert_eq!(
            format_date_examples("date", "01/24/2013 12:00:00 AM [5]"),
            "01/24/2013 12:00:00 AM [5]"
        );
        // Non-date content type → unchanged.
        assert_eq!(
            format_date_examples("category", "alpha [9]\nbeta [3]"),
            "alpha [9]\nbeta [3]"
        );
        // <ALL_UNIQUE> sentinel and empty input → unchanged.
        assert_eq!(
            format_date_examples("date:%m/%d/%Y", "<ALL_UNIQUE>"),
            "<ALL_UNIQUE>"
        );
        assert_eq!(format_date_examples("date:%m/%d/%Y", ""), "");
        // Unparseable frequency-bucket values pass through, real dates still format.
        assert_eq!(
            format_date_examples("date:%m/%d/%Y", "Other… [4091]\n01/24/2013 12:00:00 AM [5]"),
            "Other… [4091]\n01/24/2013 [5]"
        );
    }

    #[test]
    fn parse_llm_response_ignores_content_type_when_flag_off() {
        let json = r#"{
            "name": {"label": "Name", "description": "the name", "content_type": "first_name"}
        }"#;
        let fields = vec!["name".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, false).unwrap();
        let f = parsed.get("name").unwrap();
        assert_eq!(f.label, "Name");
        assert_eq!(f.description, "the name");
        assert!(
            f.content_type.is_empty(),
            "content_type must stay empty when infer_content_type is false"
        );
    }

    #[test]
    fn parse_llm_response_extracts_valid_content_type() {
        let json = r#"{
            "email_addr": {"label": "Email", "description": "an email", "content_type": "email"}
        }"#;
        let fields = vec!["email_addr".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert_eq!(parsed.get("email_addr").unwrap().content_type, "email");
    }

    #[test]
    fn parse_llm_response_normalizes_content_type_casing() {
        // LLMs don't reliably echo casing; a valid-but-differently-cased token must be
        // accepted and stored in its normalized lowercase form, not coerced to "unknown".
        let json = r#"{
            "a": {"label": "A", "description": "d", "content_type": "Email"},
            "b": {"label": "B", "description": "d", "content_type": "FIRST_NAME"},
            "c": {"label": "C", "description": "d", "content_type": "  Mac_Address  "}
        }"#;
        let fields = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert_eq!(parsed.get("a").unwrap().content_type, "email");
        assert_eq!(parsed.get("b").unwrap().content_type, "first_name");
        assert_eq!(parsed.get("c").unwrap().content_type, "mac_address");
    }

    #[test]
    fn parse_llm_response_rejects_llm_supplied_unique_id() {
        // `unique_id` is in CONTENT_TYPE_VOCAB but must be REJECTED from LLM
        // output: it is only valid when stamped deterministically based on the
        // `<ALL_UNIQUE>` sentinel. Accepting it from LLM input would let
        // non-ALL_UNIQUE fields be misclassified. The parser drops it (empties
        // the field); combine_dictionary_entries then coerces to "unknown".
        let json = r#"{
            "a": {"label": "A", "description": "d", "content_type": "unique_id"},
            "b": {"label": "B", "description": "d", "content_type": "UNIQUE_ID"},
            "c": {"label": "C", "description": "d", "content_type": "  Unique_ID  "}
        }"#;
        let fields = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert!(
            parsed.get("a").unwrap().content_type.is_empty(),
            "literal 'unique_id' must be stripped from LLM output"
        );
        assert!(
            parsed.get("b").unwrap().content_type.is_empty(),
            "uppercased 'UNIQUE_ID' must also be stripped (lowercased before check)"
        );
        assert!(
            parsed.get("c").unwrap().content_type.is_empty(),
            "padded/cased 'unique_id' must also be stripped after trim+lowercase"
        );
    }

    /// An `LlmDictField` carrying only null-sentinel proposals.
    fn llm_nulls(tokens: &[&str]) -> LlmDictField {
        LlmDictField {
            null_values: tokens.iter().map(ToString::to_string).collect(),
            gauge_range: None,
            ..Default::default()
        }
    }

    fn freq(field: &str, value: &str, count: u64, rank: f64) -> FrequencyRecord {
        FrequencyRecord {
            field: field.to_string(),
            value: value.to_string(),
            count,
            percentage: 1.0,
            rank,
        }
    }

    #[test]
    fn null_values_confirmed_only_for_observed_string_values() {
        // The happy path: a String column, a token the LLM proposed that qsv can
        // see in the data. This is the only combination that earns the trusted bucket.
        let (confirmed, candidates) = classify_null_values(
            "String",
            &["ok", "NULL", "pending"],
            &["NULL".to_string()],
            "(NULL)",
        );
        assert_eq!(confirmed, vec!["NULL"]);
        assert!(candidates.is_empty());
    }

    #[test]
    fn numeric_sentinel_is_never_confirmed_even_when_observed() {
        // THE load-bearing test. -999 is right there in the frequency distribution of
        // an Integer column, so a naive "did we see it?" check would confirm it. It
        // must NOT be: -999 parses as a valid integer and no scan can tell it from a
        // real reading. If this ever fails, a consumer could auto-blank real data.
        let (confirmed, candidates) = classify_null_values(
            "Integer",
            &["-999", "12", "45"],
            &["-999".to_string()],
            "(NULL)",
        );
        assert!(
            confirmed.is_empty(),
            "a numeric sentinel is structurally unconfirmable and must never reach null_values"
        );
        assert_eq!(candidates, vec!["-999"]);
    }

    #[test]
    fn unobserved_token_is_demoted_to_candidate() {
        // The model proposed a sentinel this column does not contain. qsv cannot
        // confirm it, so it is reported as a guess rather than silently trusted.
        let (confirmed, candidates) =
            classify_null_values("String", &["ok", "pending"], &["N/A".to_string()], "(NULL)");
        assert!(confirmed.is_empty());
        assert_eq!(candidates, vec!["N/A"]);
    }

    #[test]
    fn confirmation_emits_every_observed_casing_not_the_models_echo() {
        // A consumer must know every spelling it would have to blank, so the dictionary
        // names every observed casing - and names the spellings found IN THE DATA, never
        // the model's (here lowercase) echo of them.
        let (confirmed, candidates) = classify_null_values(
            "String",
            &["NULL", "null", "NuLl", "ok"],
            &["null".to_string()],
            "(NULL)",
        );
        assert_eq!(confirmed, vec!["NULL", "null", "NuLl"]);
        assert!(candidates.is_empty());
    }

    #[test]
    fn confirmation_sees_sentinels_beyond_the_displayed_examples() {
        // Regression (roborev 3559): confirmation must run against the FULL frequency
        // records, not the `--num-examples`-capped, display-truncated `freq_details`.
        // Here the sentinel sits at rank 7 - well past the default 5 examples - and is
        // long enough that the display copy would have been truncated with an ellipsis.
        // Sourcing from `examples` would falsely demote it to a candidate.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            ..blank_entry("status")
        }];
        let mut freqs: Vec<FrequencyRecord> = (1..=6)
            .map(|i| freq("status", &format!("value{i}"), 100 - i, i as f64))
            .collect();
        freqs.push(freq("status", "NOT APPLICABLE AT THIS SITE", 3, 7.0));
        let mut proposals = HashMap::new();
        proposals.insert(
            "status".to_string(),
            vec!["not applicable at this site".to_string()],
        );

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert_eq!(
            entries[0].null_values,
            vec!["NOT APPLICABLE AT THIS SITE"],
            "a sentinel past --num-examples, and longer than --truncate-str, must still confirm"
        );
        assert!(entries[0].null_candidates.is_empty());
    }

    #[test]
    fn aggregation_buckets_and_the_null_row_can_never_confirm_a_sentinel() {
        // `frequency` stamps aggregation buckets ("Other") with rank 0.0, and emits a
        // null row whose value is --null-text and whose count equals the column's null
        // count. Neither is a faithful datum: admitting either would let the literal
        // null label, or a rolled-up bucket name, masquerade as a confirmed sentinel.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            null_count: 3,
            ..blank_entry("status")
        }];
        let freqs = vec![
            freq("status", "ok", 90, 1.0),
            freq("status", "Other", 5, 0.0), // rank-0 aggregation bucket
            freq("status", "(NULL)", 3, 2.0), // the emitted null row (count == null_count)
        ];
        let mut proposals = HashMap::new();
        proposals.insert(
            "status".to_string(),
            vec!["Other".to_string(), "(NULL)".to_string()],
        );

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert!(
            entries[0].null_values.is_empty(),
            "neither an aggregation bucket nor the null row may confirm a sentinel, got {:?}",
            entries[0].null_values
        );
        // "Other" survives as an unconfirmable guess; "(NULL)" is frequency's label for
        // cells qsv already counts as null, so it is dropped rather than reported.
        assert_eq!(entries[0].null_candidates, vec!["Other"]);
    }

    #[test]
    fn a_proposal_echoing_frequencys_null_label_is_dropped_not_reported() {
        // Observed on real data: after `denull --apply` cleaned a column, the model saw
        // frequency's "(NULL)" null-row label in the Frequency Distribution and proposed
        // it as a sentinel for every cleaned column. Those cells are ALREADY null, so the
        // proposal is noise - drop it outright rather than emit a confirm_required
        // candidate against data that has no sentinels left.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            null_count: 5,
            ..blank_entry("depth")
        }];
        let freqs = vec![
            freq("depth", "12", 90, 1.0),
            freq("depth", "(NULL)", 5, 2.0),
        ];
        let mut proposals = HashMap::new();
        proposals.insert("depth".to_string(), vec!["(NULL)".to_string()]);

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert!(entries[0].null_values.is_empty());
        assert!(
            entries[0].null_candidates.is_empty(),
            "frequency's null label must be dropped, not reported as a candidate; got {:?}",
            entries[0].null_candidates
        );
    }

    #[test]
    fn a_non_numeric_proposal_for_a_numeric_column_is_impossible_and_dropped() {
        // Observed on real data: after cleaning, the model proposed "NULL" for the now
        // Integer-typed DataQuality. If any cell held that text, stats could not have typed
        // the column Integer - so it is impossible, not merely unobserved. Dropping it beats
        // asking a human to confirm a cell that cannot exist.
        let (confirmed, candidates) =
            classify_null_values("Integer", &["1", "2", "3"], &["NULL".to_string()], "(NULL)");
        assert!(confirmed.is_empty());
        assert!(
            candidates.is_empty(),
            "a non-numeric token cannot occur in a numeric column; got {candidates:?}"
        );
    }

    #[test]
    fn a_numeric_placeholder_for_a_numeric_column_still_rides_through_as_a_candidate() {
        // The impossibility guard must not swallow the case the whole feature exists for:
        // -999 parses as an integer, so it CAN occur, and only a human can confirm it.
        let (confirmed, candidates) = classify_null_values(
            "Integer",
            &["-999", "12"],
            &["-999".to_string(), "9999".to_string()],
            "(NULL)",
        );
        assert!(confirmed.is_empty());
        assert_eq!(candidates, vec!["-999", "9999"]);
    }

    #[test]
    fn a_column_genuinely_holding_the_null_label_as_a_value_still_confirms_it() {
        // The guard must not blanket-ban the token. A String column whose real, observed
        // value happens to be the literal text "(NULL)" - with a count that is NOT the
        // column's null count, so it is not frequency's null row - confirms normally.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            null_count: 0,
            ..blank_entry("tag")
        }];
        let freqs = vec![freq("tag", "ok", 90, 1.0), freq("tag", "(NULL)", 7, 2.0)];
        let mut proposals = HashMap::new();
        proposals.insert("tag".to_string(), vec!["(NULL)".to_string()]);

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert_eq!(
            entries[0].null_values,
            vec!["(NULL)"],
            "a real observed value must confirm even when it reads like the null label"
        );
    }

    #[test]
    fn a_custom_null_text_is_also_dropped_when_it_is_only_the_label() {
        // frequency's --null-text is configurable, so the guard keys off the configured
        // value, not the literal "(NULL)".
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            null_count: 4,
            ..blank_entry("depth")
        }];
        let freqs = vec![
            freq("depth", "12", 90, 1.0),
            freq("depth", "<MISSING>", 4, 2.0),
        ];
        let mut proposals = HashMap::new();
        proposals.insert("depth".to_string(), vec!["<MISSING>".to_string()]);

        apply_null_value_classification(&mut entries, &proposals, &freqs, "<MISSING>");
        assert!(entries[0].null_values.is_empty());
        assert!(entries[0].null_candidates.is_empty());
    }

    #[test]
    fn a_real_value_containing_the_all_unique_text_can_still_confirm() {
        // Regression (roborev 3560): the observed-value set excluded any raw value
        // CONTAINING "<ALL_UNIQUE>", not just frequency's structural sentinel row. A
        // String column that literally holds that text could therefore never confirm it.
        // The sentinel is stamped rank 0.0 by `frequency`, so rank alone already excludes
        // it - the text match only ever produced this false positive.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            ..blank_entry("tag")
        }];
        let freqs = vec![
            freq("tag", "ok", 90, 1.0),
            freq("tag", "<ALL_UNIQUE>", 4, 2.0), // a real datum, ranked like any other
        ];
        let mut proposals = HashMap::new();
        proposals.insert("tag".to_string(), vec!["<ALL_UNIQUE>".to_string()]);

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert_eq!(
            entries[0].null_values,
            vec!["<ALL_UNIQUE>"],
            "a real value that merely reads like the sentinel must still be confirmable"
        );
    }

    #[test]
    fn the_all_unique_sentinel_row_is_still_excluded_by_its_rank() {
        // The other half of the same coin: frequency emits the all-unique sentinel with
        // rank 0.0 (both emit sites do), so dropping the text match must NOT let it into
        // the observed set. This also holds for a custom `--all-unique-text`, which a text
        // match would have missed entirely.
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            ..blank_entry("id")
        }];
        let freqs = vec![
            freq("id", "<ALL_UNIQUE>", 100, 0.0),
            freq("id", "CUSTOM_UNIQUE_TEXT", 100, 0.0),
        ];
        let mut proposals = HashMap::new();
        proposals.insert(
            "id".to_string(),
            vec!["<ALL_UNIQUE>".to_string(), "CUSTOM_UNIQUE_TEXT".to_string()],
        );

        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert!(
            entries[0].null_values.is_empty(),
            "a rank-0 sentinel row must never confirm, whatever its text; got {:?}",
            entries[0].null_values
        );
        assert_eq!(entries[0].null_candidates.len(), 2);
    }

    #[test]
    fn parse_llm_response_collects_null_values_and_skips_non_strings() {
        // Only JSON strings are accepted. A bare number -999 (rather than "-999") is
        // skipped, not stringified - otherwise the array's type discipline would
        // silently decide what counts as a sentinel token.
        let json = r#"{"a": {"label": "A", "description": "d",
                       "null_values": ["NULL", "  N/A  ", "", -999, null]}}"#;
        let fields = vec!["a".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, false).unwrap();
        assert_eq!(
            parsed.get("a").unwrap().null_values,
            vec!["NULL".to_string(), "N/A".to_string()],
            "strings are trimmed and kept; empties, numbers and nulls are dropped"
        );
    }

    #[test]
    fn classification_pass_stamps_confirmed_and_candidate_onto_the_entry() {
        let mut entries = vec![DictionaryEntry {
            r#type: "String".to_string(),
            ..blank_entry("status")
        }];
        let freqs = vec![
            freq("status", "NULL", 10, 1.0),
            freq("status", "ok", 90, 2.0),
        ];
        let mut proposals = HashMap::new();
        proposals.insert(
            "status".to_string(),
            vec!["NULL".to_string(), "-999".to_string()],
        );
        apply_null_value_classification(&mut entries, &proposals, &freqs, "(NULL)");
        assert_eq!(entries[0].null_values, vec!["NULL"]);
        assert_eq!(
            entries[0].null_candidates,
            vec!["-999"],
            "an unobserved token rides along as a candidate on the same column"
        );
    }

    #[test]
    fn two_pass_refine_omitting_null_values_preserves_the_baseline() {
        // The refine prompt never asks for null sentinels, so its empty list means
        // "not re-stated", NOT "retracted". Taking it unconditionally would wipe the
        // first pass's sentinels on every --two-pass run.
        let mut baseline = HashMap::new();
        baseline.insert("status".to_string(), llm_nulls(&["NULL"]));
        let mut refine = HashMap::new();
        refine.insert(
            "status".to_string(),
            LlmDictField {
                label: "Status".to_string(),
                ..Default::default()
            },
        );
        let proposals = null_value_proposals(&baseline, Some(&refine));
        assert_eq!(
            proposals.get("status"),
            Some(&vec!["NULL".to_string()]),
            "a refine pass that omits null_values must not wipe the baseline's sentinels"
        );
    }

    #[test]
    fn two_pass_refine_restating_null_values_overrides_the_baseline() {
        let mut baseline = HashMap::new();
        baseline.insert("status".to_string(), llm_nulls(&["NULL"]));
        let mut refine = HashMap::new();
        refine.insert("status".to_string(), llm_nulls(&["N/A"]));
        let proposals = null_value_proposals(&baseline, Some(&refine));
        assert_eq!(proposals.get("status"), Some(&vec!["N/A".to_string()]));
    }

    #[test]
    fn combine_refuses_llm_supplied_unique_id_for_non_all_unique_field() {
        // Even if a future caller bypasses parse_llm_dictionary_response and
        // hands combine_dictionary_entries an LlmDictField with content_type =
        // "unique_id", combine must refuse to copy it onto a field whose
        // code-derived entry was empty (i.e. not ALL_UNIQUE). Such fields fall
        // through to "unknown" when the flag is on.
        let code_entries = vec![blank_entry("not_unique")];
        let mut llm = HashMap::new();
        llm.insert(
            "not_unique".to_string(),
            LlmDictField {
                label:        "X".to_string(),
                description:  "y".to_string(),
                content_type: "unique_id".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code_entries, &llm, true);
        assert_eq!(
            combined[0].content_type, "unknown",
            "smuggled LLM 'unique_id' on a non-ALL_UNIQUE field must be rejected and fall back to \
             'unknown'"
        );
    }

    #[test]
    fn parse_gauge_range_shape_validation() {
        // Only a well-formed [min, max] of two finite numbers with min < max survives.
        let names: Vec<String> = ["a", "b", "c", "d", "e", "f"]
            .iter()
            .map(ToString::to_string)
            .collect();
        let resp = r#"{
            "a": {"label":"A","description":"d","gauge_range":[0,100]},
            "b": {"label":"B","description":"d","gauge_range":[100,0]},
            "c": {"label":"C","description":"d","gauge_range":[5]},
            "d": {"label":"D","description":"d"},
            "e": {"label":"E","description":"d","gauge_range":["x","y"]},
            "f": {"label":"F","description":"d","gauge_range":[0,0]}
        }"#;
        let got = parse_llm_dictionary_response(resp, &names, true).unwrap();
        assert_eq!(got["a"].gauge_range, Some([0.0, 100.0]));
        assert_eq!(got["b"].gauge_range, None, "min >= max rejected");
        assert_eq!(got["c"].gauge_range, None, "wrong arity rejected");
        assert_eq!(got["d"].gauge_range, None, "missing key -> None");
        assert_eq!(got["e"].gauge_range, None, "non-numeric rejected");
        assert_eq!(got["f"].gauge_range, None, "min == max rejected");
    }

    #[test]
    fn parse_gauge_range_gated_off_without_infer_content_type() {
        let names = vec!["a".to_string()];
        let resp = r#"{"a":{"label":"A","description":"d","gauge_range":[0,100]}}"#;
        let got = parse_llm_dictionary_response(resp, &names, false).unwrap();
        assert_eq!(got["a"].gauge_range, None);
    }

    /// A numeric MEASURE whose observed [min, max] fits inside the proposed scale keeps it.
    #[test]
    fn combine_keeps_gauge_range_for_measure_when_observed_fits() {
        let mut e = blank_entry("score");
        e.r#type = "Float".to_string();
        e.min = "10".to_string();
        e.max = "90".to_string();
        let mut llm = HashMap::new();
        llm.insert(
            "score".to_string(),
            LlmDictField {
                role: "measure".to_string(),
                gauge_range: Some([0.0, 100.0]),
                ..Default::default()
            },
        );
        let out = combine_dictionary_entries(vec![e], &llm, true);
        assert_eq!(out[0].role, "measure");
        assert_eq!(out[0].gauge_range, Some([0.0, 100.0]));
    }

    /// A range the observed data spills past is DROPPED (a misleading gauge is worse than none).
    #[test]
    fn combine_drops_gauge_range_when_observed_exceeds_scale() {
        let mut e = blank_entry("pct");
        e.r#type = "Float".to_string();
        e.min = "10".to_string();
        e.max = "90".to_string(); // spills past [0, 50]
        let mut llm = HashMap::new();
        llm.insert(
            "pct".to_string(),
            LlmDictField {
                role: "measure".to_string(),
                gauge_range: Some([0.0, 50.0]),
                ..Default::default()
            },
        );
        let out = combine_dictionary_entries(vec![e], &llm, true);
        assert_eq!(out[0].gauge_range, None);
    }

    /// A gauge on a non-measure (dimension) is DROPPED regardless of the numbers.
    #[test]
    fn combine_drops_gauge_range_for_non_measure() {
        let e = blank_entry("category"); // String -> coerces to dimension
        let mut llm = HashMap::new();
        llm.insert(
            "category".to_string(),
            LlmDictField {
                role: "dimension".to_string(),
                gauge_range: Some([0.0, 100.0]),
                ..Default::default()
            },
        );
        let out = combine_dictionary_entries(vec![e], &llm, true);
        assert_eq!(out[0].role, "dimension");
        assert_eq!(out[0].gauge_range, None);
    }

    /// A String column the LLM mis-tags `role: measure` must NOT keep a gauge, even when
    /// its lexicographic min/max happen to parse as floats (roborev #3606).
    #[test]
    fn combine_drops_gauge_range_for_non_numeric_type() {
        let mut e = blank_entry("code"); // r#type stays "String"
        e.min = "10".to_string(); // numeric-looking strings that parse as f64
        e.max = "90".to_string();
        let mut llm = HashMap::new();
        llm.insert(
            "code".to_string(),
            LlmDictField {
                role: "measure".to_string(),
                gauge_range: Some([0.0, 100.0]),
                ..Default::default()
            },
        );
        let out = combine_dictionary_entries(vec![e], &llm, true);
        assert_eq!(out[0].r#type, "String");
        assert_eq!(
            out[0].gauge_range, None,
            "a non-numeric column must not emit a gauge even if role==measure"
        );
    }

    #[test]
    fn parse_llm_response_drops_out_of_vocab_content_type() {
        // An out-of-vocabulary token is left empty by parsing; combine_dictionary_entries
        // is what coerces it to "unknown".
        let json = r#"{
            "mystery": {"label": "X", "description": "y", "content_type": "made_up_token"}
        }"#;
        let fields = vec!["mystery".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert!(parsed.get("mystery").unwrap().content_type.is_empty());
    }

    #[test]
    fn normalize_duration_token_handles_all_forms() {
        // Bare token is the trivial accept.
        assert_eq!(
            normalize_duration_token("duration").as_deref(),
            Some("duration")
        );
        // Well-formed positive integer suffix: preserved verbatim.
        assert_eq!(
            normalize_duration_token("duration:3600").as_deref(),
            Some("duration:3600")
        );
        // Whitespace around the number is tolerated.
        assert_eq!(
            normalize_duration_token("duration: 18000").as_deref(),
            Some("duration:18000")
        );
        // Malformed suffixes degrade gracefully to bare "duration" rather
        // than dropping the classification entirely — the LLM picked
        // "duration" correctly, only the cap is bad.
        assert_eq!(
            normalize_duration_token("duration:0").as_deref(),
            Some("duration")
        );
        assert_eq!(
            normalize_duration_token("duration:-5").as_deref(),
            Some("duration")
        );
        assert_eq!(
            normalize_duration_token("duration:abc").as_deref(),
            Some("duration")
        );
        assert_eq!(
            normalize_duration_token("duration:").as_deref(),
            Some("duration")
        );
        // Non-duration tokens return None so the caller falls through to
        // the regular vocab check.
        assert_eq!(normalize_duration_token("time"), None);
        assert_eq!(normalize_duration_token("email"), None);
        assert_eq!(normalize_duration_token(""), None);
    }

    #[test]
    fn parse_llm_response_accepts_duration_suffix() {
        let json = r#"{
            "elapsed":    {"label": "E", "description": "d", "content_type": "duration:3600"},
            "race_time":  {"label": "R", "description": "d", "content_type": "Duration: 18000"},
            "bare":       {"label": "B", "description": "d", "content_type": "duration"},
            "bad_cap":    {"label": "X", "description": "d", "content_type": "duration:0"},
            "bad_suffix": {"label": "Y", "description": "d", "content_type": "duration:abc"}
        }"#;
        let fields = vec![
            "elapsed".to_string(),
            "race_time".to_string(),
            "bare".to_string(),
            "bad_cap".to_string(),
            "bad_suffix".to_string(),
        ];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert_eq!(parsed.get("elapsed").unwrap().content_type, "duration:3600");
        // Casing + inner whitespace normalized.
        assert_eq!(
            parsed.get("race_time").unwrap().content_type,
            "duration:18000"
        );
        assert_eq!(parsed.get("bare").unwrap().content_type, "duration");
        // Malformed suffix collapses to bare "duration" rather than empty.
        assert_eq!(parsed.get("bad_cap").unwrap().content_type, "duration");
        assert_eq!(parsed.get("bad_suffix").unwrap().content_type, "duration");
    }

    #[test]
    fn parse_llm_response_missing_content_type_is_empty() {
        // A missing content_type key is left empty by parsing; combine_dictionary_entries
        // coerces it to "unknown" when the flag is set.
        let json = r#"{ "f": {"label": "F", "description": "d"} }"#;
        let fields = vec!["f".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert!(parsed.get("f").unwrap().content_type.is_empty());
    }

    #[test]
    fn combine_copies_content_type_onto_entry() {
        let code_entries = vec![blank_entry("col_a"), blank_entry("col_b")];
        let mut llm = HashMap::new();
        llm.insert(
            "col_a".to_string(),
            LlmDictField {
                label:        "A".to_string(),
                description:  "desc a".to_string(),
                content_type: "city".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        // infer_content_type = false: pure copy, no "unknown" coercion.
        let combined = combine_dictionary_entries(code_entries, &llm, false);
        assert_eq!(combined[0].label, "A");
        assert_eq!(combined[0].description, "desc a");
        assert_eq!(combined[0].content_type, "city");
        // col_b had no LLM entry, so its content_type stays empty when the flag is off.
        assert!(combined[1].content_type.is_empty());
    }

    #[test]
    fn combine_fills_unknown_for_empty_content_type_when_flag_on() {
        // With --infer-content-type set, every entry must end up with a non-empty
        // content_type: a valid token is kept; a field the LLM left empty (e.g. an
        // out-of-vocab token dropped by parsing) or omitted entirely falls back to "unknown".
        let code_entries = vec![
            blank_entry("kept"),
            blank_entry("emptied"),
            blank_entry("omitted"),
        ];
        let mut llm = HashMap::new();
        llm.insert(
            "kept".to_string(),
            LlmDictField {
                label:        "K".to_string(),
                description:  "d".to_string(),
                content_type: "city".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        llm.insert(
            "emptied".to_string(),
            LlmDictField {
                label:        "E".to_string(),
                description:  "d".to_string(),
                content_type: String::new(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        // "omitted" is intentionally absent from the LLM map.
        let combined = combine_dictionary_entries(code_entries, &llm, true);
        assert_eq!(combined[0].content_type, "city");
        assert_eq!(combined[1].content_type, "unknown");
        assert_eq!(combined[2].content_type, "unknown");
    }

    #[test]
    fn combine_preserves_preset_unique_id_over_llm_value() {
        // generate_code_based_dictionary stamps "unique_id" deterministically on
        // ALL_UNIQUE fields; combine_dictionary_entries must keep that value even
        // when the LLM returned a different (in-vocab) token for the same field.
        let mut preset = blank_entry("pk");
        preset.content_type = "unique_id".to_string();
        let code_entries = vec![preset, blank_entry("other")];
        let mut llm = HashMap::new();
        llm.insert(
            "pk".to_string(),
            LlmDictField {
                label:        "Primary Key".to_string(),
                description:  "row identifier".to_string(),
                content_type: "uuid".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        llm.insert(
            "other".to_string(),
            LlmDictField {
                label:        "Other".to_string(),
                description:  "city field".to_string(),
                content_type: "city".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code_entries, &llm, true);
        // Deterministic "unique_id" wins over the LLM's "uuid".
        assert_eq!(combined[0].content_type, "unique_id");
        // Label/Description still flow through from the LLM.
        assert_eq!(combined[0].label, "Primary Key");
        assert_eq!(combined[0].description, "row identifier");
        // Non-ALL_UNIQUE field gets the LLM's token unchanged.
        assert_eq!(combined[1].content_type, "city");
    }

    #[test]
    fn generate_marks_all_unique_field_as_unique_id() {
        // Field "id" has the <ALL_UNIQUE> sentinel in its frequency table, so
        // when infer_content_type is on, generate_code_based_dictionary must
        // pre-set its content_type to "unique_id". The peer field "category"
        // has a normal frequency distribution and must stay empty.
        let stats = vec![
            StatsRecord {
                field:       "id".to_string(),
                r#type:      "Integer".to_string(),
                cardinality: 1000,
                nullcount:   0,
                min:         "1".to_string(),
                max:         "1000".to_string(),
                addl_cols:   IndexMap::new(),
            },
            StatsRecord {
                field:       "category".to_string(),
                r#type:      "String".to_string(),
                cardinality: 2,
                nullcount:   0,
                min:         "a".to_string(),
                max:         "b".to_string(),
                addl_cols:   IndexMap::new(),
            },
        ];
        let frequencies = vec![
            FrequencyRecord {
                field:      "id".to_string(),
                value:      "<ALL_UNIQUE>".to_string(),
                count:      1000,
                percentage: 100.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "category".to_string(),
                value:      "a".to_string(),
                count:      600,
                percentage: 60.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "category".to_string(),
                value:      "b".to_string(),
                count:      400,
                percentage: 40.0,
                rank:       2.0,
            },
        ];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert_eq!(entries[0].name, "id");
        assert_eq!(entries[0].content_type, "unique_id");
        assert_eq!(entries[1].name, "category");
        assert!(
            entries[1].content_type.is_empty(),
            "non-ALL_UNIQUE field must leave content_type empty for LLM fill"
        );
    }

    #[test]
    fn generate_does_not_mark_truncated_frequency_as_unique_id() {
        // Truncated/custom frequency (e.g. `--limit 1 --no-other`) emits a single top
        // row whose `count` happens to equal the column cardinality, but `percentage`
        // is below 100 because the column is NOT unique. The structural detector must
        // require percentage == 100, so neither content_type nor is_unique_id is set.
        let stats = vec![StatsRecord {
            field:       "code".to_string(),
            r#type:      "String".to_string(),
            cardinality: 3,
            nullcount:   0,
            min:         "a".to_string(),
            max:         "c".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        // Single emitted row: count(3) == cardinality(3) but percentage 30 (< 100),
        // i.e. the top value covers 3 of ~10 rows; the rest were truncated.
        let frequencies = vec![FrequencyRecord {
            field:      "code".to_string(),
            value:      "a".to_string(),
            count:      3,
            percentage: 30.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert_eq!(entries[0].name, "code");
        assert!(
            !entries[0].is_unique_id,
            "truncated single-row frequency with percentage < 100 must not be a unique id"
        );
        assert!(
            entries[0].content_type.is_empty(),
            "truncated single-row frequency must not be stamped unique_id"
        );
    }

    #[test]
    fn generate_skips_unique_id_when_infer_content_type_off() {
        // When --infer-content-type is OFF, the content_type column is suppressed
        // entirely, so we must not pre-set "unique_id" even for ALL_UNIQUE fields.
        let stats = vec![StatsRecord {
            field:       "id".to_string(),
            r#type:      "Integer".to_string(),
            cardinality: 100,
            nullcount:   0,
            min:         "1".to_string(),
            max:         "100".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "id".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      100,
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], false);
        assert!(entries[0].content_type.is_empty());
    }

    #[test]
    fn generate_does_not_mislabel_constant_field_with_all_unique_value() {
        // Pathological: a field whose only value is literally the string
        // "<ALL_UNIQUE>". qsv frequency emits a single row with value=
        // "<ALL_UNIQUE>" and count==row_count, but stats.cardinality==1.
        // Structural detection (count == cardinality, cardinality > 1)
        // correctly excludes this.
        let stats = vec![StatsRecord {
            field:       "weird".to_string(),
            r#type:      "String".to_string(),
            cardinality: 1,
            nullcount:   0,
            min:         "<ALL_UNIQUE>".to_string(),
            max:         "<ALL_UNIQUE>".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "weird".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      500,
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert!(
            entries[0].content_type.is_empty(),
            "a constant-value field whose value happens to be the string '<ALL_UNIQUE>' must NOT \
             be classified as unique_id"
        );
    }

    #[test]
    fn generate_detects_unique_id_with_custom_all_unique_text() {
        // If qsv frequency was run with `--all-unique-text` set to a custom
        // string (e.g. "<UNIQUE>"), the sentinel row's text differs but the
        // structural invariant (one row, count == cardinality, cardinality > 1,
        // no nulls) still holds. Detection must succeed regardless of the
        // sentinel text.
        let stats = vec![StatsRecord {
            field:       "pk".to_string(),
            r#type:      "Integer".to_string(),
            cardinality: 250,
            nullcount:   0,
            min:         "1".to_string(),
            max:         "250".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "pk".to_string(),
            value:      "<UNIQUE>".to_string(), // user-customized sentinel text
            count:      250,
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert_eq!(
            entries[0].content_type, "unique_id",
            "text-independent detection must classify ALL_UNIQUE even when frequency's sentinel \
             text was customized"
        );
    }

    #[test]
    fn generate_does_not_misclassify_high_cardinality_as_unique_id() {
        // HIGH_CARDINALITY fields also produce a single frequency row with
        // count==row_count and percentage==100.0, but their cardinality is
        // strictly less than row_count (some values repeat). The
        // `count == cardinality` check correctly excludes them.
        let stats = vec![StatsRecord {
            field:       "city".to_string(),
            r#type:      "String".to_string(),
            cardinality: 800, // many distinct values, but with repeats
            nullcount:   0,
            min:         "Aachen".to_string(),
            max:         "Zurich".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "city".to_string(),
            value:      "<HIGH_CARDINALITY>".to_string(),
            count:      10_000, // row_count >> cardinality
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert!(
            entries[0].content_type.is_empty(),
            "HIGH_CARDINALITY field must not be classified as unique_id"
        );
    }

    #[test]
    fn generate_does_not_mislabel_unique_id_when_nulls_present() {
        // Semantic contract: "every row has a distinct non-null value". A
        // field where cardinality == count but nullcount > 0 doesn't qualify
        // (some rows have no value at all). qsv frequency wouldn't emit the
        // ALL_UNIQUE sentinel for this case anyway, but the explicit
        // nullcount==0 check provides defense in depth for hand-crafted or
        // cached frequency input.
        let stats = vec![StatsRecord {
            field:       "maybe_id".to_string(),
            r#type:      "Integer".to_string(),
            cardinality: 95,
            nullcount:   5, // 5 rows are NULL
            min:         "1".to_string(),
            max:         "95".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "maybe_id".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      95,
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert!(
            entries[0].content_type.is_empty(),
            "a field with nulls must NOT be classified as unique_id even if non-null cardinality \
             matches the frequency count"
        );
    }

    #[test]
    fn combine_with_baseline_preserves_baseline_when_refine_omits_field() {
        // The single biggest correctness pitfall for --two-pass: if the LLM's refine
        // response leaves out a field, that field's Label/Description must inherit the
        // first-pass values rather than reverting to code-derived defaults.
        let code_entries = vec![blank_entry("kept"), blank_entry("refined")];

        let mut baseline = HashMap::new();
        baseline.insert(
            "kept".to_string(),
            LlmDictField {
                label:        "Baseline Label".to_string(),
                description:  "Baseline description.".to_string(),
                content_type: "email".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        baseline.insert(
            "refined".to_string(),
            LlmDictField {
                label:        "Old Label".to_string(),
                description:  "Old description.".to_string(),
                content_type: "free_text".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        // Refine pass only returns "refined" — "kept" is intentionally absent.
        let mut refine = HashMap::new();
        refine.insert(
            "refined".to_string(),
            LlmDictField {
                label:        "New Label".to_string(),
                description:  "New description with cross-field context.".to_string(),
                content_type: "address_street".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        let combined =
            combine_dictionary_entries_with_baseline(code_entries, &baseline, &refine, true);

        // "kept" inherits baseline values verbatim.
        assert_eq!(combined[0].label, "Baseline Label");
        assert_eq!(combined[0].description, "Baseline description.");
        assert_eq!(combined[0].content_type, "email");
        // "refined" gets refine-pass overrides.
        assert_eq!(combined[1].label, "New Label");
        assert_eq!(
            combined[1].description,
            "New description with cross-field context."
        );
        assert_eq!(combined[1].content_type, "address_street");
    }

    /// Two-pass: the refine prompt never asks for `gauge_range`, so the refine pass
    /// re-emits the measure field (label/description) but with NO gauge — exactly what
    /// the pipeline produces. The first-pass gauge scale MUST survive that overlay
    /// (this is the line most at risk from a future refactor of the baseline merge).
    #[test]
    fn two_pass_refine_omitting_gauge_range_preserves_the_baseline() {
        let mut score = blank_entry("score");
        score.r#type = "Float".to_string();
        score.min = "12".to_string();
        score.max = "88".to_string();
        let code_entries = vec![score];

        let mut baseline = HashMap::new();
        baseline.insert(
            "score".to_string(),
            LlmDictField {
                label: "Score".to_string(),
                description: "First-pass description.".to_string(),
                role: "measure".to_string(),
                gauge_range: Some([0.0, 100.0]),
                ..Default::default()
            },
        );

        // Refine RETURNS the field (present in the map) but omits gauge_range — the
        // realistic case, distinct from the field being absent entirely.
        let mut refine = HashMap::new();
        refine.insert(
            "score".to_string(),
            LlmDictField {
                label: "Score".to_string(),
                description: "Refined description.".to_string(),
                role: "measure".to_string(),
                gauge_range: None,
                ..Default::default()
            },
        );

        let combined =
            combine_dictionary_entries_with_baseline(code_entries, &baseline, &refine, true);
        assert_eq!(combined[0].description, "Refined description.");
        assert_eq!(
            combined[0].gauge_range,
            Some([0.0, 100.0]),
            "baseline gauge scale must survive a refine pass that omits it"
        );
    }

    /// Two-pass: a refine pass that DOES re-propose a gauge scale overrides the baseline
    /// (and is still verified against the observed range).
    #[test]
    fn two_pass_refine_restating_gauge_range_overrides_the_baseline() {
        let mut score = blank_entry("score");
        score.r#type = "Float".to_string();
        score.min = "1".to_string();
        score.max = "4".to_string();
        let code_entries = vec![score];

        let mut baseline = HashMap::new();
        baseline.insert(
            "score".to_string(),
            LlmDictField {
                role: "measure".to_string(),
                gauge_range: Some([0.0, 100.0]),
                ..Default::default()
            },
        );
        let mut refine = HashMap::new();
        refine.insert(
            "score".to_string(),
            LlmDictField {
                role: "measure".to_string(),
                gauge_range: Some([0.0, 5.0]),
                ..Default::default()
            },
        );

        let combined =
            combine_dictionary_entries_with_baseline(code_entries, &baseline, &refine, true);
        assert_eq!(combined[0].gauge_range, Some([0.0, 5.0]));
    }

    #[test]
    fn combine_with_baseline_rejects_refine_supplied_unique_id() {
        // The deterministic "unique_id" stamp (cardinality == rowcount) must survive
        // the refine pass even if the LLM tries to overwrite it with a different vocab
        // token — and the refine pass also cannot smuggle in a fabricated "unique_id"
        // for a non-ALL_UNIQUE field. Mirrors the single-pass guarantees in
        // combine_dictionary_entries.
        let mut pk = blank_entry("pk");
        pk.content_type = "unique_id".to_string();
        let code_entries = vec![pk, blank_entry("other")];

        // Baseline is empty: this asserts the refine pass is gated independently.
        let baseline = HashMap::new();

        let mut refine = HashMap::new();
        refine.insert(
            "pk".to_string(),
            LlmDictField {
                label:        "Primary Key".to_string(),
                description:  "row id".to_string(),
                // Refine pass tries to overwrite the deterministic stamp with "uuid".
                content_type: "uuid".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        refine.insert(
            "other".to_string(),
            LlmDictField {
                label:        "Other".to_string(),
                description:  "not unique".to_string(),
                // Refine pass tries to smuggle "unique_id" onto a non-ALL_UNIQUE field.
                content_type: "unique_id".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        let combined =
            combine_dictionary_entries_with_baseline(code_entries, &baseline, &refine, true);

        assert_eq!(
            combined[0].content_type, "unique_id",
            "deterministic 'unique_id' stamp must survive refine-pass overwrite"
        );
        assert_eq!(
            combined[1].content_type, "unknown",
            "refine-supplied 'unique_id' for a non-ALL_UNIQUE field must be rejected and coerced \
             to 'unknown' (the infer_content_type=true default)"
        );
    }

    #[test]
    fn combine_with_baseline_refine_overrides_valid_baseline_content_type() {
        // The whole point of --two-pass: the refine pass can upgrade a baseline content_type
        // to a better vocab token once it sees cross-field context (e.g. "free_text" ->
        // "address_street" after recognizing sibling city / state / zip columns).
        let code_entries = vec![blank_entry("street1")];

        let mut baseline = HashMap::new();
        baseline.insert(
            "street1".to_string(),
            LlmDictField {
                label:        "Street 1".to_string(),
                description:  "First street line".to_string(),
                content_type: "free_text".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        let mut refine = HashMap::new();
        refine.insert(
            "street1".to_string(),
            LlmDictField {
                label:        "Street Address".to_string(),
                description:  "Street component of the mailing address.".to_string(),
                content_type: "address_street".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        let combined =
            combine_dictionary_entries_with_baseline(code_entries, &baseline, &refine, true);
        assert_eq!(combined[0].content_type, "address_street");
        assert_eq!(combined[0].label, "Street Address");
    }

    #[test]
    fn combine_with_baseline_matches_single_pass_when_refine_empty() {
        // Sanity check: with an empty refine map, the baseline-preserving variant must
        // produce the same result as the single-pass combine. Guards against accidental
        // divergence between the two functions during future refactors.
        let mut pk = blank_entry("pk");
        pk.content_type = "unique_id".to_string();
        let code_entries = vec![pk, blank_entry("city")];

        let mut baseline = HashMap::new();
        baseline.insert(
            "city".to_string(),
            LlmDictField {
                label:        "City".to_string(),
                description:  "City name".to_string(),
                content_type: "city".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );

        let single_pass = combine_dictionary_entries(code_entries.clone(), &baseline, true);
        let two_pass = combine_dictionary_entries_with_baseline(
            code_entries,
            &baseline,
            &HashMap::new(),
            true,
        );

        assert_eq!(single_pass.len(), two_pass.len());
        for (a, b) in single_pass.iter().zip(two_pass.iter()) {
            assert_eq!(a.name, b.name);
            assert_eq!(a.label, b.label);
            assert_eq!(a.description, b.description);
            assert_eq!(a.content_type, b.content_type);
        }
    }

    fn entry_with_content_type(name: &str, content_type: &str) -> DictionaryEntry {
        DictionaryEntry {
            content_type: content_type.to_string(),
            ..blank_entry(name)
        }
    }

    #[test]
    fn normalize_datetime_token_handles_all_forms() {
        // Bare tokens accepted.
        assert_eq!(normalize_datetime_token("date").as_deref(), Some("date"));
        assert_eq!(
            normalize_datetime_token("datetime").as_deref(),
            Some("datetime")
        );
        // The token prefix is case-folded.
        assert_eq!(
            normalize_datetime_token("DateTime").as_deref(),
            Some("datetime")
        );
        // A valid strftime suffix is preserved verbatim, including its case.
        assert_eq!(
            normalize_datetime_token("date:%Y-%m-%d").as_deref(),
            Some("date:%Y-%m-%d")
        );
        assert_eq!(
            normalize_datetime_token("datetime:%m/%d/%Y %I:%M:%S %p").as_deref(),
            Some("datetime:%m/%d/%Y %I:%M:%S %p")
        );
        // Colons inside the format survive — split_once only consumes the first ':'.
        assert_eq!(
            normalize_datetime_token("datetime:%H:%M:%S").as_deref(),
            Some("datetime:%H:%M:%S")
        );
        // A malformed strftime suffix degrades to the bare token.
        assert_eq!(normalize_datetime_token("date:%Q").as_deref(), Some("date"));
        // An empty suffix degrades to the bare token.
        assert_eq!(normalize_datetime_token("date:").as_deref(), Some("date"));
        // Non-date tokens return None so the caller falls through to the vocab check.
        assert_eq!(normalize_datetime_token("time"), None);
        assert_eq!(normalize_datetime_token("duration:3600"), None);
        assert_eq!(normalize_datetime_token("email"), None);
        assert_eq!(normalize_datetime_token(""), None);
    }

    #[test]
    fn parse_llm_response_preserves_date_format_casing() {
        // The date/datetime <fmt> suffix is case-sensitive (%m month vs %M minute,
        // %p AM/PM) and must survive the parser's general lowercasing.
        let json = r#"{
            "created": {"label": "C", "description": "d", "content_type": "datetime:%m/%d/%Y %I:%M:%S %p"},
            "born":    {"label": "B", "description": "d", "content_type": "DATE:%Y-%m-%d"}
        }"#;
        let fields = vec!["created".to_string(), "born".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert_eq!(
            parsed.get("created").unwrap().content_type,
            "datetime:%m/%d/%Y %I:%M:%S %p"
        );
        // token prefix case-folded, format suffix preserved verbatim
        assert_eq!(parsed.get("born").unwrap().content_type, "date:%Y-%m-%d");
    }

    #[test]
    fn parse_llm_relationships_extracts_and_validates() {
        let field_names = vec![
            "created_date".to_string(),
            "closed_date".to_string(),
            "city".to_string(),
            "state".to_string(),
        ];
        let response = r#"{
            "created_date": {"label": "Created", "description": "x"},
            "relationships": [
                {"kind": "ordered", "members": ["created_date", "closed_date"], "anchor": "created_date"},
                {"kind": "joint", "members": ["city", "state"]},
                {"kind": "bogus", "members": ["city", "state"]},
                {"kind": "joint", "members": ["city", "missing_col"]},
                {"kind": "correlated", "members": ["city"]}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        // bogus kind, unknown member, and the 1-member group are all dropped.
        assert_eq!(rels.len(), 2);
        assert_eq!(rels[0]["kind"], "ordered");
        assert_eq!(
            rels[0]["members"],
            serde_json::json!(["created_date", "closed_date"])
        );
        assert_eq!(rels[0]["anchor"], "created_date");
        assert_eq!(rels[1]["kind"], "joint");
        // `anchor` is only kept for `ordered` groups.
        assert!(rels[1].get("anchor").is_none());
    }

    #[test]
    fn parse_llm_relationships_absent_is_empty() {
        let field_names = vec!["a".to_string(), "b".to_string()];
        let response = r#"{"a": {"label": "A", "description": "x"}}"#;
        assert!(parse_llm_relationships(response, &field_names).is_empty());
    }

    #[test]
    fn parse_llm_relationships_drops_invalid_anchor() {
        let field_names = vec!["lo".to_string(), "hi".to_string()];
        let response = r#"{
            "relationships": [
                {"kind": "ordered", "members": ["lo", "hi"], "anchor": "not_a_column"}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        // The relationship survives (members are valid) but the bad anchor is dropped.
        assert_eq!(rels.len(), 1);
        assert!(rels[0].get("anchor").is_none());
    }

    #[test]
    fn parse_llm_relationships_accepts_column_shaped_pipeline() {
        let field_names = vec![
            "planned".to_string(),
            "committed".to_string(),
            "spent".to_string(),
        ];
        let response = r#"{
            "relationships": [
                {"kind": "pipeline", "members": ["planned", "committed", "spent"]}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        assert_eq!(rels.len(), 1);
        assert_eq!(rels[0]["kind"], "pipeline");
        // member ORDER is the pipeline order and must survive verbatim -- reordering
        // it would invert the funnel.
        assert_eq!(
            rels[0]["members"],
            serde_json::json!(["planned", "committed", "spent"])
        );
        // a column-shaped pipeline carries none of the row-encoding keys
        assert!(rels[0].get("stage_column").is_none());
    }

    #[test]
    fn parse_llm_relationships_drops_a_pipeline_repeating_a_column() {
        let field_names = vec!["planned".to_string(), "spent".to_string()];
        let response = r#"{
            "relationships": [
                {"kind": "pipeline", "members": ["planned", "planned"]}
            ]
        }"#;
        assert!(parse_llm_relationships(response, &field_names).is_empty());
    }

    #[test]
    fn parse_llm_relationships_accepts_row_encoded_pipeline() {
        let field_names = vec!["status".to_string(), "amount".to_string()];
        let response = r#"{
            "relationships": [
                {"kind": "pipeline", "stage_column": "status",
                 "stages": ["Impression", "Click", "Conversion"],
                 "value_column": "amount"}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        assert_eq!(rels.len(), 1);
        assert_eq!(rels[0]["stage_column"], "status");
        assert_eq!(rels[0]["value_column"], "amount");
        assert_eq!(
            rels[0]["stages"],
            serde_json::json!(["Impression", "Click", "Conversion"])
        );
        // members are SYNTHESIZED from the column references, never taken from the LLM
        assert_eq!(rels[0]["members"], serde_json::json!(["status", "amount"]));
    }

    #[test]
    fn parse_llm_relationships_row_pipeline_without_value_column_counts_rows() {
        let field_names = vec!["status".to_string(), "amount".to_string()];
        let response = r#"{
            "relationships": [
                {"kind": "pipeline", "stage_column": "status", "stages": ["Open", "Closed"]}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        assert_eq!(rels.len(), 1);
        assert!(rels[0].get("value_column").is_none());
        assert_eq!(rels[0]["members"], serde_json::json!(["status"]));
    }

    #[test]
    fn parse_llm_relationships_drops_malformed_row_pipelines() {
        let field_names = vec!["status".to_string(), "amount".to_string()];
        for (case, response) in [
            (
                "stage_column is not a real column",
                r#"{"relationships":[{"kind":"pipeline","stage_column":"nope","stages":["a","b"]}]}"#,
            ),
            (
                "fewer than two stages",
                r#"{"relationships":[{"kind":"pipeline","stage_column":"status","stages":["a"]}]}"#,
            ),
            (
                "duplicate stages",
                r#"{"relationships":[{"kind":"pipeline","stage_column":"status","stages":["a","a"]}]}"#,
            ),
            (
                "value_column is not a real column",
                r#"{"relationships":[{"kind":"pipeline","stage_column":"status","stages":["a","b"],"value_column":"nope"}]}"#,
            ),
            (
                "value_column duplicates stage_column",
                r#"{"relationships":[{"kind":"pipeline","stage_column":"status","stages":["a","b"],"value_column":"status"}]}"#,
            ),
        ] {
            assert!(
                parse_llm_relationships(response, &field_names).is_empty(),
                "should have dropped the entry: {case}"
            );
        }
    }

    #[test]
    fn parse_llm_relationships_always_emits_members() {
        // `synthesize`'s SynthRelationship::members has no serde default, and the
        // dictionary is deserialized in a single pass -- an entry without `members`
        // does not degrade, it HARD-ERRORS `synthesize --dictionary`. Every emitted
        // relationship must therefore carry the key, including the row encoding,
        // where the LLM supplies no usable member list at all.
        let field_names = vec![
            "status".to_string(),
            "amount".to_string(),
            "lo".to_string(),
            "hi".to_string(),
        ];
        let response = r#"{
            "relationships": [
                {"kind": "pipeline", "stage_column": "status", "stages": ["a", "b"]},
                {"kind": "pipeline", "stage_column": "status", "stages": ["a", "b"],
                 "value_column": "amount"},
                {"kind": "pipeline", "members": ["lo", "hi"]},
                {"kind": "ordered", "members": ["lo", "hi"]},
                {"kind": "joint", "members": ["lo", "hi"]}
            ]
        }"#;
        let rels = parse_llm_relationships(response, &field_names);
        assert_eq!(rels.len(), 5);
        for rel in &rels {
            let members = rel
                .get("members")
                .unwrap_or_else(|| panic!("every relationship must carry `members`: {rel}"));
            assert!(
                members.as_array().is_some_and(|m| !m.is_empty()),
                "`members` must be a non-empty array: {rel}"
            );
        }
    }

    #[test]
    fn generate_stamps_date_datetime_from_stats_type() {
        let stats = vec![
            StatsRecord {
                field:       "created".to_string(),
                r#type:      "DateTime".to_string(),
                cardinality: 50,
                nullcount:   0,
                min:         "2020-01-01T00:00:00+00:00".to_string(),
                max:         "2020-12-31T00:00:00+00:00".to_string(),
                addl_cols:   IndexMap::new(),
            },
            StatsRecord {
                field:       "born".to_string(),
                r#type:      "Date".to_string(),
                cardinality: 50,
                nullcount:   0,
                min:         "1990-01-01".to_string(),
                max:         "1999-12-31".to_string(),
                addl_cols:   IndexMap::new(),
            },
            StatsRecord {
                field:       "name".to_string(),
                r#type:      "String".to_string(),
                cardinality: 50,
                nullcount:   0,
                min:         "a".to_string(),
                max:         "z".to_string(),
                addl_cols:   IndexMap::new(),
            },
        ];
        // non-ALL_UNIQUE frequency rows so nothing is stamped unique_id
        let frequencies = vec![
            FrequencyRecord {
                field:      "created".to_string(),
                value:      "x".to_string(),
                count:      10,
                percentage: 20.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "born".to_string(),
                value:      "y".to_string(),
                count:      10,
                percentage: 20.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "name".to_string(),
                value:      "z".to_string(),
                count:      10,
                percentage: 20.0,
                rank:       1.0,
            },
        ];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert_eq!(entries[0].content_type, "datetime");
        assert_eq!(entries[1].content_type, "date");
        assert!(entries[2].content_type.is_empty());

        // with infer_content_type off, nothing is stamped
        let off = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], false);
        assert!(off.iter().all(|e| e.content_type.is_empty()));
    }

    #[test]
    fn generate_unique_id_wins_over_date_stamp() {
        // An ALL_UNIQUE DateTime column (a unique timestamp key) is stamped
        // `unique_id`, not `datetime` — the `unique_id` stamp keeps priority.
        let stats = vec![StatsRecord {
            field:       "ts_pk".to_string(),
            r#type:      "DateTime".to_string(),
            cardinality: 1000,
            nullcount:   0,
            min:         "2020-01-01T00:00:00+00:00".to_string(),
            max:         "2020-12-31T00:00:00+00:00".to_string(),
            addl_cols:   IndexMap::new(),
        }];
        let frequencies = vec![FrequencyRecord {
            field:      "ts_pk".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      1000,
            percentage: 100.0,
            rank:       1.0,
        }];
        let entries = generate_code_based_dictionary(&stats, &frequencies, 10, 5, 25, &[], true);
        assert_eq!(entries[0].content_type, "unique_id");
    }

    #[test]
    fn combine_merges_date_format_suffix_onto_stamped_token() {
        // code stamped a bare "datetime"; the LLM contributes the strftime suffix
        let code = vec![entry_with_content_type("created", "datetime")];
        let mut llm = HashMap::new();
        llm.insert(
            "created".to_string(),
            LlmDictField {
                label:        "Created".to_string(),
                description:  "d".to_string(),
                content_type: "datetime:%m/%d/%Y".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        assert_eq!(combined[0].content_type, "datetime:%m/%d/%Y");
    }

    #[test]
    fn combine_keeps_bare_date_token_when_llm_base_mismatches() {
        // code stamped bare "date"; the LLM disagrees on the base token —
        // the deterministic classification wins, the mismatched suffix is dropped
        let code = vec![entry_with_content_type("born", "date")];
        let mut llm = HashMap::new();
        llm.insert(
            "born".to_string(),
            LlmDictField {
                label:        "Born".to_string(),
                description:  "d".to_string(),
                content_type: "datetime:%Y-%m-%dT%H:%M:%S".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        assert_eq!(combined[0].content_type, "date");
    }

    #[test]
    fn combine_drops_llm_date_token_on_non_date_column() {
        // the LLM may not introduce a date/datetime token onto a column the
        // stats did not type as such — only the deterministic stamp may
        let code = vec![entry_with_content_type("note", "")];
        let mut llm = HashMap::new();
        llm.insert(
            "note".to_string(),
            LlmDictField {
                label:        "Note".to_string(),
                description:  "d".to_string(),
                content_type: "date:%Y-%m-%d".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        // dropped → falls back to "unknown"
        assert_eq!(combined[0].content_type, "unknown");
    }

    #[test]
    fn combine_with_baseline_refine_corrects_date_format() {
        // the refine pass may correct the <fmt> suffix on an already-date-typed column
        let code = vec![entry_with_content_type("created", "datetime")];
        let mut baseline = HashMap::new();
        baseline.insert(
            "created".to_string(),
            LlmDictField {
                label:        "C".to_string(),
                description:  "d".to_string(),
                content_type: "datetime:%Y/%m/%d".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let mut refine = HashMap::new();
        refine.insert(
            "created".to_string(),
            LlmDictField {
                label:        "C".to_string(),
                description:  "d".to_string(),
                content_type: "datetime:%m/%d/%Y".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries_with_baseline(code, &baseline, &refine, true);
        assert_eq!(combined[0].content_type, "datetime:%m/%d/%Y");
    }

    #[test]
    fn validate_date_formats_strips_mismatched_suffix() {
        // the entry claims %m/%d/%Y but the real sample is ISO → suffix stripped
        let mut entries = vec![entry_with_content_type("d", "date:%m/%d/%Y")];
        let freqs = vec![FrequencyRecord {
            field:      "d".to_string(),
            value:      "2020-01-15".to_string(),
            count:      5,
            percentage: 10.0,
            rank:       1.0,
        }];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "date");
    }

    #[test]
    fn validate_date_formats_keeps_matching_suffix() {
        let mut entries = vec![entry_with_content_type(
            "d",
            "datetime:%m/%d/%Y %I:%M:%S %p",
        )];
        let freqs = vec![FrequencyRecord {
            field:      "d".to_string(),
            value:      "01/24/2013 12:00:00 AM".to_string(),
            count:      5,
            percentage: 10.0,
            rank:       1.0,
        }];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "datetime:%m/%d/%Y %I:%M:%S %p");
    }

    #[test]
    fn validate_date_formats_accepts_offset_datetime() {
        // %z offset datetimes: NaiveDateTime rejects them, DateTime::parse_from_str accepts
        let mut entries = vec![entry_with_content_type("d", "datetime:%Y-%m-%dT%H:%M:%S%z")];
        let freqs = vec![FrequencyRecord {
            field:      "d".to_string(),
            value:      "2020-01-15T08:30:00+0000".to_string(),
            count:      5,
            percentage: 10.0,
            rank:       1.0,
        }];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "datetime:%Y-%m-%dT%H:%M:%S%z");
    }

    #[test]
    fn validate_date_formats_keeps_suffix_when_no_sample() {
        // an ALL_UNIQUE column has only the <ALL_UNIQUE> sentinel row — there is
        // nothing to validate against, so the suffix is left unchanged.
        //
        // `rank: 0.0` is what `frequency` actually emits for this row (verified: both
        // emit sites hardcode it, for the default AND a custom --all-unique-text). The
        // fixture previously said 1.0, which never occurs, and the test passed only
        // because `usable_samples_by_field` also matched the sentinel's literal text —
        // a match that wrongly dropped real data (see roborev 3560).
        let mut entries = vec![entry_with_content_type("d", "date:%d/%m/%Y")];
        let freqs = vec![FrequencyRecord {
            field:      "d".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      100,
            percentage: 100.0,
            rank:       0.0,
        }];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "date:%d/%m/%Y");
    }

    #[test]
    fn strip_time_from_format_keeps_date_drops_time() {
        assert_eq!(
            strip_time_from_format("%m/%d/%Y %I:%M:%S %p").as_deref(),
            Some("%m/%d/%Y")
        );
        assert_eq!(
            strip_time_from_format("%Y-%m-%dT%H:%M:%S").as_deref(),
            Some("%Y-%m-%d")
        );
        assert_eq!(
            strip_time_from_format("%d-%b-%Y %H:%M").as_deref(),
            Some("%d-%b-%Y")
        );
        // a timezone-bearing time part still truncates cleanly
        assert_eq!(
            strip_time_from_format("%Y-%m-%dT%H:%M:%S%z").as_deref(),
            Some("%Y-%m-%d")
        );
        // a `-` date/time separator is trimmed, not left dangling
        assert_eq!(
            strip_time_from_format("%Y-%m-%d-%H:%M:%S").as_deref(),
            Some("%Y-%m-%d")
        );
        // a pure-date format is returned unchanged
        assert_eq!(
            strip_time_from_format("%Y-%m-%d").as_deref(),
            Some("%Y-%m-%d")
        );
        // a pure-time format has no date prefix
        assert_eq!(strip_time_from_format("%H:%M:%S"), None);
        // datetime-compound specifiers have no clean date-only prefix
        assert_eq!(strip_time_from_format("%+"), None);
    }

    #[test]
    fn downgrade_reclassifies_all_midnight_datetime_as_date() {
        let mut entries = vec![entry_with_content_type(
            "created",
            "datetime:%m/%d/%Y %I:%M:%S %p",
        )];
        let freqs = vec![
            FrequencyRecord {
                field:      "created".to_string(),
                value:      "01/24/2013 12:00:00 AM".to_string(),
                count:      347,
                percentage: 0.03,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "created".to_string(),
                value:      "01/07/2014 12:00:00 AM".to_string(),
                count:      315,
                percentage: 0.03,
                rank:       2.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        // every sample is at midnight → date, with the time stripped from <fmt>
        assert_eq!(entries[0].content_type, "date:%m/%d/%Y");
    }

    #[test]
    fn downgrade_keeps_datetime_when_a_sample_has_a_real_time() {
        let mut entries = vec![entry_with_content_type(
            "due",
            "datetime:%m/%d/%Y %I:%M:%S %p",
        )];
        let freqs = vec![
            FrequencyRecord {
                field:      "due".to_string(),
                value:      "04/08/2015 10:00:58 AM".to_string(),
                count:      214,
                percentage: 0.06,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "due".to_string(),
                value:      "05/02/2014 12:00:00 AM".to_string(),
                count:      183,
                percentage: 0.05,
                rank:       2.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        // one sample carries a non-midnight time → stays datetime
        assert_eq!(entries[0].content_type, "datetime:%m/%d/%Y %I:%M:%S %p");
    }

    #[test]
    fn downgrade_keeps_datetime_when_a_sample_is_unparsable() {
        // a usable frequency value that does not parse with the inferred
        // format blocks the downgrade — a non-parsing real value could carry
        // a non-midnight time, so it must not be silently dropped
        let mut entries = vec![entry_with_content_type(
            "closed",
            "datetime:%m/%d/%Y %I:%M:%S %p",
        )];
        let freqs = vec![
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "11/15/2010 12:00:00 AM".to_string(),
                count:      384,
                percentage: 0.04,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "not-a-date".to_string(),
                count:      28619,
                percentage: 2.0,
                rank:       2.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "datetime:%m/%d/%Y %I:%M:%S %p");
    }

    #[test]
    fn downgrade_leaves_non_datetime_entries_alone() {
        let mut entries = vec![
            entry_with_content_type("bare_dt", "datetime"),
            entry_with_content_type("a_date", "date:%Y-%m-%d"),
            entry_with_content_type("contact", "email"),
        ];
        let freqs = vec![
            FrequencyRecord {
                field:      "bare_dt".to_string(),
                value:      "01/24/2013 12:00:00 AM".to_string(),
                count:      1,
                percentage: 0.1,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "a_date".to_string(),
                value:      "2013-01-24".to_string(),
                count:      1,
                percentage: 0.1,
                rank:       1.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        // bare `datetime` has no format to test against → unchanged
        assert_eq!(entries[0].content_type, "datetime");
        // an already-`date` entry → unchanged
        assert_eq!(entries[1].content_type, "date:%Y-%m-%d");
        // a non-date token → unchanged
        assert_eq!(entries[2].content_type, "email");
    }

    #[test]
    fn downgrade_keeps_datetime_when_no_usable_sample() {
        // an ALL_UNIQUE datetime column has no usable frequency sample
        let mut entries = vec![entry_with_content_type(
            "ts",
            "datetime:%m/%d/%Y %I:%M:%S %p",
        )];
        let freqs = vec![FrequencyRecord {
            field:      "ts".to_string(),
            value:      "<ALL_UNIQUE>".to_string(),
            count:      1000,
            percentage: 100.0,
            rank:       1.0,
        }];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "datetime:%m/%d/%Y %I:%M:%S %p");
    }

    #[test]
    fn validate_date_formats_checks_all_samples_not_just_first() {
        // `%m/%d/%Y` parses the first sample `01/02/2020` but a later sample
        // `13/02/2020` (unambiguously day-first) disproves it — validating
        // against only the first sample would wrongly keep the suffix.
        let mut entries = vec![entry_with_content_type("d", "date:%m/%d/%Y")];
        let freqs = vec![
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "01/02/2020".to_string(),
                count:      5,
                percentage: 10.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "13/02/2020".to_string(),
                count:      4,
                percentage: 8.0,
                rank:       2.0,
            },
        ];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "date");
    }

    #[test]
    fn validate_date_formats_ignores_ranked_null_rows() {
        // `frequency --pct-nulls` emits the null row with a real (non-zero)
        // rank. It is identified by its `--null-text` value AND a count equal
        // to the column's null count, so it does not count as a sample —
        // otherwise it fails to parse and strips an otherwise-valid format.
        // The null label here is custom (`<MISSING>`), threaded in as
        // `null_text`, proving the check honors a configured `--null-text`.
        let mut entries = vec![DictionaryEntry {
            null_count: 5,
            ..entry_with_content_type("d", "date:%Y-%m-%d")
        }];
        let freqs = vec![
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "2020-01-15".to_string(),
                count:      8,
                percentage: 61.5,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "<MISSING>".to_string(),
                count:      5,
                percentage: 38.5,
                rank:       2.0,
            },
        ];
        validate_date_formats(&mut entries, &freqs, "<MISSING>");
        assert_eq!(entries[0].content_type, "date:%Y-%m-%d");
    }

    #[test]
    fn downgrade_ignores_ranked_null_rows() {
        // a ranked null row — its `--null-text` value plus a count equal to
        // the column's null count — must not count as a sample: it would fail
        // to parse and block the downgrade of an otherwise all-midnight
        // datetime column.
        let mut entries = vec![DictionaryEntry {
            null_count: 5,
            ..entry_with_content_type("closed", "datetime:%m/%d/%Y %I:%M:%S %p")
        }];
        let freqs = vec![
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "11/15/2010 12:00:00 AM".to_string(),
                count:      8,
                percentage: 61.5,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "(NULL)".to_string(),
                count:      5,
                percentage: 38.5,
                rank:       2.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "date:%m/%d/%Y");
    }

    #[test]
    fn validate_date_formats_keeps_real_sample_sharing_null_count() {
        // a real, non-null value whose count merely equals the column's null
        // count must still be validated — identifying the null row by count
        // alone would drop it. Here that sample (`2020-01-15`, ISO) disproves
        // the inferred `%m/%d/%Y` suffix; only the genuine `(NULL)` row, which
        // also matches `--null-text`, is excluded.
        let mut entries = vec![DictionaryEntry {
            null_count: 5,
            ..entry_with_content_type("d", "date:%m/%d/%Y")
        }];
        let freqs = vec![
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "2020-01-15".to_string(),
                count:      5,
                percentage: 35.7,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "d".to_string(),
                value:      "(NULL)".to_string(),
                count:      5,
                percentage: 35.7,
                rank:       2.0,
            },
        ];
        validate_date_formats(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "date");
    }

    #[test]
    fn downgrade_keeps_datetime_when_non_midnight_sample_shares_null_count() {
        // a real non-midnight value whose count merely equals the column's
        // null count must still block the downgrade — identifying the null row
        // by count alone would drop it and wrongly reclassify the column as
        // `date`. Only the genuine `(NULL)` row is excluded.
        let mut entries = vec![DictionaryEntry {
            null_count: 5,
            ..entry_with_content_type("closed", "datetime:%m/%d/%Y %I:%M:%S %p")
        }];
        let freqs = vec![
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "11/15/2010 12:00:00 AM".to_string(),
                count:      8,
                percentage: 44.4,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "04/08/2015 10:00:58 AM".to_string(),
                count:      5,
                percentage: 27.8,
                rank:       2.0,
            },
            FrequencyRecord {
                field:      "closed".to_string(),
                value:      "(NULL)".to_string(),
                count:      5,
                percentage: 27.8,
                rank:       3.0,
            },
        ];
        downgrade_all_midnight_datetime_columns(&mut entries, &freqs, "(NULL)");
        assert_eq!(entries[0].content_type, "datetime:%m/%d/%Y %I:%M:%S %p");
    }

    #[test]
    fn parse_llm_response_extracts_validated_role_and_concept() {
        let json = r#"{
          "zip": {"label":"ZIP","description":"postal code","content_type":"zip_code","role":"dimension","concept":"geo.zip_code"},
          "bad": {"label":"B","description":"d","role":"sideways","concept":"not.a.concept"},
          "sk":  {"label":"K","description":"key","concept":"id.surrogate_key","role":"identifier"}
        }"#;
        let fields = vec!["zip".to_string(), "bad".to_string(), "sk".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, true).unwrap();
        assert_eq!(parsed["zip"].role, "dimension");
        assert_eq!(parsed["zip"].concept, "geo.zip_code");
        // Invalid role + out-of-vocab concept are dropped to empty (the seed/fallback fills them).
        assert_eq!(parsed["bad"].role, "");
        assert_eq!(parsed["bad"].concept, "");
        // `id.surrogate_key` is reserved/deterministic — rejected from LLM input (like unique_id).
        assert_eq!(parsed["sk"].concept, "");
        assert_eq!(parsed["sk"].role, "identifier");
    }

    #[test]
    fn parse_llm_response_ignores_role_concept_when_flag_off() {
        let json = r#"{"zip":{"label":"ZIP","description":"d","role":"dimension","concept":"geo.zip_code"}}"#;
        let fields = vec!["zip".to_string()];
        let parsed = parse_llm_dictionary_response(json, &fields, false).unwrap();
        assert_eq!(parsed["zip"].role, "");
        assert_eq!(parsed["zip"].concept, "");
    }

    #[test]
    fn parse_llm_grain_reads_top_level_string() {
        assert_eq!(
            parse_llm_grain(r#"{"grain":"one row = one trip"}"#).as_deref(),
            Some("one row = one trip")
        );
        assert_eq!(parse_llm_grain(r#"{"grain":"   "}"#), None);
        assert_eq!(parse_llm_grain(r#"{"foo":1}"#), None);
        assert_eq!(parse_llm_grain("not json at all"), None);
    }

    #[test]
    fn parse_llm_grain_unit_reads_top_level_string() {
        assert_eq!(
            parse_llm_grain_unit(r#"{"grain":"one row = one trip","grain_unit":"trip"}"#)
                .as_deref(),
            Some("trip")
        );
        // localized unit rides through verbatim -- the whole point of the field (issue #4321)
        assert_eq!(
            parse_llm_grain_unit(r#"{"grain_unit":"registro de coleta"}"#).as_deref(),
            Some("registro de coleta")
        );
        assert_eq!(parse_llm_grain_unit(r#"{"grain_unit":"  "}"#), None);
        // grain present but no grain_unit -> None, so viz falls back to parsing the sentence
        assert_eq!(
            parse_llm_grain_unit(r#"{"grain":"one row = one trip"}"#),
            None
        );
        assert_eq!(parse_llm_grain_unit("not json at all"), None);
    }

    #[test]
    fn concept_from_content_type_maps_known_tokens() {
        assert_eq!(concept_from_content_type("zip_code"), Some("geo.zip_code"));
        assert_eq!(
            concept_from_content_type("unique_id"),
            Some("id.surrogate_key")
        );
        assert_eq!(
            concept_from_content_type("datetime"),
            Some("time.event_timestamp")
        );
        assert_eq!(concept_from_content_type("free_text"), None);
    }

    #[test]
    fn is_linkable_concept_excludes_pii_and_unknown() {
        assert!(is_linkable_concept("geo.zip_code"));
        assert!(is_linkable_concept("nyc.bbl"));
        assert!(is_linkable_concept("id.foreign_key"));
        assert!(!is_linkable_concept("pii.email"));
        assert!(!is_linkable_concept("category.status"));
        assert!(!is_linkable_concept("unknown"));
        assert!(!is_linkable_concept(""));
    }

    #[test]
    fn generate_code_based_dictionary_seeds_concept_and_role() {
        let stats = vec![
            StatsRecord {
                field:       "id".to_string(),
                r#type:      "Integer".to_string(),
                cardinality: 3,
                nullcount:   0,
                min:         "1".to_string(),
                max:         "3".to_string(),
                addl_cols:   IndexMap::new(),
            },
            StatsRecord {
                field:       "created".to_string(),
                r#type:      "Date".to_string(),
                cardinality: 2,
                nullcount:   0,
                min:         "2020-01-01".to_string(),
                max:         "2020-12-31".to_string(),
                addl_cols:   IndexMap::new(),
            },
        ];
        let freqs = vec![
            FrequencyRecord {
                field:      "id".to_string(),
                value:      "<ALL_UNIQUE>".to_string(),
                count:      3,
                percentage: 100.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "created".to_string(),
                value:      "2020-01-01".to_string(),
                count:      1,
                percentage: 50.0,
                rank:       1.0,
            },
            FrequencyRecord {
                field:      "created".to_string(),
                value:      "2020-12-31".to_string(),
                count:      1,
                percentage: 50.0,
                rank:       2.0,
            },
        ];
        let entries = generate_code_based_dictionary(&stats, &freqs, 50, 5, 0, &[], true);

        let id = entries.iter().find(|e| e.name == "id").unwrap();
        assert!(id.is_unique_id);
        assert_eq!(id.concept, "id.surrogate_key");
        assert_eq!(id.role, "identifier");

        let created = entries.iter().find(|e| e.name == "created").unwrap();
        assert_eq!(created.concept, "time.date");
        assert_eq!(created.role, "timestamp");

        // With the flag off, no concept/role is seeded.
        let off = generate_code_based_dictionary(&stats, &freqs, 50, 5, 0, &[], false);
        assert_eq!(off.iter().find(|e| e.name == "id").unwrap().concept, "");
        assert_eq!(off.iter().find(|e| e.name == "id").unwrap().role, "");
    }

    #[test]
    fn combine_backfills_concept_from_content_type_when_llm_omits_it() {
        // A model returns a valid content_type ("zip_code") but omits "concept".
        // The merge path must backfill the deterministic concept mapping
        // ("geo.zip_code") rather than coercing straight to "unknown".
        let code = vec![blank_entry("zip")];
        let mut llm = HashMap::new();
        llm.insert(
            "zip".to_string(),
            LlmDictField {
                label:        "ZIP".to_string(),
                description:  "postal code".to_string(),
                content_type: "zip_code".to_string(),
                concept:      String::new(),
                role:         "dimension".to_string(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        assert_eq!(combined[0].content_type, "zip_code");
        assert_eq!(
            combined[0].concept, "geo.zip_code",
            "concept must backfill from the merged content_type, not fall back to unknown"
        );

        // Same backfill when the model emits the literal "unknown" concept (a valid
        // vocab token) alongside a mappable content_type.
        let code = vec![blank_entry("zip2")];
        let mut llm = HashMap::new();
        llm.insert(
            "zip2".to_string(),
            LlmDictField {
                label:        "ZIP".to_string(),
                description:  "postal code".to_string(),
                content_type: "zip_code".to_string(),
                concept:      "unknown".to_string(),
                role:         "dimension".to_string(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        assert_eq!(
            combined[0].concept, "geo.zip_code",
            "a literal \"unknown\" concept must still backfill from content_type"
        );

        // No content_type mapping → still "unknown".
        let code = vec![blank_entry("notes")];
        let mut llm = HashMap::new();
        llm.insert(
            "notes".to_string(),
            LlmDictField {
                label:        "Notes".to_string(),
                description:  "free text".to_string(),
                content_type: "free_text".to_string(),
                concept:      String::new(),
                role:         String::new(),
                null_values:  Vec::new(),
                gauge_range:  None,
            },
        );
        let combined = combine_dictionary_entries(code, &llm, true);
        assert_eq!(combined[0].concept, "unknown");
    }
}
