holger-server-lib 0.6.9

//! Retention / GC **planner** — the pure function that decides which artifacts a
//! repository should keep and which it should reclaim.
//!
//! This is the answer to the Artifactory/Nexus "retention policy" table stakes
//! (`.nornir/feature-complete-gaps.md` §4, the highest operability gap), shipped
//! the way the parity design demands: a **pure, deterministic function over
//! artifact metadata** — no I/O, no scheduler, no deletion. [`plan_retention`]
//! takes a slice of [`ArtifactRecord`] plus a [`RetentionPolicy`] and returns a
//! [`RetentionPlan`]: a keep/delete verdict per artifact, each annotated with the
//! [`RetentionReason`]s that produced it, plus the reclaimable-byte total.
//!
//! Why a planner and not a deleter: holger's storage is immutable, blake3-verified
//! znippy archives (retention at the *archive-set* level, never in-archive edits).
//! The plan is the decision; carrying it out belongs to a jera job that drops whole
//! superseded generations, and the decision is historized to skade — exactly the
//! reuse note in the gap doc. Keeping the planner pure means it is trivially
//! testable, re-runnable on a policy change, and identical whether it drives a
//! `--dry-run` CLI preview, a gRPC verb, or the eventual jera executor.
//!
//! ## Policy dimensions (all optional, all composable)
//!
//! * `keep_last_n`          — keep the N newest **versions per name**, reclaim the rest.
//! * `max_age_secs`         — reclaim anything older than this (needs a `created_at`).
//! * `keep_min_per_name`    — a floor: never drop below this many versions of a name,
//!                            even when age would.
//! * `max_total_bytes`      — a store-size quota: after the above, evict oldest-first
//!                            until the kept set fits, honouring the floors.
//! * `protect_latest_per_name` — the single newest version of every name is never
//!                            reclaimed (default `true`, the safety net).
//!
//! Ordering is deterministic end to end (group by `(namespace, name)`, newest-first
//! within a group by `created_at` then version), so the same inputs always yield the
//! same plan — a property the tests pin.

use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

use traits::ArtifactId;

fn default_true() -> bool {
    true
}

/// One artifact considered for retention. Pure data — the planner never does I/O;
/// callers gather these from a backend's `list()` (see
/// `Holger::plan_repo_retention`) or synthesize them in tests/benches.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactRecord {
    /// The artifact coordinate. Records are grouped by `(namespace, name)`; the
    /// `version` breaks newest-first ties when `created_at` is absent or equal.
    pub id: ArtifactId,
    /// Size on disk (uncompressed bytes) — the reclaim/quota accounting unit.
    pub size_bytes: u64,
    /// Ingest time (unix seconds). `None` when the backend can't supply one
    /// (immutable znippy archives carry no per-file mtime today — see the deferred
    /// note in `holger_agents_problems.md`); age policies then no-op for this
    /// record while `keep_last_n` / `max_total_bytes` still apply, ordered by
    /// version.
    pub created_at: Option<i64>,
    /// Protected from reclamation by a **custom property** (`RetentionPolicy::
    /// protect_property`). The pure planner honours this flag — a `true` record is
    /// never selected for deletion by any rule (superseded / aged / quota) — but
    /// never computes it: the property lookup lives in the adapter
    /// (`Holger::plan_repo_retention`), which reads the `PropertyStore` and sets
    /// this, keeping the planner I/O-free. Default `false`.
    #[serde(default)]
    pub protected_by_property: bool,
}

impl ArtifactRecord {
    /// Build a record from a `list()` entry with no known ingest time. `size_bytes`
    /// is clamped at zero (the proto carries an `i64`).
    pub fn from_entry(entry: traits::ArtifactEntry) -> Self {
        Self {
            id: entry.id,
            size_bytes: entry.size_bytes.max(0) as u64,
            created_at: None,
            protected_by_property: false,
        }
    }

    /// Mark this record as protected by a custom property (builder-style, used by
    /// the adapter after a `PropertyStore` lookup).
    pub fn protected_by_property(mut self, protected: bool) -> Self {
        self.protected_by_property = protected;
        self
    }

    fn group_key(&self) -> (Option<String>, String) {
        (self.id.namespace.clone(), self.id.name.clone())
    }
}

/// A retention policy: which artifacts to keep. Every dimension is optional and
/// they compose (an artifact is reclaimed if any rule selects it, unless a
/// protection rule saves it). Deserializes straight from RON.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionPolicy {
    /// Keep the N newest **versions per name**; reclaim the rest. `None` = no bound.
    #[serde(default)]
    pub keep_last_n: Option<usize>,
    /// Reclaim records older than this many seconds. Needs `created_at`; a record
    /// with none is never aged out. `None` = no age bound.
    #[serde(default)]
    pub max_age_secs: Option<i64>,
    /// Floor: never drop below this many versions of a name, even when age would.
    /// `None` = no floor. Independent of `protect_latest_per_name`.
    #[serde(default)]
    pub keep_min_per_name: Option<usize>,
    /// Store-size quota (bytes): after the per-name rules, evict oldest-first until
    /// the kept set fits under this, honouring the protection floors. `None` = no
    /// quota.
    #[serde(default)]
    pub max_total_bytes: Option<u64>,
    /// Never reclaim the single newest version of any name. Default `true` — the
    /// safety net that stops an aggressive `max_age`/quota from emptying a repo.
    #[serde(default = "default_true")]
    pub protect_latest_per_name: bool,
    /// Protect any artifact carrying this custom **property** `(key, value)` from
    /// reclamation — the "pin this blob" escape hatch (e.g. `("keep","true")`, or
    /// `("keep","")` for "any value under `keep`"). The pure planner does not read
    /// this (it has no store); the adapter (`Holger::plan_repo_retention`) resolves
    /// it against the `PropertyStore` and sets `ArtifactRecord::protected_by_property`.
    /// `None` = no property protection. `.nornir/property-policy-design.md`.
    #[serde(default)]
    pub protect_property: Option<(String, String)>,
}

impl Default for RetentionPolicy {
    fn default() -> Self {
        Self {
            keep_last_n: None,
            max_age_secs: None,
            keep_min_per_name: None,
            max_total_bytes: None,
            protect_latest_per_name: true,
            protect_property: None,
        }
    }
}

impl RetentionPolicy {
    /// A policy that selects nothing for deletion (the identity plan) — the base
    /// for the builder methods below.
    pub fn new() -> Self {
        Self::default()
    }
    /// Keep the N newest versions per name.
    pub fn keep_last(mut self, n: usize) -> Self {
        self.keep_last_n = Some(n);
        self
    }
    /// Reclaim records older than `secs`.
    pub fn max_age(mut self, secs: i64) -> Self {
        self.max_age_secs = Some(secs);
        self
    }
    /// Never drop below `n` versions per name.
    pub fn keep_min(mut self, n: usize) -> Self {
        self.keep_min_per_name = Some(n);
        self
    }
    /// Cap the kept set at `bytes` total (oldest-first eviction).
    pub fn quota(mut self, bytes: u64) -> Self {
        self.max_total_bytes = Some(bytes);
        self
    }
    /// Toggle the "never delete the latest version of a name" safety net.
    pub fn protect_latest(mut self, on: bool) -> Self {
        self.protect_latest_per_name = on;
        self
    }
    /// Pin any artifact carrying the custom property `(key, value)` from
    /// reclamation (an empty `value` = "the key is present, any value"). The
    /// adapter resolves this against the `PropertyStore`.
    pub fn protect_property_kv(mut self, key: &str, value: &str) -> Self {
        self.protect_property = Some((key.to_string(), value.to_string()));
        self
    }
    /// True when the policy would never select anything for deletion regardless of
    /// input (all bounds unset). The plan is then the pure keep-everything identity.
    pub fn is_noop(&self) -> bool {
        self.keep_last_n.is_none()
            && self.max_age_secs.is_none()
            && self.max_total_bytes.is_none()
    }
}

/// Why the planner kept or reclaimed a record. A kept record carries the
/// protection/keep reasons; a reclaimed one carries the rule(s) that selected it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RetentionReason {
    /// KEEP — the single newest version of its name (`protect_latest_per_name`).
    ProtectedLatest,
    /// KEEP — carries the `protect_property` custom property (e.g. `keep=true`).
    ProtectedByProperty,
    /// KEEP — within the `keep_min_per_name` floor.
    MinPerNameFloor,
    /// KEEP — within the `keep_last_n` newest versions.
    WithinKeepLastN,
    /// KEEP — younger than `max_age_secs`.
    WithinMaxAge,
    /// KEEP — no deletion rule selected it.
    NoPolicyMatched,
    /// DELETE — beyond `keep_last_n` (a newer version supersedes it).
    SupersededByNewer,
    /// DELETE — older than `max_age_secs`.
    AgedOut,
    /// DELETE — evicted to bring the kept set under `max_total_bytes`.
    QuotaEvicted,
}

impl RetentionReason {
    /// Short stable slug for CLI/JSON output.
    pub fn as_str(&self) -> &'static str {
        match self {
            RetentionReason::ProtectedLatest => "protected-latest",
            RetentionReason::ProtectedByProperty => "protected-by-property",
            RetentionReason::MinPerNameFloor => "min-per-name-floor",
            RetentionReason::WithinKeepLastN => "within-keep-last-n",
            RetentionReason::WithinMaxAge => "within-max-age",
            RetentionReason::NoPolicyMatched => "no-policy-matched",
            RetentionReason::SupersededByNewer => "superseded-by-newer",
            RetentionReason::AgedOut => "aged-out",
            RetentionReason::QuotaEvicted => "quota-evicted",
        }
    }
}

/// The per-artifact verdict: keep or delete, plus the reason(s).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionDecision {
    pub record: ArtifactRecord,
    /// `true` = reclaim this artifact; `false` = keep it.
    pub delete: bool,
    /// The reasons that produced the verdict (deletion rules when `delete`, keep/
    /// protection reasons otherwise).
    pub reasons: Vec<RetentionReason>,
}

/// A full retention plan: a verdict per input record plus the roll-up totals. The
/// decisions are in a deterministic order (by `(namespace, name)`, newest-first
/// within a name).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionPlan {
    pub decisions: Vec<RetentionDecision>,
    /// How many records are kept.
    pub kept_count: usize,
    /// How many records are selected for reclamation.
    pub delete_count: usize,
    /// Total bytes of the kept records.
    pub kept_bytes: u64,
    /// Total bytes that carrying out the plan would reclaim.
    pub reclaimed_bytes: u64,
    /// Number of distinct `(namespace, name)` groups evaluated.
    pub name_count: usize,
    /// The `now` the plan was evaluated at (unix seconds) — stamped so a stored
    /// plan is reproducible.
    pub evaluated_at: i64,
}

impl RetentionPlan {
    /// The records this plan would reclaim, in plan order.
    pub fn to_delete(&self) -> impl Iterator<Item = &ArtifactRecord> {
        self.decisions.iter().filter(|d| d.delete).map(|d| &d.record)
    }
    /// The records this plan would keep, in plan order.
    pub fn to_keep(&self) -> impl Iterator<Item = &ArtifactRecord> {
        self.decisions.iter().filter(|d| !d.delete).map(|d| &d.record)
    }
}

// ============================================================================
// Executor contract (pure data — the executor lives on the `Holger` facade,
// `Holger::execute_repo_retention`, so this module stays I/O-free). The planner
// decides; the executor consumes a `RetentionPlan` and acts.
// ============================================================================

/// How to carry out a [`RetentionPlan`]. **Dry-run by default** — nothing is
/// deleted unless `dry_run` is explicitly set to `false`. This is the gate: a
/// preview costs nothing and drops nothing; an actual reclaim is an opt-in.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GcOptions {
    /// `true` (the default) ⇒ report what *would* be reclaimed, unlink nothing.
    /// `false` ⇒ actually drop the plan's delete-set from the store.
    pub dry_run: bool,
}

impl Default for GcOptions {
    fn default() -> Self {
        Self { dry_run: true }
    }
}

impl GcOptions {
    /// The safe default: a dry-run that deletes nothing.
    pub fn dry_run() -> Self {
        Self { dry_run: true }
    }
    /// Opt in to an actual reclaim — the executor will unlink the plan's
    /// delete-set (each safely, content-id-verified).
    pub fn execute() -> Self {
        Self { dry_run: false }
    }
}

/// What the executor did (or, in a dry-run, would do) with one artifact.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum GcOutcome {
    /// Bytes were actually reclaimed from the store.
    Deleted,
    /// Dry-run: these bytes *would* be reclaimed (nothing was unlinked).
    WouldDelete,
    /// Not acted on — the reason is carried (protected by the plan, already
    /// absent, …). Never an error, just a no-op the operator should see.
    Skipped(String),
    /// A delete was attempted but failed (content-id mismatch, IO error, or a
    /// backend that does not support in-place delete). The store is unchanged for
    /// this artifact; the reason is carried.
    Failed(String),
}

impl GcOutcome {
    /// Short stable slug for CLI/JSON output.
    pub fn as_str(&self) -> &'static str {
        match self {
            GcOutcome::Deleted => "deleted",
            GcOutcome::WouldDelete => "would-delete",
            GcOutcome::Skipped(_) => "skipped",
            GcOutcome::Failed(_) => "failed",
        }
    }
}

/// The executor's per-artifact record — the plan's coordinate plus what became
/// of it and the content id that was verified before any unlink.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GcAction {
    pub id: ArtifactId,
    pub size_bytes: u64,
    /// The lowercase-hex SHA-256 the executor computed for the stored bytes and
    /// verified against the backend before unlinking. `None` when the artifact
    /// was absent or protected (never read).
    pub content_id: Option<String>,
    pub outcome: GcOutcome,
    /// The plan's reasons this coordinate was in the delete-set.
    pub reasons: Vec<RetentionReason>,
}

/// The result of carrying out (or previewing) a [`RetentionPlan`] against one
/// repository — the GC record historized to the warehouse/audit plane.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GcReport {
    pub repository: String,
    /// Whether this was a preview (`true`) or an actual reclaim (`false`).
    pub dry_run: bool,
    pub actions: Vec<GcAction>,
    /// Artifacts actually reclaimed (0 in a dry-run).
    pub deleted_count: usize,
    /// Bytes actually reclaimed (0 in a dry-run).
    pub reclaimed_bytes: u64,
    /// Artifacts the executor would reclaim in a dry-run (0 outside a dry-run).
    pub would_delete_count: usize,
    /// Artifacts left alone (protected, already absent).
    pub skipped_count: usize,
    /// Artifacts a delete was attempted for but failed.
    pub failed_count: usize,
    /// The `evaluated_at` of the plan that produced this report.
    pub evaluated_at: i64,
}

/// Compare two version strings newest-first-friendly: split on `.`, `-`, `+`,
/// compare segment by segment, numerically when both segments parse as integers
/// (so `10` > `9`), lexically otherwise. A pure prerelease-agnostic ordering — it
/// is only a *tie-break* for grouping, not a semver authority.
fn cmp_version(a: &str, b: &str) -> std::cmp::Ordering {
    use std::cmp::Ordering;
    let split = |s: &str| -> Vec<String> {
        s.split(['.', '-', '+', '_'])
            .filter(|p| !p.is_empty())
            .map(|p| p.to_string())
            .collect()
    };
    let (sa, sb) = (split(a), split(b));
    for (x, y) in sa.iter().zip(sb.iter()) {
        let ord = match (x.parse::<u64>(), y.parse::<u64>()) {
            (Ok(nx), Ok(ny)) => nx.cmp(&ny),
            _ => x.cmp(y),
        };
        if ord != Ordering::Equal {
            return ord;
        }
    }
    // Longer version (more segments) sorts as the greater when the shared prefix
    // is equal (1.2.1 > 1.2), keeping the order total + stable.
    sa.len().cmp(&sb.len())
}

/// The pure planner. Groups `records` by `(namespace, name)`, orders each group
/// newest-first, applies the per-name rules (`protect_latest`, `keep_min`,
/// `keep_last_n`, `max_age`), then the global `max_total_bytes` quota (oldest-first
/// eviction, honouring the floors), and returns the [`RetentionPlan`].
///
/// Deterministic: the same `(records, policy, now)` always yields the same plan.
/// Total function: an empty input yields an empty plan; a no-op policy keeps
/// everything.
pub fn plan_retention(
    records: &[ArtifactRecord],
    policy: &RetentionPolicy,
    now: i64,
) -> RetentionPlan {
    // 1. Group by (namespace, name), deterministic order via BTreeMap. Each group
    //    holds indices into `records` so we can rebuild the plan in a stable order.
    let mut groups: BTreeMap<(Option<String>, String), Vec<usize>> = BTreeMap::new();
    for (i, r) in records.iter().enumerate() {
        groups.entry(r.group_key()).or_default().push(i);
    }
    let name_count = groups.len();

    // 2. Order each group newest-first: created_at desc (unknown time sorts oldest),
    //    then version desc, then a stable version-string tie-break.
    for idxs in groups.values_mut() {
        idxs.sort_by(|&x, &y| {
            let (rx, ry) = (&records[x], &records[y]);
            let tx = rx.created_at.unwrap_or(i64::MIN);
            let ty = ry.created_at.unwrap_or(i64::MIN);
            ty.cmp(&tx)
                .then_with(|| cmp_version(&ry.id.version, &rx.id.version))
                .then_with(|| ry.id.version.cmp(&rx.id.version))
        });
    }

    // 3. Per-name rules → an initial decision per record (keyed by original index).
    let mut delete = vec![false; records.len()];
    let mut reasons: Vec<Vec<RetentionReason>> = vec![Vec::new(); records.len()];
    // Track protection so the quota pass can't evict a protected/floored record.
    let mut protected = vec![false; records.len()];

    for idxs in groups.values() {
        for (pos, &i) in idxs.iter().enumerate() {
            let r = &records[i];
            let protected_latest = policy.protect_latest_per_name && pos == 0;
            let within_floor = policy.keep_min_per_name.is_some_and(|m| pos < m);
            // A custom-property pin (adapter-computed) is a hard protection too, so
            // no rule — superseded / aged / quota — can ever reclaim it.
            protected[i] = protected_latest || within_floor || r.protected_by_property;

            let superseded = policy.keep_last_n.is_some_and(|n| pos >= n);
            let aged = match (policy.max_age_secs, r.created_at) {
                (Some(max), Some(c)) => now.saturating_sub(c) > max,
                _ => false,
            };

            let should_delete = (superseded || aged) && !protected[i];
            delete[i] = should_delete;

            if should_delete {
                if superseded {
                    reasons[i].push(RetentionReason::SupersededByNewer);
                }
                if aged {
                    reasons[i].push(RetentionReason::AgedOut);
                }
            } else {
                if protected_latest {
                    reasons[i].push(RetentionReason::ProtectedLatest);
                }
                if r.protected_by_property {
                    reasons[i].push(RetentionReason::ProtectedByProperty);
                }
                if within_floor {
                    reasons[i].push(RetentionReason::MinPerNameFloor);
                }
                if let Some(n) = policy.keep_last_n {
                    if pos < n {
                        reasons[i].push(RetentionReason::WithinKeepLastN);
                    }
                }
                if policy.max_age_secs.is_some() && !aged && r.created_at.is_some() {
                    reasons[i].push(RetentionReason::WithinMaxAge);
                }
                if reasons[i].is_empty() {
                    reasons[i].push(RetentionReason::NoPolicyMatched);
                }
            }
        }
    }

    // 4. Quota pass: if the kept set still exceeds `max_total_bytes`, evict the
    //    kept, non-protected records oldest-first (created_at asc, then version asc)
    //    until it fits (or nothing evictable remains).
    if let Some(quota) = policy.max_total_bytes {
        let mut kept_bytes: u64 = records
            .iter()
            .enumerate()
            .filter(|(i, _)| !delete[*i])
            .map(|(_, r)| r.size_bytes)
            .sum();

        if kept_bytes > quota {
            // Eviction order: oldest first (unknown time = oldest = evicted first),
            // then oldest version, then a stable index tie-break.
            let mut evictable: Vec<usize> = (0..records.len())
                .filter(|&i| !delete[i] && !protected[i])
                .collect();
            evictable.sort_by(|&x, &y| {
                let (rx, ry) = (&records[x], &records[y]);
                let tx = rx.created_at.unwrap_or(i64::MIN);
                let ty = ry.created_at.unwrap_or(i64::MIN);
                tx.cmp(&ty)
                    .then_with(|| cmp_version(&rx.id.version, &ry.id.version))
                    .then_with(|| x.cmp(&y))
            });
            for i in evictable {
                if kept_bytes <= quota {
                    break;
                }
                delete[i] = true;
                reasons[i].clear();
                reasons[i].push(RetentionReason::QuotaEvicted);
                kept_bytes = kept_bytes.saturating_sub(records[i].size_bytes);
            }
        }
    }

    // 5. Assemble the plan in the deterministic group/newest-first order.
    let mut decisions = Vec::with_capacity(records.len());
    let mut kept_count = 0usize;
    let mut delete_count = 0usize;
    let mut kept_bytes = 0u64;
    let mut reclaimed_bytes = 0u64;
    for idxs in groups.values() {
        for &i in idxs {
            let d = delete[i];
            if d {
                delete_count += 1;
                reclaimed_bytes += records[i].size_bytes;
            } else {
                kept_count += 1;
                kept_bytes += records[i].size_bytes;
            }
            decisions.push(RetentionDecision {
                record: records[i].clone(),
                delete: d,
                reasons: std::mem::take(&mut reasons[i]),
            });
        }
    }

    RetentionPlan {
        decisions,
        kept_count,
        delete_count,
        kept_bytes,
        reclaimed_bytes,
        name_count,
        evaluated_at: now,
    }
}

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

    fn rec(name: &str, version: &str, size: u64, created_at: Option<i64>) -> ArtifactRecord {
        ArtifactRecord {
            id: ArtifactId {
                namespace: None,
                name: name.into(),
                version: version.into(),
            },
            size_bytes: size,
            created_at,
            protected_by_property: false,
        }
    }

    fn ns_rec(ns: &str, name: &str, version: &str, size: u64) -> ArtifactRecord {
        ArtifactRecord {
            id: ArtifactId {
                namespace: Some(ns.into()),
                name: name.into(),
                version: version.into(),
            },
            size_bytes: size,
            created_at: None,
            protected_by_property: false,
        }
    }

    #[test]
    fn empty_input_yields_empty_plan() {
        let plan = plan_retention(&[], &RetentionPolicy::new().keep_last(3), 1000);
        assert_eq!(plan.decisions.len(), 0);
        assert_eq!(plan.kept_count, 0);
        assert_eq!(plan.delete_count, 0);
        assert_eq!(plan.name_count, 0);
    }

    #[test]
    fn noop_policy_keeps_everything() {
        let recs = vec![
            rec("a", "1.0.0", 10, None),
            rec("a", "2.0.0", 10, None),
            rec("a", "3.0.0", 10, None),
        ];
        let plan = plan_retention(&recs, &RetentionPolicy::new(), 1000);
        assert!(RetentionPolicy::new().is_noop());
        assert_eq!(plan.kept_count, 3);
        assert_eq!(plan.delete_count, 0);
        assert_eq!(plan.reclaimed_bytes, 0);
    }

    #[test]
    fn keep_last_n_reclaims_older_versions_per_name() {
        // Five versions of `a`, keep the 2 newest → reclaim 3 oldest.
        let recs = vec![
            rec("a", "1.0.0", 100, None),
            rec("a", "2.0.0", 100, None),
            rec("a", "3.0.0", 100, None),
            rec("a", "4.0.0", 100, None),
            rec("a", "5.0.0", 100, None),
        ];
        let plan = plan_retention(&recs, &RetentionPolicy::new().keep_last(2), 1000);
        assert_eq!(plan.kept_count, 2);
        assert_eq!(plan.delete_count, 3);
        assert_eq!(plan.reclaimed_bytes, 300);
        // The two kept are the newest versions.
        let kept: Vec<_> = plan.to_keep().map(|r| r.id.version.clone()).collect();
        assert_eq!(kept, vec!["5.0.0", "4.0.0"]);
        // Version ordering is numeric, not lexical (10 > 9).
        let recs2 = vec![
            rec("b", "1.9.0", 1, None),
            rec("b", "1.10.0", 1, None),
        ];
        let p2 = plan_retention(&recs2, &RetentionPolicy::new().keep_last(1), 1000);
        assert_eq!(p2.to_keep().next().unwrap().id.version, "1.10.0");
    }

    #[test]
    fn keep_last_n_is_per_name_not_global() {
        let recs = vec![
            rec("a", "1.0.0", 1, None),
            rec("a", "2.0.0", 1, None),
            rec("b", "1.0.0", 1, None),
            rec("b", "2.0.0", 1, None),
        ];
        let plan = plan_retention(&recs, &RetentionPolicy::new().keep_last(1), 1000);
        assert_eq!(plan.name_count, 2);
        // One newest kept per name → 2 kept, 2 reclaimed.
        assert_eq!(plan.kept_count, 2);
        assert_eq!(plan.delete_count, 2);
    }

    #[test]
    fn max_age_reclaims_old_and_keeps_young() {
        let now = 10_000;
        let recs = vec![
            rec("a", "1.0.0", 1, Some(now - 5000)), // old
            rec("a", "2.0.0", 1, Some(now - 100)),  // young
            rec("a", "3.0.0", 1, Some(now - 50)),   // youngest
        ];
        // max_age 1000s, and turn OFF latest-protection so age can bite freely.
        let policy = RetentionPolicy::new().max_age(1000).protect_latest(false);
        let plan = plan_retention(&recs, &policy, now);
        assert_eq!(plan.delete_count, 1);
        let del: Vec<_> = plan.to_delete().map(|r| r.id.version.clone()).collect();
        assert_eq!(del, vec!["1.0.0"]);
        assert!(plan
            .decisions
            .iter()
            .find(|d| d.record.id.version == "1.0.0")
            .unwrap()
            .reasons
            .contains(&RetentionReason::AgedOut));
    }

    #[test]
    fn protect_latest_saves_the_newest_even_when_aged() {
        let now = 10_000;
        // Both ancient; only one version of `a`.
        let recs = vec![rec("a", "1.0.0", 1, Some(now - 9000))];
        let policy = RetentionPolicy::new().max_age(10); // everything is "old"
        let plan = plan_retention(&recs, &policy, now);
        // protect_latest defaults ON → the lone newest survives.
        assert_eq!(plan.delete_count, 0);
        assert!(plan.decisions[0]
            .reasons
            .contains(&RetentionReason::ProtectedLatest));
    }

    #[test]
    fn keep_min_per_name_is_a_floor_against_age() {
        let now = 10_000;
        // Three old versions; max_age would drop all, but keep_min 2 floors it, and
        // protect_latest is off so the floor (not protection) is what saves them.
        let recs = vec![
            rec("a", "1.0.0", 1, Some(now - 9000)),
            rec("a", "2.0.0", 1, Some(now - 8000)),
            rec("a", "3.0.0", 1, Some(now - 7000)),
        ];
        let policy = RetentionPolicy::new()
            .max_age(10)
            .keep_min(2)
            .protect_latest(false);
        let plan = plan_retention(&recs, &policy, now);
        // 2 newest floored-kept, oldest aged out.
        assert_eq!(plan.kept_count, 2);
        assert_eq!(plan.delete_count, 1);
        let kept: Vec<_> = plan.to_keep().map(|r| r.id.version.clone()).collect();
        assert_eq!(kept, vec!["3.0.0", "2.0.0"]);
        assert!(plan
            .decisions
            .iter()
            .find(|d| d.record.id.version == "2.0.0")
            .unwrap()
            .reasons
            .contains(&RetentionReason::MinPerNameFloor));
    }

    #[test]
    fn quota_evicts_oldest_until_under_bound() {
        let now = 10_000;
        // Four artifacts of distinct names, 100 bytes each = 400 total. Quota 250 →
        // must evict 2 (oldest first). protect_latest is per-name; distinct names so
        // each is a "latest" — turn it off so the quota can actually act.
        let recs = vec![
            rec("a", "1.0.0", 100, Some(now - 4000)),
            rec("b", "1.0.0", 100, Some(now - 3000)),
            rec("c", "1.0.0", 100, Some(now - 2000)),
            rec("d", "1.0.0", 100, Some(now - 1000)),
        ];
        let policy = RetentionPolicy::new().quota(250).protect_latest(false);
        let plan = plan_retention(&recs, &policy, now);
        assert!(plan.kept_bytes <= 250, "kept {} > 250", plan.kept_bytes);
        assert_eq!(plan.delete_count, 2);
        // Oldest two (a, b) evicted.
        let del: Vec<_> = plan.to_delete().map(|r| r.id.name.clone()).collect();
        assert_eq!(del, vec!["a", "b"]);
        assert!(plan
            .decisions
            .iter()
            .find(|d| d.record.id.name == "a")
            .unwrap()
            .reasons
            .contains(&RetentionReason::QuotaEvicted));
    }

    #[test]
    fn quota_never_evicts_a_protected_latest() {
        let now = 10_000;
        // One name, two versions, 100 bytes each. Quota 50 (< one artifact). The
        // newest is protected → it must survive even though the quota can't be met.
        let recs = vec![
            rec("a", "1.0.0", 100, Some(now - 2000)),
            rec("a", "2.0.0", 100, Some(now - 1000)),
        ];
        let policy = RetentionPolicy::new().quota(50); // protect_latest defaults ON
        let plan = plan_retention(&recs, &policy, now);
        // Older evicted; newest protected stays despite the quota being unmet.
        assert_eq!(plan.delete_count, 1);
        let kept: Vec<_> = plan.to_keep().map(|r| r.id.version.clone()).collect();
        assert_eq!(kept, vec!["2.0.0"]);
        assert!(plan.kept_bytes > 50, "protected latest must survive the quota");
    }

    #[test]
    fn combined_policy_composes_keep_last_and_quota() {
        let now = 10_000;
        // Name `a`: 4 versions. keep_last 3 drops v1; then quota trims more oldest.
        let recs = vec![
            rec("a", "1.0.0", 100, Some(now - 4000)),
            rec("a", "2.0.0", 100, Some(now - 3000)),
            rec("a", "3.0.0", 100, Some(now - 2000)),
            rec("a", "4.0.0", 100, Some(now - 1000)),
        ];
        let policy = RetentionPolicy::new()
            .keep_last(3)
            .quota(150)
            .protect_latest(false);
        let plan = plan_retention(&recs, &policy, now);
        // keep_last 3 removes v1 (superseded); quota 150 then evicts v2 & v3 oldest
        // -first, leaving only v4 (100 <= 150).
        assert!(plan.kept_bytes <= 150);
        let kept: Vec<_> = plan.to_keep().map(|r| r.id.version.clone()).collect();
        assert_eq!(kept, vec!["4.0.0"]);
        // v1's reason is superseded (per-name rule), v2/v3 are quota-evicted.
        let v1 = plan.decisions.iter().find(|d| d.record.id.version == "1.0.0").unwrap();
        assert!(v1.reasons.contains(&RetentionReason::SupersededByNewer));
    }

    #[test]
    fn namespace_separates_groups() {
        // Same name, different namespace = different groups; keep_last 1 keeps one
        // newest per (namespace, name).
        let recs = vec![
            ns_rec("org1", "lib", "1.0.0", 1),
            ns_rec("org1", "lib", "2.0.0", 1),
            ns_rec("org2", "lib", "1.0.0", 1),
            ns_rec("org2", "lib", "2.0.0", 1),
        ];
        let plan = plan_retention(&recs, &RetentionPolicy::new().keep_last(1), 1000);
        assert_eq!(plan.name_count, 2);
        assert_eq!(plan.kept_count, 2);
        assert_eq!(plan.delete_count, 2);
    }

    #[test]
    fn plan_is_deterministic() {
        let recs = vec![
            rec("z", "3.0.0", 7, Some(300)),
            rec("a", "1.0.0", 3, Some(100)),
            rec("a", "2.0.0", 3, Some(200)),
            rec("z", "1.0.0", 7, Some(100)),
        ];
        let policy = RetentionPolicy::new().keep_last(1).quota(20).protect_latest(false);
        let p1 = plan_retention(&recs, &policy, 1000);
        let p2 = plan_retention(&recs, &policy, 1000);
        assert_eq!(p1, p2, "same inputs must produce byte-identical plans");
    }

    #[test]
    fn record_without_created_at_is_never_aged_out() {
        // A record with no ingest time can't be aged out, but keep_last still applies.
        let recs = vec![
            rec("a", "1.0.0", 1, None),
            rec("a", "2.0.0", 1, None),
        ];
        let policy = RetentionPolicy::new().max_age(1).protect_latest(false);
        let plan = plan_retention(&recs, &policy, 1_000_000);
        assert_eq!(plan.delete_count, 0, "no created_at → age never bites");
    }

    #[test]
    fn cmp_version_orders_numeric_segments() {
        use std::cmp::Ordering;
        assert_eq!(cmp_version("1.10.0", "1.9.0"), Ordering::Greater);
        assert_eq!(cmp_version("2.0.0", "10.0.0"), Ordering::Less);
        assert_eq!(cmp_version("1.2.0", "1.2"), Ordering::Greater);
        assert_eq!(cmp_version("1.0.0", "1.0.0"), Ordering::Equal);
    }

    /// RED-when-broken (property-policy §): a blob PINNED by a custom property
    /// survives a `keep_last_n=1` sweep that reclaims its untagged sibling. Without
    /// the `protected_by_property` honouring, the older pinned version would be
    /// SupersededByNewer and deleted.
    #[test]
    fn property_pinned_record_survives_a_sweep_that_reclaims_the_untagged() {
        // Two versions of "a"; keep_last_n=1 would normally reclaim the older 1.0.0.
        // Pin 1.0.0 with a property; it must be KEPT while an untagged 1.0.0 would go.
        let pinned = rec("a", "1.0.0", 1, None).protected_by_property(true);
        let newer = rec("a", "2.0.0", 1, None);
        let policy = RetentionPolicy::new().keep_last(1).protect_latest(false);

        let plan = plan_retention(&[pinned.clone(), newer.clone()], &policy, 1_000);
        let older = plan
            .decisions
            .iter()
            .find(|d| d.record.id.version == "1.0.0")
            .unwrap();
        assert!(!older.delete, "the property-pinned older version must be KEPT");
        assert!(
            older.reasons.contains(&RetentionReason::ProtectedByProperty),
            "kept for the property pin"
        );

        // Control (RED-when-broken): the SAME shape UNtagged is reclaimed.
        let untagged = rec("a", "1.0.0", 1, None); // not pinned
        let plan2 = plan_retention(&[untagged, newer], &policy, 1_000);
        let older2 = plan2
            .decisions
            .iter()
            .find(|d| d.record.id.version == "1.0.0")
            .unwrap();
        assert!(older2.delete, "the untagged older version IS reclaimed (control)");
    }

    /// A property pin also beats a byte quota: a pinned record is never quota-evicted.
    #[test]
    fn property_pinned_record_is_not_quota_evicted() {
        // Two names, each one version, 100 bytes; quota=100 forces one eviction.
        let pinned = rec("keepme", "1.0.0", 100, Some(10)).protected_by_property(true);
        let evictable = rec("dropme", "1.0.0", 100, Some(20));
        // protect_latest off so only the pin can save "keepme".
        let policy = RetentionPolicy::new().quota(100).protect_latest(false);
        let plan = plan_retention(&[pinned, evictable], &policy, 1_000);
        let keep = plan.decisions.iter().find(|d| d.record.id.name == "keepme").unwrap();
        assert!(!keep.delete, "a property-pinned record must never be quota-evicted");
    }
}