tachyon-types 0.4.0

//! Resource derivation and lease / gating policy for the run queue.
//!
//! A run targets a set of inventory groups — its *resources*. The lease queue
//! serializes runs that share a resource; the scoped MCP gateway rejects runs
//! whose resources fall outside an allowed set. Both operate on the **derived**
//! resource set, never the raw `--limit` string, so a single-host limit that
//! lands inside a managed cluster is correctly treated as touching that
//! cluster (the slip-through the SOW calls out explicitly).
//!
//! Everything here is pure: it takes an [`Inventory`] snapshot plus plain
//! patterns and returns plain data, so the queue/admission logic is
//! unit-testable without ansible or a running server.

use std::collections::{BTreeSet, HashMap, HashSet};

use crate::inventory::{HostGroup, Inventory};
use crate::run::{Run, RunStatus};
use crate::RunId;

/// Inventory groups that never represent a lease resource: `all` spans every
/// host (locking it would make every run conflict) and `ungrouped` is
/// ansible's catch-all for hosts with no group.
const NON_RESOURCE_GROUPS: [&str; 2] = ["all", "ungrouped"];

/// Minimal shell-style glob: `*` matches any run of characters (including
/// empty), `?` matches exactly one. Case-sensitive, no character classes —
/// enough for matching inventory group names like `cluster_1node_*`.
pub fn glob_match(pattern: &str, text: &str) -> bool {
    let p = pattern.as_bytes();
    let t = text.as_bytes();
    let (mut pi, mut ti) = (0usize, 0usize);
    let (mut star, mut mark) = (None, 0usize);
    while ti < t.len() {
        if pi < p.len() && (p[pi] == b'?' || p[pi] == t[ti]) {
            pi += 1;
            ti += 1;
        } else if pi < p.len() && p[pi] == b'*' {
            star = Some(pi);
            mark = ti;
            pi += 1;
        } else if let Some(s) = star {
            pi = s + 1;
            mark += 1;
            ti = mark;
        } else {
            return false;
        }
    }
    while pi < p.len() && p[pi] == b'*' {
        pi += 1;
    }
    pi == p.len()
}

/// Transitively collect every host reachable from `group`, following children.
/// `seen` guards against cyclic group graphs.
fn collect_group_hosts(
    index: &HashMap<&str, &HostGroup>,
    group: &str,
    out: &mut HashSet<String>,
    seen: &mut HashSet<String>,
) {
    if !seen.insert(group.to_string()) {
        return;
    }
    if let Some(g) = index.get(group) {
        for h in &g.hosts {
            out.insert(h.clone());
        }
        for child in &g.children {
            collect_group_hosts(index, child, out, seen);
        }
    }
}

/// Map every group name to its transitive host set.
fn group_host_closures(inv: &Inventory) -> HashMap<String, HashSet<String>> {
    let index: HashMap<&str, &HostGroup> =
        inv.groups.iter().map(|g| (g.name.as_str(), g)).collect();
    let mut closures = HashMap::new();
    for g in &inv.groups {
        let mut hosts = HashSet::new();
        let mut seen = HashSet::new();
        collect_group_hosts(&index, &g.name, &mut hosts, &mut seen);
        closures.insert(g.name.clone(), hosts);
    }
    closures
}

/// Resolve an ansible `--limit` expression to the set of hosts it targets.
///
/// Handles the realistic cases — a bare group name, a bare host name, and
/// `:`/`,`-separated unions of those, plus `*`/`?` globs over group and host
/// names. The intersection (`:&`) and exclusion (`:!`) operators are treated
/// as unions (the leading operator char is stripped): this can only *widen*
/// the resolved set, so a lease never under-locks. Unknown tokens (not a group
/// or host in inventory) are ignored — ansible would not target them either.
/// A missing/empty limit means "all hosts".
fn resolve_target_hosts(
    limit: Option<&str>,
    all_hosts: &HashSet<String>,
    group_hosts: &HashMap<String, HashSet<String>>,
) -> HashSet<String> {
    let limit = match limit {
        Some(s) if !s.trim().is_empty() => s,
        _ => return all_hosts.clone(),
    };

    let mut target = HashSet::new();
    for raw in limit.split([':', ',']) {
        let tok = raw.trim().trim_start_matches(|c| c == '&' || c == '!').trim();
        if tok.is_empty() {
            continue;
        }
        if tok == "all" || tok == "*" {
            return all_hosts.clone();
        }
        if tok.contains('*') || tok.contains('?') {
            for (gname, hosts) in group_hosts {
                if glob_match(tok, gname) {
                    target.extend(hosts.iter().cloned());
                }
            }
            for h in all_hosts {
                if glob_match(tok, h) {
                    target.insert(h.clone());
                }
            }
            continue;
        }
        if let Some(hosts) = group_hosts.get(tok) {
            target.extend(hosts.iter().cloned());
            continue;
        }
        if all_hosts.contains(tok) {
            target.insert(tok.to_string());
        }
        // else: token names neither a known group nor host — ignore.
    }
    target
}

/// Derive the resource set a run touches: every inventory group (except the
/// non-resource `all`/`ungrouped`) whose transitive host set intersects the
/// hosts the run's `limit` targets. Returned sorted and de-duplicated.
pub fn derive_resources(limit: Option<&str>, inv: &Inventory) -> Vec<String> {
    let group_hosts = group_host_closures(inv);
    let all_hosts: HashSet<String> = inv.hosts.iter().map(|h| h.name.clone()).collect();
    let target = resolve_target_hosts(limit, &all_hosts, &group_hosts);

    let mut resources = BTreeSet::new();
    for g in &inv.groups {
        if NON_RESOURCE_GROUPS.contains(&g.name.as_str()) {
            continue;
        }
        if let Some(hosts) = group_hosts.get(&g.name) {
            if !hosts.is_disjoint(&target) {
                resources.insert(g.name.clone());
            }
        }
    }
    resources.into_iter().collect()
}

/// Narrows a run's derived resources to the groups that actually carry a lease.
///
/// The command handler stores *every* touched group on the run (including
/// umbrella parents like an `all`-of-clusters group). Leasing at that raw
/// granularity would make sibling clusters contend through their shared parent.
/// Configuring `lease_groups` (glob patterns naming the real cluster groups)
/// restricts contention to those. An empty filter is the identity — every
/// touched group leases — which is the structural fallback when no patterns are
/// configured.
#[derive(Debug, Clone, Default)]
pub struct LeaseFilter {
    patterns: Vec<String>,
}

impl LeaseFilter {
    pub fn new(patterns: Vec<String>) -> Self {
        Self { patterns }
    }

    /// The lease-significant subset of `resources`.
    pub fn apply(&self, resources: &[String]) -> BTreeSet<String> {
        if self.patterns.is_empty() {
            return resources.iter().cloned().collect();
        }
        resources
            .iter()
            .filter(|r| self.patterns.iter().any(|p| glob_match(p, r)))
            .cloned()
            .collect()
    }
}

/// Pure admission step. Given the current set of runs and the lease filter,
/// return the ids of `Queued` runs that may be promoted to `Running` now:
/// processed oldest-first, a queued run is admitted only when its
/// lease-significant resources are disjoint from those already held by a
/// `Running` run *and* from those claimed by an earlier-admitted run in this
/// same pass. The lease is implicit — "the Running run holding the resource" —
/// and releases when the run finishes (its status leaves `Running`).
pub fn select_admissions(runs: &[Run], lease: &LeaseFilter) -> Vec<RunId> {
    let mut held: BTreeSet<String> = BTreeSet::new();
    for r in runs.iter().filter(|r| r.status == RunStatus::Running) {
        held.extend(lease.apply(&r.resources));
    }

    let mut queued: Vec<&Run> = runs
        .iter()
        .filter(|r| r.status == RunStatus::Queued)
        .collect();
    // FIFO by submission time, with a stable id tiebreak for same-instant runs.
    queued.sort_by(|a, b| {
        a.started_at
            .cmp(&b.started_at)
            .then_with(|| a.id.0.to_string().cmp(&b.id.0.to_string()))
    });

    let mut admit = Vec::new();
    for r in queued {
        let res = lease.apply(&r.resources);
        if res.is_disjoint(&held) {
            held.extend(res);
            admit.push(r.id.clone());
        }
    }
    admit
}

/// Allow/deny policy for the scoped MCP gateway, evaluated over a run's
/// *derived* resources. `deny` wins: a run is rejected if any resource matches
/// a deny pattern. If `allow` is non-empty, every resource must additionally
/// match some allow pattern. Empty allow + empty deny is allow-all.
///
/// `significant` is the **policy dimension**: when set, the `allow` check is
/// evaluated only over the derived resources whose group names match a
/// significant pattern, and the ambient co-membership umbrella groups (`ue`,
/// `ue_content`, `windows`, a `clusters` parent…) are filtered out before the
/// check — so they never have to be enumerated in `allow`. This makes
/// `allow=cluster_1node_*` over `significant=cluster_*` a true default-deny: a
/// target that resolves to *no* significant resource cannot be proven in scope
/// and is rejected. `deny` is unaffected — it still spans every derived
/// resource, which is the conservative direction. Empty `significant` preserves
/// the legacy behavior (allow is checked over every derived resource).
///
/// It mirrors the lease's [`LeaseFilter`] (which names the groups that carry a
/// *lease*); here the patterns name the groups that are *significant for
/// policy*. The two dimensions are configured independently.
#[derive(Debug, Clone, Default)]
pub struct ResourcePolicy {
    pub allow: Vec<String>,
    pub deny: Vec<String>,
    pub significant: Vec<String>,
}

impl ResourcePolicy {
    pub fn new(allow: Vec<String>, deny: Vec<String>) -> Self {
        Self {
            allow,
            deny,
            significant: Vec::new(),
        }
    }

    /// Set the policy-significant group patterns (the policy dimension).
    pub fn with_significant(mut self, significant: Vec<String>) -> Self {
        self.significant = significant;
        self
    }

    /// `Ok(())` if the resource set is permitted, `Err(reason)` otherwise.
    pub fn evaluate(&self, resources: &[String]) -> Result<(), String> {
        // Deny wins, and spans every derived resource (conservative — an
        // ambient umbrella group naming a denied cluster still rejects).
        for r in resources {
            if let Some(p) = self.deny.iter().find(|p| glob_match(p, r)) {
                return Err(format!(
                    "target resolves to resource `{r}`, denied by policy (deny pattern `{p}`)"
                ));
            }
        }

        if self.allow.is_empty() {
            return Ok(());
        }

        // Legacy mode (no policy dimension configured): every derived resource
        // must match an allow pattern.
        if self.significant.is_empty() {
            for r in resources {
                if !self.allow.iter().any(|p| glob_match(p, r)) {
                    return Err(format!(
                        "target resolves to resource `{r}`, which is outside the allowed set"
                    ));
                }
            }
            return Ok(());
        }

        // Policy-dimension mode: evaluate `allow` only over the significant
        // resources, and require at least one — a target touching nothing
        // significant cannot be proven in scope, so default-deny.
        let significant: Vec<&String> = resources
            .iter()
            .filter(|r| self.significant.iter().any(|p| glob_match(p, r)))
            .collect();
        if significant.is_empty() {
            return Err(
                "target resolves to no policy-significant resource; denied by default-deny policy"
                    .to_string(),
            );
        }
        for r in significant {
            if !self.allow.iter().any(|p| glob_match(p, r)) {
                return Err(format!(
                    "target resolves to resource `{r}`, which is outside the allowed set"
                ));
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::inventory::{Host, HostGroup};
    use crate::{InventoryId, PlaybookId, RunId};

    fn grp(name: &str, children: &[&str], hosts: &[&str]) -> HostGroup {
        HostGroup {
            name: name.to_string(),
            children: children.iter().map(|s| s.to_string()).collect(),
            hosts: hosts.iter().map(|s| s.to_string()).collect(),
        }
    }

    fn host(name: &str) -> Host {
        Host {
            name: name.to_string(),
            ansible_host: None,
        }
    }

    /// Inventory with two single-node clusters and one 8-node cluster, all under
    /// a shared `clusters` parent plus the conventional `all` group.
    fn inventory() -> Inventory {
        Inventory {
            groups: vec![
                grp("cluster_1node_a", &[], &["n1a"]),
                grp("cluster_1node_b", &[], &["n1b"]),
                grp(
                    "cluster_8node",
                    &[],
                    &["ue-content-01", "ue-content-02", "ue-content-03"],
                ),
                grp(
                    "clusters",
                    &["cluster_1node_a", "cluster_1node_b", "cluster_8node"],
                    &[],
                ),
                grp(
                    "all",
                    &["clusters"],
                    &[],
                ),
            ],
            hosts: vec![
                host("n1a"),
                host("n1b"),
                host("ue-content-01"),
                host("ue-content-02"),
                host("ue-content-03"),
            ],
            id: InventoryId::from("default".to_string()),
        }
    }

    fn run(id: &str, status: RunStatus, started_at: &str, resources: &[&str]) -> Run {
        Run {
            playbook_id: PlaybookId::from("pb".to_string()),
            status,
            started_at: started_at.to_string(),
            finished_at: None,
            exit_code: None,
            extra_vars: None,
            limit: None,
            host_stats: Default::default(),
            resources: resources.iter().map(|s| s.to_string()).collect(),
            client_id: None,
            id: RunId::from(id.to_string()),
        }
    }

    #[test]
    fn glob_matches() {
        assert!(glob_match("cluster_1node_*", "cluster_1node_a"));
        assert!(glob_match("cluster_1node_*", "cluster_1node_b"));
        assert!(!glob_match("cluster_1node_*", "cluster_8node"));
        assert!(!glob_match("cluster_1node_*", "clusters"));
        assert!(glob_match("*", "anything"));
        assert!(glob_match("ue-content-0?", "ue-content-03"));
        assert!(!glob_match("ue-content-0?", "ue-content-10"));
        assert!(glob_match("cluster_8node", "cluster_8node"));
    }

    #[test]
    fn group_limit_resolves_to_group_and_parent() {
        let inv = inventory();
        // A group limit touches the group itself and its umbrella parent, but
        // never the non-resource `all` group.
        assert_eq!(
            derive_resources(Some("cluster_1node_a"), &inv),
            vec!["cluster_1node_a".to_string(), "clusters".to_string()]
        );
    }

    #[test]
    fn single_host_limit_locks_owning_cluster() {
        let inv = inventory();
        // SOW acceptance: a single host inside the live cluster locks the whole
        // cluster group (and parent), not just the one host.
        assert_eq!(
            derive_resources(Some("ue-content-03"), &inv),
            vec!["cluster_8node".to_string(), "clusters".to_string()]
        );
    }

    #[test]
    fn empty_limit_locks_everything() {
        let inv = inventory();
        let res = derive_resources(None, &inv);
        assert!(res.contains(&"cluster_1node_a".to_string()));
        assert!(res.contains(&"cluster_1node_b".to_string()));
        assert!(res.contains(&"cluster_8node".to_string()));
        assert!(res.contains(&"clusters".to_string()));
        assert!(!res.contains(&"all".to_string()));
    }

    #[test]
    fn glob_and_union_limits() {
        let inv = inventory();
        assert_eq!(
            derive_resources(Some("cluster_1node_*"), &inv),
            vec![
                "cluster_1node_a".to_string(),
                "cluster_1node_b".to_string(),
                "clusters".to_string()
            ]
        );
        // `:` union of two host names → both owning clusters.
        let res = derive_resources(Some("n1a:ue-content-01"), &inv);
        assert!(res.contains(&"cluster_1node_a".to_string()));
        assert!(res.contains(&"cluster_8node".to_string()));
    }

    #[test]
    fn exclusion_operator_is_conservative() {
        let inv = inventory();
        // `cluster_8node:!ue-content-03` still locks the whole cluster (the
        // exclusion is ignored → never under-locks).
        let res = derive_resources(Some("cluster_8node:!ue-content-03"), &inv);
        assert!(res.contains(&"cluster_8node".to_string()));
    }

    #[test]
    fn unknown_token_locks_nothing() {
        let inv = inventory();
        assert!(derive_resources(Some("not-a-real-host"), &inv).is_empty());
    }

    #[test]
    fn lease_filter_isolates_sibling_clusters() {
        let inv = inventory();
        let a = derive_resources(Some("cluster_1node_a"), &inv);
        let b = derive_resources(Some("cluster_1node_b"), &inv);

        // Without a lease filter the shared `clusters` parent makes siblings contend.
        let identity = LeaseFilter::default();
        assert!(!identity.apply(&a).is_disjoint(&identity.apply(&b)));

        // Naming the real cluster groups isolates them.
        let lease = LeaseFilter::new(vec![
            "cluster_1node_*".to_string(),
            "cluster_8node".to_string(),
        ]);
        assert!(lease.apply(&a).is_disjoint(&lease.apply(&b)));
    }

    #[test]
    fn admission_serializes_same_resource() {
        let lease = LeaseFilter::new(vec!["cluster_*".to_string()]);
        let runs = vec![
            run("r1", RunStatus::Queued, "2026-06-09T00:00:01Z", &["cluster_8node"]),
            run("r2", RunStatus::Queued, "2026-06-09T00:00:02Z", &["cluster_8node"]),
        ];
        // Only the oldest queued run on a contended resource is admitted.
        assert_eq!(select_admissions(&runs, &lease), vec![RunId::from("r1".to_string())]);
    }

    #[test]
    fn admission_parallelizes_distinct_resources() {
        let lease = LeaseFilter::new(vec!["cluster_*".to_string()]);
        let runs = vec![
            run("r1", RunStatus::Queued, "2026-06-09T00:00:01Z", &["cluster_1node_a"]),
            run("r2", RunStatus::Queued, "2026-06-09T00:00:02Z", &["cluster_1node_b"]),
        ];
        let admitted = select_admissions(&runs, &lease);
        assert_eq!(admitted.len(), 2);
    }

    #[test]
    fn admission_respects_running_lease_holder() {
        let lease = LeaseFilter::new(vec!["cluster_*".to_string()]);
        let runs = vec![
            run("r1", RunStatus::Running, "2026-06-09T00:00:01Z", &["cluster_8node"]),
            run("r2", RunStatus::Queued, "2026-06-09T00:00:02Z", &["cluster_8node"]),
            run("r3", RunStatus::Queued, "2026-06-09T00:00:03Z", &["cluster_1node_a"]),
        ];
        // r2 blocked by the running r1; r3 on a free resource is admitted.
        assert_eq!(select_admissions(&runs, &lease), vec![RunId::from("r3".to_string())]);
    }

    #[test]
    fn policy_denies_derived_live_cluster() {
        let inv = inventory();
        let policy = ResourcePolicy::new(vec!["cluster_1node_*".to_string()], vec!["cluster_8node".to_string()]);

        // A single live host slips past a raw-string check but its derived
        // resources include cluster_8node → denied.
        let live = derive_resources(Some("ue-content-03"), &inv);
        assert!(policy.evaluate(&live).is_err());

        // A dev single-node cluster is allowed. (Its parent `clusters` is not in
        // the allow set, so deny-only policy is the robust config here.)
        let dev = derive_resources(Some("cluster_1node_a"), &inv);
        let deny_only = ResourcePolicy::new(vec![], vec!["cluster_8node*".to_string()]);
        assert!(deny_only.evaluate(&dev).is_ok());
    }

    #[test]
    fn policy_dimension_makes_allow_a_clean_default_deny() {
        let inv = inventory();
        // significant=cluster_*  allow=cluster_1node_*  : the umbrella `clusters`
        // parent is filtered out before the allow check, so it need not appear in
        // the allow set. A pure allow-list with no deny is now sufficient.
        let policy = ResourcePolicy::new(vec!["cluster_1node_*".to_string()], vec![])
            .with_significant(vec!["cluster_*".to_string()]);

        // Dev single-node cluster → allowed, despite `clusters` being a derived
        // resource (it's not significant).
        let dev = derive_resources(Some("cluster_1node_a"), &inv);
        assert!(dev.contains(&"clusters".to_string()));
        assert!(policy.evaluate(&dev).is_ok());

        // Prod 8-node cluster (even via a single live host) → denied, no blocklist
        // needed: cluster_8node is significant but outside the allow set.
        let prod = derive_resources(Some("ue-content-03"), &inv);
        assert!(policy.evaluate(&prod).is_err());
    }

    #[test]
    fn policy_dimension_denies_target_with_no_significant_resource() {
        // A target that resolves only to ambient (non-significant) groups can't be
        // proven in scope under a configured policy dimension → default-deny.
        let policy = ResourcePolicy::new(vec!["cluster_1node_*".to_string()], vec![])
            .with_significant(vec!["cluster_*".to_string()]);
        // No cluster_* resource present.
        let ambient = vec!["ue_content".to_string(), "windows".to_string()];
        assert!(policy.evaluate(&ambient).is_err());

        // With an empty allow list there's nothing to default-deny against.
        let deny_only =
            ResourcePolicy::new(vec![], vec![]).with_significant(vec!["cluster_*".to_string()]);
        assert!(deny_only.evaluate(&ambient).is_ok());
    }

    #[test]
    fn deny_still_spans_all_resources_under_policy_dimension() {
        // deny is conservative: it matches across every derived resource, not just
        // the significant ones, so an ambient-group deny still rejects.
        let policy = ResourcePolicy::new(vec![], vec!["ue_content".to_string()])
            .with_significant(vec!["cluster_*".to_string()]);
        let resources = vec!["cluster_1node_a".to_string(), "ue_content".to_string()];
        assert!(policy.evaluate(&resources).is_err());
    }
}