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

jobq: delete DagView/NodeView, the second projection of one graph

Browse source 
Two views of the same graph existed: the typed `DagView`/`NodeView`
(`/api/state.rebuild_queue`, the `QueueDag` socket request, and the
`RebuildQueueChanged` payload) and `hive-jobq-wire`'s generic
`GraphNode` (`/api/jobq/graph`, `QueueNodes`). Every consumer has moved
to the generic one, so the typed pair is deleted rather than kept in
agreement with it.

What that removes, beyond the types: the `QueueDag` request and
`HostResponse::dags`; `Queue::snapshot`; `dag_view`, `visible_dags`,
`shown_on_wire`, `dag_finished_at` and `containers`; and the
`rebuild_queue` field on `/api/state`. `RebuildQueueChanged` keeps its
seq and loses its payload — nothing read it, and shipping the graph
both on an event and on an endpoint is the duplication this issue is
about. It stays an event rather than becoming a poll because
push-on-change is what every other live surface here does.

Two behaviours came out simpler for a structural reason. `await_dags`
needed two rules — settled means "gone from the snapshot" *or* "present
with every node terminal" — because the typed view evicted finished
groups; the generic view doesn't, so pending is just "some node isn't
terminal". And `state_of` in the tests no longer derives a roll-up at
all: a group root's own state is the scheduler's answer.

That second one found a bug. `cancelled_dag_still_runs_its_approval
tail` asserted the group reads `Cancelled` while the tail it exists to
protect was still pending — `rollup_state` flattened the surviving
child away and called the group settled. The root reads `Finishing`,
which is what the scheduler documents: own logic done, children still
running. The test now asserts that, with the reasoning inline so it
doesn't get "fixed" back.

Kept: `Source`, `State`, `PermPayload` and the `NodeId` alias in
`hive-host-sock::jobs` — shared vocabulary, still used by hivectl.
This commit is contained in:
atlas 2026-08-03 21:07:58 +02:00 • committed by mara
commit f707c60f90
10 changed files with 131 additions and 615 deletions
Download patch file Download diff file Expand all files Collapse all files

2
hive-c0re/src/coordinator.rs
View file

@ -640,10 +640,8 @@ impl Coordinator {
/// wrappers below) and the worker so every state transition
/// surfaces on the dashboard without extra plumbing.
pub fn emit_rebuild_queue_snapshot(self: &Arc<Self>) {
let queue = self.job_queue.snapshot();
self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged {
seq: self.next_seq(),
queue,
});
}

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

@ -78,13 +78,6 @@ pub(super) struct StateSnapshot {
/// disabled "updating…" state; live transitions arrive via the
/// `MetaUpdateRunning` event.
meta_update_running: bool,
/// Current state of the global job queue — pending + running DAGs
/// (rebuild / meta-update / spawn / power ops) with their per-node
/// breakdowns, plus the most recent few terminal DAGs the queue
/// retains for history. Live transitions arrive via the
/// `RebuildQueueChanged` event. See `job_queue/`. Field name kept
/// from the old flat queue for wire compatibility.
rebuild_queue: Vec<crate::job_queue::DagView>,
/// Whether the hive-forge container is up. When true the dashboard
/// links each container's config + each approval's commit into the
/// forge's `agent-configs` repos.
@ -424,7 +417,6 @@ pub(super) async fn api_state(
question_history,
tombstones,
port_conflicts,
rebuild_queue: state.coord.job_queue.snapshot(),
forge_present: crate::forge::is_present().await,
matrix_gui_enabled: std::env::var_os("HIVE_MATRIX_GUI_ENABLED").is_some_and(|v| {
// Accept any truthy string ("1", "true", "yes") since the

29
hive-c0re/src/dashboard_events.rs
View file

@ -8,7 +8,6 @@ use serde::Serialize;
use crate::container_view::ContainerView;
use crate::dashboard::{MetaInputView, TombstoneView};
use crate::job_queue::DagView;
use chrono::{DateTime, Utc};
#[derive(Debug, Clone, Serialize)]
@ -217,15 +216,20 @@ pub enum DashboardEvent {
/// when the active-run count crosses 0, so concurrent updates flip
/// the flag exactly once.
MetaUpdateRunning { seq: u64, running: bool },
/// Full snapshot of the rebuild queue (`hive-c0re::rebuild_queue`)
/// — every entry, in enqueue order, including the few most-recent
/// terminal entries the queue retains for history. Same
/// snapshot-shape rationale as `TombstonesChanged` /
/// `MetaInputsChanged`: the list is small, snapshot semantics avoid
/// the add/remove races a per-row event would have, and the
/// dashboard renders each DAG's multi-agent shape from its `nodes`
/// (grouped by `NodeView::agent`) — no cross-DAG grouping needed.
RebuildQueueChanged { seq: u64, queue: Vec<DagView> },
/// The job queue changed — **a bare trigger, no payload.**
///
/// It used to carry a full typed snapshot of the queue. Nothing reads
/// that any more: the dashboard renders the queue from
/// `GET /api/jobq/graph`, so the event's whole job is telling a client
/// *when* to re-fetch. Shipping the graph twice — once here, once on
/// the endpoint — is two projections to keep in agreement for a client
/// that only ever used one.
///
/// Kept as an event rather than deleted in favour of polling because
/// push-on-change is what every other live surface here does
/// (transients, container state, approvals), and a poll loop would be
/// slower for the same information.
RebuildQueueChanged { seq: u64 },
/// Full snapshot of all scheduled prompts. Emitted after every
/// operator mutation (new / edit / cancel / fire-now) and after the
/// worker fires or rearms a row. Same snapshot-shape rationale as
@ -421,10 +425,7 @@ mod tests {
seq: 1,
running: false,
},
DashboardEvent::RebuildQueueChanged {
seq: 1,
queue: Vec::new(),
},
DashboardEvent::RebuildQueueChanged { seq: 1 },
DashboardEvent::SchedulesChanged {
seq: 1,
schedules: Vec::new(),

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

@ -39,15 +39,14 @@ mod tests;
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Utc};
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::{Graph, NodeId};
use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify;
pub use hive_jobq::TerminalState;
pub use model::{DagView, NodeKind, PermPayload, Source, State};
pub use model::{NodeKind, PermPayload, Source, State};
use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload
@ -380,17 +379,6 @@ impl JobQueue {
hive_jobq_wire::state_rollup(inner.graph(), visible_roots(&inner))
}
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
#[must_use]
pub fn snapshot(&self) -> Vec<DagView> {
let inner = self.lock();
let mut ids = visible_dags(&inner);
ids.sort_unstable_by_key(|c| c.get());
ids.into_iter()
.filter_map(|c| dag_view(&inner, c))
.collect()
}
/// One or more nodes plus their live subtrees, as generic wire nodes —
/// the `QueueNodes` polling surface behind `hivectl`'s wait/progress
/// loop. Sibling of [`Self::snapshot`] (which serves the same graph
@ -410,8 +398,8 @@ impl JobQueue {
/// the result rather than erroring the whole batch — some ids in a
/// batch may already be evicted while others are still live. Today
/// that only happens for a genuinely unknown id: nothing prunes the
/// graph yet (bounded-prune is a Stage-C follow-up, see
/// [`visible_dags`]), so a *completed* DAG's nodes keep riding here
/// graph yet (bounded-prune is a Stage-C follow-up; [`visible_roots`]
/// bounds the *view*, not the graph), so a completed group's nodes keep riding here
/// with a terminal `state` rather than disappearing — callers
/// watching for "done" should read the root's `state`, not absence.
#[must_use]
@ -431,191 +419,13 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
.find_map(|n| (n.id.get() == id).then_some(n.id))
}
/// Project a DAG into its wire [`DagView`]: a near-raw view of the
/// container's work nodes, with `Done` nodes excluded. Lifecycle
/// (`state` / `started_at` / `finished_at` / `error`) is read straight
/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
/// state, and DAG timestamps from the node set. Non-derivable per-node
/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
/// `None` when every work node is `Done` or `Skipped` — a fully-settled
/// DAG drops out of the snapshot entirely (a `Failed` one lingers until
/// aged out).
fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
// Read straight off the container's payload: the domain fields have a
// single home there, so an intermediate owned copy of them was a second
// type describing the same data rather than a grouping side-table.
//
// `created_at` is NOT among them — it comes off the container node itself
// below, where the graph stamps it for every node. Keeping a payload copy
// would record one instant in two places.
let node = sched.graph().node(container)?;
let NodeKind::Dag { source, reason } = &node.payload else {
return None;
};
let all: Vec<_> = sched.graph().descendants(container).collect();
// DAG-level timestamps are taken over *all* subtree nodes (including the
// `Done` ones excluded from the wire) — the client can't derive them
// from a `Done`-filtered node set, so the host computes them here.
let mut started: Vec<DateTime<Utc>> = Vec::new();
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for node in &all {
if let Some(s) = node.started_at {
started.push(s);
}
if let Some(f) = node.finished_at {
finished.push(f);
}
}
// Decide which nodes ride the wire *before* projecting any of them: a
// `NodeView` costs a `build_logs` lookup, so building one for a node
// that's about to be dropped would be a query per finished step.
let shown = shown_on_wire(&all.iter().map(|n| n.state).collect::<Vec<_>>())?;
let mut nodes = Vec::new();
for node in shown.into_iter().map(|i| all[i]) {
let id = node.id;
let deps: Vec<u64> = node
.deps
.iter()
.filter_map(|d| match d {
Dep::Node { id, .. } => Some(id.get()),
Dep::Resource { .. } => None,
})
.collect();
// Non-derivable per-node payload rides the node that owns it. Every
// deploy phase carries the approval id, but only the subtree root
// projects it onto the wire — hanging the approval link off all of
// them would render the same card once per phase.
let approval_id = match &node.payload {
NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
_ => None,
};
let inputs = match &node.payload {
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
_ => Vec::new(),
};
// Looked up from the log row itself (`build_logs.node_id`), not a
// host-side map. One indexed query per node in the snapshot; the
// node set is bounded by `MAX_HISTORY_DAGS` and the store is a
// local sqlite file, so this is cheaper than the lock contention
// a second shared map would reintroduce.
let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get()));
// `node.parent` is the structural jobq parent. Top-level nodes
// have `parent == Some(container)` (direct children of the Dag
// container); those become `parent: None` on the wire since the
// container itself is not part of the work-node payload. Sub-nodes
// carry the id of their containing parent work-node.
let parent = node
.parent
.filter(|&p| p != container)
.map(hive_jobq::NodeId::get);
nodes.push(NodeView {
id: id.get(),
agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(),
deps,
state: node.state,
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
approval_id,
inputs,
build_log_id,
parent,
});
}
let is_terminal = sched.graph().is_settled(container) == Some(true);
Some(DagView {
id: container.get(),
source: *source,
reason: reason.clone(),
created_at: node.created_at,
started_at: started.into_iter().min(),
finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
nodes,
})
}
/// 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.
///
/// Two separate decisions, and conflating them pins every completed deploy in
/// the queue view forever:
/// - **`Done` drops off the wire.** A finished step isn't interesting.
/// `Skipped` stays: which branch a run *didn't* take is the readable half of
/// an outcome-branched DAG.
/// - **`Skipped` alone doesn't hold a DAG in the snapshot.** So "the node list
/// is non-empty" and "there's still something here worth showing" are
/// different questions, and only the second one may drop the DAG.
///
/// Takes states rather than projected nodes so the caller can skip the work of
/// projecting what it's about to discard, and so this is testable without a
/// graph — the states it keys on are ones only a run can produce.
fn shown_on_wire(states: &[State]) -> Option<Vec<usize>> {
let worth_showing = states
.iter()
.any(|s| !matches!(s, State::Done | State::Skipped));
if !worth_showing {
return None;
}
Some(
states
.iter()
.enumerate()
.filter(|(_, s)| !matches!(s, State::Done))
.map(|(i, _)| i)
.collect(),
)
}
/// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.
fn dag_finished_at(sched: &Sched, container: NodeId) -> i64 {
sched
.graph()
.descendants(container)
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max()
.unwrap_or(0)
}
/// Every DAG container node id in the graph.
fn containers(sched: &Sched) -> Vec<NodeId> {
sched
.graph()
.nodes()
.filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
.map(|n| n.id)
.collect()
}
/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for
/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up);
/// this filter is what bounds what the dashboard sees.
fn visible_dags(sched: &Sched) -> Vec<NodeId> {
let mut live: Vec<NodeId> = Vec::new();
let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new();
for c in containers(sched) {
if sched.graph().is_settled(c) == Some(true) {
terminal.push((c, dag_finished_at(sched, c), c.get()));
} else {
live.push(c);
}
}
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.
/// Selected *structurally* — a root is a node with no parent. The typed
/// projection this replaced keyed on `NodeKind::Dag` instead, which made the
/// visible set depend on one host node kind; nothing here knows what a node
/// means.
///
/// **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
@ -636,10 +446,13 @@ fn visible_roots(sched: &Sched) -> Vec<NodeId> {
}
/// 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.
/// **and** its descendants.
///
/// The root itself counts, because a 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. (The typed path this replaced read
/// descendants only, and could get away with it: its roots were always DAG
/// containers, which always have children.)
fn group_finished_at(sched: &Sched, root: NodeId) -> i64 {
sched
.graph()
@ -657,8 +470,8 @@ fn group_finished_at(sched: &Sched, root: NodeId) -> i64 {
.unwrap_or(0)
}
/// [`visible_dags`]'s policy, split from the graph it reads: keep every live
/// DAG, plus the newest `cap` terminal ones.
/// [`visible_roots`]'s policy, split from the graph it reads: keep every live
/// group, plus the newest `cap` terminal ones.
///
/// `terminal` rows are `(handle, finished_at, tiebreak)`. The tiebreak orders
/// DAGs that settled inside the same wall-clock second — which is *most* of

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

@ -1,10 +1,10 @@
//! Data model for the generic job-DAG queue: node kinds (the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store.
//! The serialized *views* — `DagView` / `NodeView` plus the `Source` /
//! `State` / `PermPayload` wire enums — live in `hive_host_sock::jobs`
//! (they travel on the host admin socket and the dashboard channels
//! served off the same snapshot, and nowhere else) and are re-exported
//! here for the queue's internal use.
//! The `Source` / `State` / `PermPayload` wire enums live in
//! `hive_host_sock::jobs` (they travel on the host admin socket) and are
//! re-exported here for the queue's internal use. The graph itself is
//! served through `hive_jobq_wire`'s generic projection — there is no
//! second, typed view of it any more.
//!
//! Two levels: the **DAG** is the unit of cancel / approval-resolution
//! and the dashboard group; the **node** is the unit of scheduling /
@ -12,7 +12,7 @@
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design.
pub use hive_host_sock::jobs::{DagView, PermPayload, Source, State};
pub use hive_host_sock::jobs::{PermPayload, Source, State};
use serde::Serialize;
use hive_jobq::TerminalState;

143
hive-c0re/src/job_queue/tests.rs
View file

@ -284,14 +284,27 @@ fn declared_shape_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<Declared> {
}
fn state_of(q: &JobQueue, dag_id: u64) -> State {
// A DAG whose nodes have all settled `Done` or `Skipped` drops out of the
// snapshot — absence is the completion signal, so map it to `Done`.
// Otherwise derive the roll-up from the node set, exactly as every wire
// consumer does.
q.snapshot()
// A group root's own state *is* its subtree's roll-up — that is the
// scheduler's contract (`Finishing` until children settle, then the
// rolled-up outcome), so there is nothing to derive here any more.
// A root aged out of the retained history reads `Done`: it settled, or
// it would still be live.
q.graph_snapshot()
.iter()
.find(|d| d.id == dag_id)
.map_or(State::Done, DagView::rollup_state)
.find(|n| n.id == dag_id)
.map_or(State::Done, |n| n.state)
}
/// How many groups the queue is showing — roots, not nodes.
///
/// `graph_snapshot` is flat (every node under every visible root), so a test
/// asking "how many DAGs" counts the parentless ones. Counting rows would
/// count steps, which is a different number: one rebuild is ~7 nodes.
fn dag_count(q: &JobQueue) -> usize {
q.graph_snapshot()
.iter()
.filter(|n| n.parent.is_none())
.count()
}
// ---- submit (dedup removed — every submit is a fresh DAG) ----
@ -302,7 +315,7 @@ fn submit_assigns_distinct_ids() {
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let second = submit(&q, "second", |builder| rebuild(builder, "agent-b"));
assert_ne!(first, second);
assert_eq!(q.snapshot().len(), 2);
assert_eq!(dag_count(&q), 2);
}
/// Submit-time dedup was removed with the agent-per-node refactor (a
@ -316,7 +329,7 @@ fn identical_resubmit_is_a_distinct_dag() {
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a"));
assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
assert_eq!(q.snapshot().len(), 2);
assert_eq!(dag_count(&q), 2);
}
#[test]
@ -329,7 +342,7 @@ fn distinct_submits_never_collapse() {
});
assert_ne!(rebuild_a, rebuild_b);
assert_ne!(rebuild_a, restart_a);
assert_eq!(q.snapshot().len(), 3);
assert_eq!(dag_count(&q), 3);
}
#[test]
@ -349,7 +362,7 @@ fn resubmit_while_running_is_new_dag() {
rebuild(builder, "agent-a");
});
assert_ne!(a, again);
assert_eq!(q.snapshot().len(), 2);
assert_eq!(dag_count(&q), 2);
}
// ---- malformed specs: no longer expressible ----
@ -488,50 +501,6 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
);
}
/// Which nodes ride the wire, and whether a DAG is still worth showing.
///
/// This used to submit a rebuild, claim and complete all seven of its nodes,
/// then assert the DAG was absent from `snapshot()` — the scheduler, the
/// roll-up and the whole projection standing in for one predicate over a list
/// of states. `shown_on_wire` is that predicate, so the cases can be named
/// instead of arranged.
#[test]
fn skipped_nodes_ride_the_wire_but_do_not_hold_a_settled_dag_in_the_snapshot() {
// Nothing left worth showing → the DAG drops out of the snapshot, and its
// absence is what signals completion.
assert_eq!(shown_on_wire(&[State::Done, State::Done]), None);
// The case the old test was built around: a clean run whose not-taken
// failure branch is still in the graph as `Skipped`. "The node list is
// non-empty" and "there's something here worth showing" are different
// questions, and conflating them pins every completed deploy in the queue
// view forever.
assert_eq!(
shown_on_wire(&[State::Done, State::Skipped, State::Done]),
None,
"a skipped branch does not keep a finished DAG alive"
);
// A `Failed` DAG lingers — and takes its skipped branches with it, which is
// how an operator sees which steps the run never reached.
assert_eq!(
shown_on_wire(&[State::Done, State::Failed, State::Skipped, State::Done]),
Some(vec![1, 2]),
"done nodes drop off, the failure and what it ruled out stay"
);
// Live DAGs keep everything but their finished steps.
assert_eq!(
shown_on_wire(&[State::Done, State::Running, State::Pending]),
Some(vec![1, 2])
);
assert_eq!(
shown_on_wire(&[State::Cancelled, State::Skipped]),
Some(vec![0, 1]),
"a cancelled DAG is still worth showing to whoever cancelled it"
);
// An empty DAG has nothing worth showing either — no special case needed,
// but it is the one input where "any" and "all" disagree, so it is pinned.
assert_eq!(shown_on_wire(&[]), None);
}
// ---- build slots ----
#[test]
@ -590,7 +559,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide restart is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
assert_eq!(dag_count(&q), 1);
// Each agent's subgraph head (StopForUpdate, since both are running) is a
// group root with no deps, so nothing orders them against each other; and
// each declares only its OWN agent's lease, so nothing makes them contend.
@ -626,7 +595,7 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
stop_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
assert_eq!(dag_count(&q), 1);
// Same declared story as the restart case above: each agent's subgraph head
// is a group root with no node-deps, holding only its own agent's lease.
// Independent roots on disjoint resources is what "concurrently" means at
@ -665,7 +634,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
);
});
// One DAG spanning both agents.
assert_eq!(q.snapshot().len(), 1);
assert_eq!(dag_count(&q), 1);
// The fold is a *declared* difference, readable the moment submit returns:
// both agents get a `SetWanted(Up)` group root, but the fresh agent's
// subgraph ends at the Reconcile behind it while the stale agent's carries
@ -712,13 +681,12 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
submit::restart_nodes(builder, &[("down2".to_owned(), false)], true);
});
let shape = |id: u64| -> Vec<String> {
q.snapshot()
// The group's work nodes: its subtree minus the container itself,
// which the generic view carries as an ordinary node.
q.node_subtrees(&[id])
.iter()
.find(|d| d.id == id)
.expect("dag")
.nodes
.iter()
.map(|n| n.kind.clone())
.filter(|n| n.id != id)
.map(|n| n.payload.label.clone())
.collect()
};
assert_eq!(
@ -827,11 +795,13 @@ fn rebuild_chain_nodes_suppress_crash_watch() {
// c0re's declaration of *which* edge is which is asserted in
// `rebuild_chain_is_declared_serial` — the reconcile's accepted outcome
// set is right there in the shape table.
// 2. the wire projection — `Done` off, `Skipped` on. That is
// `shown_on_wire`, tested directly above.
// 3. the roll-up — a DAG with a failed node reads `Failed`, and `Skipped`
// contributes nothing. That is `DagView::rollup_state`, which lives in
// hive-host-sock and is now tested there, next to the invariant.
// 2. the wire projection — which nodes ride and which are filtered out.
// **That decision no longer exists:** the generic view serves every node
// under a visible root, terminal ones included, so there is no predicate
// left to test. The `Done`-off/`Skipped`-on filter died with `DagView`.
// 3. the roll-up — a group with a failed node reads `Failed`. That is the
// root node's own `state`, stamped by `hive_jobq`'s settle loop and
// tested there; nothing re-derives it host-side any more.
//
// Reconstructing all three from one arranged run made none of them
// individually legible, and the arrangement was the only reason this module
@ -1035,13 +1005,21 @@ fn cancel_drops_one_agents_branch_leaving_the_rest() {
let id = submit(&q, "r", |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// Per-agent subgraphs are independent roots; find agent-a's.
let snap = q.snapshot();
let dag = snap.iter().find(|d| d.id == id).expect("dag in snapshot");
let a_root = dag
.nodes
// Per-agent subgraphs hang directly off the container, one per agent.
// ⚠️ `parent` is the *graph* parent here, not a DAG-relative one: what
// the typed view called a parentless group root is a direct child of the
// container node in the generic view.
let snap = q.node_subtrees(&[id]);
let a_root = snap
.iter()
.find(|n| n.agent == "agent-a" && n.parent.is_none())
.find(|n| {
n.parent == Some(id)
&& n.payload
.data
.get("agent")
.and_then(serde_json::Value::as_str)
== Some("agent-a")
})
.expect("agent-a has a group root");
assert!(q.cancel(a_root.id), "an interior/group root cancels alone");
@ -1154,11 +1132,18 @@ fn cancelled_dag_still_runs_its_approval_tail() {
),
"only the cancelled-outcome tail is spared, got {spared:?}"
);
assert_eq!(state_of(&q, id), State::Cancelled);
// An unrelated DAG landing in the same graph doesn't disturb this one's
// roll-up — the snapshot is per-DAG, not a global state machine.
// ⚠️ `Finishing`, not `Cancelled` — and the change is a **fix**, not a
// regression. This used to read the host-side `DagView::rollup_state`,
// which flattened the spared tail away and reported the group settled
// while a node of it was still pending. The root's own state is the
// scheduler's answer: `Finishing` means "own logic done, children still
// running", and the tail this test exists to protect *is* such a child.
// A group that still has work to do does not read terminal.
assert_eq!(state_of(&q, id), State::Finishing);
// An unrelated group landing in the same graph doesn't disturb this one's
// state — a root rolls up its own subtree, not the graph.
let _other = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
assert_eq!(state_of(&q, id), State::Cancelled);
assert_eq!(state_of(&q, id), State::Finishing);
}
// ---- approval deploy subtree ----

28
hive-c0re/src/server.rs
View file

@ -150,16 +150,6 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
HostRequest::Rebuild { name } => {
submit_single(&coord, name.as_str(), Verb::Rebuild).await
}
HostRequest::QueueDag { id } => {
// A multi-step op is one DAG now (no fan-out children to gather).
let dags = coord
.job_queue
.snapshot()
.into_iter()
.filter(|d| d.id == *id)
.collect();
HostResponse::dags(dags)
}
HostRequest::QueueNodes { ids } => {
HostResponse::nodes(coord.job_queue.node_subtrees(ids))
}
@ -901,15 +891,15 @@ async fn handle_stop(
async fn await_dags(coord: &Arc<Coordinator>, ids: &[u64], timeout: std::time::Duration) {
let deadline = std::time::Instant::now() + timeout;
loop {
let snap = coord.job_queue.snapshot();
// A DAG has settled when it's either gone from the snapshot (fully
// `Done` DAGs drop out) or still present but with every node terminal
// (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has
// a non-terminal node.
let pending = ids.iter().any(|id| {
snap.iter()
.any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal()))
});
// Every node under the named roots, terminal ones included — unlike the
// old typed snapshot, a settled group does not drop out of this view.
// So "pending" is simply "some node hasn't finished", with no second
// rule for the disappeared case.
let pending = coord
.job_queue
.node_subtrees(ids)
.iter()
.any(|n| !n.state.is_terminal());
if !pending {
return;
}