From a8728ac532d87d3a82a47b6de06cc2a321058766 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 11:34:31 +0200 Subject: [PATCH 1/2] job_queue: move the jobs wire types to hive-host-sock The DagView / NodeView / Source / State / PermPayload types only ever travel on the host admin socket and the dashboard channels hive-c0re serves off the same snapshot; their whole consumer set is hive-c0re, hivectl and the socket protocol crate itself. Living in hive-sh4re made the five other crates that depend on it carry job-queue types they never name. Pure move: git mv of the module plus the import sweep, no type changes. hive-sh4re keeps its own chrono (wire_time still needs it). --- Cargo.lock | 1 + hive-c0re/src/job_queue/mod.rs | 2 +- hive-c0re/src/job_queue/model.rs | 11 ++++--- hive-host-sock/Cargo.toml | 1 + {hive-sh4re => hive-host-sock}/src/jobs.rs | 0 hive-host-sock/src/lib.rs | 14 ++++++--- hive-sh4re/src/lib.rs | 1 - hivectl/src/dag_progress.rs | 36 ++++++++++++---------- 8 files changed, 39 insertions(+), 27 deletions(-) rename {hive-sh4re => hive-host-sock}/src/jobs.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index ea942228..a1ab379c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1694,6 +1694,7 @@ dependencies = [ name = "hive-host-sock" version = "0.1.0" dependencies = [ + "chrono", "hive-sh4re", "hive-types", "serde", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 6ebcc9f1..4a96b858 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -39,10 +39,10 @@ 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, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; -use hive_sh4re::jobs::NodeView; use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 5d353d31..63aae51f 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -1,9 +1,10 @@ //! 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. +//! The serialized *views* — `DagView` / `NodeView` plus the `Source` / +//! `State` / `PermPayload` wire enums — live in `hive_host_sock::jobs` +//! (they travel on the host admin socket and the dashboard channels +//! served off the same snapshot, and nowhere else) and are re-exported +//! here for the queue's internal use. //! //! Two levels: the **DAG** is the unit of cancel / approval-resolution //! and the dashboard group; the **node** is the unit of scheduling / @@ -11,7 +12,7 @@ //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. -pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State}; +pub use hive_host_sock::jobs::{DagView, NodeId, PermPayload, Source, State}; use serde::Serialize; use crate::coordinator::TransientKind; diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index cdb83f53..902b2ab7 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -8,6 +8,7 @@ readme = "README.md" workspace = true [dependencies] +chrono.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true diff --git a/hive-sh4re/src/jobs.rs b/hive-host-sock/src/jobs.rs similarity index 100% rename from hive-sh4re/src/jobs.rs rename to hive-host-sock/src/jobs.rs diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 0427c6ee..518a43b0 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -2,16 +2,22 @@ //! //! The host-control protocol spoken between `hivectl` and the `hive-c0re` //! daemon. Re-homed out of `hive-sh4re` so a standalone `hivectl` can depend -//! on just this protocol crate instead of the whole daemon crate. The shared -//! payload types it references (`Approval`, `AgentStatusRow`, `jobs::DagView`) -//! stay in `hive-sh4re`. +//! on just this protocol crate instead of the whole daemon crate. The +//! job-queue wire types ([`jobs`]) travel on this socket and the dashboard +//! channels `hive-c0re` serves off the same snapshot, and on nothing else — +//! their whole consumer set is `hive-c0re` + `hivectl` + this crate — so they +//! live here rather than in the shared crate. The payload types genuinely +//! shared with the *agent* sockets (`Approval`, `AgentStatusRow`) stay in +//! `hive-sh4re`. use std::path::PathBuf; -use hive_sh4re::{AgentStatusRow, Approval, jobs}; +use hive_sh4re::{AgentStatusRow, Approval}; use hive_types::Ident; use serde::{Deserialize, Serialize}; +pub mod jobs; + // ── Shared hive layout facts ────────────────────────────────────────────── // Paths + names both the `hive-c0re` daemon and the host-side `hivectl` CLI // must agree on. Homed here (the protocol crate both sides already depend on) diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index ac49b852..3be84fa7 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -5,7 +5,6 @@ use hive_types::Ident; use serde::{Deserialize, Serialize}; pub mod assets; -pub mod jobs; pub mod paths; pub mod wire_time; diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index 1b30b4bf..ec32b6d8 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -66,7 +66,7 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { // may still be running — keep watching so the operator // sees whether the agent came back. if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.rollup_state() == hive_sh4re::jobs::State::Failed { + if d.rollup_state() == hive_host_sock::jobs::State::Failed { failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); } } else { @@ -104,9 +104,9 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { // insertion order groups a DAG's nodes right under its header. let mut dag_bars: std::collections::HashMap = std::collections::HashMap::new(); - let mut node_bars: std::collections::HashMap<(u64, hive_sh4re::jobs::NodeId), ProgressBar> = + let mut node_bars: std::collections::HashMap<(u64, hive_host_sock::jobs::NodeId), ProgressBar> = std::collections::HashMap::new(); - let mut node_done: std::collections::HashSet<(u64, hive_sh4re::jobs::NodeId)> = + let mut node_done: std::collections::HashSet<(u64, hive_host_sock::jobs::NodeId)> = std::collections::HashSet::new(); let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); @@ -163,7 +163,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { } } if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.rollup_state() == hive_sh4re::jobs::State::Failed { + if d.rollup_state() == hive_host_sock::jobs::State::Failed { failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); } } else { @@ -205,7 +205,7 @@ fn finish_wait(mut failed: Vec) -> Result<()> { /// Distinct agents across a DAG's nodes, comma-joined for display — the /// per-node replacement for the old DAG-level `agent` field. Single-agent /// DAGs render one name; a hive-wide DAG lists each. -fn dag_agents(d: &hive_sh4re::jobs::DagView) -> String { +fn dag_agents(d: &hive_host_sock::jobs::DagView) -> String { let mut seen: Vec<&str> = Vec::new(); for n in &d.nodes { if !seen.contains(&n.agent.as_str()) { @@ -227,7 +227,7 @@ fn now_unix() -> i64 { /// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`) /// through `finished_at` or `now`. The wire carries these as RFC3339 /// `DateTime`; compare in unix seconds against `now`. -fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 { +fn dag_elapsed(d: &hive_host_sock::jobs::DagView, now: i64) -> i64 { let start = d .started_at .map_or_else(|| d.created_at.timestamp(), |t| t.timestamp()); @@ -236,7 +236,7 @@ fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 { /// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0 /// when it hasn't started. -fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 { +fn node_elapsed(n: &hive_host_sock::jobs::NodeView, now: i64) -> i64 { match n.started_at { Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), None => 0, @@ -255,7 +255,11 @@ fn fmt_dur(secs: i64) -> String { /// One animated node line: kind, an `(after …)` marker for a fan-in node /// (>1 dep), its elapsed timer, and a truncated error tail. -fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String { +fn node_line( + d: &hive_host_sock::jobs::DagView, + n: &hive_host_sock::jobs::NodeView, + now: i64, +) -> String { use std::fmt::Write as _; let mut s = n.kind.clone(); if n.deps.len() > 1 { @@ -280,13 +284,13 @@ fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: s } -fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { +fn state_glyph(state: hive_host_sock::jobs::State) -> &'static str { match state { - hive_sh4re::jobs::State::Queued => "⏸", - hive_sh4re::jobs::State::Running => "▶", - hive_sh4re::jobs::State::Done => "✔", - hive_sh4re::jobs::State::Failed => "✖", - hive_sh4re::jobs::State::Cancelled => "⊘", + hive_host_sock::jobs::State::Queued => "⏸", + hive_host_sock::jobs::State::Running => "▶", + hive_host_sock::jobs::State::Done => "✔", + hive_host_sock::jobs::State::Failed => "✖", + hive_host_sock::jobs::State::Cancelled => "⊘", } } @@ -294,7 +298,7 @@ fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { /// node chain — the CLI twin of the dashboard's queue card. The header shows /// what the backend sends (`source` + the raw node kinds); only the roll-up /// state glyph is derived from the node set. Used by the plain (non-TTY) path. -fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { +fn render_dag_line(d: &hive_host_sock::jobs::DagView) -> String { use std::fmt::Write as _; let mut out = format!( "{} {} {:<12}", @@ -315,7 +319,7 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { #[cfg(test)] mod tests { - use hive_sh4re::jobs::{DagView, NodeView, Source, State}; + use hive_host_sock::jobs::{DagView, NodeView, Source, State}; use hive_sh4re::wire_time::from_secs; use super::render_dag_line; From ca7146e4f08a520b3848ee7bbf49a311bf8ba0c1 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 12:37:03 +0200 Subject: [PATCH 2/2] refactor(#2756): declare the terminal hook instead of inferring it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Template` was a DAG-level enum that three different things read back out: `terminal_hook()` mapped it to a side effect, the retention pass bucketed history by it, and a tracing field printed it. None of those needed a *label* — they needed the two facts the label happened to encode. So the enum was a lossy stand-in for intent, and every new DAG shape had to pick the variant whose inferred behaviour matched, whether or not the name fit (`reparent` rode `MetaUpdate` for exactly this reason, with a 10-line comment apologising for it). Replace the inference with a declaration: `DagSpec.hook: Option`. Only the builder assembling a DAG knows why it did so, so only the builder can say what should happen when it settles. `run_terminal_hook` becomes a field read, and `reparent`'s apology becomes `hook: None`. Hook assignment is byte-identical to the old precedence rule (`approval_id.is_some()` wins, then `Rebuild | PermChange`), checked site by site; `meta_update` is the only builder with a variable approval id and so the only remaining conditional. Retention loses the per-template bucket with the enum that keyed it. The dashboard renders one recent-builds list, so one flat newest-first cap (`MAX_HISTORY_DAGS`) bounds it. `HISTORY_GRACE_SECS` goes too — it existed to stop a burst of same-template DAGs evicting each other inside one poll interval, which is not a failure mode a flat cap has. That takes `snapshot_capped()` and the `snapshot_no_grace()` test hook with it. The queue is runtime-only (empty graph on boot), so the serde changes carry no migration risk. --- hive-c0re/src/job_queue/exec.rs | 10 +-- hive-c0re/src/job_queue/mod.rs | 113 +++++++-------------------- hive-c0re/src/job_queue/model.rs | 85 +++++++------------- hive-c0re/src/job_queue/scheduler.rs | 1 - hive-c0re/src/job_queue/submit.rs | 29 ++----- hive-c0re/src/job_queue/templates.rs | 39 ++++----- hive-c0re/src/job_queue/tests.rs | 71 +++++++---------- hive-c0re/src/workers/auto_update.rs | 6 +- 8 files changed, 113 insertions(+), 241 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 0adc028b..178b55a2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -97,18 +97,18 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await, NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // Pure grouping container — no work; completing it lets it reach - // `Finishing` so its child template nodes start. The DAG's terminal + // `Finishing` so its child work nodes start. The DAG's terminal // hook fires (inline, via `run_terminal_hook`) when the container itself // rolls up terminal — not as a scheduled node. NodeKind::Dag { .. } => Ok(NodeOutput::default()), } } -/// Run a settled DAG's inline terminal hook, dispatched off its rolled-up -/// summary — the container-terminal replacement for the old per-DAG hook node. -/// Always best-effort: a hook failure is logged inside, never surfaced. +/// Run a settled DAG's inline terminal hook — the container-terminal +/// replacement for the old per-DAG hook node. Always best-effort: a hook +/// failure is logged inside, never surfaced. pub(crate) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { - match super::terminal_hook(terminal.template, terminal.approval_id) { + match terminal.hook { Some(super::HookKind::ResolveApproval) => { crate::actions::resolve_approval_dag(coord, terminal).await; } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 4a96b858..287493fa 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -11,15 +11,15 @@ //! 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 +//! 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 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. +//! container rolls up terminal, off the [`HookKind`] the *builder* stated on +//! the spec: approval-resolve or `Rebuilt`-emit. 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 @@ -48,18 +48,14 @@ use tokio::sync::Notify; use crate::coordinator::TransientKind; pub use model::{ - DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template, + DagSpec, DagView, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source, State, }; 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; +/// 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; @@ -74,7 +70,6 @@ pub struct Claim { /// 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, pub inputs: Vec, /// Transient pill kind for the lease window (from the spec). Whether the @@ -88,7 +83,8 @@ pub struct Claim { /// revert). Computed on demand from live graph state, not drained. #[derive(Debug, Clone)] pub struct TerminalDag { - pub template: Template, + /// The side effect to fire, carried from the DAG container's payload. + pub hook: Option, /// Distinct agents this DAG's nodes targeted (one for a single-agent DAG). pub agents: Vec, pub approval_id: Option, @@ -110,7 +106,7 @@ struct NodeRuntime { /// 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, + hook: Option, source: Source, reason: String, transient: Option, @@ -161,37 +157,6 @@ fn to_crate_when(when: DepWhen) -> JobDepWhen { } } -/// 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) -> Option { - 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 { @@ -289,7 +254,7 @@ impl JobQueue { .sched .append( NodeKind::Dag { - template: spec.template, + hook: spec.hook, source: spec.source, reason: spec.reason, transient: spec.transient, @@ -380,7 +345,6 @@ impl JobQueue { node_id: id, kind, agent, - template: meta.template, approval_id: meta.approval_id, inputs: meta.inputs, transient: meta.transient, @@ -535,14 +499,8 @@ impl JobQueue { /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. #[must_use] pub fn snapshot(&self) -> Vec { - 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 { let inner = self.lock(); - let mut ids = inner.visible_dags(grace_cutoff); + 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() } @@ -558,14 +516,6 @@ impl JobQueue { .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 { - self.snapshot_capped(i64::MAX) - } } impl QueueInner { @@ -616,7 +566,7 @@ impl QueueInner { /// not a stored side-table. fn dag_meta(&self, container: NodeId) -> Option { let NodeKind::Dag { - template, + hook, source, reason, transient, @@ -628,7 +578,7 @@ impl QueueInner { return None; }; Some(DagMeta { - template: *template, + hook: *hook, source: *source, reason: reason.clone(), transient: *transient, @@ -708,7 +658,7 @@ impl QueueInner { fn terminal_dag(&self, container: NodeId) -> Option { let meta = self.dag_meta(container)?; Some(TerminalDag { - template: meta.template, + hook: meta.hook, agents: self.dag_agents(container), approval_id: meta.approval_id, state: self.dag_rollup(container), @@ -829,33 +779,24 @@ impl QueueInner { } /// 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 { + /// 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 { let mut live: Vec = Vec::new(); - let mut terminal: Vec<(NodeId, Template, i64)> = Vec::new(); + let mut terminal: Vec<(NodeId, 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))); - } + terminal.push((c, 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 = HashMap::new(); + // 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; - 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.extend(terminal.into_iter().map(|(c, _)| c)); kept } } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 63aae51f..03ea621f 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,55 +17,25 @@ use serde::Serialize; use crate::coordinator::TransientKind; -/// What a DAG *means* — the request-level shape. Internal to the queue now: -/// it drives the terminal-hook dispatch ([`crate::job_queue`]'s `dag_hook`) -/// and the meta-update dedup key, and is **no longer sent on the wire** — the -/// dashboard derives a DAG's label from its node kinds (see `DagView`). The -/// `NodeKind::Dag` container carries it in its payload. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] +/// The inline side effect a settled DAG fires when its container node rolls +/// up terminal (there is no hook *node*). Stated explicitly by the builder in +/// `templates.rs` / `submit.rs` rather than inferred from a DAG-level enum: +/// only the builder knows why it assembled the DAG, so only the builder can +/// say what should happen at the end of it. +/// +/// A cancelled DAG deliberately gets **no** compensating hook. [`super::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. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] -pub enum Template { - /// Rebuild one agent's container (prebuild → stop → profile-swap → - /// reconcile). - Rebuild, - /// Bump meta flake locks; grows a rebuild subgraph per affected - /// agent into the same DAG on completion. - MetaUpdate, - /// First-deploy spawn (approval-driven). - Spawn, - /// Mechanical stop + converge to `wanted = Up` (a restart). - Restart, - /// Signal → drain → mechanical stop → converge to `wanted = Up` — a - /// graceful restart as one atomic DAG. - GracefulRestart, - /// Perm-file commit followed by the rebuild subgraph. - PermChange, - /// Quiesce the harness, drain, then stop (`wanted = Offline`). - GracefulStop, - /// Converge to `wanted = Up`. - Start, - /// Converge to `wanted = Offline`. - Stop, - /// Boot-time config sweep as one DAG. - Boot, -} - -impl Template { - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Template::Rebuild => "rebuild", - Template::MetaUpdate => "meta_update", - Template::Spawn => "spawn", - Template::Restart => "restart", - Template::GracefulRestart => "graceful_restart", - Template::PermChange => "perm_change", - Template::GracefulStop => "graceful_stop", - Template::Start => "start", - Template::Stop => "stop", - Template::Boot => "boot", - } - } +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, } /// When a dependency edge is considered satisfied. @@ -293,15 +263,16 @@ pub enum NodeKind { /// is skipped.) SetWanted { agent: String, up: bool }, /// The **DAG container** node: one per submitted DAG, carrying the group's - /// domain metadata. Every template node hangs *under* it (its subtree), so + /// domain metadata. Every node hangs *under* it (its subtree), so /// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the /// DAG state, and it reaching terminal **is** the completion signal that - /// fires the DAG's inline hook (approval-resolve / rebuilt-emit / - /// intent-revert, dispatched off `template`). Pure grouping — lease- and + /// fires the DAG's inline `hook`. Pure grouping — lease- and /// build-slot-exempt; the executor instant-completes it (`Done`) so it /// reaches `Finishing` and its children start. Dag { - template: Template, + /// The side effect to run when this DAG settles, or `None` for a DAG + /// with none (power op, meta-update, boot). + hook: Option, source: Source, reason: String, transient: Option, @@ -475,14 +446,16 @@ pub struct NodeSpec { /// ([`NodeKind::WritePermFile`]), not this generic spec. #[derive(Debug, Clone)] pub struct DagSpec { - pub template: Template, + /// The inline side effect to fire when this DAG settles. Explicit — the + /// builder assembling the DAG is the only thing that knows its intent. + pub hook: Option, pub source: Source, /// Free-form "why". pub reason: String, - /// Fires the approval-resolution hook on DAG terminal. + /// The approval row [`HookKind::ResolveApproval`] resolves. Set together + /// with that hook; carried separately because the hook needs the id. pub approval_id: Option, - /// `MetaUpdate`-only: the inputs to bump (also part of the dedup - /// key for that template). Display copy lives on the DAG. + /// Meta-update only: the inputs to bump. Display copy lives on the DAG. pub inputs: Vec, /// Dashboard transient pill (and crash-watch suppression) held for /// the lease window — from lease acquisition to DAG terminal. diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 850d6bb2..87536ef9 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -53,7 +53,6 @@ pub async fn run_worker(coord: Arc) { node = claim.node_id.get(), kind = claim.kind.as_str(), agent = %claim.agent, - template = claim.template.as_str(), "job_queue: node running" ); let coord = Arc::clone(&coord); diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index c9648162..3584793c 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -24,7 +24,7 @@ use std::sync::Arc; -use super::model::{DagSpec, Dep, NodeKind, NodeSpec, Template}; +use super::model::{DagSpec, Dep, NodeKind, NodeSpec}; use super::templates::{after_ok, child, node, rebuild_nodes}; use super::{Source, templates}; use crate::coordinator::{Coordinator, TransientKind}; @@ -188,16 +188,17 @@ fn concat_subgraphs(chains: Vec>) -> Vec { out } -/// Wrap assembled power-op `nodes` in a `DagSpec`. +/// Wrap assembled power-op `nodes` in a `DagSpec`. No terminal hook: a power +/// op's effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to +/// do once they settle. fn power_dag( - template: Template, transient: TransientKind, source: Source, reason: String, nodes: Vec, ) -> DagSpec { DagSpec { - template, + hook: None, source, reason, approval_id: None, @@ -223,13 +224,7 @@ pub(crate) fn stop_spec( .iter() .map(|(agent, running)| stop_chain(agent, graceful, *running)) .collect(); - let template = if graceful { - Template::GracefulStop - } else { - Template::Stop - }; power_dag( - template, TransientKind::Stopping, source, reason, @@ -256,13 +251,7 @@ pub(crate) fn start_spec( } else { TransientKind::Starting }; - power_dag( - Template::Start, - transient, - source, - reason, - concat_subgraphs(chains), - ) + power_dag(transient, source, reason, concat_subgraphs(chains)) } /// Assemble the restart DAG from explicit `(agent, running)` targets. @@ -276,13 +265,7 @@ pub(crate) fn restart_spec( .iter() .map(|(agent, running)| restart_chain(agent, graceful, *running)) .collect(); - let template = if graceful { - Template::GracefulRestart - } else { - Template::Restart - }; power_dag( - template, TransientKind::Restarting, source, reason, diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 5aec3c0a..1b85f338 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -30,7 +30,7 @@ use anyhow::{Result, bail}; -use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source, Template}; +use super::model::{DagSpec, Dep, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source}; use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with @@ -175,7 +175,7 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec { /// children. pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { DagSpec { - template: Template::Rebuild, + hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, @@ -206,7 +206,7 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { let a = || agent.to_owned(); DagSpec { - template: Template::Rebuild, + hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -235,14 +235,13 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec /// in the queue tests); production paths no longer emit a bare reconcile. #[cfg(test)] pub fn reconcile_only( - template: Template, agent: &str, source: Source, reason: String, transient: Option, ) -> DagSpec { DagSpec { - template, + hook: None, source, reason, approval_id: None, @@ -268,7 +267,7 @@ pub fn reconcile_only( /// container was never created). pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { - template: Template::Spawn, + hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -299,7 +298,7 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay )]; nodes.extend(rebuild_nodes(agent, true, 1)); DagSpec { - template: Template::PermChange, + hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, @@ -326,7 +325,9 @@ pub fn meta_update( approval_id: Option, ) -> DagSpec { DagSpec { - template: Template::MetaUpdate, + // The bump itself has no side effect; an approval-driven one still has + // its row to resolve. + hook: approval_id.map(|_| HookKind::ResolveApproval), source, reason, approval_id, @@ -350,24 +351,14 @@ pub fn meta_update( /// (dashboard tree, ``/`` sentinel routing, permission /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one -/// off of) and near-instant. -/// -/// Rides `Template::MetaUpdate` rather than a dedicated variant because -/// `Template` is being removed and nothing should dispatch on it — a fresh -/// variant would just be more surface to delete later. `Template` is already -/// internal-only (not on `DagView`'s wire shape — the dashboard derives its -/// label from `nodes`), so the choice of stand-in variant only affects -/// `terminal_hook` dispatch (`MetaUpdate` resolves to `None`, same as a -/// dedicated variant would) and the per-template history-retention bucket — -/// both cosmetic. Swap this to whatever the eventual node-kind-derived -/// dispatch lands with, whenever it lands. +/// off of) and near-instant. No terminal hook: the write is the whole effect. pub fn reparent( moves: Vec<(hive_types::Ident, Option)>, source: Source, reason: String, ) -> DagSpec { DagSpec { - template: Template::MetaUpdate, + hook: None, source, reason, approval_id: None, @@ -388,7 +379,7 @@ pub fn reparent( /// old queue's documented "circular dep silently deadlocks forever" caveat. pub fn validate(spec: &DagSpec) -> Result<()> { if spec.nodes.is_empty() { - bail!("dag spec {:?} has no nodes", spec.template); + bail!("dag spec {:?} has no nodes", spec.reason); } let n = spec.nodes.len(); let mut graph = petgraph::graph::DiGraph::::new(); @@ -404,14 +395,14 @@ pub fn validate(spec: &DagSpec) -> Result<()> { { bail!( "dag spec {:?} node {i} has invalid parent {p} (must be an earlier node)", - spec.template + spec.reason ); } for dep in &node.deps { let Some(&dep_idx) = usize::try_from(dep.on).ok().and_then(|i| idx.get(i)) else { bail!( "dag spec {:?} node {i} depends on unknown node {}", - spec.template, + spec.reason, dep.on ); }; @@ -419,7 +410,7 @@ pub fn validate(spec: &DagSpec) -> Result<()> { } } if petgraph::algo::toposort(&graph, None).is_err() { - bail!("dag spec {:?} contains a dependency cycle", spec.template); + bail!("dag spec {:?} contains a dependency cycle", spec.reason); } Ok(()) } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 30a08535..d3b7b5ce 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -300,13 +300,7 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { let restart = submit(&q, restart_online(&["agent-a"], false, "restart")); let stop = submit( &q, - templates::reconcile_only( - Template::Stop, - "agent-a", - Source::Manual, - "stop".to_owned(), - None, - ), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), ); // Restart's first node (StopForUpdate) takes the lease; stop's // Reconcile must wait even though slots are free. @@ -337,13 +331,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { submit(&q, rebuild("agent-a", "rebuild")); let stop = submit( &q, - templates::reconcile_only( - Template::Stop, - "agent-a", - Source::Manual, - "stop".to_owned(), - None, - ), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), ); // Both DAGs' heads are lease-independent of each other: the rebuild's // MetaSync (meta window) and the stop's Reconcile (agent lease). @@ -603,7 +591,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // the emitter and its LOCAL 0-based deps are rebased onto the DAG. let q = JobQueue::new(4); let spec = DagSpec { - template: Template::Boot, + hook: None, source: Source::AutoUpdate, reason: "sweep".to_owned(), approval_id: None, @@ -831,13 +819,7 @@ fn failed_reconcile_marks_dag_failed() { let q = JobQueue::new(1); let id = submit( &q, - templates::reconcile_only( - Template::Start, - "agent-a", - Source::Manual, - "start".to_owned(), - None, - ), + templates::reconcile_only("agent-a", Source::Manual, "start".to_owned(), None), ); let c = claim_one(&q); q.complete_node(id, c.node_id, Err("start failed".to_owned())); @@ -917,8 +899,7 @@ fn cancelled_power_op_fires_no_hook() { let summary = q.cancel(id).expect("cancelled while queued"); assert_eq!(summary.state, State::Cancelled); assert_eq!( - terminal_hook(summary.template, summary.approval_id), - None, + summary.hook, None, "cancelled {name} (graceful={graceful}, running={running}) must \ fire no hook — no node of it ever ran" ); @@ -951,13 +932,7 @@ fn dag_settles_terminal_and_releases_lease_after_work() { // immediately. let next = submit( &q, - templates::reconcile_only( - Template::Stop, - "agent-a", - Source::Manual, - "stop".to_owned(), - None, - ), + templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned(), None), ); let c = claim_one(&q); assert_eq!(c.dag_id, next); @@ -1220,14 +1195,20 @@ fn set_build_log_id_links_running_node() { ); } +/// History retention is a **flat** newest-first cap over all terminal DAGs +/// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window. +/// The dashboard renders one recent-builds list, so one number bounds it — +/// and with no bucketing there's nothing for a burst of same-shaped DAGs to +/// evict early, which is what the grace window used to paper over. #[test] -fn history_evicts_old_terminals_per_template() { +fn history_evicts_oldest_terminals_past_flat_cap() { + const OVERFLOW: usize = 8; let q = JobQueue::new(1); - for i in 0..8 { + let mut ids = Vec::new(); + for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) { let id = submit( &q, templates::reconcile_only( - Template::Start, &format!("agent-{i}"), Source::Manual, "start".to_owned(), @@ -1241,17 +1222,19 @@ fn history_evicts_old_terminals_per_template() { // node rolls the container up terminal (its inline hook fires off the // returned summary — no terminal-hook node). q.complete_node(id, c.node_id, Err("boom".to_owned())); + ids.push(id); + } + let kept: std::collections::HashSet = q.snapshot().iter().map(|d| d.id).collect(); + assert_eq!(kept.len(), MAX_HISTORY_DAGS, "flat history cap"); + // Newest-first: the oldest `OVERFLOW` fall off, everything after survives. + // These DAGs settle within the same wall-clock second, so this also pins + // the `NodeId`-descending tiebreak that orders them when `finished_at` ties. + for old in &ids[..OVERFLOW] { + assert!(!kept.contains(old), "oldest terminal {old} evicted"); + } + for recent in &ids[OVERFLOW..] { + assert!(kept.contains(recent), "recent terminal {recent} retained"); } - // Fresh terminals are inside the grace window: nothing evicts yet, - // so a ~1s QueueDag poller can still observe every terminal state - // (a broad stop/start settles many same-template DAGs at once). - assert_eq!( - q.snapshot().len(), - 8, - "grace window protects fresh terminals" - ); - // Past the grace window the per-template cap applies. - assert_eq!(q.snapshot_no_grace().len(), 5, "per-template history cap"); assert_eq!(q.live_count(), 0); } diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 8db07795..0ba9460a 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -303,7 +303,7 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source, Template}; + use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source}; // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. if !any_stale && drifted.is_empty() { @@ -343,7 +343,9 @@ fn submit_boot_tree( } let spec = DagSpec { - template: Template::Boot, + // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they + // land; the boot DAG as a whole has no terminal side effect. + hook: None, source: Source::AutoUpdate, reason, approval_id: None,