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

//! Generic job-DAG queue + desired-state reconciliation — the host-side
//! wrapper over the domain-agnostic [`hive_jobq`] scheduler. Jobs are nodes in
//! per-request DAGs (see [`templates`]); the special cases (graceful-stop
//! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG
//! *shapes* over the shared node primitives ([`model::NodeKind`]).
//!
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up
//! settle loop; this module maps hive-c0re's concepts onto it:
//! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant
//!   carries the agent it targets ([`NodeKind::agent`]); the two resource
//!   classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
//!   subtree-held), derived per node by [`NodeKind::resource_deps`];
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
//!   carrying the group's metadata, with the template's nodes hung under it as
//!   its subtree (the **parent axis** groups; `deps` order). So the container's
//!   `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership
//!   is a graph walk — there are no host grouping side-tables. The lease is owned
//!   by a subtree root and borrowed by its descendants (continuity);
//! - per-DAG terminal work runs **inline** ([`exec::run_terminal_hook`]) when the
//!   container rolls up terminal — dispatched off its template
//!   ([`terminal_hook`]): approval-resolve, `Rebuilt`-emit, or power-intent
//!   revert. No terminal-hook node, no drained event stream.
//!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task
//! ([`scheduler::run_worker`]) drives it; concurrency comes from the build-slot
//! capacity, not multiple workers. Design: `docs/coordinator.md::Job queue`.

pub mod exec;
pub mod model;
pub mod resource;
pub mod scheduler;
pub mod submit;
pub mod templates;
#[cfg(test)]
mod tests;

use std::collections::HashMap;
use std::sync::Mutex;

use chrono::{DateTime, Utc};
use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState};
use hive_sh4re::jobs::NodeView;
use hive_sh4re::wire_time::now_unix;
use tokio::sync::Notify;

use crate::coordinator::TransientKind;
pub use model::{
    DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template,
};
use resource::Resource;

/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain per
/// template in the snapshot, matching the old per-kind history cap.
const MAX_HISTORY_PER_TEMPLATE: usize = 5;

/// Terminal DAGs younger than this are exempt from the per-template history
/// cap, so a burst of same-template DAGs that settle within one `QueueDag`
/// poll interval isn't evicted before the poller observes their terminal state.
const HISTORY_GRACE_SECS: i64 = 300;

/// Cap on stored node error strings.
const MAX_ERROR_LEN: usize = 2_000;

/// A node claimed for execution — everything the executor needs, snapshotted at
/// claim time.
#[derive(Debug, Clone)]
pub struct Claim {
    pub dag_id: u64,
    pub node_id: NodeId,
    pub kind: NodeKind,
    /// The agent this node targets (its own, not a DAG-level field). Empty for
    /// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
    pub agent: String,
    pub template: Template,
    pub approval_id: Option<i64>,
    pub inputs: Vec<String>,
    /// Transient pill kind for the lease window (from the spec). Whether the
    /// pill is currently shown is derived from live lease ownership
    /// ([`JobQueue::held_transients`]), not a per-claim edge.
    pub transient: Option<TransientKind>,
}

/// Summary of a DAG's terminal roll-up — the input to the terminal node's
/// executor (approval resolution, `Rebuilt` emission, cancelled-power-op intent
/// revert). Computed on demand from live graph state, not drained.
#[derive(Debug, Clone)]
pub struct TerminalDag {
    pub template: Template,
    /// Distinct agents this DAG's nodes targeted (one for a single-agent DAG).
    pub agents: Vec<String>,
    pub approval_id: Option<i64>,
    pub state: State,
    /// First failed node's error when `state == Failed`.
    pub error: Option<String>,
}

/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node`
/// itself now, so only the build-log row link remains host-side (the
/// client fetches the log by node id).
#[derive(Debug, Default, Clone)]
struct NodeRuntime {
    build_log_id: Option<i64>,
}

/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
/// Derived on read from the container node — the data has a single home (the
/// node payload); this is not a stored side-table.
struct DagMeta {
    template: Template,
    source: Source,
    reason: String,
    transient: Option<TransientKind>,
    approval_id: Option<i64>,
    inputs: Vec<String>,
    created_at: i64,
}

/// The mutable queue state behind the mutex: the crate scheduler plus the
/// per-node runtime metadata the graph can't carry. A **DAG is a single
/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
/// is the DAG state, and there are no grouping side-tables: membership + meta
/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] /
/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
struct QueueInner {
    sched: Scheduler<NodeKind, Resource>,
    /// Per-node runtime metadata (the build-log id) — mutable after
    /// insert, so it can't ride the immutable node payload.
    node_rt: HashMap<NodeId, NodeRuntime>,
}

/// The queue. Lives on `Coordinator` (one per hive-c0re process); a single
/// scheduler task ([`scheduler::run_worker`]) drives it.
pub struct JobQueue {
    inner: Mutex<QueueInner>,
    /// Wakes the scheduler when something new arrives or state changed.
    pub(crate) notify: Notify,
}

impl std::fmt::Debug for JobQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JobQueue").finish_non_exhaustive()
    }
}

impl Default for JobQueue {
    fn default() -> Self {
        Self::new(1)
    }
}

/// Map a spec dependency edge kind onto the crate's.
fn to_crate_when(when: DepWhen) -> JobDepWhen {
    match when {
        DepWhen::AfterOk => JobDepWhen::AfterOk,
        DepWhen::AfterAny => JobDepWhen::AfterAny,
    }
}

/// The inline terminal-hook a settled DAG fires — dispatched off its container's
/// template + approval id when the container rolls up terminal (no hook node).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HookKind {
    /// Approval-driven DAG (spawn / opaque deploy): resolve the approval row.
    ResolveApproval,
    /// Rebuild / perm-change: emit one `Rebuilt` manager event per agent.
    EmitRebuilt,
}

/// The terminal hook a DAG needs, from its template + approval id — or `None`
/// for a DAG with no terminal side effect (power-op, meta-update, boot, bare
/// reconcile).
///
/// A cancelled DAG deliberately gets **no** compensating hook. [`JobQueue::cancel`]
/// refuses unless every work node is still `Pending`, and a cancel *cascade*
/// rolls up `Failed` (see `dag_rollup`), never `Cancelled` — so on a
/// `Cancelled` DAG no node ever executed and there is nothing to undo. A power
/// op's `SetWanted` head provably never ran, so its intent is still whatever
/// the operator last set it to.
#[must_use]
pub fn terminal_hook(template: Template, approval_id: Option<i64>) -> Option<HookKind> {
    if approval_id.is_some() {
        return Some(HookKind::ResolveApproval);
    }
    match template {
        Template::Rebuild | Template::PermChange => Some(HookKind::EmitRebuilt),
        _ => None,
    }
}

/// Map a crate node state onto the wire state (`Pending` ↔ `Queued`;
/// `Finishing` — own logic done, sub-nodes still running — reads as `Running`).
fn to_wire_state(state: JobState) -> State {
    match state {
        JobState::Pending => State::Queued,
        JobState::Running | JobState::Finishing => State::Running,
        JobState::Done => State::Done,
        JobState::Failed => State::Failed,
        JobState::Cancelled => State::Cancelled,
    }
}

/// Insert `nodes` into the shared graph, honouring the spec's explicit **parent
/// axis**: a node with `parent = None` is a top-level group root (re-parented to
/// `group_parent`, which is `None` for `submit` and the emitting node for
/// `append_subgraph`); a node with `parent = Some(idx)` becomes a child of the
/// already-inserted node at spec index `idx`. `deps` are translated to crate
/// `Dep::Node` edges verbatim — templates declare the parent axis + sibling
/// ordering directly, so there is no dep-on-root to drop and no lease to hoist:
/// each node declares its own `Dep::Resource`, and the crate's borrow model
/// keeps a resource continuous across a subtree (a root owns it, descendants
/// borrow it). Independent group roots (multiple `parent = None` nodes) carry no
/// cross-links, so a multi-agent DAG's per-agent subgraphs run concurrently, each
/// on its own lease. Records per-node `node_rt`. Returns the inserted ids
/// (index-aligned with `nodes`). A node with `parent = None` is re-parented to
/// `group_parent` (the DAG container for a template, or the emitting node for a
/// runtime-appended subgraph); a node's `parent` / dep targets must precede it
/// in `nodes` (submit-time `validate` enforces density + acyclicity).
///
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
    inner: &mut QueueInner,
    nodes: &[NodeSpec],
    group_parent: Option<NodeId>,
) -> anyhow::Result<Vec<NodeId>> {
    let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len());
    for ns in nodes {
        let payload = ns.kind.clone();
        let mut deps = payload.resource_deps();
        for d in &ns.deps {
            deps.push(Dep::Node {
                id: ids[dep_index(d.on)],
                when: to_crate_when(d.when),
            });
        }
        let parent = match ns.parent {
            Some(idx) => Some(ids[dep_index(idx)]),
            None => group_parent,
        };
        let id = inner
            .sched
            .append(payload, deps, parent)
            .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
        ids.push(id);
        inner.node_rt.insert(id, NodeRuntime::default());
    }
    Ok(ids)
}

impl JobQueue {
    #[must_use]
    pub fn new(build_slots: usize) -> Self {
        let mut table = ResourceTable::new();
        table.set_capacity(
            Resource::BuildSlot,
            u32::try_from(build_slots.max(1)).unwrap_or(u32::MAX),
        );
        Self {
            inner: Mutex::new(QueueInner {
                sched: Scheduler::new(Graph::new(), table),
                node_rt: HashMap::new(),
            }),
            notify: Notify::new(),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, QueueInner> {
        self.inner.lock().expect("job_queue mutex poisoned")
    }

    /// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container
    /// node** carrying the group's metadata, then inserts the template's nodes as
    /// its subtree (their roots re-parented to the container). Returns the
    /// container's id as the DAG id — its rolled-up state is the DAG state and it
    /// reaching terminal fires the DAG's inline hook.
    ///
    /// # Errors
    /// Propagates the spec-validation error (empty / cyclic / bad parent) or a
    /// graph-insert error (dependencies that aren't dependency-topological).
    pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> {
        templates::validate(&spec)?;
        let mut inner = self.lock();
        let container = inner
            .sched
            .append(
                NodeKind::Dag {
                    template: spec.template,
                    source: spec.source,
                    reason: spec.reason,
                    transient: spec.transient,
                    approval_id: spec.approval_id,
                    inputs: spec.inputs,
                    created_at: now_unix(),
                },
                Vec::new(),
                None,
            )
            .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
        inner.node_rt.insert(container, NodeRuntime::default());
        insert_group(&mut inner, &spec.nodes, Some(container))?;
        // Settle the container's own (no-op) logic immediately so it parks in
        // `Finishing` and its children become runnable — it never needs claiming
        // or executing, and stays out of `claim_ready`. It rolls up terminal when
        // its whole subtree settles (that's the DAG-done signal).
        inner.sched.complete(container, Outcome::Done);
        drop(inner);
        self.notify.notify_one();
        Ok(container.get())
    }

    /// Append a whole *subgraph* into a live DAG at runtime — the single
    /// in-DAG-growth primitive. The subgraph is inserted as a [`insert_group`]
    /// rooted under `dep_on` (the emitting node): the subgraph's own root becomes
    /// a *child* of `dep_on`, its steps children of that root, and the group's
    /// agent lease is hoisted onto that root. Ordering root→`dep_on` is the parent
    /// gate — the children run once `dep_on` reaches `Finishing`. Because the
    /// emitting node stays `Finishing` until this appended subtree is terminal and
    /// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
    /// settling early with no explicit wiring. Returns the new node ids; empty if
    /// the DAG is gone or `nodes` is empty.
    pub fn append_subgraph(&self, dag_id: u64, nodes: &[NodeSpec], dep_on: NodeId) -> Vec<NodeId> {
        if nodes.is_empty() {
            return Vec::new();
        }
        let mut inner = self.lock();
        if inner.container(dag_id).is_none() {
            return Vec::new();
        }
        // Insert the subgraph as a group rooted under the emitting node: the
        // subgraph's own root becomes a child of `dep_on`, its steps children of
        // that root. No terminal-node wiring — roll-up carries terminality: the
        // emitter stays `Finishing` until this appended subtree settles, and the
        // container node rolls up terminal only once its whole subtree (incl. this
        // appended work) has settled, so the DAG hook waits for free.
        let ids = match insert_group(&mut inner, nodes, Some(dep_on)) {
            Ok(ids) => ids,
            Err(e) => {
                tracing::error!(
                    dag = dag_id,
                    error = %e,
                    "job_queue: append_subgraph insert failed"
                );
                return Vec::new();
            }
        };
        drop(inner);
        self.notify.notify_one();
        ids
    }

    /// Claim every currently-runnable node, acquiring its resources, and mark it
    /// `Running`. Delegates readiness + resource acquisition to the crate's
    /// settle loop; builds a [`Claim`] per started node from its payload + its
    /// DAG container's metadata. The container node itself is claimed like any
    /// other (its executor is an instant no-op that lets its subtree start).
    pub fn claim_ready(&self) -> Vec<Claim> {
        let mut inner = self.lock();
        let inner = &mut *inner;
        let started = inner.sched.settle();
        let mut claims = Vec::with_capacity(started.len());
        for id in started {
            let Some(node) = inner.sched.graph().node(id) else {
                continue;
            };
            let kind = node.payload.clone();
            let agent = node.payload.agent().to_owned();
            let Some(container) = inner.dag_of(id) else {
                continue;
            };
            let Some(meta) = inner.dag_meta(container) else {
                continue;
            };
            claims.push(Claim {
                dag_id: container.get(),
                node_id: id,
                kind,
                agent,
                template: meta.template,
                approval_id: meta.approval_id,
                inputs: meta.inputs,
                transient: meta.transient,
            });
            // `started_at` is stamped on the graph `Node` by the scheduler's
            // transition to `Running` — no host-side copy needed.
        }
        claims
    }

    /// Mark a claimed node terminal, recording its outcome + (truncated) error.
    /// The crate releases the node's build slot immediately and cascades the
    /// `AfterOk` failure cancellation + subtree lease release. Returns the DAG's
    /// terminal summary **iff** this completion rolled its container terminal —
    /// the scheduler runs the DAG's inline hook off it.
    pub fn complete_node(
        &self,
        _dag_id: u64,
        node_id: NodeId,
        result: Result<(), String>,
    ) -> Option<TerminalDag> {
        let mut inner = self.lock();
        // The failure reason + `finished_at` are stamped onto the graph `Node`
        // by the scheduler (the reason rides `Outcome::Failed`); no host-side
        // copy, so there is nothing to clear here.
        let outcome = match result {
            Ok(()) => Outcome::Done,
            Err(e) => Outcome::Failed(truncate_error(&e)),
        };
        let container = inner.dag_of(node_id);
        inner.sched.complete(node_id, outcome);
        // If this completion rolled the DAG's container up to a terminal state,
        // hand its summary back so the scheduler fires the inline hook once.
        let terminal = container
            .filter(|&c| c != node_id && inner.dag_is_terminal(c))
            .and_then(|c| inner.terminal_dag(c));
        drop(inner);
        self.notify.notify_one();
        terminal
    }

    /// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
    /// so each is cancelled. `None` once any work node is running or terminal —
    /// an in-flight nix build isn't interruptible. Otherwise the container is
    /// rolled up so the DAG settles (wire state `Cancelled`) and its terminal
    /// summary is returned — the caller fires the inline hook (power-intent
    /// revert / approval resolution) off it.
    pub fn cancel(&self, dag_id: u64) -> Option<TerminalDag> {
        let mut inner = self.lock();
        let container = inner.container(dag_id)?;
        let work = inner.subtree(container);
        let all_pending = work.iter().all(|&id| {
            inner
                .sched
                .graph()
                .node(id)
                .is_some_and(|n| n.state == JobState::Pending)
        });
        if !all_pending {
            return None;
        }
        for id in work {
            inner.sched.cancel_node(id);
        }
        // The container was settled to `Finishing` at submit; completing it again
        // now re-runs the roll-up with its children all `Cancelled`, driving it to
        // a terminal state synchronously within this lock — so the caller reads
        // the terminal summary immediately instead of waiting for the scheduler
        // loop to observe the cancellation. `dag_rollup` reports `Cancelled` to
        // the wire (a container whose children all cancelled).
        inner.sched.complete(container, Outcome::Done);
        let terminal = inner.terminal_dag(container);
        drop(inner);
        self.notify.notify_one();
        terminal
    }

    /// Link a `build_logs` row to a specific `Running` node.
    pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
        let mut inner = self.lock();
        if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
            return false;
        }
        inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
        true
    }

    /// The `build_logs` row id linked to the wire node id `node_id`, if any —
    /// the lookup behind the `GET /api/build-log/<node_id>` query endpoint (the
    /// client fetches a node's captured build output on demand rather than
    /// receiving it inline). Takes the raw wire `u64` (the endpoint's path
    /// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the
    /// matching id — the map is small (live + recently-terminal nodes).
    #[must_use]
    pub fn build_log_id_of(&self, node_id: u64) -> Option<i64> {
        self.lock()
            .node_rt
            .iter()
            .find(|(nid, _)| nid.get() == node_id)
            .and_then(|(_, rt)| rt.build_log_id)
    }

    /// The first failed node's error in `dag_id`, if any has failed yet.
    ///
    /// Unlike the roll-up summary this is readable *mid-flight*, which is the
    /// point: a compensation node runs `AfterAny` its subject, so when it asks,
    /// the DAG is still `Finishing` (the compensation node itself is running)
    /// while the node it is compensating for has already settled `Failed`. That
    /// lets the compensation annotate its bookkeeping with the reason the deploy
    /// failed, instead of having the error handed down from the node that hit
    /// it. `None` when nothing has failed — the ordinary success path.
    #[must_use]
    pub fn first_error(&self, dag_id: u64) -> Option<String> {
        let inner = self.lock();
        let container = inner.container(dag_id)?;
        inner.dag_first_error(container)
    }

    /// A DAG's terminal roll-up summary, computed on demand from its container.
    /// `None` if the DAG id is unknown. Test-only — production reads the summary
    /// `complete_node` returns when the container rolls up terminal.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option<TerminalDag> {
        let inner = self.lock();
        let container = inner.container(dag_id)?;
        inner.terminal_dag(container)
    }

    /// The `(dag_id, agent, kind)` triples for every per-agent lease currently
    /// held by a DAG that carries a transient pill — the live transient-pill
    /// set, a pull query over crate resource ownership (replaces the old
    /// lease-release event stream). A DAG with no transient kind is omitted.
    #[must_use]
    pub fn held_transients(&self) -> Vec<(u64, String, TransientKind)> {
        let inner = self.lock();
        inner
            .sched
            .resource_state()
            .into_iter()
            .filter_map(|(res, holder)| {
                let Resource::Agent(agent) = res else {
                    return None;
                };
                let container = inner.dag_of(holder)?;
                let kind = inner.dag_meta(container)?.transient?;
                Some((container.get(), agent, kind))
            })
            .collect()
    }

    /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
    #[must_use]
    pub fn snapshot(&self) -> Vec<DagView> {
        self.snapshot_capped(now_unix() - HISTORY_GRACE_SECS)
    }

    /// Snapshot the visible DAG set (live + newest-per-template terminal, terminal
    /// ones after `grace_cutoff` always kept), sorted by container id.
    fn snapshot_capped(&self, grace_cutoff: i64) -> Vec<DagView> {
        let inner = self.lock();
        let mut ids = inner.visible_dags(grace_cutoff);
        ids.sort_unstable_by_key(|c| c.get());
        ids.into_iter().filter_map(|c| inner.dag_view(c)).collect()
    }

    /// Number of live (non-terminal) DAGs — tests + diagnostics.
    #[cfg(test)]
    #[must_use]
    pub fn live_count(&self) -> usize {
        let inner = self.lock();
        inner
            .containers()
            .into_iter()
            .filter(|&c| !inner.dag_is_terminal(c))
            .count()
    }

    /// Test hook: snapshot with the history grace window disabled, so the
    /// per-template cap applies to just-finished terminal DAGs too.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn snapshot_no_grace(&self) -> Vec<DagView> {
        self.snapshot_capped(i64::MAX)
    }
}

impl QueueInner {
    /// Whether `id` is a `Running` node.
    fn node_running(&self, id: NodeId) -> bool {
        self.sched
            .graph()
            .node(id)
            .is_some_and(|n| n.state == JobState::Running)
    }

    /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
    /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search.
    fn container(&self, dag_id: u64) -> Option<NodeId> {
        self.sched.graph().nodes().find_map(|n| {
            (n.parent.is_none()
                && n.id.get() == dag_id
                && matches!(n.payload, NodeKind::Dag { .. }))
            .then_some(n.id)
        })
    }

    /// The DAG container a node belongs to — walk its parent chain to the root
    /// (`parent == None`), which is the container. Returns `id` itself for a
    /// container node.
    fn dag_of(&self, id: NodeId) -> Option<NodeId> {
        let mut cur = id;
        loop {
            match self.sched.graph().node(cur)?.parent {
                Some(p) => cur = p,
                None => return Some(cur),
            }
        }
    }

    /// The DAG's work nodes — its `container`'s subtree, excluding the container.
    fn subtree(&self, container: NodeId) -> Vec<NodeId> {
        self.sched
            .graph()
            .nodes()
            .filter(|n| n.id != container && self.dag_of(n.id) == Some(container))
            .map(|n| n.id)
            .collect()
    }

    /// The container's carried domain metadata as an owned read-view. The data
    /// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
    /// not a stored side-table.
    fn dag_meta(&self, container: NodeId) -> Option<DagMeta> {
        let NodeKind::Dag {
            template,
            source,
            reason,
            transient,
            approval_id,
            inputs,
            created_at,
        } = &self.sched.graph().node(container)?.payload
        else {
            return None;
        };
        Some(DagMeta {
            template: *template,
            source: *source,
            reason: reason.clone(),
            transient: *transient,
            approval_id: *approval_id,
            inputs: inputs.clone(),
            created_at: *created_at,
        })
    }

    /// Roll-up state over a DAG's work nodes: `Failed` if any failed; else
    /// `Running` if any running; else `Queued` if any queued; else `Cancelled`
    /// if any cancelled; else `Done`. (Kept eager over the subtree — a failed
    /// child shows `Failed` immediately, before the container finishes rolling
    /// up — matching the pre-container behaviour.)
    fn dag_rollup(&self, container: NodeId) -> State {
        let mut any_running = false;
        let mut any_queued = false;
        let mut any_cancelled = false;
        for id in self.subtree(container) {
            match self.sched.graph().node(id).map(|n| n.state) {
                Some(JobState::Failed) => return State::Failed,
                Some(JobState::Running | JobState::Finishing) => any_running = true,
                Some(JobState::Pending) => any_queued = true,
                Some(JobState::Cancelled) => any_cancelled = true,
                Some(JobState::Done) | None => {}
            }
        }
        if any_running {
            State::Running
        } else if any_queued {
            State::Queued
        } else if any_cancelled {
            State::Cancelled
        } else {
            State::Done
        }
    }

    /// True when the DAG has settled — its container has rolled up terminal
    /// (equivalent to every work node being terminal).
    fn dag_is_terminal(&self, container: NodeId) -> bool {
        self.sched
            .graph()
            .node(container)
            .is_some_and(|n| n.state.is_terminal())
    }

    /// Distinct agents a DAG's work nodes target, in first-seen order.
    fn dag_agents(&self, container: NodeId) -> Vec<String> {
        let mut seen: Vec<String> = Vec::new();
        for id in self.subtree(container) {
            if let Some(n) = self.sched.graph().node(id) {
                let agent = n.payload.agent();
                if !agent.is_empty() && !seen.iter().any(|s| s == agent) {
                    seen.push(agent.to_owned());
                }
            }
        }
        seen
    }

    /// First failed work node's error (read off the graph `Node`), for the
    /// terminal roll-up summary the inline hook consumes.
    fn dag_first_error(&self, container: NodeId) -> Option<String> {
        for id in self.subtree(container) {
            if let Some(n) = self.sched.graph().node(id)
                && n.state == JobState::Failed
                && let Some(e) = n.error.clone()
            {
                return Some(e);
            }
        }
        None
    }

    /// A DAG's terminal roll-up summary — the input to its inline hook.
    fn terminal_dag(&self, container: NodeId) -> Option<TerminalDag> {
        let meta = self.dag_meta(container)?;
        Some(TerminalDag {
            template: meta.template,
            agents: self.dag_agents(container),
            approval_id: meta.approval_id,
            state: self.dag_rollup(container),
            error: self.dag_first_error(container),
        })
    }

    /// Project a DAG into its wire [`DagView`]: a near-raw view of the
    /// container's work nodes, with `Done` nodes excluded. Lifecycle
    /// (`state` / `started_at` / `finished_at` / `error`) is read straight
    /// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
    /// state, and DAG timestamps from the node set. Non-derivable per-node
    /// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
    /// `None` when every work node is `Done` — a fully-completed DAG drops
    /// out of the snapshot entirely (a `Failed` one lingers until aged out).
    fn dag_view(&self, container: NodeId) -> Option<DagView> {
        let meta = self.dag_meta(container)?;
        let mut nodes = Vec::new();
        // DAG-level timestamps are taken over *all* subtree nodes (including the
        // `Done` ones excluded from the wire) — the client can't derive them
        // from a `Done`-filtered node set, so the host computes them here.
        let mut started: Vec<DateTime<Utc>> = Vec::new();
        let mut finished: Vec<DateTime<Utc>> = Vec::new();
        for id in self.subtree(container) {
            let Some(node) = self.sched.graph().node(id) else {
                continue;
            };
            if let Some(s) = node.started_at {
                started.push(s);
            }
            if let Some(f) = node.finished_at {
                finished.push(f);
            }
            if node.state == JobState::Done {
                continue;
            }
            let deps: Vec<u64> = node
                .deps
                .iter()
                .filter_map(|d| match d {
                    Dep::Node { id, .. } => Some(id.get()),
                    Dep::Resource { .. } => None,
                })
                .collect();
            // Non-derivable per-node payload rides the node that owns it. For a
            // deploy that's the subtree root: the phases below it are ordinary
            // nodes, and hanging the approval link off all four would render the
            // same card four times.
            let approval_id = matches!(node.payload, NodeKind::DeployWindow { .. })
                .then_some(meta.approval_id)
                .flatten();
            let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) {
                meta.inputs.clone()
            } else {
                Vec::new()
            };
            let has_log = self.node_rt.get(&id).and_then(|r| r.build_log_id).is_some();
            // `node.parent` is the structural jobq parent. Top-level nodes
            // have `parent == Some(container)` (direct children of the Dag
            // container); those become `parent: None` on the wire since the
            // container itself is not part of the work-node payload. Sub-nodes
            // carry the id of their containing parent work-node.
            let parent = node
                .parent
                .filter(|&p| p != container)
                .map(hive_jobq::NodeId::get);
            nodes.push(NodeView {
                id: id.get(),
                agent: node.payload.agent().to_owned(),
                kind: node.payload.as_str().to_owned(),
                deps,
                state: to_wire_state(node.state),
                started_at: node.started_at,
                finished_at: node.finished_at,
                error: node.error.clone(),
                approval_id,
                inputs,
                has_log,
                parent,
            });
        }
        if nodes.is_empty() {
            return None;
        }
        let is_terminal = self.dag_is_terminal(container);
        Some(DagView {
            id: container.get(),
            source: meta.source,
            reason: meta.reason.clone(),
            created_at: hive_sh4re::wire_time::from_secs(meta.created_at),
            started_at: started.into_iter().min(),
            finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
            nodes,
        })
    }

    /// When a DAG's work node finishes on `finished_at` — the max over its
    /// subtree (read off the graph `Node`, as unix seconds), for the history
    /// cap ordering.
    fn dag_finished_at(&self, container: NodeId) -> i64 {
        self.subtree(container)
            .iter()
            .filter_map(|id| self.sched.graph().node(*id))
            .filter_map(|n| n.finished_at)
            .map(|t| t.timestamp())
            .max()
            .unwrap_or(0)
    }

    /// Every DAG container node id in the graph.
    fn containers(&self) -> Vec<NodeId> {
        self.sched
            .graph()
            .nodes()
            .filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
            .map(|n| n.id)
            .collect()
    }

    /// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
    /// plus the newest [`MAX_HISTORY_PER_TEMPLATE`] terminal DAGs per template
    /// (terminal DAGs finished after `grace_cutoff` are always kept). Crate nodes
    /// for evicted DAGs linger in the graph (bounded-prune is a Stage-C
    /// follow-up); this filter is what bounds what the dashboard sees.
    fn visible_dags(&self, grace_cutoff: i64) -> Vec<NodeId> {
        let mut live: Vec<NodeId> = Vec::new();
        let mut terminal: Vec<(NodeId, Template, i64)> = Vec::new();
        for c in self.containers() {
            if self.dag_is_terminal(c) {
                if let Some(meta) = self.dag_meta(c) {
                    terminal.push((c, meta.template, self.dag_finished_at(c)));
                }
            } else {
                live.push(c);
            }
        }
        // Newest first so the per-template cap keeps the most recent.
        terminal.sort_by(|a, b| b.2.cmp(&a.2).then(b.0.get().cmp(&a.0.get())));
        let mut counts: HashMap<Template, usize> = HashMap::new();
        let mut kept = live;
        for (c, template, finished) in terminal {
            let n = counts.entry(template).or_insert(0);
            *n += 1;
            if *n <= MAX_HISTORY_PER_TEMPLATE || finished > grace_cutoff {
                kept.push(c);
            }
        }
        kept
    }
}

/// A spec dependency index (`Dep.on`, a wire `u64`) as a `usize` for indexing
/// into the node/id vectors. `templates::validate` guarantees it's in range.
fn dep_index(on: u64) -> usize {
    usize::try_from(on).unwrap_or(usize::MAX)
}

/// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`.
fn truncate_error(e: &str) -> String {
    if e.len() <= MAX_ERROR_LEN {
        return e.to_owned();
    }
    let cut = (0..=MAX_ERROR_LEN)
        .rev()
        .find(|i| e.is_char_boundary(*i))
        .unwrap_or(0);
    let mut msg = e[..cut].to_owned();
    msg.push('…');
    msg
}