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).
This commit is contained in:
atlas 2026-07-27 11:34:31 +02:00 committed by mara
commit a8728ac532
8 changed files with 39 additions and 27 deletions

1
Cargo.lock generated
View file

@ -1694,6 +1694,7 @@ dependencies = [
name = "hive-host-sock" name = "hive-host-sock"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"chrono",
"hive-sh4re", "hive-sh4re",
"hive-types", "hive-types",
"serde", "serde",

View file

@ -39,10 +39,10 @@ use std::collections::HashMap;
use std::sync::Mutex; use std::sync::Mutex;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use hive_host_sock::jobs::NodeView;
use hive_jobq::resources::ResourceTable; use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; 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 hive_sh4re::wire_time::now_unix;
use tokio::sync::Notify; use tokio::sync::Notify;

View file

@ -1,9 +1,10 @@
//! Data model for the generic job-DAG queue: node kinds (the primitive //! Data model for the generic job-DAG queue: node kinds (the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store. //! operations), dependency edges, and the runtime `Dag` / `Node` store.
//! The serialized *views* — `DagView` / `NodeView` plus the `Template` //! The serialized *views* — `DagView` / `NodeView` plus the `Source` /
//! / `Source` / `State` / `PermPayload` wire enums — live in //! `State` / `PermPayload` wire enums — live in `hive_host_sock::jobs`
//! `hive_sh4re::jobs` (wire types belong to the shared crate) and are //! (they travel on the host admin socket and the dashboard channels
//! re-exported here for the queue's internal use. //! 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 //! Two levels: the **DAG** is the unit of cancel / approval-resolution
//! and the dashboard group; the **node** is the unit of scheduling / //! 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 //! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design. //! 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 serde::Serialize;
use crate::coordinator::TransientKind; use crate::coordinator::TransientKind;

View file

@ -8,6 +8,7 @@ readme = "README.md"
workspace = true workspace = true
[dependencies] [dependencies]
chrono.workspace = true
hive-sh4re.workspace = true hive-sh4re.workspace = true
hive-types.workspace = true hive-types.workspace = true
serde.workspace = true serde.workspace = true

View file

@ -2,16 +2,22 @@
//! //!
//! The host-control protocol spoken between `hivectl` and the `hive-c0re` //! The host-control protocol spoken between `hivectl` and the `hive-c0re`
//! daemon. Re-homed out of `hive-sh4re` so a standalone `hivectl` can depend //! 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 //! on just this protocol crate instead of the whole daemon crate. The
//! payload types it references (`Approval`, `AgentStatusRow`, `jobs::DagView`) //! job-queue wire types ([`jobs`]) travel on this socket and the dashboard
//! stay in `hive-sh4re`. //! 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 std::path::PathBuf;
use hive_sh4re::{AgentStatusRow, Approval, jobs}; use hive_sh4re::{AgentStatusRow, Approval};
use hive_types::Ident; use hive_types::Ident;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
pub mod jobs;
// ── Shared hive layout facts ────────────────────────────────────────────── // ── Shared hive layout facts ──────────────────────────────────────────────
// Paths + names both the `hive-c0re` daemon and the host-side `hivectl` CLI // 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) // must agree on. Homed here (the protocol crate both sides already depend on)

View file

@ -5,7 +5,6 @@ use hive_types::Ident;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
pub mod assets; pub mod assets;
pub mod jobs;
pub mod paths; pub mod paths;
pub mod wire_time; pub mod wire_time;

View file

@ -66,7 +66,7 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec<u64>) -> Result<()> {
// may still be running — keep watching so the operator // may still be running — keep watching so the operator
// sees whether the agent came back. // sees whether the agent came back.
if d.nodes.iter().all(|n| n.state.is_terminal()) { 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))); failed.push(format!("{} {}", d.source.as_str(), dag_agents(d)));
} }
} else { } else {
@ -104,9 +104,9 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
// insertion order groups a DAG's nodes right under its header. // insertion order groups a DAG's nodes right under its header.
let mut dag_bars: std::collections::HashMap<u64, ProgressBar> = let mut dag_bars: std::collections::HashMap<u64, ProgressBar> =
std::collections::HashMap::new(); 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(); 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(); std::collections::HashSet::new();
let mut pending: std::collections::BTreeSet<u64> = ids.into_iter().collect(); let mut pending: std::collections::BTreeSet<u64> = ids.into_iter().collect();
@ -163,7 +163,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
} }
} }
if d.nodes.iter().all(|n| n.state.is_terminal()) { 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))); failed.push(format!("{} {}", d.source.as_str(), dag_agents(d)));
} }
} else { } else {
@ -205,7 +205,7 @@ fn finish_wait(mut failed: Vec<String>) -> Result<()> {
/// Distinct agents across a DAG's nodes, comma-joined for display — the /// Distinct agents across a DAG's nodes, comma-joined for display — the
/// per-node replacement for the old DAG-level `agent` field. Single-agent /// per-node replacement for the old DAG-level `agent` field. Single-agent
/// DAGs render one name; a hive-wide DAG lists each. /// 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(); let mut seen: Vec<&str> = Vec::new();
for n in &d.nodes { for n in &d.nodes {
if !seen.contains(&n.agent.as_str()) { 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`) /// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`)
/// through `finished_at` or `now`. The wire carries these as RFC3339 /// through `finished_at` or `now`. The wire carries these as RFC3339
/// `DateTime<Utc>`; compare in unix seconds against `now`. /// `DateTime<Utc>`; 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 let start = d
.started_at .started_at
.map_or_else(|| d.created_at.timestamp(), |t| t.timestamp()); .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 /// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0
/// when it hasn't started. /// 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 { match n.started_at {
Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0),
None => 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 /// One animated node line: kind, an `(after …)` marker for a fan-in node
/// (>1 dep), its elapsed timer, and a truncated error tail. /// (>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 _; use std::fmt::Write as _;
let mut s = n.kind.clone(); let mut s = n.kind.clone();
if n.deps.len() > 1 { if n.deps.len() > 1 {
@ -280,13 +284,13 @@ fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now:
s s
} }
fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { fn state_glyph(state: hive_host_sock::jobs::State) -> &'static str {
match state { match state {
hive_sh4re::jobs::State::Queued => "", hive_host_sock::jobs::State::Queued => "",
hive_sh4re::jobs::State::Running => "", hive_host_sock::jobs::State::Running => "",
hive_sh4re::jobs::State::Done => "", hive_host_sock::jobs::State::Done => "",
hive_sh4re::jobs::State::Failed => "", hive_host_sock::jobs::State::Failed => "",
hive_sh4re::jobs::State::Cancelled => "", 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 /// 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 /// 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. /// 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 _; use std::fmt::Write as _;
let mut out = format!( let mut out = format!(
"{} {} {:<12}", "{} {} {:<12}",
@ -315,7 +319,7 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
#[cfg(test)] #[cfg(test)]
mod tests { 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 hive_sh4re::wire_time::from_secs;
use super::render_dag_line; use super::render_dag_line;