c0re: history retention is a policy, split from the graph it reads

`visible_dags` mixed two things: walking the graph to classify containers
live-vs-terminal, and the sort-and-truncate that decides what the
dashboard sees. `retain_history` is the second half, generic over the
handle so it is reachable without a graph at all — a `NodeId` cannot be
fabricated, so a test forced to pass real ones could only get them by
submitting and running DAGs.

Which is exactly what the old test did: `MAX_HISTORY_DAGS + 8` submits,
claim and fail each node, read the ids back out of a snapshot — the
scheduler, the roll-up and the wire projection all standing in the path of
a policy that reads none of them. And it only ever exercised the tiebreak,
because every DAG in that loop settled inside the same wall-clock second,
so `finished_at` tied on all of them. Eviction *by time* — the actual
policy — had no coverage. It does now, along with the live-never-competes
case.

`live_count` and `templates::reconcile_only` were both test-only and lose
their last caller here.

hive-c0re/src/job_queue/mod.rs

@@ -363,17 +363,6 @@ impl JobQueue {
             .filter_map(|c| dag_view(&inner, c))
             .collect()
     }
-
-    /// Number of live (non-terminal) DAGs — tests + diagnostics.
-    #[cfg(test)]
-    #[must_use]
-    pub fn live_count(&self) -> usize {
-        let inner = self.lock();
-        containers(&inner)
-            .into_iter()
-            .filter(|&c| inner.graph().is_settled(c) == Some(false))
-            .count()
-    }
 }
 
 /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
@@ -535,19 +524,34 @@ fn containers(sched: &Sched) -> Vec<NodeId> {
 /// 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)> = 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)));
+            terminal.push((c, dag_finished_at(sched, c), c.get()));
         } else {
             live.push(c);
         }
     }
+    retain_history(live, terminal, MAX_HISTORY_DAGS)
+}
+
+/// [`visible_dags`]'s policy, split from the graph it reads: keep every live
+/// DAG, 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
+/// them under a burst, and all of them in a test, so it is load-bearing rather
+/// than a formality.
+///
+/// Generic over the handle purely so this is reachable without a graph: a
+/// `NodeId` cannot be fabricated, so a test that had to pass real ones could
+/// only get them by submitting and running DAGs.
+fn retain_history<T>(live: Vec<T>, mut terminal: Vec<(T, i64, u64)>, cap: usize) -> Vec<T> {
     // Newest first, so truncating to the cap keeps the most recent.
-    terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get())));
+    terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.2.cmp(&a.2)));
-    terminal.truncate(MAX_HISTORY_DAGS);
+    terminal.truncate(cap);
     let mut kept = live;
-    kept.extend(terminal.into_iter().map(|(c, _)| c));
+    kept.extend(terminal.into_iter().map(|(handle, _, _)| handle));
     kept
 }
 

hive-c0re/src/job_queue/templates.rs

@@ -405,29 +405,6 @@ pub fn approval_deploy(
     }
 }
 
-/// A single `Reconcile` node that converges observed power state to the
-/// persisted intent — `wanted` is untouched (no `SetWanted`), unlike the
-/// operator `start`/`stop` templates. Test-only helper now (used to build
-/// single-node lifecycle DAGs that exercise per-agent lease serialization
-/// in the queue tests); production paths no longer emit a bare reconcile.
-#[cfg(test)]
-pub fn reconcile_only(
-    agent: &str,
-    source: Source,
-    reason: String,
-) -> DagSpec<impl FnOnce(&Job) + use<>> {
-    let agent = agent.to_owned();
-    DagSpec {
-        source,
-        reason,
-        declare: Box::new(move |b: &Job| {
-            // Name the lease before the agent string is moved into the kind.
-            let lease = Resource::Agent(agent.clone());
-            let _reconcile = b.node(NodeKind::Reconcile { agent }).needs(lease);
-        }),
-    }
-}
-
 /// First-deploy spawn (approval-driven): `Provision` (proposed/applied
 /// repos, state subvolume, meta registration) then `Create`
 /// (`nixos-container create`), drop-in write, then `Reconcile` starts

hive-c0re/src/job_queue/tests.rs

@@ -1451,36 +1451,59 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
 /// The dashboard renders one recent-builds list, so one number bounds it —
 /// and with no bucketing there's nothing for a burst of same-shaped DAGs to
 /// evict early, which is what the grace window used to paper over.
+///
+/// `retain_history` is a sort-and-truncate over `(handle, finished_at,
+/// tiebreak)`. This used to submit `MAX_HISTORY_DAGS + 8` DAGs, claim and fail
+/// each one's node, then read the ids back out of a snapshot — the scheduler,
+/// the roll-up and the wire projection all in the path of a policy that reads
+/// none of them.
+///
+/// Calling it directly also reaches the half the round-trip never could: every
+/// DAG in that loop settled inside the same wall-clock second, so `finished_at`
+/// tied on all of them and **only** the tiebreak was ever exercised. Eviction
+/// by time — the actual policy — went untested.
 #[test]
-fn history_evicts_oldest_terminals_past_flat_cap() {
+fn history_retains_live_dags_and_the_newest_terminals() {
-    const OVERFLOW: usize = 8;
+    const CAP: usize = 3;
-    let q = JobQueue::new(1);
+    // Live DAGs are kept whole, regardless of the cap.
-    let mut ids = Vec::new();
+    assert_eq!(
-    for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) {
+        retain_history(vec!["live-a", "live-b"], vec![], CAP),
-        let id = submit(
+        vec!["live-a", "live-b"],
-            &q,
+    );
-            templates::reconcile_only(&format!("agent-{i}"), Source::Manual, "start".to_owned()),
+    // Terminals: newest `finished_at` first, oldest evicted past the cap.
-        );
+    assert_eq!(
-        let c = claim_one(&q);
+        retain_history(
-        // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG
+            vec![],
-        // drops off the wire entirely, but a `Failed` one is retained (+
+            vec![
-        // history-capped) so the operator can still triage it. Completing the
+                ("oldest", 100, 1),
-        // node rolls the container up terminal.
+                ("newest", 400, 2),
-        q.complete_node(c.node_id, Err("boom".to_owned()));
+                ("middle", 200, 3),
-        ids.push(id);
+                ("later", 300, 4),
-    }
+            ],
-    let kept: std::collections::HashSet<u64> = q.snapshot().iter().map(|d| d.id).collect();
+            CAP,
-    assert_eq!(kept.len(), MAX_HISTORY_DAGS, "flat history cap");
+        ),
-    // Newest-first: the oldest `OVERFLOW` fall off, everything after survives.
+        vec!["newest", "later", "middle"],
-    // These DAGs settle within the same wall-clock second, so this also pins
+        "the oldest terminal falls off"
-    // the `NodeId`-descending tiebreak that orders them when `finished_at` ties.
+    );
-    for old in &ids[..OVERFLOW] {
+    // Same second → the tiebreak decides, descending, so the DAG inserted
-        assert!(!kept.contains(old), "oldest terminal {old} evicted");
+    // last wins. This is the *common* case: a burst settles together.
-    }
+    assert_eq!(
-    for recent in &ids[OVERFLOW..] {
+        retain_history(
-        assert!(kept.contains(recent), "recent terminal {recent} retained");
+            vec![],
-    }
+            vec![("a", 100, 1), ("b", 100, 2), ("c", 100, 3), ("d", 100, 4)],
-    assert_eq!(q.live_count(), 0);
+            CAP,
+        ),
+        vec!["d", "c", "b"],
+    );
+    // A live DAG never competes with history for the cap.
+    assert_eq!(
+        retain_history(
+            vec!["live"],
+            vec![("a", 100, 1), ("b", 200, 2), ("c", 300, 3), ("d", 400, 4)],
+            CAP,
+        ),
+        vec!["live", "d", "c", "b"],
+    );
 }
 
 #[test]