vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

jobq-wire: move the generic graph projection into its own crate

Browse source 
The wire types were in hive-host-sock, which is the host *socket* crate — so
anything living there is core-shaped by construction, and the projection had
quietly grown two core dependencies to match: it selected roots by matching
NodeKind::Dag, and rendered payloads through free functions in hive-c0re that
nothing obliged a second host to write.

hive-jobq is the wrong home too. That crate is the scheduler — logic — and
folding presentation in means every consumer of it carries a JSON vocabulary
it may never serve.

So: a new hive-jobq-wire. A host implements WireNode for its payload N and
WireResource for its resource name R; GraphWire::wire_snapshot is
blanket-implemented for Graph<N, R> when both hold, and for nothing else. A
payload that has never said how it displays has no way onto the wire.

wire_snapshot takes the roots to serve rather than reading Graph::roots
itself. Nothing is ever removed from a Graph, so retention is a policy only
the host can hold; hive-c0re passes visible_roots(), which is the existing
MAX_HISTORY_DAGS bound selected structurally (a root is a node with no
parent) instead of by node kind.
This commit is contained in:
atlas 2026-08-02 23:52:32 +02:00 • committed by mara
commit 7966d5eb66
14 changed files with 564 additions and 297 deletions
Download patch file Download diff file Expand all files Collapse all files

7
CLAUDE.md
View file

@ -63,6 +63,13 @@ hand-maintained per-file tree drifts out of sync with the code.
remaining `job_queue/` module is the c0re-specific layer *over* this
crate, and is being removed in favour of it — new scheduler-shaped code
belongs here, not there.
- **`hive-jobq-wire/`** — wire types for serving a `hive-jobq` graph to a
viewer, plus the `WireNode` / `WireResource` traits a host implements to
say how its `N` and `R` render. Deliberately *not* part of `hive-jobq`:
that crate is logic, this is presentation, and folded together they
remix. `GraphWire::wire_snapshot` is blanket-implemented for any
`Graph<N, R>` whose parameters implement both — so a payload that has
never said how it displays cannot reach a viewer at all.
- **`hive-screen-mcp/`** — stdio MCP bridge for GUI agents
(`hyperhive.gui.enable`): `screenshot` via `grim`, `type_text` /
`key_press` via `wtype` (Wayland virtual-keyboard protocol), and

12
Cargo.lock generated
View file

@ -1636,6 +1636,7 @@ dependencies = [
"hive-core-agent-sock",
"hive-host-sock",
"hive-jobq",
"hive-jobq-wire",
"hive-priv-sock",
"hive-sh4re",
"hive-sock-client",
@ -1727,7 +1728,6 @@ dependencies = [
"hive-sh4re",
"hive-types",
"serde",
"serde_json",
]
[[package]]
@ -1742,6 +1742,16 @@ dependencies = [
"uuid",
]
[[package]]
name = "hive-jobq-wire"
version = "0.1.0"
dependencies = [
"chrono",
"hive-jobq",
"serde",
"serde_json",
]
[[package]]
name = "hive-matrix-mcp"
version = "0.1.0"

2
Cargo.toml
View file

@ -12,6 +12,7 @@ members = [
"hive-forge-notify",
"hive-host-sock",
"hive-jobq",
"hive-jobq-wire",
"hive-matrix-mcp",
"hive-metric",
"hive-priv",
@ -56,6 +57,7 @@ indicatif = "0.18"
hive-sh4re = { path = "hive-sh4re" }
hive-agent-sock = { path = "hive-agent-sock" }
hive-jobq = { path = "hive-jobq" }
hive-jobq-wire = { path = "hive-jobq-wire" }
hive-core-agent-sock = { path = "hive-core-agent-sock" }
hive-claude = "0.1"
hive-host-sock = { path = "hive-host-sock" }

1
hive-c0re/Cargo.toml
View file

@ -36,6 +36,7 @@ hive-core-agent-sock.workspace = true
hive-sh4re.workspace = true
hive-host-sock.workspace = true
hive-jobq.workspace = true
hive-jobq-wire.workspace = true
hive-priv-sock.workspace = true
hive-agent-sock.workspace = true
hive-sock-client.workspace = true

4
hive-c0re/src/dashboard/state_snapshot.rs
View file

@ -661,14 +661,14 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
one opaque per-node payload. Group roots ride as ordinary nodes \
(`parent: null`) and `Done` nodes are not filtered — a consumer \
renders the graph without knowing what any node means. Each element \
is a `hive_host_sock::graph::GraphNode`",
is a `hive_jobq_wire::GraphNode`",
body = serde_json::Value),
),
tag = "state_snapshot"
)]
pub(super) async fn jobq_graph(
State(state): State<AppState>,
) -> axum::Json<Vec<hive_host_sock::graph::GraphNode>> {
) -> axum::Json<Vec<hive_jobq_wire::GraphNode>> {
axum::Json(state.coord.job_queue.graph_snapshot())
}

145
hive-c0re/src/job_queue/mod.rs
View file

@ -39,11 +39,11 @@ mod tests;
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Utc};
use hive_host_sock::graph::{GraphDep, GraphNode, NodePayload};
use hive_host_sock::jobs::NodeView;
use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Dep, Graph, NodeId};
use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify;
pub use hive_jobq::TerminalState;
@ -363,26 +363,13 @@ impl JobQueue {
/// finished step is visible rather than vanishing from the payload, which
/// is what makes a fast rebuild render as a single node).
///
/// Retention is the *same* bound as [`Self::snapshot`] — every live group
/// plus the newest terminal ones, via [`visible_dags`]. Serving the raw
/// graph instead would grow without limit: evicted groups' nodes linger
/// until bounded pruning lands.
/// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies
/// is *which* groups to show — see [`visible_roots`] for why the graph
/// can't decide that for itself.
#[must_use]
pub fn graph_snapshot(&self) -> Vec<GraphNode> {
let inner = self.lock();
let mut roots = visible_dags(&inner);
roots.sort_unstable_by_key(|root| root.get());
roots
.into_iter()
.flat_map(|root| {
inner
.graph()
.node(root)
.into_iter()
.chain(inner.graph().descendants(root))
.map(graph_node)
})
.collect()
inner.graph().wire_snapshot(visible_roots(&inner))
}
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
@ -519,79 +506,6 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
})
}
/// Project one graph node onto the generic wire shape.
///
/// Deliberately near-total: everything `hive_jobq` records is carried, and
/// the only editorial decision is what goes in the opaque payload. That is
/// the inverse of [`dag_view`], which decides what a consumer is allowed to
/// see — a viewer that can render *any* graph cannot have that decided for it.
fn graph_node(node: &hive_jobq::Node<NodeKind, Resource>) -> GraphNode {
GraphNode {
id: node.id.get(),
parent: node.parent.map(NodeId::get),
state: node.state,
deps: node.deps.iter().map(graph_dep).collect(),
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
payload: NodePayload {
label: node.payload.as_str().to_owned(),
data: node_data(node.id, &node.payload),
},
}
}
fn graph_dep(dep: &Dep<Resource>) -> GraphDep {
match dep {
Dep::Node { id, when } => GraphDep::Node {
id: id.get(),
accepts: [
TerminalState::Done,
TerminalState::Failed,
TerminalState::Cancelled,
TerminalState::Skipped,
]
.into_iter()
.filter(|outcome| when.accepts(*outcome))
.collect(),
},
Dep::Resource { name, count } => GraphDep::Resource {
name: name.wire_name(),
count: *count,
},
}
}
/// hive-c0re's domain data for one node, as the opaque payload slot.
///
/// Every field here used to be a named column on `NodeView`, meaningful for
/// one node kind and `null` on all the others. As free-form data it costs the
/// wire type nothing and a generic consumer renders it without knowing what
/// any of it means.
fn node_data(id: NodeId, kind: &NodeKind) -> serde_json::Value {
let mut data = serde_json::Map::new();
let agent = kind.agent();
if !agent.is_empty() {
data.insert("agent".to_owned(), agent.into());
}
if let NodeKind::DeployWindow { approval_id, .. } = kind {
data.insert("approval_id".to_owned(), (*approval_id).into());
}
if let NodeKind::MetaLock { inputs, .. } = kind
&& !inputs.is_empty()
{
data.insert("inputs".to_owned(), inputs.clone().into());
}
if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())) {
data.insert("build_log_id".to_owned(), log.into());
}
if data.is_empty() {
serde_json::Value::Null
} else {
serde_json::Value::Object(data)
}
}
/// Which of a DAG's work nodes ride the wire, by index into `states` — or
/// `None` when the DAG has nothing left worth showing and drops out of the
/// snapshot entirely.
@ -665,6 +579,55 @@ fn visible_dags(sched: &Sched) -> Vec<NodeId> {
retain_history(live, terminal, MAX_HISTORY_DAGS)
}
/// The visible **group** set for [`Queue::graph_snapshot`]: every live group
/// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
///
/// Same policy as [`visible_dags`], selected *structurally* — a root is a node
/// with no parent. The `DagView` path next door keys on `NodeKind::Dag`
/// instead, which is fine for a projection that already only means anything to
/// hive-c0re, but would make the generic endpoint depend on one node kind that
/// is itself slated for removal.
///
/// **This bound is load-bearing, not tidiness.** Nothing ever removes a node
/// from the graph (bounded pruning is a Stage-C follow-up), so serving
/// `graph.roots()` directly would grow the payload without limit for the whole
/// uptime of the daemon.
fn visible_roots(sched: &Sched) -> Vec<NodeId> {
let roots: Vec<NodeId> = sched.graph().roots().map(|n| n.id).collect();
let mut live: Vec<NodeId> = Vec::new();
let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new();
for root in roots {
if sched.graph().is_settled(root) == Some(true) {
terminal.push((root, group_finished_at(sched, root), root.get()));
} else {
live.push(root);
}
}
retain_history(live, terminal, MAX_HISTORY_DAGS)
}
/// When a whole group last finished: the newest `finished_at` across the root
/// **and** its descendants. Unlike [`dag_finished_at`] the root itself counts,
/// because a generic group root can be an ordinary node with no children at
/// all — reading only descendants would date every such group to the epoch and
/// evict it first.
fn group_finished_at(sched: &Sched, root: NodeId) -> i64 {
sched
.graph()
.node(root)
.and_then(|n| n.finished_at)
.into_iter()
.chain(
sched
.graph()
.descendants(root)
.filter_map(|n| n.finished_at),
)
.map(|t| t.timestamp())
.max()
.unwrap_or(0)
}
/// [`visible_dags`]'s policy, split from the graph it reads: keep every live
/// DAG, plus the newest `cap` terminal ones.
///

38
hive-c0re/src/job_queue/model.rs
View file

@ -283,6 +283,44 @@ pub enum NodeKind {
},
}
/// How a hive-c0re node describes itself to a generic graph viewer.
///
/// Every field in [`WireNode::data`] here used to be a named column on
/// `NodeView`, meaningful for one node kind and `null` on all the others. As
/// free-form data it costs the wire type nothing, and a generic consumer
/// renders it without knowing what any of it means.
impl hive_jobq_wire::WireNode for NodeKind {
fn label(&self) -> String {
self.as_str().to_owned()
}
fn data(&self, id: hive_jobq_wire::WireId) -> serde_json::Value {
let mut data = serde_json::Map::new();
let agent = self.agent();
if !agent.is_empty() {
data.insert("agent".to_owned(), agent.into());
}
if let NodeKind::DeployWindow { approval_id, .. } = self {
data.insert("approval_id".to_owned(), (*approval_id).into());
}
if let NodeKind::MetaLock { inputs, .. } = self
&& !inputs.is_empty()
{
data.insert("inputs".to_owned(), inputs.clone().into());
}
// Not in the payload at all — the build log is keyed on node identity
// in a side table, which is why `data` is handed the id.
if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id)) {
data.insert("build_log_id".to_owned(), log.into());
}
if data.is_empty() {
serde_json::Value::Null
} else {
serde_json::Value::Object(data)
}
}
}
impl NodeKind {
/// Wire string for `NodeView.kind`.
pub fn as_str(&self) -> &'static str {

17
hive-c0re/src/job_queue/resource.rs
View file

@ -58,15 +58,14 @@ pub enum Resource {
MetaWindow,
}
impl Resource {
/// This resource's name on the generic graph wire.
///
/// `hive_jobq` is generic over the resource type, so a viewer that can
/// render any graph gets a string here rather than this enum. The
/// `agent:` prefix keeps the per-agent leases from colliding with a
/// hypothetical global resource that happens to share an agent's name.
#[must_use]
pub fn wire_name(&self) -> String {
/// This resource's name on the generic graph wire.
///
/// `hive_jobq` is generic over the resource type, so a viewer that can render
/// any graph gets a string here rather than this enum. The `agent:` prefix
/// keeps the per-agent leases from colliding with a hypothetical global
/// resource that happens to share an agent's name.
impl hive_jobq_wire::WireResource for Resource {
fn name(&self) -> String {
match self {
Resource::BuildSlot => "build-slot".to_owned(),
Resource::Agent(agent) => format!("agent:{agent}"),

1
hive-host-sock/Cargo.toml
View file

@ -13,4 +13,3 @@ hive-jobq.workspace = true
hive-sh4re.workspace = true
hive-types.workspace = true
serde.workspace = true
serde_json.workspace = true

192
hive-host-sock/src/graph.rs
View file

@ -1,192 +0,0 @@
//! Generic jobq graph wire types — a `hive_jobq` graph serialised without
//! knowing what its nodes mean.
//!
//! [`super::jobs`]'s `DagView` / `NodeView` are hive-c0re's *domain*
//! projection: they carry `approval_id`, `inputs`, `build_log_id` and an
//! `agent`, each meaningful for a subset of one specific node kind. Anything
//! built on those can only ever display hive-c0re's queue.
//!
//! These types carry what `hive_jobq::Node` itself carries — identity, the
//! parent tree, dependency edges, lifecycle — and push everything
//! domain-specific into one opaque [`NodePayload::data`] slot the consumer
//! renders without branching on. That is the crate boundary made visible:
//! `hive-jobq` owns structure, its host owns meaning.
//!
//! There is deliberately **no roll-up field**. A group root's own [`State`]
//! *is* its subtree's answer: `Finishing` means "own logic done, children
//! still running", and the terminal states are the rolled-up outcome. A
//! separate field would be a lossier copy of a value already on the wire —
//! lossier because it would have to flatten `Running` and `Finishing` together.
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
pub use hive_jobq::{State, TerminalState};
/// Node id, carried verbatim from `hive_jobq::NodeId`: globally unique across
/// the whole graph, not per group. Opaque to consumers — they group by
/// [`GraphNode::parent`] and match dependency edges, nothing more.
pub type NodeId = u64;
/// One node, serialised near-raw from `hive_jobq::Node`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphNode {
pub id: NodeId,
/// Structural parent, or `None` for a group root.
///
/// A group root is an **ordinary node here** — nothing is hidden, so a
/// consumer needs no special case for "the container", and the root's
/// own `state` answers "how is this whole group doing".
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent: Option<NodeId>,
pub state: State,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub deps: Vec<GraphDep>,
/// When the node entered `Running`. `None` until it starts; a node that
/// never ran keeps `None`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<DateTime<Utc>>,
/// When the node reached a terminal state. `None` while non-terminal.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<DateTime<Utc>>,
/// Failure reason, set only when this node's *own* logic failed — a node
/// that rolled up `Failed` from a child carries none of its own.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub payload: NodePayload,
}
/// What a node *is*, in terms the graph layer does not interpret.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodePayload {
/// Short tag the consumer displays verbatim. **Not** for matching on: a
/// generic viewer that branches on this has stopped being generic.
pub label: String,
/// Domain data, rendered generically (as key/value, a details pane, a
/// tooltip — the consumer's choice). Everything a specific host wants to
/// say about a node beyond its label lives here, so adding a field costs
/// the wire type nothing.
#[serde(default, skip_serializing_if = "serde_json::Value::is_null")]
pub data: serde_json::Value,
}
/// What must hold before a node runs.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum GraphDep {
/// Depend on another node finishing acceptably.
Node {
/// The node depended on. May name a node the consumer has not been
/// sent (a filtered view); treat an absent target as satisfied.
id: NodeId,
/// **The outcomes that satisfy this edge, as a set** — not a
/// strong/weak flag.
///
/// A template routinely emits several tails edged on the *same*
/// upstream node, distinguished only by which outcomes each accepts
/// (one for `Done`, one for `Failed`/`Cancelled`, …). Collapsing this
/// to a boolean renders those as identical nodes.
accepts: Vec<TerminalState>,
},
/// Need `count` units of a named resource. Named, not typed: the resource
/// vocabulary belongs to the host, so it rides as a string.
Resource { name: String, count: u32 },
}
#[cfg(test)]
mod tests {
use super::{GraphDep, GraphNode, NodePayload, State, TerminalState};
fn node(id: u64, parent: Option<u64>, state: State, deps: Vec<GraphDep>) -> GraphNode {
GraphNode {
id,
parent,
state,
deps,
started_at: None,
finished_at: None,
error: None,
payload: NodePayload {
label: "reconcile".to_owned(),
data: serde_json::Value::Null,
},
}
}
#[test]
fn a_node_round_trips_through_json() {
let before = node(
7,
Some(3),
State::Running,
vec![
GraphDep::Node {
id: 3,
accepts: vec![TerminalState::Done],
},
GraphDep::Resource {
name: "build-slot".to_owned(),
count: 1,
},
],
);
let json = serde_json::to_string(&before).expect("serialises");
let after: GraphNode = serde_json::from_str(&json).expect("round trips");
assert_eq!(format!("{after:?}"), format!("{before:?}"));
}
#[test]
fn two_tails_on_one_upstream_are_distinguishable_by_their_accepted_set() {
// The reason `accepts` is a set and not a strong/weak bool: these two
// are the same shape and the same upstream, and *only* the outcome
// set tells them apart.
let ok_tail = node(
10,
None,
State::Pending,
vec![GraphDep::Node {
id: 4,
accepts: vec![TerminalState::Done],
}],
);
let fail_tail = node(
11,
None,
State::Pending,
vec![GraphDep::Node {
id: 4,
accepts: vec![TerminalState::Failed, TerminalState::Cancelled],
}],
);
let render = |n: &GraphNode| match n.deps.first() {
Some(GraphDep::Node { accepts, .. }) => format!("{accepts:?}"),
_ => "none".to_owned(),
};
assert_ne!(render(&ok_tail), render(&fail_tail));
}
#[test]
fn a_group_root_is_an_ordinary_node_and_carries_the_groups_answer() {
// No roll-up field: `Finishing` on the root says "own logic done,
// children still running", and a terminal root state is the roll-up.
let root = node(1, None, State::Finishing, Vec::new());
assert!(root.parent.is_none(), "a group root is just parentless");
assert_eq!(root.state, State::Finishing);
let done_root = node(2, None, State::Failed, Vec::new());
assert_eq!(
done_root.state,
State::Failed,
"the root's own state is the subtree's rolled-up outcome"
);
}
#[test]
fn an_empty_payload_slot_is_omitted_from_the_wire() {
// The opaque slot costs nothing when a host has nothing to say.
let json =
serde_json::to_string(&node(1, None, State::Pending, Vec::new())).expect("serialises");
assert!(!json.contains("\"data\""), "null data is skipped: {json}");
assert!(json.contains("\"label\""), "the label always rides: {json}");
}
}

1
hive-host-sock/src/lib.rs
View file

@ -16,7 +16,6 @@ use hive_sh4re::{AgentStatusRow, Approval};
use hive_types::Ident;
use serde::{Deserialize, Serialize};
pub mod graph;
pub mod jobs;
// ── Shared hive layout facts ──────────────────────────────────────────────

14
hive-jobq-wire/Cargo.toml Normal file
View file

@ -0,0 +1,14 @@
[package]
name = "hive-jobq-wire"
edition.workspace = true
version.workspace = true
readme = "README.md"
[lints]
workspace = true
[dependencies]
chrono = { workspace = true }
hive-jobq = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }

16
hive-jobq-wire/README.md Normal file
View file

@ -0,0 +1,16 @@
# hive-jobq-wire
Wire types for serving a [`hive-jobq`](../hive-jobq) graph to a viewer, plus
the traits a host implements to say how its graph renders.
**Why this is not part of `hive-jobq`.** The scheduler crate is logic: a graph,
a resource pool, a run loop. Presentation is a different concern with a
different audience, and folding it in means every consumer of the scheduler
also carries a JSON vocabulary it may never serve — the two get remixed and
stay that way. A separate crate keeps that boundary where it can be seen.
A host implements `WireNode` for its node payload `N` and `WireResource` for
its resource name `R`. `GraphWire::wire_snapshot` is then blanket-implemented
for `hive_jobq::Graph<N, R>` — so the projection exists exactly when both types
have said how they render, and a payload that hasn't cannot reach a viewer at
all.

411
hive-jobq-wire/src/lib.rs Normal file
View file

@ -0,0 +1,411 @@
//! A [`hive_jobq`] graph, serialised without knowing what its nodes mean.
//!
//! A [`Graph`] is generic over its node payload `N` and resource name `R`, so
//! the scheduler can carry a host's domain types without interpreting them.
//! This crate extends that to the wire: [`GraphNode`] carries what a [`Node`]
//! itself carries — identity, the parent tree, dependency edges, lifecycle —
//! and pushes everything domain-specific into one opaque [`NodePayload::data`]
//! slot the consumer renders without branching on.
//!
//! **Why a separate crate.** `hive-jobq` is logic; this is presentation. Held
//! together they remix, and every consumer of the scheduler ends up carrying a
//! JSON vocabulary it may never serve.
//!
//! A host says how its `N` and `R` render by implementing [`WireNode`] and
//! [`WireResource`]. [`GraphWire`] is then blanket-implemented for any
//! `Graph<N, R>` whose two type parameters do — so the projection exists
//! exactly when both have answered, and a payload that has never said how it
//! displays cannot reach a viewer at all. The alternative (a projection written
//! next to each host's endpoint) lets the second host ship a graph of
//! unlabelled integers.
//!
//! There is deliberately **no roll-up field**. A group root's own [`State`]
//! *is* its subtree's answer: `Finishing` means "own logic done, children still
//! running", and the terminal states are the rolled-up outcome. A separate
//! field would be a lossier copy of a value already on the wire — lossier
//! because it would have to flatten `Running` and `Finishing` together.
//!
//! [`Node`]: hive_jobq::Node
use chrono::{DateTime, Utc};
use hive_jobq::{Dep, Graph, NodeId, State, TerminalState};
use serde::{Deserialize, Serialize};
/// Node id, carried verbatim from [`hive_jobq::NodeId`]: globally unique across
/// the whole graph, not per group. Opaque to consumers — they group by
/// [`GraphNode::parent`] and match dependency edges, nothing more.
pub type WireId = u64;
/// How a node payload `N` describes itself to a generic viewer.
///
/// Half of [`GraphWire`]'s bound, which is the enforcement: no impl, no
/// projection — a payload that has not answered these two questions has no way
/// onto the wire.
pub trait WireNode {
/// Short tag the consumer displays verbatim. **Not** for matching on: a
/// generic viewer that branches on this has stopped being generic.
fn label(&self) -> String;
/// Domain data for this node, rendered generically (key/value, a details
/// pane, a tooltip — the consumer's choice). Return
/// [`serde_json::Value::Null`] when there is nothing to add; it is then
/// omitted from the wire entirely.
///
/// Takes the node's `id` because a host's extra data is not always *in* the
/// payload — it may be keyed on node identity in a side table (a build log,
/// an artifact store) that only the host can resolve.
fn data(&self, id: WireId) -> serde_json::Value;
}
/// How a resource name `R` rides the wire.
///
/// Resource deps are named rather than typed on the wire: the resource
/// vocabulary belongs to the host, and a viewer only ever groups by the string.
pub trait WireResource {
/// Stable display name for this resource.
fn name(&self) -> String;
}
/// One node, serialised near-raw from [`hive_jobq::Node`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphNode {
pub id: WireId,
/// Structural parent, or `None` for a group root.
///
/// A group root is an **ordinary node here** — nothing is hidden, so a
/// consumer needs no special case for "the container", and the root's own
/// `state` answers "how is this whole group doing".
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent: Option<WireId>,
pub state: State,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub deps: Vec<GraphDep>,
/// When the node entered `Running`. `None` until it starts; a node that
/// never ran keeps `None`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<DateTime<Utc>>,
/// When the node reached a terminal state. `None` while non-terminal.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<DateTime<Utc>>,
/// Failure reason, set only when this node's *own* logic failed — a node
/// that rolled up `Failed` from a child carries none of its own.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub payload: NodePayload,
}
/// What a node *is*, in terms the graph layer does not interpret.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodePayload {
/// From [`WireNode::label`].
pub label: String,
/// From [`WireNode::data`]. Everything a specific host wants to say about a
/// node beyond its label lives here, so adding a field costs the wire type
/// nothing.
#[serde(default, skip_serializing_if = "serde_json::Value::is_null")]
pub data: serde_json::Value,
}
/// What must hold before a node runs.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum GraphDep {
/// Depend on another node finishing acceptably.
Node {
/// The node depended on. May name a node the consumer has not been sent
/// (a filtered view); treat an absent target as satisfied.
id: WireId,
/// **The outcomes that satisfy this edge, as a set** — not a
/// strong/weak flag.
///
/// A host routinely emits several tails edged on the *same* upstream
/// node, distinguished only by which outcomes each accepts (one for
/// `Done`, one for `Failed`/`Cancelled`, …). Collapsing this to a
/// boolean renders those as identical nodes.
accepts: Vec<TerminalState>,
},
/// Need `count` units of a named resource, per [`WireResource::name`].
Resource { name: String, count: u32 },
}
/// Serving a [`Graph`] to a viewer.
///
/// Blanket-implemented for every `Graph<N, R>` whose payload and resource types
/// implement [`WireNode`] / [`WireResource`], and for no others — an extension
/// trait rather than an inherent method because the graph lives in the
/// scheduler crate and this projection deliberately does not.
pub trait GraphWire {
/// `roots` and all their descendants as [`GraphNode`]s, ordered by root id
/// with each root immediately followed by its own subtree.
///
/// Takes the roots rather than reading [`Graph::roots`] itself, because
/// *which* groups to show is the host's policy and the graph has no opinion
/// on it — nothing in a `Graph` is ever removed, so a host that keeps
/// finished work bounded must bound it at this call. Pass `graph.roots()`
/// ids to serve everything.
///
/// What is *not* the host's business is picking those roots by matching on
/// `N`: a root is a node with no parent, which is structure. Keying a
/// generic view on one domain's node vocabulary is exactly what this crate
/// exists to avoid.
fn wire_snapshot(&self, roots: impl IntoIterator<Item = NodeId>) -> Vec<GraphNode>;
}
impl<N: WireNode, R: WireResource> GraphWire for Graph<N, R> {
fn wire_snapshot(&self, roots: impl IntoIterator<Item = NodeId>) -> Vec<GraphNode> {
let mut roots: Vec<NodeId> = roots.into_iter().collect();
roots.sort_unstable_by_key(|id| id.get());
roots
.into_iter()
.flat_map(|root| {
self.node(root)
.into_iter()
.chain(self.descendants(root))
.map(wire_node)
})
.collect()
}
}
fn wire_node<N: WireNode, R: WireResource>(node: &hive_jobq::Node<N, R>) -> GraphNode {
let id = node.id.get();
GraphNode {
id,
parent: node.parent.map(NodeId::get),
state: node.state,
deps: node.deps.iter().map(wire_dep).collect(),
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
payload: NodePayload {
label: node.payload.label(),
data: node.payload.data(id),
},
}
}
fn wire_dep<R: WireResource>(dep: &Dep<R>) -> GraphDep {
match dep {
Dep::Node { id, when } => GraphDep::Node {
id: id.get(),
accepts: [
TerminalState::Done,
TerminalState::Failed,
TerminalState::Cancelled,
TerminalState::Skipped,
]
.into_iter()
.filter(|outcome| when.accepts(*outcome))
.collect(),
},
Dep::Resource { name, count } => GraphDep::Resource {
name: name.name(),
count: *count,
},
}
}
#[cfg(test)]
mod tests {
use hive_jobq::Graph;
use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::Scheduler;
use super::{
GraphDep, GraphNode, GraphWire, NodePayload, State, TerminalState, WireId, WireNode,
};
/// A scheduler over the test payloads, built through the same public API a
/// host uses.
///
/// These tests originally reached for `Graph::insert` — which is
/// `pub(crate)` to `hive-jobq`, so they only ever compiled while this
/// module still lived *inside* that crate. That they now have to go through
/// `insert_job` is the point rather than an inconvenience: it proves the
/// projection needs no privileged access to a graph it does not own.
fn sched() -> Scheduler<&'static str, String> {
Scheduler::new(Graph::new(), ResourceTable::new())
}
// The test payloads. `&str` labels itself and has nothing extra to say;
// `String` is its own resource name.
impl WireNode for &str {
fn label(&self) -> String {
(*self).to_owned()
}
fn data(&self, id: WireId) -> serde_json::Value {
// Keyed on identity, not on the payload — the shape a host with a
// side table (build logs, artifacts) needs.
serde_json::json!({ "seen_id": id })
}
}
impl super::WireResource for String {
fn name(&self) -> String {
self.clone()
}
}
/// The bound is the *caller's*: `wire_snapshot` serialises the roots it is
/// handed and their subtrees, and nothing else. Nothing in a `Graph` is
/// ever removed, so a host that couldn't restrict this would have no way to
/// stop the payload growing for its whole uptime.
#[test]
fn only_the_requested_roots_and_their_descendants_ride() {
let mut sched = sched();
let shown = sched
.insert_job(None, |job| {
let root = job.node("dag");
let child = job.node("prebuild").part_of(root);
let grandchild = job.node("swap").part_of(child);
vec![root.guid(), child.guid(), grandchild.guid()]
})
.expect("job inserts");
let evicted = sched
.insert_job(None, |job| vec![job.node("old-dag").guid()])
.expect("job inserts");
let wire = sched.graph().wire_snapshot([shown[0]]);
let ids: Vec<WireId> = wire.iter().map(|n| n.id).collect();
assert_eq!(
ids,
shown.iter().map(|id| id.get()).collect::<Vec<_>>(),
"the root, then its whole subtree"
);
assert!(
!ids.contains(&evicted[0].get()),
"a root the host did not ask for stays off the wire"
);
assert_eq!(wire[0].parent, None, "a group root is parentless");
assert_eq!(wire[1].parent, Some(shown[0].get()));
}
/// `label` / `data` / `name` all come from the host's impls — the graph
/// layer never invents a rendering for a payload it cannot read.
#[test]
fn payload_and_resource_rendering_come_from_the_host_impls() {
let mut sched = sched();
let ids = sched
.insert_job(None, |job| {
let first = job.node("reconcile");
let second = job
.node("swap")
.after_ok(first)
.needs_units("build-slot".to_owned(), 2);
vec![first.guid(), second.guid()]
})
.expect("job inserts");
let (first, second) = (ids[0], ids[1]);
let wire = sched.graph().wire_snapshot([first, second]);
assert_eq!(wire[0].payload.label, "reconcile");
assert_eq!(wire[0].payload.data["seen_id"], first.get());
match &wire[1].deps[..] {
[
GraphDep::Node { id, accepts },
GraphDep::Resource { name, count },
] => {
assert_eq!(*id, first.get());
assert_eq!(accepts, &vec![TerminalState::Done]);
assert_eq!(name, "build-slot", "the host's WireResource::name");
assert_eq!(*count, 2);
}
other => panic!("expected a node dep then a resource dep, got {other:?}"),
}
}
fn node(id: WireId, parent: Option<WireId>, state: State, deps: Vec<GraphDep>) -> GraphNode {
GraphNode {
id,
parent,
state,
deps,
started_at: None,
finished_at: None,
error: None,
payload: NodePayload {
label: "reconcile".to_owned(),
data: serde_json::Value::Null,
},
}
}
#[test]
fn a_node_round_trips_through_json() {
let before = node(
7,
Some(3),
State::Running,
vec![
GraphDep::Node {
id: 3,
accepts: vec![TerminalState::Done],
},
GraphDep::Resource {
name: "build-slot".to_owned(),
count: 1,
},
],
);
let json = serde_json::to_string(&before).expect("serialises");
let after: GraphNode = serde_json::from_str(&json).expect("round trips");
assert_eq!(format!("{after:?}"), format!("{before:?}"));
}
#[test]
fn two_tails_on_one_upstream_are_distinguishable_by_their_accepted_set() {
// The reason `accepts` is a set and not a strong/weak bool: these two
// are the same shape and the same upstream, and *only* the outcome
// set tells them apart.
let ok_tail = node(
10,
None,
State::Pending,
vec![GraphDep::Node {
id: 4,
accepts: vec![TerminalState::Done],
}],
);
let fail_tail = node(
11,
None,
State::Pending,
vec![GraphDep::Node {
id: 4,
accepts: vec![TerminalState::Failed, TerminalState::Cancelled],
}],
);
let render = |n: &GraphNode| match n.deps.first() {
Some(GraphDep::Node { accepts, .. }) => format!("{accepts:?}"),
_ => "none".to_owned(),
};
assert_ne!(render(&ok_tail), render(&fail_tail));
}
#[test]
fn a_group_root_is_an_ordinary_node_and_carries_the_groups_answer() {
// No roll-up field: `Finishing` on the root says "own logic done,
// children still running", and a terminal root state is the roll-up.
let root = node(1, None, State::Finishing, Vec::new());
assert!(root.parent.is_none(), "a group root is just parentless");
assert_eq!(root.state, State::Finishing);
let done_root = node(2, None, State::Failed, Vec::new());
assert_eq!(
done_root.state,
State::Failed,
"the root's own state is the subtree's rolled-up outcome"
);
}
#[test]
fn an_empty_payload_slot_is_omitted_from_the_wire() {
// The opaque slot costs nothing when a host has nothing to say.
let json =
serde_json::to_string(&node(1, None, State::Pending, Vec::new())).expect("serialises");
assert!(!json.contains("\"data\""), "null data is skipped: {json}");
assert!(json.contains("\"label\""), "the label always rides: {json}");
}
}