diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index 8ad220d9..6b990f93 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -127,8 +127,10 @@ pub(super) async fn post_rebuild_queue_cancel( State(state): State, AxumPath(id): AxumPath, ) -> Response { - let cancelled = state.coord.job_queue.cancel(id); - if cancelled { + if let Some(terminal) = state.coord.job_queue.cancel(id) { + // Fire the DAG's inline terminal hook (power-op intent revert / approval + // resolution) off the cancel roll-up, then surface the flip live. + crate::job_queue::exec::run_terminal_hook(&state.coord, &terminal).await; state.coord.emit_rebuild_queue_snapshot(); axum::Json(serde_json::json!({"cancelled": true})).into_response() } else { diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 0c7d9d08..83bb8014 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -111,7 +111,7 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< /// Run a settled DAG's inline terminal hook, dispatched off its rolled-up /// summary — the container-terminal replacement for the old per-DAG hook node. /// Always best-effort: a hook failure is logged inside, never surfaced. -pub(super) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { +pub(crate) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { match super::terminal_hook(terminal.template, terminal.approval_id) { Some(super::HookKind::ResolveApproval) => { crate::actions::resolve_approval_dag(coord, terminal).await; diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 1d15acdd..b6abc7f6 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -312,6 +312,11 @@ impl JobQueue { .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; inner.node_rt.insert(container, NodeRuntime::default()); insert_group(&mut inner, &spec.nodes, Some(container))?; + // Settle the container's own (no-op) logic immediately so it parks in + // `Finishing` and its children become runnable — it never needs claiming + // or executing, and stays out of `claim_ready`. It rolls up terminal when + // its whole subtree settles (that's the DAG-done signal). + inner.sched.complete(container, Outcome::Done); drop(inner); self.notify.notify_one(); Ok(container.get()) @@ -434,21 +439,17 @@ impl JobQueue { terminal } - /// Cancel a DAG that hasn't started yet: the container + every work node is - /// still `Pending`, so each is cancelled. No-op (`false`) once any node is - /// running or terminal — an in-flight nix build isn't interruptible. The - /// cancel rolls the container up to `Cancelled`, so its inline hook (approval - /// resolution, power-intent revert) still fires. - pub fn cancel(&self, dag_id: u64) -> bool { + /// Cancel a DAG that hasn't started yet: every work node is still `Pending`, + /// so each is cancelled. `None` once any work node is running or terminal — + /// an in-flight nix build isn't interruptible. Otherwise the container is + /// rolled up so the DAG settles (wire state `Cancelled`) and its terminal + /// summary is returned — the caller fires the inline hook (power-intent + /// revert / approval resolution) off it. + pub fn cancel(&self, dag_id: u64) -> Option { let mut inner = self.lock(); - let inner = &mut *inner; - let Some(container) = inner.container(dag_id) else { - return false; - }; - let ids: Vec = std::iter::once(container) - .chain(inner.subtree(container)) - .collect(); - let all_pending = ids.iter().all(|&id| { + let container = inner.container(dag_id)?; + let work = inner.subtree(container); + let all_pending = work.iter().all(|&id| { inner .sched .graph() @@ -456,13 +457,18 @@ impl JobQueue { .is_some_and(|n| n.state == JobState::Pending) }); if !all_pending { - return false; + return None; } - for id in ids { + for id in work { inner.sched.cancel_node(id); } + // Roll the container up so the DAG reaches a terminal state (all children + // now `Cancelled`); `dag_rollup` reports `Cancelled` to the wire. + inner.sched.complete(container, Outcome::Done); + let terminal = inner.terminal_dag(container); + drop(inner); self.notify.notify_one(); - true + terminal } /// Set the step label on a `Running` node. Returns `true` when it changed. diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 2d184803..0063b7fc 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -280,24 +280,13 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { assert_eq!(second.dag_id, restart); assert_eq!(second.kind.as_str(), "reconcile"); q.complete_node(restart, second.node_id, Ok(())); - // Restart's work is terminal → its lease releases. Its terminal node and - // stop's now-unblocked Reconcile both become ready in the same pass. - let ready = q.claim_ready(); - let restart_fin = ready - .iter() - .find(|c| c.dag_id == restart && c.kind.as_str() == "revert_intent") - .expect("restart finalize ready"); - q.complete_node(restart, restart_fin.node_id, Ok(())); - let third = ready - .iter() - .find(|c| c.dag_id == stop) - .expect("stop reconcile ready once the lease is freed"); + // Restart's work is terminal → its lease releases, so stop's now-unblocked + // Reconcile becomes ready (restart's inline hook fired off the returned + // summary — no terminal-hook node). + let third = claim_one(&q); + assert_eq!(third.dag_id, stop); assert_eq!(third.kind.as_str(), "reconcile"); q.complete_node(stop, third.node_id, Ok(())); - // stop's terminal node (revert-intent) then runs. - let stop_fin = claim_one(&q); - assert_eq!(stop_fin.kind.as_str(), "revert_intent"); - q.complete_node(stop, stop_fin.node_id, Ok(())); assert_eq!(state_of(&q, restart), State::Done); assert_eq!(state_of(&q, stop), State::Done); } @@ -340,15 +329,10 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .expect("reconcile claim") .clone(); q.complete_node(stop, reconcile.node_id, Ok(())); - // stop's Reconcile done → its lease frees (rebuild's StopForUpdate unblocks) - // and its terminal node becomes ready; both surface in the same pass. + // stop's Reconcile done → its lease frees, so rebuild's StopForUpdate + // unblocks. (stop's DAG rolls up terminal; its inline hook fires off the + // returned summary — no terminal-hook node in the claim set.) let after = q.claim_ready(); - assert!( - after - .iter() - .any(|c| c.dag_id == stop && c.kind.as_str() == "revert_intent"), - "stop DAG's finalize runs once its work settles" - ); let sfu = after .iter() .find(|c| c.kind.as_str() == "stop_for_update") @@ -788,13 +772,11 @@ fn failed_reconcile_marks_dag_failed() { fn cancel_clears_queued_dag() { let q = JobQueue::new(1); let id = submit(&q, rebuild("agent-a", "r")); - assert!(q.cancel(id)); + // Cancel returns the terminal summary (state `Cancelled`) — the inline hook + // fires off it at the caller; there's no terminal-hook node to claim. + let terminal = q.cancel(id).expect("cancelled"); + assert_eq!(terminal.state, State::Cancelled); assert_eq!(state_of(&q, id), State::Cancelled); - // The terminal node's weak edges are satisfied by the cancelled (terminal) - // work nodes, so it still runs its hooks — it's the one thing left claimable. - let fin = claim_one(&q); - assert_eq!(fin.kind.as_str(), "emit_rebuilt"); - q.complete_node(id, fin.node_id, Ok(())); assert!(q.claim_ready().is_empty()); } @@ -803,32 +785,32 @@ fn cancel_refuses_running_dag() { let q = JobQueue::new(1); let id = submit(&q, rebuild("agent-a", "r")); let _ = claim_one(&q); - assert!(!q.cancel(id)); + assert!(q.cancel(id).is_none()); assert_eq!(state_of(&q, id), State::Running); } // ---- terminal reporting + lease release ---- #[test] -fn terminal_node_runs_after_work_settles_and_lease_released() { +fn dag_settles_terminal_and_releases_lease_after_work() { let q = JobQueue::new(1); let id = submit(&q, restart_online(&["agent-a"], false, "r")); - // restart = StopForUpdate → Reconcile; the terminal node (which weak-deps on - // the chain tail) isn't runnable until the whole chain is terminal. + // restart = StopForUpdate → Reconcile. let stop = claim_one(&q); + assert_eq!(stop.kind.as_str(), "stop_for_update"); q.complete_node(id, stop.node_id, Ok(())); let rec = claim_one(&q); assert_eq!(rec.kind.as_str(), "reconcile"); - q.complete_node(id, rec.node_id, Ok(())); - // Work settled → the terminal node is now the runnable one; it carries the - // terminal roll-up the hook consumes via `terminal_summary`. - let fin = claim_one(&q); - assert_eq!(fin.kind.as_str(), "revert_intent"); - let summary = q.terminal_summary(id).expect("terminal summary"); + // Completing the last work node rolls the container up terminal and returns + // the summary the inline hook consumes — there is no terminal-hook node. + let summary = q + .complete_node(id, rec.node_id, Ok(())) + .expect("terminal summary"); assert_eq!(summary.state, State::Done); - q.complete_node(id, fin.node_id, Ok(())); - // Lease released (freed when the work chain settled, ahead of finalize): a - // new DAG for the agent claims immediately. + assert!(q.claim_ready().is_empty(), "no terminal-hook node to claim"); + assert_eq!(state_of(&q, id), State::Done); + // Lease released when the work chain settled: a new DAG for the agent claims + // immediately. let next = submit( &q, templates::reconcile_only( @@ -853,16 +835,11 @@ fn cancelled_dag_finalizes_with_terminal_rollup() { &q, templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), ); - assert!(q.cancel(id)); - // The terminal node's weak edges still fire on a fully-cancelled DAG, so its - // hook (approval resolution) runs — surfaced here as a claimable resolve- - // approval node whose `terminal_summary` is Cancelled + carries the approval id. - let fin = claim_one(&q); - assert_eq!(fin.kind.as_str(), "resolve_approval"); - let summary = q.terminal_summary(id).expect("terminal summary"); + // Cancel rolls the DAG up terminal and returns its summary — the inline hook + // (approval resolution) runs off it at the caller. Cancelled + approval id 7. + let summary = q.cancel(id).expect("cancelled"); assert_eq!(summary.state, State::Cancelled); assert_eq!(summary.approval_id, Some(7)); - q.complete_node(id, fin.node_id, Ok(())); // The cancelled DAG's summary stays available (until history-trimmed) and // unrelated later activity doesn't disturb it. let other = submit(&q, rebuild("agent-b", "r")); @@ -931,12 +908,9 @@ fn history_evicts_old_terminals_per_template() { ), ); let c = claim_one(&q); + // Completing the single work node rolls the container up terminal (its + // inline hook fires off the returned summary — no terminal-hook node). q.complete_node(id, c.node_id, Ok(())); - // Drain the DAG's terminal node too, so the next iteration's claim sees - // only its own work (the terminal node is excluded from the view + rollup). - let fin = claim_one(&q); - assert_eq!(fin.kind.as_str(), "revert_intent"); - q.complete_node(id, fin.node_id, Ok(())); } // Fresh terminals are inside the grace window: nothing evicts yet, // so a ~1s QueueDag poller can still observe every terminal state @@ -947,8 +921,7 @@ fn history_evicts_old_terminals_per_template() { "grace window protects fresh terminals" ); // Past the grace window the per-template cap applies. - q.trim_ignoring_grace(); - assert_eq!(q.snapshot().len(), 5, "per-template history cap"); + assert_eq!(q.snapshot_no_grace().len(), 5, "per-template history cap"); assert_eq!(q.live_count(), 0); }