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), declared per node at its construction site;
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
//!   carrying the group's metadata, with the work 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 is an ordinary **tail node**
//!   ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//!   appends in [`templates`], edged onto the DAG's other group roots by the
//!   outcome it reports. Templates emit one tail per outcome and the graph runs
//!   exactly one, so nothing branches at runtime.
//!
//! 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_host_sock::jobs::NodeView;
use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Dep, Graph, NodeId};
use tokio::sync::Notify;

pub use hive_jobq::TerminalState;
pub use model::{DagSpec, DagView, NodeKind, PermPayload, Source, State};
use resource::Resource;

/// A job under construction: `hive_jobq`'s builder over this queue's payload
/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into a
/// borrowed one; only `hive_jobq` can make or insert it.
pub type Job = hive_jobq::JobBuilder<NodeKind, Resource>;

/// A handle to one node a template declared — where its edges, grouping and
/// resources are declared. `Copy`; naming a node as a dependency does not
/// consume the ability to name it again.
pub type Handle<'a> = hive_jobq::NodeRef<'a, NodeKind, Resource>;

/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
/// retains, newest first. A flat cap over the whole sorted list: the
/// dashboard renders one recent-builds list, so one number bounds it.
const MAX_HISTORY_DAGS: usize = 50;

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

/// One live transient pill, derived from a running node.
///
/// A named struct rather than a tuple because three of its four fields are
/// easy to confuse at a call site: two are strings and two answer questions
/// nobody should have to guess at ("is this the agent or the label?", "does
/// this bool mean deliberate or running?").
#[derive(Debug, Clone)]
pub struct RunningTransient {
    /// The agent whose lease the node declared.
    pub agent: String,
    /// The node's own wire tag, rendered as the pill.
    pub label: String,
    /// Whether this operation is expected to take the container down — the
    /// crash watcher's input. See [`NodeKind::takes_container_down`].
    pub takes_container_down: bool,
    /// When the node started running, so the dashboard can tick elapsed
    /// seconds. Taken from the node itself, which is the true start of the
    /// operation rather than the moment a watcher noticed it.
    pub since: DateTime<Utc>,
}

/// 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,
}

/// 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 {
    source: Source,
    reason: String,
    created_at: DateTime<Utc>,
}

/// 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::dag_meta`] +
/// the `hive_jobq::Graph` accessors). 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)
    }
}

/// A node runner's `Result` as the scheduler's [`Outcome`].
///
/// The failure reason + `finished_at` are stamped onto the graph `Node` by the
/// scheduler (the reason rides `Outcome::Failed`); there is no host-side copy,
/// so nothing needs clearing on success.
fn outcome_of(result: Result<(), String>) -> Outcome {
    match result {
        Ok(()) => Outcome::Done,
        Err(e) => Outcome::Failed(truncate_error(&e)),
    }
}

/// Insert a declared `job` into the shared graph and record its per-node
/// `node_rt`, returning the inserted ids.
///
/// A node that declared no parent hangs under `group_parent` — the DAG
/// container for a template, the emitting node for a runtime-appended
/// subgraph. 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 resources, and the crate's borrow model keeps a resource continuous
/// across a subtree (a root owns it, descendants borrow it). Independent group
/// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
/// concurrently, each on its own lease.
///
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
    inner: &mut QueueInner,
    declare: impl FnOnce(&Job),
    group_parent: Option<NodeId>,
) -> anyhow::Result<()> {
    inner
        .sched
        .insert_job(group_parent, |b| {
            declare(b);
            // c0re names no handles: a DAG is addressed by its container node,
            // which `submit` inserts itself, and nothing downstream looks an
            // individual step up by id.
            Vec::new()
        })
        .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
    Ok(())
}

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: insert a [`NodeKind::Dag`] **container node** carrying the
    /// group's metadata, then insert 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.
    ///
    /// Takes the spec's recipe by generic, not as a boxed closure: a spec
    /// travels from the template that built it directly into this call, so
    /// there is nothing to allocate for.
    ///
    /// # Errors
    /// Propagates a graph-insert error (dependencies that aren't
    /// dependency-topological).
    pub fn submit<F: FnOnce(&Job)>(&self, spec: DagSpec<F>) -> anyhow::Result<u64> {
        let mut inner = self.lock();
        let container = inner
            .sched
            .append(
                NodeKind::Dag {
                    source: spec.source,
                    reason: spec.reason,
                    created_at: Utc::now(),
                },
                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.declare, 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())
    }

    /// 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.sched.graph().root_of(id) else {
                continue;
            };
            claims.push(Claim {
                dag_id: container.get(),
                node_id: id,
                kind,
                agent,
            });
            // `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.
    ///
    /// Nothing is returned: a DAG's terminal side effects are its own tail nodes
    /// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the
    /// scheduler claims and runs like any other node.
    pub fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
        // Deliberately not `complete_node_growing(.., self.new_job())`: that
        // would take the lock twice (once to mint an empty builder, once to
        // complete) to express "grew nothing". The shared part is the outcome
        // mapping, and that's a free fn.
        let mut inner = self.lock();
        inner.sched.complete(node_id, outcome_of(result));
        drop(inner);
        self.notify.notify_one();
    }

    /// A builder for a node to declare more work into while it runs.
    ///
    /// Handed to [`exec::run_node`] and returned to
    /// [`JobQueue::complete_node_growing`]. Only `hive_jobq` can construct one,
    /// which is why this goes through the scheduler rather than
    /// `Job::default()`.
    #[must_use]
    pub fn new_job(&self) -> Job {
        self.lock().sched.new_job()
    }

    /// [`JobQueue::complete_node`] plus the work the node declared while it ran.
    ///
    /// `grown` is inserted **under `node_id`** before the completion, so the DAG
    /// cannot roll terminal with the appended work still pending — the property
    /// the old two-call `append_subgraph` + `complete_node` sequence had to
    /// arrange by hand at every call site.
    pub fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
        let mut inner = self.lock();
        // A rejected grown job is logged, not propagated: the node's own work
        // already ran, and refusing to complete it here would both misreport
        // that and wedge the DAG on a node stuck `Running`.
        if let Err(e) = inner
            .sched
            .complete_growing(node_id, outcome_of(result), grown)
        {
            tracing::error!(
                node = node_id.get(),
                error = %e,
                "job_queue: work grown by a completing node was rejected"
            );
        }
        drop(inner);
        self.notify.notify_one();
    }

    /// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
    /// so each is cancelled. `false` once any work node is running or terminal —
    /// an in-flight nix build isn't interruptible.
    ///
    /// **Nodes that explicitly observe cancellation are spared** — a node whose
    /// edge names [`hive_jobq::TerminalState::Cancelled`] is asking to run when
    /// the work it follows was dropped, which is exactly what an approval tail
    /// needs: cancel the work, and the tail still fires to resolve the approval
    /// row rather than leaving it dangling forever.
    ///
    /// Nothing is special-cased by node kind. `AFTER_ANY` deliberately does *not*
    /// accept `Cancelled`, so an ordinary weak-edged step (rebuild's `Reconcile`,
    /// say) is cancelled along with everything else — there is nothing to converge
    /// when no node ever ran. Only a node that named `Cancelled` survives, and it
    /// survives because it asked to.
    ///
    /// `id` names **any node**, not specifically a DAG. Cancelling a group root
    /// drops that whole group (the cascade is the scheduler's), which is what
    /// the dashboard's whole-DAG cancel does; cancelling an interior node drops
    /// just that branch. Nothing here knows about DAGs.
    pub fn cancel(&self, id: u64) -> bool {
        let mut inner = self.lock();
        let Some(node) = inner.sched.graph().resolve_id(id) else {
            return false;
        };
        if !inner.sched.cancel_node(node) {
            return false;
        }
        drop(inner);
        self.notify.notify_one();
        true
    }

    /// 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.sched.graph().root_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
            .sched
            .graph()
            .first_error(container)
            .map(ToOwned::to_owned)
    }

    /// `(agent, label, takes_container_down)` for the live transient-pill set,
    /// recomputed from the nodes **actually running** — not from an intent a
    /// template declared at submit time. (A rebuild used to report `rebuilding`
    /// for its whole life: prebuild, stop, swap, tail and reconcile alike.)
    ///
    /// A node lights a pill when it is `Running` **and declares the agent's
    /// resource itself**. Declaring is the test, not targeting — `Prebuild` /
    /// `MetaSync` name an agent but are lease-exempt on purpose, since the
    /// container keeps serving through them. Nor is it the lease *owner*:
    /// `resource_state()` answers "who holds the slot", a different question.
    ///
    /// `label` is the node's own wire tag ([`NodeKind::as_str`]), the vocabulary
    /// [`NodeView::kind`] already ships, so a pill and a DAG node name an
    /// operation identically. `takes_container_down` is the crash watcher's
    /// input, carried rather than inferred from the label — a `Start` pill and a
    /// `Stop` pill are both pills; only one means a vanished container is
    /// expected.
    ///
    /// Read off the node's **declared** resource edges, not off its kind. Those
    /// are the same thing now that every construction site states what it holds,
    /// and the distinction is the whole point: `Start` / `Stop` / `PostSwap` run
    /// inside a lease-holding ancestor, and while the declaration was derived
    /// from the kind they re-declared nothing and lit no pill. Asking the node
    /// what it holds cannot go stale that way. An agent's lease is cap-1, so at
    /// most one entry per agent.
    #[must_use]
    pub fn running_transients(&self) -> Vec<RunningTransient> {
        let inner = self.lock();
        inner
            .sched
            .graph()
            .nodes()
            .filter(|n| matches!(n.state, State::Running))
            .filter_map(|n| {
                let agent = n.deps.iter().find_map(|dep| match dep {
                    hive_jobq::Dep::Resource {
                        name: Resource::Agent(a),
                        ..
                    } => Some(a.clone()),
                    _ => None,
                })?;
                Some(RunningTransient {
                    agent,
                    label: n.payload.as_str().to_owned(),
                    takes_container_down: n.payload.takes_container_down(),
                    // `started_at` is set when a node enters `Running`, and this
                    // only sees `Running` nodes — the fallback is unreachable in
                    // practice, and "just now" is the honest answer if it isn't.
                    since: n.started_at.unwrap_or_else(Utc::now),
                })
            })
            .collect()
    }

    /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
    #[must_use]
    pub fn snapshot(&self) -> Vec<DagView> {
        let inner = self.lock();
        let mut ids = inner.visible_dags();
        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.sched.graph().is_settled(c) == Some(false))
            .count()
    }
}

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 == State::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 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 {
            source,
            reason,
            created_at,
        } = &self.sched.graph().node(container)?.payload
        else {
            return None;
        };
        Some(DagMeta {
            source: *source,
            reason: reason.clone(),
            created_at: *created_at,
        })
    }

    /// 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` or `Skipped` — a fully-settled
    /// 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();
        // Whether anything in this DAG still has an outcome worth showing.
        // Kept separate from `nodes` being non-empty: skipped nodes ride the
        // wire so the dashboard can mark the branches that weren't taken, but
        // they must not by themselves hold a finished DAG in the snapshot.
        let mut any_unsettled = false;
        // 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 node in self.sched.graph().descendants(container) {
            let id = node.id;
            if let Some(s) = node.started_at {
                started.push(s);
            }
            if let Some(f) = node.finished_at {
                finished.push(f);
            }
            // `Done` nodes drop off the wire — a finished step isn't
            // interesting. `Skipped` ones stay: which branch a run *didn't*
            // take is the readable half of an outcome-branched DAG.
            if matches!(node.state, State::Done) {
                continue;
            }
            any_unsettled |= !matches!(node.state, State::Skipped);
            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. Every
            // deploy phase carries the approval id, but only the subtree root
            // projects it onto the wire — hanging the approval link off all of
            // them would render the same card once per phase.
            let approval_id = match &node.payload {
                NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
                _ => None,
            };
            let inputs = match &node.payload {
                NodeKind::MetaLock { inputs, .. } => inputs.clone(),
                _ => Vec::new(),
            };
            let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id);
            // `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: node.state,
                started_at: node.started_at,
                finished_at: node.finished_at,
                error: node.error.clone(),
                approval_id,
                inputs,
                build_log_id,
                parent,
            });
        }
        if !any_unsettled {
            return None;
        }
        let is_terminal = self.sched.graph().is_settled(container) == Some(true);
        Some(DagView {
            id: container.get(),
            source: meta.source,
            reason: meta.reason.clone(),
            created_at: 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.sched
            .graph()
            .descendants(container)
            .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_DAGS`] terminal ones. 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) -> Vec<NodeId> {
        let mut live: Vec<NodeId> = Vec::new();
        let mut terminal: Vec<(NodeId, i64)> = Vec::new();
        for c in self.containers() {
            if self.sched.graph().is_settled(c) == Some(true) {
                terminal.push((c, self.dag_finished_at(c)));
            } else {
                live.push(c);
            }
        }
        // Newest first, so truncating to the cap keeps the most recent.
        terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get())));
        terminal.truncate(MAX_HISTORY_DAGS);
        let mut kept = live;
        kept.extend(terminal.into_iter().map(|(c, _)| c));
        kept
    }
}

/// 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
}