From 792d7cb3048c658029c16d879638ba0dd11e9dce Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 22:45:29 +0200 Subject: [PATCH 1/5] hive-host-sock: a jobq graph wire type that doesn't know what a node is MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `jobs::NodeView` can only ever display hive-c0re's queue. Five of its fields are domain knowledge: `approval_id` is only ever on a `DeployWindow`, `inputs` only on a `MetaLock`, `build_log_id` only on the nix-heavy kinds, `agent` is derived from the payload, and `kind` is a payload tag consumers branch on. A component built against that shape cannot render a second jobq. `graph::GraphNode` is `hive_jobq::Node` with both generics erased: the crate's own field set, with everything domain-specific in one opaque `payload.data` slot the consumer renders without branching on. That is the crate boundary made visible — hive-jobq owns structure, its host owns meaning — and it is the same split #2957 drew inside the code. Two details that are easy to get wrong and are pinned by tests: `GraphDep::Node` carries `accepts` as the **set** of terminal outcomes, not a strong/weak flag. A template emits its tails as a pair edged on the same upstream node, and the only thing telling them apart is which outcomes each accepts; collapsing that renders two structurally different nodes identically. There is **no roll-up field**. A group root ships as an ordinary node with `parent: None`, and its own `state` is its subtree's answer — `Finishing` means "own logic done, children still running", the terminal states are the rolled-up outcome. A separate field would be a lossier copy: `DagView::rollup_state` flattens `Running` and `Finishing` into one, which is exactly the distinction a viewer wants. `State` and `TerminalState` are re-exported from `hive-jobq` rather than redeclared, so they cannot drift from the scheduler that produces them. --- Cargo.lock | 1 + hive-host-sock/Cargo.toml | 1 + hive-host-sock/src/graph.rs | 192 ++++++++++++++++++++++++++++++++++++ hive-host-sock/src/lib.rs | 1 + 4 files changed, 195 insertions(+) create mode 100644 hive-host-sock/src/graph.rs diff --git a/Cargo.lock b/Cargo.lock index 644146c2..00118c83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1727,6 +1727,7 @@ dependencies = [ "hive-sh4re", "hive-types", "serde", + "serde_json", ] [[package]] diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index ffd069fd..425d890d 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -13,3 +13,4 @@ hive-jobq.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true +serde_json.workspace = true diff --git a/hive-host-sock/src/graph.rs b/hive-host-sock/src/graph.rs new file mode 100644 index 00000000..d98853ed --- /dev/null +++ b/hive-host-sock/src/graph.rs @@ -0,0 +1,192 @@ +//! 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, + pub state: State, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub deps: Vec, + /// 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>, + /// When the node reached a terminal state. `None` while non-terminal. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option>, + /// 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, + 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, + }, + /// 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, state: State, deps: Vec) -> 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}"); + } +} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 2363e3db..8524d14f 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -16,6 +16,7 @@ use hive_sh4re::{AgentStatusRow, Approval}; use hive_types::Ident; use serde::{Deserialize, Serialize}; +pub mod graph; pub mod jobs; // ── Shared hive layout facts ────────────────────────────────────────────── From f357f988679200509d28459a71d4c2ec29497df7 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 22:59:44 +0200 Subject: [PATCH 2/5] c0re: serve the jobq graph generically at /api/jobq/graph MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `graph_node` projects a `hive_jobq::Node` onto the wire type from the parent commit: everything the crate records, with hive-c0re's own fields (`agent`, `approval_id`, `inputs`, `build_log_id`) collected into the opaque payload slot instead of standing as named columns. A node with nothing domain-specific to say serialises no `data` key at all, so the slot costs nothing when it is unused. `graph_dep` turns a `DepWhen` into the set of outcomes that satisfy the edge by asking it about each of the four terminal states, rather than leaking the bitflags representation onto the wire. `Resource::wire_name` gives the resource vocabulary a string form — hive-jobq is generic over the resource type, so a viewer that renders any graph cannot be handed this enum. The `agent:` prefix keeps per-agent leases from colliding with a global resource sharing an agent's name. `graph_snapshot` deliberately reuses `visible_dags` for retention: every live group plus the newest terminal ones. Serving the raw graph would grow without bound — evicted groups' nodes linger until bounded pruning lands. Within a retained group nothing is filtered: group roots ride as ordinary nodes, and `Done` nodes stay, which is the projection defect behind the "rebuild shows a single node" report. The endpoint lands in the same commit rather than after it. Without a consumer the whole projection is dead code, and a wire type nobody produces cannot be reviewed for whether it says the right things. Its OpenAPI body is `serde_json::Value`, matching `api_state`: no type in `hive-host-sock` derives `ToSchema`, and that crate stays dependency-lean on purpose. --- hive-c0re/src/dashboard/mod.rs | 4 +- hive-c0re/src/dashboard/state_snapshot.rs | 21 +++++ hive-c0re/src/job_queue/mod.rs | 104 ++++++++++++++++++++++ hive-c0re/src/job_queue/resource.rs | 17 ++++ 4 files changed, 145 insertions(+), 1 deletion(-) diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index f3b3c563..edc20408 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -205,6 +205,7 @@ pub async fn serve( .routes(routes!(build_logs::get_build_log_stream)) .routes(routes!(state_snapshot::dashboard_stream)) .routes(routes!(state_snapshot::dashboard_history)) + .routes(routes!(state_snapshot::jobq_graph)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from // the nix store (see the module doc comment above `ApiDoc`). `api` @@ -448,6 +449,7 @@ mod router_build_probe { .routes(routes!(meta_inputs::post_meta_update)) .routes(routes!(build_logs::get_build_log_stream)) .routes(routes!(state_snapshot::dashboard_stream)) - .routes(routes!(state_snapshot::dashboard_history)); + .routes(routes!(state_snapshot::dashboard_history)) + .routes(routes!(state_snapshot::jobq_graph)); } } diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index ad9b8362..09763b60 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -651,6 +651,27 @@ fn build_approval_views(approvals: Vec) -> Vec { out } +#[utoipa::path( + get, + path = "/api/jobq/graph", + responses( + (status = 200, description = "every node of every retained job group, \ + as generic `hive_jobq` graph nodes: identity, the parent tree, \ + dependency edges with their accepted-outcome sets, lifecycle, and \ + 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`", + body = serde_json::Value), + ), + tag = "state_snapshot" +)] +pub(super) async fn jobq_graph( + State(state): State, +) -> axum::Json> { + axum::Json(state.coord.job_queue.graph_snapshot()) +} + #[utoipa::path( get, path = "/api/dashboard/history", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 597a564f..f42da480 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -39,6 +39,7 @@ 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}; @@ -354,6 +355,36 @@ impl JobQueue { .collect() } + /// Every node of every visible group, as generic graph nodes. + /// + /// **Nothing is hidden.** Group roots ride as ordinary nodes (so a + /// consumer needs no special case for "the container" and reads the + /// root's own `state` as the group's answer), and `Done` nodes stay (so a + /// 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. + #[must_use] + pub fn graph_snapshot(&self) -> Vec { + 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() + } + /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. #[must_use] pub fn snapshot(&self) -> Vec { @@ -488,6 +519,79 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option { }) } +/// 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) -> 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) -> 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. diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index e86a69c3..c8500983 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -57,3 +57,20 @@ pub enum Resource { /// held across the holder's whole subtree). 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 { + match self { + Resource::BuildSlot => "build-slot".to_owned(), + Resource::Agent(agent) => format!("agent:{agent}"), + Resource::MetaWindow => "meta-window".to_owned(), + } + } +} From 7966d5eb660ca21311a5f1abb8444b153968a773 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 23:52:32 +0200 Subject: [PATCH 3/5] jobq-wire: move the generic graph projection into its own crate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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. --- CLAUDE.md | 7 + Cargo.lock | 12 +- Cargo.toml | 2 + hive-c0re/Cargo.toml | 1 + hive-c0re/src/dashboard/state_snapshot.rs | 4 +- hive-c0re/src/job_queue/mod.rs | 145 +++----- hive-c0re/src/job_queue/model.rs | 38 ++ hive-c0re/src/job_queue/resource.rs | 17 +- hive-host-sock/Cargo.toml | 1 - hive-host-sock/src/graph.rs | 192 ---------- hive-host-sock/src/lib.rs | 1 - hive-jobq-wire/Cargo.toml | 14 + hive-jobq-wire/README.md | 16 + hive-jobq-wire/src/lib.rs | 411 ++++++++++++++++++++++ 14 files changed, 564 insertions(+), 297 deletions(-) delete mode 100644 hive-host-sock/src/graph.rs create mode 100644 hive-jobq-wire/Cargo.toml create mode 100644 hive-jobq-wire/README.md create mode 100644 hive-jobq-wire/src/lib.rs diff --git a/CLAUDE.md b/CLAUDE.md index b88d9d3b..20201153 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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` 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 diff --git a/Cargo.lock b/Cargo.lock index 00118c83..ddbeb5ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index 56a33612..50181767 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index b87de707..54e97769 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -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 diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 09763b60..f6c58682 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -661,14 +661,14 @@ fn build_approval_views(approvals: Vec) -> Vec { 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, -) -> axum::Json> { +) -> axum::Json> { axum::Json(state.coord.job_queue.graph_snapshot()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f42da480..561d8c67 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -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 { 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 { }) } -/// 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) -> 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) -> 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 { 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 { + let roots: Vec = sched.graph().roots().map(|n| n.id).collect(); + let mut live: Vec = 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. /// diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index b95f89e8..263ccf2f 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -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 { diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index c8500983..dbb3fdc0 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -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}"), diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index 425d890d..ffd069fd 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -13,4 +13,3 @@ hive-jobq.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true -serde_json.workspace = true diff --git a/hive-host-sock/src/graph.rs b/hive-host-sock/src/graph.rs deleted file mode 100644 index d98853ed..00000000 --- a/hive-host-sock/src/graph.rs +++ /dev/null @@ -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, - pub state: State, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub deps: Vec, - /// 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>, - /// When the node reached a terminal state. `None` while non-terminal. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option>, - /// 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, - 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, - }, - /// 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, state: State, deps: Vec) -> 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}"); - } -} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 8524d14f..2363e3db 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -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 ────────────────────────────────────────────── diff --git a/hive-jobq-wire/Cargo.toml b/hive-jobq-wire/Cargo.toml new file mode 100644 index 00000000..6a0606fe --- /dev/null +++ b/hive-jobq-wire/Cargo.toml @@ -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 } diff --git a/hive-jobq-wire/README.md b/hive-jobq-wire/README.md new file mode 100644 index 00000000..78031a4b --- /dev/null +++ b/hive-jobq-wire/README.md @@ -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` — 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. diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs new file mode 100644 index 00000000..fcd18724 --- /dev/null +++ b/hive-jobq-wire/src/lib.rs @@ -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` 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, + pub state: State, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub deps: Vec, + /// 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>, + /// When the node reached a terminal state. `None` while non-terminal. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option>, + /// 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, + 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, + }, + /// 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` 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) -> Vec; +} + +impl GraphWire for Graph { + fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec { + let mut roots: Vec = 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(node: &hive_jobq::Node) -> 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(dep: &Dep) -> 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 = wire.iter().map(|n| n.id).collect(); + assert_eq!( + ids, + shown.iter().map(|id| id.get()).collect::>(), + "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, state: State, deps: Vec) -> 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}"); + } +} From fe0906c0439c780cd021b59fc75fb9eca16a5e64 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 00:14:37 +0200 Subject: [PATCH 4/5] jobq-wire: name every outcome, keep the enum spellings, document the schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things, all from review: Accepted outcomes were built from a hand-listed [Done, Failed, Cancelled, Skipped] array. Exhaustive today, silently short the day someone adds a variant — the new outcome would vanish from every edge that accepts it. BitFlags::ALL asks the type instead. TerminalState carried rename_all = "snake_case" while its sibling State did not, so one enum shipped "done" and the other "Done". A rename is a second spelling of a name that then has to be kept in agreement by hand; both now serialise their variant names verbatim. Nothing else reads TerminalState off a wire, so no consumer moves. GraphDep's tag values likewise. The endpoint documented its body as serde_json::Value, which tells a spec reader nothing. hive-jobq-wire now derives ToSchema. State and TerminalState are foreign types here and utoipa stays out of the scheduler crate, so the schema points at local mirror enums. A mirror that drifts is worse than none: the conversions are exhaustive (a new upstream variant fails the build) and a test asserts each documented name equals the serialised one, since an exhaustive match still compiles when only the spellings diverge. --- Cargo.lock | 2 + hive-c0re/src/dashboard/state_snapshot.rs | 5 +- hive-jobq-wire/Cargo.toml | 4 + hive-jobq-wire/src/lib.rs | 149 ++++++++++++++++++++-- hive-jobq/src/lib.rs | 6 +- 5 files changed, 148 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ddbeb5ec..93b98034 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1747,9 +1747,11 @@ name = "hive-jobq-wire" version = "0.1.0" dependencies = [ "chrono", + "enumflags2", "hive-jobq", "serde", "serde_json", + "utoipa", ] [[package]] diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index f6c58682..ee668f79 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -660,9 +660,8 @@ fn build_approval_views(approvals: Vec) -> Vec { dependency edges with their accepted-outcome sets, lifecycle, and \ 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_jobq_wire::GraphNode`", - body = serde_json::Value), + renders the graph without knowing what any node means.", + body = Vec), ), tag = "state_snapshot" )] diff --git a/hive-jobq-wire/Cargo.toml b/hive-jobq-wire/Cargo.toml index 6a0606fe..b2605877 100644 --- a/hive-jobq-wire/Cargo.toml +++ b/hive-jobq-wire/Cargo.toml @@ -9,6 +9,10 @@ workspace = true [dependencies] chrono = { workspace = true } +enumflags2 = { workspace = true } hive-jobq = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +# `chrono` for the lifecycle timestamps' schema; the workspace default carries +# only `axum_extras`, and features are additive so this affects nothing else. +utoipa = { workspace = true, features = ["chrono"] } diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs index fcd18724..19ef493b 100644 --- a/hive-jobq-wire/src/lib.rs +++ b/hive-jobq-wire/src/lib.rs @@ -28,8 +28,74 @@ //! [`Node`]: hive_jobq::Node use chrono::{DateTime, Utc}; +use enumflags2::BitFlags; use hive_jobq::{Dep, Graph, NodeId, State, TerminalState}; use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +/// `OpenAPI` mirror of [`hive_jobq::State`]. +/// +/// Exists **only** so the generated spec can enumerate the states: `State` is +/// a foreign type, so neither `ToSchema` nor a newtype around it can be +/// implemented here, and `utoipa` is deliberately not a dependency of the +/// scheduler crate. Referenced via `#[schema(value_type = …)]`; nothing is ever +/// serialised through it, so it cannot change the wire. +/// +/// [`StateSchema::of`] is what keeps it honest — an exhaustive match, so adding +/// a variant upstream **fails to compile here** instead of quietly leaving the +/// documented enum short. +#[derive(Debug, Clone, Copy, ToSchema)] +pub enum StateSchema { + Pending, + Running, + Finishing, + Done, + Failed, + Cancelled, + Skipped, +} + +impl StateSchema { + /// The mirror variant for `state`. Public because the mirror type is: a + /// consumer that renders the documented enum needs the same mapping, and a + /// constructor nobody outside can call is a schema nobody can check. + #[must_use] + pub fn of(state: State) -> Self { + match state { + State::Pending => Self::Pending, + State::Running => Self::Running, + State::Finishing => Self::Finishing, + State::Done => Self::Done, + State::Failed => Self::Failed, + State::Cancelled => Self::Cancelled, + State::Skipped => Self::Skipped, + } + } +} + +/// `OpenAPI` mirror of [`hive_jobq::TerminalState`] — same rationale and same +/// exhaustive-match guard as [`StateSchema`]. +#[derive(Debug, Clone, Copy, ToSchema)] +pub enum TerminalStateSchema { + Done, + Failed, + Cancelled, + Skipped, +} + +impl TerminalStateSchema { + /// The mirror variant for `outcome`. Public for the same reason as + /// [`StateSchema::of`]. + #[must_use] + pub fn of(outcome: TerminalState) -> Self { + match outcome { + TerminalState::Done => Self::Done, + TerminalState::Failed => Self::Failed, + TerminalState::Cancelled => Self::Cancelled, + TerminalState::Skipped => Self::Skipped, + } + } +} /// Node id, carried verbatim from [`hive_jobq::NodeId`]: globally unique across /// the whole graph, not per group. Opaque to consumers — they group by @@ -67,8 +133,9 @@ pub trait WireResource { } /// One node, serialised near-raw from [`hive_jobq::Node`]. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct GraphNode { + /// Unique across the whole graph, not per group. pub id: WireId, /// Structural parent, or `None` for a group root. /// @@ -77,7 +144,12 @@ pub struct GraphNode { /// `state` answers "how is this whole group doing". #[serde(default, skip_serializing_if = "Option::is_none")] pub parent: Option, + /// Lifecycle state. On a group root this is also the subtree's answer: + /// `Finishing` = own logic done, children still running; a terminal value + /// is the rolled-up outcome. + #[schema(value_type = StateSchema)] pub state: State, + /// What must hold before this node runs. Omitted when empty. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub deps: Vec, /// When the node entered `Running`. `None` until it starts; a node that @@ -95,7 +167,7 @@ pub struct GraphNode { } /// What a node *is*, in terms the graph layer does not interpret. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct NodePayload { /// From [`WireNode::label`]. pub label: String, @@ -107,8 +179,12 @@ pub struct NodePayload { } /// What must hold before a node runs. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "snake_case", tag = "kind")] +/// +/// Externally tagged on `kind`, whose values are the **variant names verbatim** +/// (`"Node"`, `"Resource"`) — no `rename_all`. A rename is a second spelling of +/// the same name that has to be kept in agreement with the Rust one by hand. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +#[serde(tag = "kind")] pub enum GraphDep { /// Depend on another node finishing acceptably. Node { @@ -122,6 +198,10 @@ pub enum GraphDep { /// 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. + /// + /// Every accepted outcome is **named**, so a client never shifts bits + /// to read an edge. + #[schema(value_type = Vec)] accepts: Vec, }, /// Need `count` units of a named resource, per [`WireResource::name`]. @@ -188,15 +268,14 @@ fn wire_dep(dep: &Dep) -> 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(), + // Every outcome the type knows about, not a hand-listed set: a + // variant added to `TerminalState` is then named on the wire + // automatically, where a literal array would have silently + // dropped it from every edge that accepts it. + accepts: BitFlags::::ALL + .iter() + .filter(|outcome| when.accepts(*outcome)) + .collect(), }, Dep::Resource { name, count } => GraphDep::Resource { name: name.name(), @@ -212,9 +291,51 @@ mod tests { use hive_jobq::scheduler::Scheduler; use super::{ - GraphDep, GraphNode, GraphWire, NodePayload, State, TerminalState, WireId, WireNode, + BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, TerminalState, + TerminalStateSchema, WireId, WireNode, }; + /// The mirrors document what the wire actually says — a mirror that + /// disagreed would be worse than none, since the spec is all a consumer + /// has. + /// + /// `StateSchema::of` is the *compile-time* half (exhaustive, so a new + /// upstream variant breaks the build). This is the *value* half: the + /// documented variant name has to equal the serialised one. Both are + /// needed — an exhaustive match still compiles if the names diverge, which + /// is exactly what a stray `rename_all` upstream would do. + #[test] + fn the_openapi_mirrors_name_states_exactly_as_the_wire_does() { + let states = [ + State::Pending, + State::Running, + State::Finishing, + State::Done, + State::Failed, + State::Cancelled, + State::Skipped, + ]; + for state in states { + let on_the_wire = serde_json::to_string(&state).expect("serialises"); + let documented = format!("{:?}", StateSchema::of(state)); + assert_eq!( + on_the_wire, + format!("\"{documented}\""), + "{state:?} is documented as {documented} but serialises as {on_the_wire}" + ); + } + + for outcome in BitFlags::::ALL { + let on_the_wire = serde_json::to_string(&outcome).expect("serialises"); + let documented = format!("{:?}", TerminalStateSchema::of(outcome)); + assert_eq!( + on_the_wire, + format!("\"{documented}\""), + "{outcome:?} is documented as {documented} but serialises as {on_the_wire}" + ); + } + } + /// A scheduler over the test payloads, built through the same public API a /// host uses. /// diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 838c0c29..cccd55a1 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -62,10 +62,14 @@ impl NodeId { /// How a node finished. The terminal subset of [`State`], as its own type so an /// edge condition cannot name `Pending` / `Running` / `Finishing` — those are /// meaningless in a dependency and are better unrepresentable than rejected. +/// +/// Serialises as its **variant name verbatim** (`"Done"`, `"Cancelled"`), with +/// no `rename_all` transformation — matching [`State`], whose names the +/// dashboard already matches on. A rename is a second spelling of the same +/// value that has to be kept in agreement by hand. #[enumflags2::bitflags] #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "snake_case")] pub enum TerminalState { /// Own logic succeeded and every sub-node did too. Done, From b8627e7e9420e155336105e1f116af38fda1ed81 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 00:17:52 +0200 Subject: [PATCH 5/5] jobq-wire: take utoipa's chrono feature from the workspace The per-crate features = ["chrono"] override was a local deviation with a comment defending it. The feature belongs in the workspace list, where every crate sees the same utoipa. --- Cargo.toml | 2 +- hive-jobq-wire/Cargo.toml | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 50181767..ea41f781 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -116,5 +116,5 @@ matrix-sdk = { version = "0.18", default-features = false, features = [ futures-util = "0.3" hmac = "0.13" sha2 = "0.11" -utoipa = { version = "5", features = ["axum_extras"] } +utoipa = { version = "5", features = ["axum_extras", "chrono"] } utoipa-axum = "0.2" diff --git a/hive-jobq-wire/Cargo.toml b/hive-jobq-wire/Cargo.toml index b2605877..eeda67f3 100644 --- a/hive-jobq-wire/Cargo.toml +++ b/hive-jobq-wire/Cargo.toml @@ -13,6 +13,4 @@ enumflags2 = { workspace = true } hive-jobq = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -# `chrono` for the lifecycle timestamps' schema; the workspace default carries -# only `axum_extras`, and features are additive so this affects nothing else. -utoipa = { workspace = true, features = ["chrono"] } +utoipa = { workspace = true }