hyperhive/hive-c0re/src/job_queue/model.rs

//! Data model for the generic job-DAG queue: node kinds (the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store.
//! The serialized *views* — `DagView` / `NodeView` plus the `Template`
//! / `Source` / `State` / `PermPayload` wire enums — live in
//! `hive_sh4re::jobs` (wire types belong to the shared crate) and are
//! re-exported here for the queue's internal use.
//!
//! Two levels: the **DAG** is the unit of dedup / cancel /
//! approval-resolution and the dashboard group; the **node** is the
//! unit of scheduling / execution / build-log / step label. See
//! `docs/coordinator.md::Job queue` for the full design.

pub use hive_sh4re::jobs::{DagView, NodeId, NodeView, PermPayload, Source, State, Template};
use serde::Serialize;

/// Dedup compares the perm *type*, not the value — a tool-groups
/// change and a capabilities change for the same agent are distinct
/// operations that must not collapse.
pub(super) fn perm_payload_same_type(a: Option<&PermPayload>, b: Option<&PermPayload>) -> bool {
    matches!(
        (a, b),
        (
            Some(PermPayload::ToolGroups { .. }),
            Some(PermPayload::ToolGroups { .. })
        ) | (
            Some(PermPayload::Capabilities { .. }),
            Some(PermPayload::Capabilities { .. })
        ) | (
            Some(PermPayload::Combined { .. }),
            Some(PermPayload::Combined { .. })
        ) | (None, None)
    )
}

/// When a dependency edge is considered satisfied.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DepWhen {
    /// Dep must reach `Done`. A `Failed` / `Cancelled` dep cancels this
    /// node (cancel-downstream).
    AfterOk,
    /// Dep must merely reach a terminal state (ok *or* fail). Used only
    /// by `rebuild`'s tail `Reconcile` so the recovery-start runs even
    /// when `Swap` failed.
    AfterAny,
}

/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
/// the per-agent lease + dedup, never from edges between DAGs).
#[derive(Debug, Clone, Copy, Serialize)]
pub struct Dep {
    pub on: NodeId,
    pub when: DepWhen,
}

/// The primitive operations — each kind maps to one executor fn in
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta.rs`
/// code. Concurrency is gated by two resource classes (see
/// [`NodeKind::needs_build_slot`] / [`NodeKind::needs_lease`]); the
/// meta *repo* is serialized by `meta::META_LOCK` inside the wrapped
/// functions themselves, which is why there is no `GitCommit` node —
/// a standalone commit node would open a dirty-working-tree window
/// between nodes that the fused `meta.rs` ops deliberately close.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum NodeKind {
    /// Out-of-band toplevel build while the container keeps serving:
    /// meta `sync_agents`, optional per-agent relock, then
    /// `lifecycle::prebuild_toplevel`. `relock = false` only for
    /// meta-update cascade rebuilds (re-locking would revert the bump
    /// the cascade just committed).
    Prebuild { relock: bool },
    /// `nixos-container update` profile-swap (requires the container
    /// stopped). Re-applies nspawn flags + resource limits first —
    /// rebuild is the reconcile verb — and carries the post-rebuild
    /// bookkeeping tail (rev marker, forge/matrix sync, kick, rescan).
    Swap,
    /// First-spawn `nixos-container create` plus the pre-create
    /// provisioning (proposed/applied repos, state subvolume, meta
    /// registration).
    Create,
    /// Meta flake lock bump. `sweep = false`: `meta::lock_update`
    /// (commit fused, under `META_LOCK`) with the DAG's `inputs`;
    /// `sweep = true`: `meta::lock_update_hyperhive`, *non-fatal* (a
    /// failed boot-time bump must not cancel the fan-out rebuilds).
    /// On success the scheduler appends child `Rebuild` DAGs: the
    /// precomputed `fanout` list when present (boot sweep), else the
    /// post-bump affected set (`meta_update_cascade_agents`).
    MetaLock {
        sweep: bool,
        fanout: Option<Vec<String>>,
    },
    /// Idempotent power converge: read `wanted` + observed state;
    /// start if `Up` & down (with cold-start fallback), stop if
    /// `Offline` & up, else noop.
    Reconcile,
    /// Mechanical `nixos-container stop` for the profile swap. Never
    /// touches `wanted`. Noop if already stopped.
    StopForUpdate,
    /// Set the graceful-stop fence + kick the harness so it runs one
    /// stop-checkpoint turn.
    Signal,
    /// Await the harness clearing the fence, bounded by
    /// `GRACEFUL_STOP_TIMEOUT`. Resolves ok either way — the
    /// downstream `Reconcile` performs the actual stop.
    Drain,
    /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
    WriteDropin,
    /// Commit `tool-groups.json` / `capabilities.json` per the DAG's
    /// `perm_payload` (commit fused under `META_LOCK`).
    WritePermFile,
    /// Opaque approval deploy pipeline (`ApplyCommit` /
    /// `MergeConfigPr`): the two-phase prepare/finalize/abort meta
    /// deploy stays inside `actions.rs` in v1 — deliberately not
    /// modeled as scheduler nodes (see the design doc §9).
    ApprovalDeploy,
}

impl NodeKind {
    /// Wire string for `NodeView.kind`.
    pub fn as_str(&self) -> &'static str {
        match self {
            NodeKind::Prebuild { .. } => "prebuild",
            NodeKind::Swap => "swap",
            NodeKind::Create => "create",
            NodeKind::MetaLock { .. } => "meta_lock",
            NodeKind::Reconcile => "reconcile",
            NodeKind::StopForUpdate => "stop_for_update",
            NodeKind::Signal => "signal",
            NodeKind::Drain => "drain",
            NodeKind::WriteDropin => "write_dropin",
            NodeKind::WritePermFile => "write_perm_file",
            NodeKind::ApprovalDeploy => "approval_deploy",
        }
    }

    /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits
    /// for the node's duration.
    pub fn needs_build_slot(&self) -> bool {
        matches!(
            self,
            NodeKind::Prebuild { .. }
                | NodeKind::Swap
                | NodeKind::Create
                | NodeKind::MetaLock { .. }
                | NodeKind::ApprovalDeploy
        )
    }

    /// Container-affecting kinds require the DAG to hold the agent's
    /// lifecycle lease (acquired at the first such node, held until the
    /// DAG is terminal). Lease-exempt kinds (`Prebuild`, `MetaLock`,
    /// `WritePermFile`) touch the store / meta repo, not the running
    /// container — which is exactly why a `Prebuild` can overlap
    /// another DAG's work on the same agent.
    pub fn needs_lease(&self) -> bool {
        matches!(
            self,
            NodeKind::Swap
                | NodeKind::Create
                | NodeKind::Reconcile
                | NodeKind::StopForUpdate
                | NodeKind::Signal
                | NodeKind::Drain
                | NodeKind::WriteDropin
                | NodeKind::ApprovalDeploy
        )
    }
}

/// One schedulable unit inside a DAG.
#[derive(Debug, Clone)]
pub struct Node {
    pub id: NodeId,
    pub kind: NodeKind,
    pub deps: Vec<Dep>,
    pub state: State,
    /// Live sub-label while `Running` (kept for parity with the old
    /// per-entry `step`).
    pub step: Option<String>,
    /// Row id of the `build_logs` entry this node opened (`Prebuild` /
    /// `Swap` / `ApprovalDeploy`), for the dashboard's live-stream link.
    pub build_log_id: Option<i64>,
    pub started_at: Option<i64>,
    pub finished_at: Option<i64>,
    /// Populated when `state == Failed` (truncated by the queue).
    pub error: Option<String>,
}

/// Submit-time spec for one node.
#[derive(Debug, Clone)]
pub struct NodeSpec {
    pub kind: NodeKind,
    pub deps: Vec<Dep>,
}

/// Submit-time spec for a whole DAG. Built by `templates.rs`; validated
/// (cycle rejection) and dedup'd by `JobQueue::submit`.
#[derive(Debug, Clone)]
pub struct DagSpec {
    pub template: Template,
    /// Primary target agent, or `"hyperhive"` for meta-level DAGs.
    pub agent: String,
    pub source: Source,
    /// Free-form "why"; dedup appends "also requested by …" lines.
    pub reason: String,
    /// Cascade grouping (meta-update / sweep children).
    pub parent_id: Option<u64>,
    /// Fires the approval-resolution hook on DAG terminal.
    pub approval_id: Option<i64>,
    /// `MetaUpdate`-only: the inputs to bump (also part of the dedup
    /// key for that template). Display copy lives on the DAG.
    pub inputs: Vec<String>,
    /// `PermChange`-only payload.
    pub perm_payload: Option<PermPayload>,
    /// Dashboard transient pill (and crash-watch suppression) held for
    /// the lease window — from lease acquisition to DAG terminal.
    pub transient: Option<crate::coordinator::TransientKind>,
    pub nodes: Vec<NodeSpec>,
}

/// A live DAG in the queue.
#[derive(Debug, Clone)]
pub struct Dag {
    pub id: u64,
    pub template: Template,
    pub agent: String,
    pub source: Source,
    pub reason: String,
    pub parent_id: Option<u64>,
    pub approval_id: Option<i64>,
    pub inputs: Vec<String>,
    pub perm_payload: Option<PermPayload>,
    pub transient: Option<crate::coordinator::TransientKind>,
    pub created_at: i64,
    pub nodes: Vec<Node>,
    /// Terminal roll-up already reported to the scheduler's hooks
    /// (approval resolution, transient release). Internal bookkeeping,
    /// never serialized.
    pub terminal_reported: bool,
}

impl Dag {
    /// Roll-up state: `Failed` if any node failed; else `Running` if
    /// any running; else `Queued` if any queued; else `Cancelled` if
    /// any cancelled; else `Done`.
    pub fn rollup(&self) -> State {
        let mut any_cancelled = false;
        let mut any_queued = false;
        let mut any_running = false;
        for n in &self.nodes {
            match n.state {
                State::Failed => return State::Failed,
                State::Running => any_running = true,
                State::Queued => any_queued = true,
                State::Cancelled => any_cancelled = true,
                State::Done => {}
            }
        }
        if any_running {
            State::Running
        } else if any_queued {
            State::Queued
        } else if any_cancelled {
            State::Cancelled
        } else {
            State::Done
        }
    }

    /// True when every node is terminal.
    pub fn is_terminal(&self) -> bool {
        self.nodes.iter().all(|n| n.state.is_terminal())
    }

    /// First failed node's error, for the roll-up `error` field.
    pub fn first_error(&self) -> Option<&str> {
        self.nodes
            .iter()
            .find(|n| n.state == State::Failed)
            .and_then(|n| n.error.as_deref())
    }

    pub fn node(&self, id: NodeId) -> Option<&Node> {
        self.nodes.iter().find(|n| n.id == id)
    }

    pub fn node_mut(&mut self, id: NodeId) -> Option<&mut Node> {
        self.nodes.iter_mut().find(|n| n.id == id)
    }
}

impl Dag {
    pub fn view(&self) -> DagView {
        let started_at = self.nodes.iter().filter_map(|n| n.started_at).min();
        let finished_at = if self.is_terminal() {
            self.nodes.iter().filter_map(|n| n.finished_at).max()
        } else {
            None
        };
        DagView {
            id: self.id,
            agent: self.agent.clone(),
            kind: self.template,
            state: self.rollup(),
            source: self.source,
            parent_id: self.parent_id,
            reason: self.reason.clone(),
            enqueued_at: self.created_at,
            started_at,
            finished_at,
            inputs: self.inputs.clone(),
            approval_id: self.approval_id,
            perm_payload: self.perm_payload.clone(),
            nodes: self
                .nodes
                .iter()
                .map(|n| NodeView {
                    id: n.id,
                    kind: n.kind.as_str().to_owned(),
                    deps: n.deps.iter().map(|d| d.on).collect(),
                    state: n.state,
                    step: n.step.clone(),
                    build_log_id: n.build_log_id,
                    started_at: n.started_at,
                    finished_at: n.finished_at,
                    error: n.error.clone(),
                })
                .collect(),
        }
    }
}