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 644146c2..93b98034 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", @@ -1741,6 +1742,18 @@ dependencies = [ "uuid", ] +[[package]] +name = "hive-jobq-wire" +version = "0.1.0" +dependencies = [ + "chrono", + "enumflags2", + "hive-jobq", + "serde", + "serde_json", + "utoipa", +] + [[package]] name = "hive-matrix-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 56a33612..ea41f781 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" } @@ -114,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-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/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..ee668f79 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -651,6 +651,26 @@ 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.", + body = Vec), + ), + 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..561d8c67 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -43,6 +43,7 @@ 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; @@ -354,6 +355,23 @@ 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). + /// + /// 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(); + inner.graph().wire_snapshot(visible_roots(&inner)) + } + /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. #[must_use] pub fn snapshot(&self) -> Vec { @@ -561,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 e86a69c3..dbb3fdc0 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -57,3 +57,19 @@ pub enum Resource { /// held across the holder's whole subtree). MetaWindow, } + +/// 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}"), + Resource::MetaWindow => "meta-window".to_owned(), + } + } +} diff --git a/hive-jobq-wire/Cargo.toml b/hive-jobq-wire/Cargo.toml new file mode 100644 index 00000000..eeda67f3 --- /dev/null +++ b/hive-jobq-wire/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "hive-jobq-wire" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +chrono = { workspace = true } +enumflags2 = { workspace = true } +hive-jobq = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +utoipa = { 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..19ef493b --- /dev/null +++ b/hive-jobq-wire/src/lib.rs @@ -0,0 +1,532 @@ +//! 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 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 +/// [`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, ToSchema)] +pub struct GraphNode { + /// Unique across the whole graph, not per group. + 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, + /// 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 + /// 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, ToSchema)] +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. +/// +/// 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 { + /// 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. + /// + /// 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`]. + 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(), + // 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(), + count: *count, + }, + } +} + +#[cfg(test)] +mod tests { + use hive_jobq::Graph; + use hive_jobq::resources::ResourceTable; + use hive_jobq::scheduler::Scheduler; + + use super::{ + 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. + /// + /// 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}"); + } +} 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,