swarm-controller: wire the swarm-level job graph, no nodes yet

Cargo.lock

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

swarm-controller/Cargo.toml

@@ -18,6 +18,14 @@ 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,16 +23,46 @@
 
 use std::os::unix::fs::PermissionsExt as _;
 use std::path::PathBuf;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 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 — uninhabited on purpose. 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 NodeKind {}
+
+impl hive_jobq_wire::WireNode for NodeKind {
+    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 [`NodeKind`].
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+enum ResourceKind {}
+
+impl hive_jobq_wire::WireResource for ResourceKind {
+    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
@@ -68,6 +98,7 @@ 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;
@@ -108,6 +139,12 @@ 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), same
+    /// choice `hive-c0re::job_queue::JobQueue` makes for the same reason.
+    /// Always `Some` graph, never gated on the swarm queue: this is process
+    /// state, not something read over the network.
+    jobq: Arc<Mutex<hive_jobq::Graph<NodeKind, ResourceKind>>>,
 }
 
 /// Env var the controller's NixOS module sets from
@@ -251,6 +288,88 @@ async fn get_hives_status(
     }
 }
 
+/// Query params for `GET /api/jobq/graph` — same shape and same rationale as
+/// `hive-c0re::dashboard::state_snapshot::JobqGraphQuery`.
+#[derive(Deserialize, utoipa::IntoParams)]
+struct JobqGraphQuery {
+    states: Option<String>,
+}
+
+/// Same parsing rules as `hive-c0re`'s own `parse_states`: an absent or
+/// entirely-unparseable query is "no filter," never "match nothing."
+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)
+}
+
+/// Same shape as `hive-c0re::job_queue::filter_nodes_by_state`: a flat
+/// filter on each node's own state, no special-casing for a root vs. a
+/// descendant. `None` (no query given) is a no-op, not "match nothing."
+fn filter_nodes_by_state(
+    nodes: Vec<hive_jobq_wire::GraphNode>,
+    states: Option<&[hive_jobq::State]>,
+) -> Vec<hive_jobq_wire::GraphNode> {
+    let Some(states) = states else {
+        return nodes;
+    };
+    nodes
+        .into_iter()
+        .filter(|n| states.contains(&n.state))
+        .collect()
+}
+
+/// Every node of every root group in the swarm-level job graph — same shape
+/// as `hive-c0re`'s `/api/jobq/graph`. Nothing filters "done and old" here
+/// the way `hive-c0re::job_queue::visible_roots` does, because nothing is
+/// old yet: this wires the graph with no nodes, so every root the graph
+/// has is worth showing.
+#[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 — same shape as hive-c0re's own `/api/jobq/graph`. \
+        `?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 = 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(filter_nodes_by_state(nodes, states.as_deref()))
+}
+
+/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves —
+/// same shape as `hive-c0re`'s own `/api/jobq/rollup`.
+#[utoipa::path(
+    get,
+    path = "/api/jobq/rollup",
+    responses((status = 200, description = "counts by lifecycle state, same shape as \
+        hive-c0re's own `/api/jobq/rollup`", 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()
@@ -334,6 +453,7 @@ 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())
@@ -341,6 +461,8 @@ 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