hyperhive/hive-host-sock/src/jobs.rs

//! Wire shapes of hive-c0re's job-DAG queue: what a queued job looks
//! like on the dashboard SSE channel (`rebuild_queue_changed`), the
//! `/api/state.rebuild_queue` snapshot, and the host admin socket's
//! `QueueDag` polling surface (`hivectl`'s wait/progress loop). The
//! queue *internals* — node kinds, dependency edges, scheduling state —
//! live in `hive-c0re::job_queue`; these are the serialized views it
//! produces. Semantics: `docs/coordinator.md::Job queue`.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Where the submit request originated — drives the "why" chip on the
/// dashboard.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Source {
    /// Operator action (dashboard button, CLI, manager tool).
    Manual,
    /// Meta-update cascade rebuild (grown into the meta-update DAG).
    MetaUpdate,
    /// Boot-time submission (the boot sweep DAG + boot reconciles).
    AutoUpdate,
    /// Crash recovery path (future use).
    CrashRecover,
    /// Operator approved a pending `Approval` row; `approval_id` on
    /// the DAG points back at the source row.
    Approval,
}

impl Source {
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Source::Manual => "manual",
            Source::MetaUpdate => "meta_update",
            Source::AutoUpdate => "auto_update",
            Source::CrashRecover => "crash_recover",
            Source::Approval => "approval",
        }
    }
}

pub use hive_jobq::State;

/// Kind-specific payload for `Template::PermChange` DAGs.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum PermPayload {
    /// Set the tool groups for one agent (`tool-groups.json`).
    ToolGroups { groups: Vec<String> },
    /// Set the capabilities for one agent (`capabilities.json`).
    Capabilities { caps: Vec<String> },
    /// Set both perm-types in one entry — the batch
    /// `POST /api/permissions` path. `None` leaves that file untouched;
    /// present fields commit together and rebuild once.
    Combined {
        groups: Option<Vec<String>>,
        caps: Option<Vec<String>>,
    },
}

/// Node id. Carries the scheduler crate's globally-monotonic node id
/// (`hive_jobq::NodeId`) verbatim on the wire — unique across all DAGs, not
/// just within one. Consumers treat it opaquely (grouping + dep matching),
/// so the widening from the old dag-local `u32` is transparent.
pub type NodeId = u64;

/// One node of a queued DAG, serialized near-raw from the scheduler
/// graph. Lifecycle (`state` / `started_at` / `finished_at` / `error`)
/// comes straight off the `hive_jobq::Node`. The client derives DAG-level
/// roll-ups (label, state, timestamps) from the node set — nothing is
/// rolled up host-side. Build logs are fetched on demand by node id
/// (`GET /api/build-log/<id>`), not carried inline.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeView {
    pub id: NodeId,
    /// The agent whose container (or meta repo, for `hyperhive` meta-level
    /// nodes) this node operates on. Agent is per-node — a single DAG can
    /// span multiple agents (e.g. a hive-wide restart), so there is no
    /// DAG-level agent field; consumers group by this.
    pub agent: String,
    /// Node primitive tag: `"prebuild"`, `"stop_for_update"`,
    /// `"swap"`, `"create"`, `"meta_lock"`, `"reconcile"`, `"signal"`,
    /// `"drain"`, `"write_dropin"`, `"write_perm_file"`,
    /// `"approval_deploy"`.
    pub kind: String,
    /// Ids of the nodes this one waits for. May reference an already-`Done`
    /// node that's been filtered out of the wire — the client treats a dep
    /// on an absent node as satisfied.
    #[serde(default)]
    pub deps: Vec<NodeId>,
    pub state: State,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started_at: Option<DateTime<Utc>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finished_at: Option<DateTime<Utc>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// Approval-queue row id — present only on the `approval_deploy` node.
    /// The client links a DAG to its pending approval through this (it is
    /// not derivable from the graph, so it rides the node that owns it).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approval_id: Option<i64>,
    /// Meta-flake inputs being bumped — present only on the `meta_lock`
    /// node. Display-only payload, not derivable from the graph.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub inputs: Vec<String>,
    /// The build-log history row id, when this node has a captured build
    /// log fetchable at `GET /api/build-log/<node id>` and deep-linkable to
    /// `/builds.html?id=<this>#buildlogs`. Only the nix-heavy nodes that
    /// stream build output set one; the client gates its log link on
    /// `.is_some()` so lock / noop / store-only nodes don't render a link
    /// that 404s.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build_log_id: Option<i64>,
    /// Structural parent in the jobq tree — `None` for top-level nodes
    /// (direct children of the DAG container). Sub-nodes carry the id of
    /// their containing parent node. The client uses this to render the
    /// recursive tree rather than inferring structure from `deps` alone.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent: Option<NodeId>,
}

/// A queued / running / failed DAG — a thin projection of one container
/// node plus its (non-`Done`) subtree from the scheduler graph. Only
/// non-derivable facts live here: `id`, `source`, `reason`, `created_at`,
/// and the node set. The client derives the card label, roll-up state, and
/// DAG timestamps from `nodes` (per-node `kind` + lifecycle) — nothing is
/// rolled up host-side. There is no DAG-level `agent`: a DAG can span
/// agents, so agent is per-[`NodeView`]; consumers group by `NodeView::agent`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DagView {
    pub id: u64,
    pub source: Source,
    pub reason: String,
    /// When the DAG was enqueued.
    pub created_at: DateTime<Utc>,
    /// When the DAG's first node started (min over *all* its nodes) — computed
    /// host-side, **not** derived on the client: `Done` nodes are excluded from
    /// `nodes` below, so the earliest-started node is usually absent from the
    /// wire and the client can't take the min itself. `None` until a node runs.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started_at: Option<DateTime<Utc>>,
    /// When the DAG finished (max `finished_at` over all its nodes), set only
    /// once the DAG has settled terminal. Host-computed for the same reason as
    /// `started_at`. `None` while the DAG is still live.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finished_at: Option<DateTime<Utc>>,
    /// Nodes of this DAG with `Done` ones excluded. `Skipped` nodes are
    /// carried so the dashboard can show which branches weren't taken, but
    /// they don't keep a DAG alive: one whose nodes are all `Done` or
    /// `Skipped` is omitted from the snapshot entirely, and its absence is
    /// what signals completion. A `Failed` DAG lingers until aged out by the
    /// history cap.
    pub nodes: Vec<NodeView>,
}

impl DagView {
    /// Roll-up state derived from the node set — the shared derivation every
    /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's
    /// JS render and the host agree: `Failed` if any node failed, else
    /// **`Cancelled` if any cancelled**, else `Running` if any running, else
    /// `Pending` if any pending, else `Done`. `Done` nodes are excluded
    /// from the wire, so a DAG that is *entirely* done isn't sent at all —
    /// its absence from the snapshot is what signals completion.
    ///
    /// `Cancelled` outranks both `Running` and `Pending` because a cancelled DAG
    /// still has its weak-edged tail node to run (it reports the cancellation),
    /// so `Pending`-then-`Running` would flicker back at the operator who just
    /// cancelled it and read as "the cancel didn't take". Outside that window
    /// the states barely co-occur: a cancel *cascade* originates at a `Failed`
    /// node, which returns early above.
    ///
    /// `Finishing` counts as running: the node's own work is done but its
    /// sub-nodes are still going, so the DAG is still in flight.
    ///
    /// `Skipped` contributes nothing: a not-taken branch is an expected part of
    /// a healthy run, so counting it would make every successful DAG roll up
    /// non-`Done`.
    ///
    /// This ordering matches `frontend/packages/dashboard/src/builds.js`'s
    /// `rollupState`. The two implementations must be edited together — they
    /// have silently disagreed before.
    #[must_use]
    pub fn rollup_state(&self) -> State {
        let mut any_running = false;
        let mut any_pending = false;
        let mut any_cancelled = false;
        for n in &self.nodes {
            match n.state {
                State::Failed => return State::Failed,
                State::Running | State::Finishing => any_running = true,
                State::Pending => any_pending = true,
                State::Cancelled => any_cancelled = true,
                State::Done | State::Skipped => {}
            }
        }
        if any_cancelled {
            State::Cancelled
        } else if any_running {
            State::Running
        } else if any_pending {
            State::Pending
        } else {
            State::Done
        }
    }
}

#[cfg(test)]
mod tests {
    use chrono::Utc;

    use super::{DagView, NodeView, Source, State};

    /// A node set carrying nothing but the states — the only input
    /// `rollup_state` reads.
    fn dag(states: &[State]) -> DagView {
        DagView {
            id: 1,
            source: Source::Manual,
            reason: "test".to_owned(),
            created_at: Utc::now(),
            started_at: None,
            finished_at: None,
            nodes: states
                .iter()
                .enumerate()
                .map(|(i, &state)| NodeView {
                    id: i as u64,
                    agent: "a".to_owned(),
                    kind: "reconcile".to_owned(),
                    deps: Vec::new(),
                    state,
                    started_at: None,
                    finished_at: None,
                    error: None,
                    approval_id: None,
                    inputs: Vec::new(),
                    build_log_id: None,
                    parent: None,
                })
                .collect(),
        }
    }

    #[test]
    fn a_failure_outranks_everything_and_skipped_counts_for_nothing() {
        // The case this replaces used to be arranged in hive-c0re by running a
        // rebuild until its Prebuild failed. Only the states ever mattered.
        assert_eq!(
            dag(&[State::Done, State::Failed, State::Skipped]).rollup_state(),
            State::Failed
        );
        // A failure wins even against a node still going — the DAG's verdict
        // is already decided.
        assert_eq!(
            dag(&[State::Running, State::Failed]).rollup_state(),
            State::Failed
        );
        // Skipped is an expected part of a healthy run: an outcome-branched
        // DAG always leaves one branch untaken, so counting it would make
        // every successful DAG roll up non-Done.
        assert_eq!(
            dag(&[State::Done, State::Skipped]).rollup_state(),
            State::Done
        );
    }

    #[test]
    fn cancelled_outranks_running_and_pending() {
        // A cancelled DAG still has its weak-edged tail node to run, so
        // Pending-then-Running would flicker back at the operator who just
        // cancelled it and read as "the cancel didn't take".
        assert_eq!(
            dag(&[State::Cancelled, State::Pending]).rollup_state(),
            State::Cancelled
        );
        assert_eq!(
            dag(&[State::Cancelled, State::Running]).rollup_state(),
            State::Cancelled
        );
    }

    #[test]
    fn finishing_still_counts_as_running() {
        // The node's own work is done but its sub-nodes are still going, so
        // the DAG is in flight. A parent parked in Finishing is the normal
        // shape of a subtree mid-run, not an edge case.
        assert_eq!(
            dag(&[State::Finishing, State::Pending]).rollup_state(),
            State::Running
        );
        assert_eq!(
            dag(&[State::Running, State::Pending]).rollup_state(),
            State::Running
        );
        assert_eq!(
            dag(&[State::Done, State::Pending]).rollup_state(),
            State::Pending
        );
    }

    #[test]
    fn an_empty_node_set_reads_done() {
        // Every node Done means every node is filtered off the wire, so this
        // is what a finished DAG actually looks like to a consumer that has
        // one in hand at all.
        assert_eq!(dag(&[]).rollup_state(), State::Done);
    }
}