Cargo.lock

@@ -4563,8 +4563,6 @@ dependencies = [
  "async-nats",
  "axum",
  "futures-util",
- "hive-jobq",
- "hive-jobq-wire",
  "serde",
  "serde_json",
  "swarm-queue-client",

hive-c0re/src/dashboard/state_snapshot.rs

@@ -580,23 +580,35 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
     out
 }
 
-/// `/api/jobq/graph` query string — the generic jobq/wire query shape (any
-/// host serving a `GraphWire` projection over HTTP takes the same one; see
-/// `hive_jobq_wire::parse_states`, which does the actual parsing this side
-/// of the query string). Today's only field is `states`: a comma-separated
-/// allow-list of `hive_jobq::State` names (`"Pending"`, `"Running"`, ...).
-/// Empty / absent ⇒ no filter (current behaviour, every visible root). Set
-/// ⇒ only **root** groups whose own state is named are served — a root's
-/// state is already its subtree's rolled-up answer (see `hive_jobq_wire`'s
-/// doc), so filtering the root filters the whole group. Unknown tokens are
-/// silently ignored (an unrecognised name matches nothing rather than
-/// erroring the whole request), mirroring `DashboardStreamQuery::kinds`
-/// above.
+/// `/api/jobq/graph` query string. Today's only field is `states`: a
+/// comma-separated allow-list of `hive_jobq::State` names (`"Pending"`,
+/// `"Running"`, ...). Empty / absent ⇒ no filter (current behaviour, every
+/// visible root). Set ⇒ only **root** groups whose own state is named are
+/// served — a root's state is already its subtree's rolled-up answer (see
+/// `hive_jobq_wire`'s doc), so filtering the root filters the whole group.
+/// Unknown tokens are silently ignored (an unrecognised name matches
+/// nothing rather than erroring the whole request), mirroring
+/// `DashboardStreamQuery::kinds` above.
 #[derive(Deserialize, Default, IntoParams)]
 pub(super) struct JobqGraphQuery {
     states: Option<String>,
 }
 
+/// Parses [`JobqGraphQuery::states`] into the list
+/// [`crate::job_queue::JobQueue::graph_snapshot`] wants. `None` when absent
+/// or when every token failed to parse — both mean "no filter" rather than
+/// "match nothing", so an empty/garbled query reads as the unfiltered call
+/// it replaces rather than an empty result set.
+fn parse_states(raw: Option<&str>) -> Option<Vec<hive_jobq::State>> {
+    let states: Vec<hive_jobq::State> = raw?
+        .split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .filter_map(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
+        .collect();
+    (!states.is_empty()).then_some(states)
+}
+
 #[utoipa::path(
     get,
     path = "/api/jobq/graph",
@@ -605,14 +617,10 @@ pub(super) struct JobqGraphQuery {
         (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. This is the generic jobq/wire shape \
-            (`hive_jobq_wire::GraphWire::wire_snapshot`), not a hive-c0re-only \
-            projection — a `swarm-controller` serving its own graph responds \
-            with the same shape at the same path. Group roots ride as \
-            ordinary nodes (`parent: null`) and `Done` nodes are not \
-            filtered by default — a consumer renders the graph without \
-            knowing what any node means. `?states=` narrows to root groups \
-            in the named states.",
+            one opaque per-node payload. Group roots ride as ordinary nodes \
+            (`parent: null`) and `Done` nodes are not filtered by default — a \
+            consumer renders the graph without knowing what any node means. \
+            `?states=` narrows to root groups in the named states.",
             body = Vec<hive_jobq_wire::GraphNode>),
     ),
     tag = "state_snapshot"
@@ -621,7 +629,7 @@ pub(super) async fn jobq_graph(
     State(state): State<AppState>,
     axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
 ) -> axum::Json<Vec<hive_jobq_wire::GraphNode>> {
-    let states = hive_jobq_wire::parse_states(q.states.as_deref());
+    let states = parse_states(q.states.as_deref());
     axum::Json(state.coord.job_queue.graph_snapshot(states.as_deref()))
 }
 
@@ -631,9 +639,7 @@ pub(super) async fn jobq_graph(
     responses(
         (status = 200, description = "counts by lifecycle state over the same \
             groups `/api/jobq/graph` serves, as `(state, nodes, roots)` \
-            triples — the generic jobq/wire roll-up shape \
-            (`hive_jobq_wire::state_rollup`), same as `/api/jobq/graph` \
-            above. Every state is present, zero counts included, in a fixed \
+            triples. Every state is present, zero counts included, in a fixed \
             order — a consumer renders a summary (\"3 running · 2 queued\") \
             without fetching the graph and without re-deriving the tally. \
             `roots` counts groups, `nodes` counts every step at any depth: one \

hive-c0re/src/job_queue/mod.rs

@@ -351,16 +351,15 @@ impl JobQueue {
     /// still-matching descendant one level higher rather than hiding or
     /// orphaning it.
     ///
-    /// The projection and state filter are both [`hive_jobq_wire`]'s; this
-    /// layer only supplies *which roots* are in view (see [`visible_roots`]
-    /// for why the graph can't decide that itself) — an older, different
-    /// question than state filtering, and neither replaces the other.
+    /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies
+    /// is *which* nodes to show — see [`visible_roots`] for why the graph
+    /// can't decide the root-visibility half of that for itself.
     #[must_use]
     pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec<GraphNode> {
         let inner = self.lock();
         let roots = visible_roots(&inner);
         let nodes = inner.graph().wire_snapshot(roots);
-        hive_jobq_wire::filter_nodes_by_state(nodes, states)
+        filter_nodes_by_state(nodes, states)
     }
 
     /// Per-state counts over the **same** groups [`JobQueue::graph_snapshot`]
@@ -415,6 +414,26 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
         .find_map(|n| (n.id.get() == id).then_some(n.id))
 }
 
+/// [`JobQueue::graph_snapshot`]'s `states` ask, applied to the already-
+/// projected node list: keeps every node — root or descendant — whose own
+/// `state` is named.
+///
+/// Applied *after* [`GraphWire::wire_snapshot`] rather than as a root
+/// pre-filter — narrowing which roots are visible at all is
+/// [`visible_roots`]'s job (a different question: how much settled work is
+/// retained, full stop); this is "of what's retained and live, which
+/// individual nodes does the caller want shown right now." `None` (or an
+/// unrecognised/absent query) is the identity filter.
+fn filter_nodes_by_state(nodes: Vec<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
+    let Some(states) = states else {
+        return nodes;
+    };
+    nodes
+        .into_iter()
+        .filter(|n| states.contains(&n.state))
+        .collect()
+}
+
 /// The visible **group** set for [`JobQueue::graph_snapshot`]: every live group
 /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
 ///

hive-c0re/src/job_queue/tests.rs

@@ -1527,6 +1527,60 @@ fn graph_snapshot_states_filters_by_node_state() {
     );
 }
 
+/// One hand-built node, bypassing the scheduler entirely — `filter_nodes_by_state`
+/// is a pure `Vec<GraphNode> -> Vec<GraphNode>` transform, so this exercises it
+/// directly rather than trying (and failing, per this module's own rule) to
+/// drive a live group into a genuinely mixed state through the real queue.
+fn node(id: u64, parent: Option<u64>, state: State) -> GraphNode {
+    GraphNode {
+        id,
+        parent,
+        state,
+        deps: Vec::new(),
+        created_at: Utc::now(),
+        started_at: None,
+        finished_at: None,
+        error: None,
+        payload: hive_jobq_wire::NodePayload {
+            label: id.to_string(),
+            data: serde_json::Value::Null,
+        },
+    }
+}
+
+/// The case `graph_snapshot_states_filters_by_node_state` above can't reach:
+/// a still-live group (root `Running`) holding a mix of already-`Done` and
+/// still-`Pending` steps. Filtering out `Done` must drop exactly the `Done`
+/// node and nothing else — the root and the `Pending` sibling both stay,
+/// even though the root itself isn't in the requested state set.
+#[test]
+fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() {
+    let nodes = vec![
+        node(1, None, State::Running),    // root: whole group still live
+        node(2, Some(1), State::Done),    // finished step, should be hidden
+        node(3, Some(1), State::Pending), // not-yet-run step, should stay
+    ];
+
+    let mut kept: Vec<u64> =
+        filter_nodes_by_state(nodes.clone(), Some(&[State::Running, State::Pending]))
+            .into_iter()
+            .map(|n| n.id)
+            .collect();
+    kept.sort_unstable();
+    assert_eq!(
+        kept,
+        vec![1, 3],
+        "Done is filtered out even though its still-live parent (root) isn't"
+    );
+
+    let mut unfiltered: Vec<u64> = filter_nodes_by_state(nodes, None)
+        .into_iter()
+        .map(|n| n.id)
+        .collect();
+    unfiltered.sort_unstable();
+    assert_eq!(unfiltered, vec![1, 2, 3], "None is the identity filter");
+}
+
 #[test]
 fn error_truncation_cuts_on_a_char_boundary() {
     // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,

hive-jobq-wire/src/lib.rs

@@ -354,46 +354,6 @@ fn state_index(state: State) -> usize {
     }
 }
 
-/// Parses a `?states=` query value (comma-separated [`State`] names, e.g.
-/// `"Running,Pending"`) into the filter [`filter_nodes_by_state`] wants.
-///
-/// Absent or entirely-unparseable input is "no filter," never "match
-/// nothing" — an empty/garbled query reads as the unfiltered call it
-/// replaces rather than as a request for zero results. Unrecognised tokens
-/// are silently dropped rather than erroring the whole request.
-///
-/// Lives here rather than in each host's own dashboard/API layer because
-/// every host that serves [`GraphWire::wire_snapshot`] over HTTP wants the
-/// same query shape — a second, independently-typed copy is exactly the
-/// kind of drift this crate exists to prevent (see the module doc comment).
-#[must_use]
-pub fn parse_states(raw: Option<&str>) -> Option<Vec<State>> {
-    let states: Vec<State> = raw?
-        .split(',')
-        .map(str::trim)
-        .filter(|s| !s.is_empty())
-        .filter_map(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
-        .collect();
-    (!states.is_empty()).then_some(states)
-}
-
-/// Keeps only the nodes whose own `state` is named in `states` — a flat
-/// filter on [`GraphNode::state`], no special-casing for a root vs. a
-/// descendant (a root's own state already *is* its subtree's rolled-up
-/// answer, per the module doc comment, so filtering the root filters the
-/// whole group). `None` is a no-op, not "match nothing," mirroring
-/// [`parse_states`]'s own absent-input rule.
-#[must_use]
-pub fn filter_nodes_by_state(nodes: Vec<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
-    let Some(states) = states else {
-        return nodes;
-    };
-    nodes
-        .into_iter()
-        .filter(|n| states.contains(&n.state))
-        .collect()
-}
-
 fn wire_node<N: WireNode, R: WireResource>(node: &hive_jobq::Node<N, R>) -> GraphNode {
     let id = node.id.get();
     GraphNode {
@@ -440,8 +400,7 @@ mod tests {
 
     use super::{
         ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema,
-        TerminalState, TerminalStateSchema, WireId, WireNode, filter_nodes_by_state, parse_states,
-        state_rollup,
+        TerminalState, TerminalStateSchema, WireId, WireNode, state_rollup,
     };
 
     /// The mirrors document what the wire actually says — a mirror that
@@ -671,50 +630,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn parse_states_covers_absent_malformed_and_valid() {
-        assert_eq!(parse_states(None), None, "no query is no filter");
-        assert_eq!(
-            parse_states(Some("")),
-            None,
-            "an empty query is no filter, not a request for zero results"
-        );
-        assert_eq!(
-            parse_states(Some("not-a-state,also-not-one")),
-            None,
-            "every token failing to parse is still no filter"
-        );
-        assert_eq!(
-            parse_states(Some(" Running , Pending ,bogus")),
-            Some(vec![State::Running, State::Pending]),
-            "trims whitespace, keeps the valid tokens, drops the unrecognised one"
-        );
-    }
-
-    #[test]
-    fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() {
-        let nodes = vec![
-            node(1, None, State::Running, Vec::new()), // root: whole group still live
-            node(2, Some(1), State::Done, Vec::new()), // finished step, should be hidden
-            node(3, Some(1), State::Pending, Vec::new()), // not-yet-run step, should stay
-        ];
-
-        let mut kept: Vec<WireId> =
-            filter_nodes_by_state(nodes.clone(), Some(&[State::Running, State::Pending]))
-                .into_iter()
-                .map(|n| n.id)
-                .collect();
-        kept.sort_unstable();
-        assert_eq!(
-            kept,
-            vec![1, 3],
-            "Done is filtered out even though its still-live parent (root) isn't"
-        );
-
-        let unfiltered = filter_nodes_by_state(nodes, None);
-        assert_eq!(unfiltered.len(), 3, "no states given is a no-op, not empty");
-    }
-
     #[test]
     fn an_empty_payload_slot_is_omitted_from_the_wire() {
         // The opaque slot costs nothing when a host has nothing to say.

swarm-controller/Cargo.toml

@@ -18,14 +18,6 @@ anyhow.workspace = true
 async-nats = { workspace = true, features = ["kv"] }
 axum.workspace = true
 futures-util.workspace = true
-# The graph itself, held directly rather than behind a c0re-style wrapper
-# module — that layering (`hive-c0re::job_queue`) is partially legacy (predates
-# `hive-jobq`'s extraction into its own crate) and this daemon does not need it
-# repeated. No `Scheduler` yet either: nothing here submits or executes a job,
-# so there is nothing to schedule — just a `Graph` for the read-only endpoints
-# to serve.
-hive-jobq.workspace = true
-hive-jobq-wire.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 # The queue connect (token mint + auth callback + reconnect) is shared with

swarm-controller/src/main.rs

@@ -23,49 +23,16 @@
 
 use std::os::unix::fs::PermissionsExt as _;
 use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use axum::{Json, extract::State, routing::get};
-use hive_jobq_wire::GraphWire as _;
 use serde::{Deserialize, Serialize};
 use utoipa::{OpenApi, ToSchema};
 use utoipa_axum::{router::OpenApiRouter, routes};
 
 mod status;
 
-/// Placeholder node payload for the swarm-level job graph — uninhabited on
-/// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource`
-/// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't
-/// land on both crates. This wires the graph and its read-only endpoints;
-/// giving it real variants waits on there being an actual job (agent
-/// creation) to run. `WireNode` is trivially satisfiable on an empty enum
-/// (`match *self {}`), so the wire machinery below is real and typechecked
-/// today, with nothing yet to put in it.
-#[derive(Clone, Debug)]
-enum SwarmNodeKind {}
-
-impl hive_jobq_wire::WireNode for SwarmNodeKind {
-    fn label(&self) -> String {
-        match *self {}
-    }
-
-    fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
-        match *self {}
-    }
-}
-
-/// Placeholder resource name — same rationale and same "no variants until a
-/// real node needs one" shape as [`SwarmNodeKind`].
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-enum SwarmResourceKind {}
-
-impl hive_jobq_wire::WireResource for SwarmResourceKind {
-    fn name(&self) -> String {
-        match *self {}
-    }
-}
-
 /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.
 ///
 /// A compiled-in default is legitimate here and is *not* the mistake that
@@ -101,7 +68,6 @@ fn socket_path() -> PathBuf {
         (name = "health", description = "liveness probe"),
         (name = "hives", description = "the swarm's hive directory"),
         (name = "links", description = "swarm service quick links"),
-        (name = "jobq", description = "the swarm-level job graph"),
     )
 )]
 struct ApiDoc;
@@ -142,11 +108,6 @@ struct AppState {
     /// that is merely *unreachable* still yields a reader, because
     /// `async-nats` reconnects underneath it.
     status: Option<Arc<status::StatusReader>>,
-    /// The swarm-level job graph. `std::sync::Mutex`, not `tokio`'s — every
-    /// lock scope below is synchronous (no `.await` while held). Always a
-    /// graph, never gated on the swarm queue: this is process state, not
-    /// something read over the network.
-    jobq: Arc<Mutex<hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>>>,
 }
 
 /// Env var the controller's NixOS module sets from
@@ -290,60 +251,6 @@ async fn get_hives_status(
     }
 }
 
-/// Query params for `GET /api/jobq/graph` — `?states=` narrows to root
-/// groups in the named states, same shape `hive_jobq_wire::parse_states`
-/// parses.
-#[derive(Deserialize, utoipa::IntoParams)]
-struct JobqGraphQuery {
-    states: Option<String>,
-}
-
-/// Every node of every root group in the swarm-level job graph. Nothing
-/// filters "done and old" here — with zero nodes ever submitted there is
-/// nothing to bound yet, so `graph.roots()` (everything) is the whole
-/// roster passed to [`hive_jobq_wire::GraphWire::wire_snapshot`].
-#[utoipa::path(
-    get,
-    path = "/api/jobq/graph",
-    params(JobqGraphQuery),
-    responses((status = 200, description = "every node of every root group, as generic \
-        `hive_jobq` graph nodes. `?states=` narrows to root groups in the named states.",
-        body = Vec<hive_jobq_wire::GraphNode>)),
-    tag = "jobq"
-)]
-async fn get_jobq_graph(
-    State(state): State<AppState>,
-    axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
-) -> Json<Vec<hive_jobq_wire::GraphNode>> {
-    let states = hive_jobq_wire::parse_states(q.states.as_deref());
-    let graph = state
-        .jobq
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
-    let nodes = graph.wire_snapshot(roots);
-    Json(hive_jobq_wire::filter_nodes_by_state(
-        nodes,
-        states.as_deref(),
-    ))
-}
-
-/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves.
-#[utoipa::path(
-    get,
-    path = "/api/jobq/rollup",
-    responses((status = 200, description = "counts by lifecycle state", body = Vec<hive_jobq_wire::StateCount>)),
-    tag = "jobq"
-)]
-async fn get_jobq_rollup(State(state): State<AppState>) -> Json<Vec<hive_jobq_wire::StateCount>> {
-    let graph = state
-        .jobq
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
-    Json(hive_jobq_wire::state_rollup(&graph, roots))
-}
-
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt()
@@ -427,7 +334,6 @@ async fn main() -> Result<()> {
         hives: Arc::new(load_hives()),
         links: Arc::new(load_links()),
         status,
-        jobq: Arc::new(Mutex::new(hive_jobq::Graph::new())),
     };
 
     let (router, api) = OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi())
@@ -435,8 +341,6 @@ async fn main() -> Result<()> {
         .routes(routes!(get_hives))
         .routes(routes!(get_hives_status))
         .routes(routes!(get_links))
-        .routes(routes!(get_jobq_graph))
-        .routes(routes!(get_jobq_rollup))
         .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). `api` is