diff --git a/docs/coordinator.md b/docs/coordinator.md index af66f329..c8df2b12 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -49,7 +49,7 @@ Cheap — no build slot: | `MetaSync` | the rebuild's meta preamble — rebuild-dir prep, idempotent meta `sync_agents`, optional per-agent relock. Holds the `MetaWindow` resource (below); deliberately its own node so the window never covers `Prebuild`'s multi-minute build | | `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop | | `StopForUpdate` | mechanical `nixos-container stop` for the profile swap; never touches `wanted`; noop if already stopped | -| `PostSwap` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the `Rebuilt` manager event is emitted by the DAG's `EmitRebuilt` tail node, not here) | +| `PostSwap` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the `Rebuilt` manager event still fires once per DAG from the terminal hook, not here) | | `Signal` | set the graceful fence + kick, so the harness runs one stop-checkpoint turn | | `Drain` | await the harness clearing the fence, bounded by the 3-min graceful-stop timeout; resolves ok either way | | `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index d00d9227..a84fdde6 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -501,26 +501,19 @@ async fn run_approval_schedule_prompt( finish_approval(coord, &approval, result, None) } -/// Resolve an approval row from how its DAG's work ended — the body of the -/// [`NodeKind::ResolveApproval`] tail node. Every approval-carrying template -/// resolves here, deploys included: the deploy pipeline is ordinary queue nodes, -/// so the work's terminal state is the authoritative outcome and there's no -/// in-node resolution to skip around. -/// -/// `state` / `error` come from the tail node's own dependency roll-up -/// ([`Claim::deps_state`] / [`Claim::deps_error`]), so this runs on the success, -/// failure **and cancel** paths alike. -/// -/// [`NodeKind::ResolveApproval`]: crate::job_queue::NodeKind::ResolveApproval -/// [`Claim::deps_state`]: crate::job_queue::Claim::deps_state -/// [`Claim::deps_error`]: crate::job_queue::Claim::deps_error +/// Terminal hook for approval-carrying DAGs — the job queue's scheduler calls +/// this exactly once when such a DAG settles terminal. Every approval-carrying +/// template resolves here, deploys included: the deploy pipeline is ordinary +/// queue nodes now, so the DAG's own terminal state is the authoritative +/// outcome and there's no in-node resolution to skip around. pub(crate) async fn resolve_approval_dag( coord: &Arc, - approval_id: i64, - state: crate::job_queue::State, - error: Option<&str>, + terminal: &crate::job_queue::TerminalDag, ) { use crate::job_queue::State; + let Some(approval_id) = terminal.approval_id else { + return; + }; let approval = match coord.approvals.get(approval_id) { Ok(Some(a)) => a, Ok(None) => { @@ -532,10 +525,16 @@ pub(crate) async fn resolve_approval_dag( return; } }; - let result: Result<()> = match state { + let result: Result<()> = match terminal.state { State::Done => Ok(()), State::Cancelled => Err(anyhow::anyhow!("cancelled before completion")), - _ => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))), + _ => Err(anyhow::anyhow!( + "{}", + terminal + .error + .clone() + .unwrap_or_else(|| "job dag failed".to_owned()) + )), }; let mut terminal_tag = None; match approval.kind { @@ -551,7 +550,8 @@ pub(crate) async fn resolve_approval_dag( } } ApprovalKind::MergeConfigPr => { - terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, state).await; + terminal_tag = + deploy_terminal_tag(approval.agent.as_str(), approval_id, terminal.state).await; // On a failed deploy, surface the failing build log back onto the // PR so the manager sees why it was rejected without leaving the // forge. Posted here rather than inside a node because this is the diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index 2cfd002f..6b990f93 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -127,9 +127,10 @@ pub(super) async fn post_rebuild_queue_cancel( State(state): State, AxumPath(id): AxumPath, ) -> Response { - if state.coord.job_queue.cancel(id) { - // Any terminal side effect is the DAG's own spared tail node, which the - // scheduler picks up on its next pass — nothing to fire from here. + 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 21ea0728..178b55a2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -95,71 +95,50 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await, NodeKind::FinalizeDeploy { .. } => run_finalize_deploy(coord, claim).await, NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await, - NodeKind::ResolveApproval { approval_id, .. } => { - run_resolve_approval(coord, claim, *approval_id).await - } - NodeKind::EmitRebuilt { .. } => Ok(run_emit_rebuilt(coord, claim)), NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // Pure grouping container — no work; completing it lets it reach - // `Finishing` so its child work nodes start. The DAG's terminal side - // effect, if any, is its own tail node in the graph. + // `Finishing` so its child work nodes start. The DAG's terminal + // hook fires (inline, via `run_terminal_hook`) when the container itself + // rolls up terminal — not as a scheduled node. NodeKind::Dag { .. } => Ok(NodeOutput::default()), } } -/// Resolve the DAG's approval row from how the work it follows ended. The -/// outcome comes off this node's own dependency roll-up, not from re-reading -/// the world. Best-effort: a resolution failure is logged inside -/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure — -/// the work already happened, and failing the tail would only misreport it. -async fn run_resolve_approval( - coord: &Arc, - claim: &Claim, - approval_id: i64, -) -> Result { - let reason = failure_reason(coord, claim); - crate::actions::resolve_approval_dag(coord, approval_id, claim.deps_state(), reason.as_deref()) - .await; - Ok(NodeOutput::default()) +/// Run a settled DAG's inline terminal hook — the container-terminal +/// replacement for the old per-DAG hook node. Always best-effort: a hook +/// failure is logged inside, never surfaced. +pub(crate) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { + match terminal.hook { + Some(super::HookKind::ResolveApproval) => { + crate::actions::resolve_approval_dag(coord, terminal).await; + } + Some(super::HookKind::EmitRebuilt) => emit_rebuilt(coord, terminal), + None => {} + } } -/// Why the work a tail node follows failed, as a human-readable string. -/// -/// Prefers the tail's own dependency error, but a dep that is a **group root** -/// rolled up `Failed` from a child carries no error of its own (the reason lives -/// on the leaf that actually failed) — and a grafted subgraph's nodes can't be -/// edged statically anyway. So fall back to the DAG's first failing node. Still -/// the queue's own graph, not the outside world. -fn failure_reason(coord: &Arc, claim: &Claim) -> Option { - claim - .deps_error() - .map(str::to_owned) - .or_else(|| coord.job_queue.first_error(claim.dag_id)) -} - -/// Emit this agent's `Rebuilt` manager event — `ok` when the work it follows is -/// `Done`, `!ok` with the failure note when it `Failed`, and nothing at all when -/// it `Cancelled` (nothing ran, so there is no rebuild to report). -fn run_emit_rebuilt(coord: &Arc, claim: &Claim) -> NodeOutput { - let agent = claim.agent.clone(); - match claim.deps_state() { - State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent, - ok: true, - note: None, - sha: None, - tag: None, - }), - State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent, - ok: false, - note: failure_reason(coord, claim), - sha: None, - tag: None, - }), - _ => {} +/// Rebuild / perm-change hook: emit one `Rebuilt` manager event per targeted +/// agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel. +fn emit_rebuilt(coord: &Arc, terminal: &super::TerminalDag) { + for agent in &terminal.agents { + match terminal.state { + State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: true, + note: None, + sha: None, + tag: None, + }), + State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: false, + note: terminal.error.clone(), + sha: None, + tag: None, + }), + _ => {} + } } - NodeOutput::default() } /// Write the agent's durable power intent — the DAG-node form of the old @@ -259,8 +238,8 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded // and the tail `Reconcile` (`AfterAny(PostSwap)`) handles recovery; here // we only refresh the observed state so dashboards reflect the failed - // swap immediately. The `Rebuilt { ok: false }` manager event is emitted by - // the DAG's `EmitRebuilt` tail (any node may be the one that failed). + // swap immediately. The `Rebuilt { ok: false }` manager event fires once + // per DAG from the terminal hook (any node may be the one that failed). if result.is_err() { coord.rescan_containers_and_emit().await; } @@ -279,8 +258,8 @@ async fn run_post_swap(coord: &Arc, claim: &Claim) -> Result, -} - /// A node claimed for execution — everything the executor needs, snapshotted at /// claim time. #[derive(Debug, Clone)] @@ -101,38 +76,21 @@ pub struct Claim { /// pill is currently shown is derived from live lease ownership /// ([`JobQueue::held_transients`]), not a per-claim edge. pub transient: Option, - /// How each node this one depends on finished. Empty for a head node. - /// See [`DepOutcome`] — this is how a tail node learns the outcome of the - /// work it follows without going back to the world to ask. - pub deps: Vec, } -impl Claim { - /// Roll this claim's dependencies up into one outcome — what a weak-edged - /// tail node acts on. `Done` only when every dep succeeded; `Failed` when any - /// failed; otherwise `Cancelled` (all terminal, none failed, so the work was - /// dropped before it ran). - /// - /// A head node has no deps and rolls up `Done` — vacuously true, and never - /// reached in practice since only tail kinds consult this. - pub fn deps_state(&self) -> State { - if self.deps.iter().all(|d| d.state == State::Done) { - State::Done - } else if self.deps.iter().any(|d| d.state == State::Failed) { - State::Failed - } else { - State::Cancelled - } - } - - /// The first failed dependency's error, for reporting *why* the work ended - /// badly. `None` when nothing failed. - pub fn deps_error(&self) -> Option<&str> { - self.deps - .iter() - .find(|d| d.state == State::Failed) - .and_then(|d| d.error.as_deref()) - } +/// Summary of a DAG's terminal roll-up — the input to the terminal node's +/// executor (approval resolution, `Rebuilt` emission, cancelled-power-op intent +/// revert). Computed on demand from live graph state, not drained. +#[derive(Debug, Clone)] +pub struct TerminalDag { + /// The side effect to fire, carried from the DAG container's payload. + pub hook: Option, + /// Distinct agents this DAG's nodes targeted (one for a single-agent DAG). + pub agents: Vec, + pub approval_id: Option, + pub state: State, + /// First failed node's error when `state == Failed`. + pub error: Option, } /// Per-node runtime metadata the crate graph doesn't carry. Lifecycle @@ -148,6 +106,7 @@ struct NodeRuntime { /// Derived on read from the container node — the data has a single home (the /// node payload); this is not a stored side-table. struct DagMeta { + hook: Option, source: Source, reason: String, transient: Option, @@ -282,7 +241,8 @@ impl JobQueue { /// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container /// node** carrying the group's metadata, then inserts the template's nodes as /// its subtree (their roots re-parented to the container). Returns the - /// container's id as the DAG id — its rolled-up state is the DAG state. + /// container's id as the DAG id — its rolled-up state is the DAG state and it + /// reaching terminal fires the DAG's inline hook. /// /// # Errors /// Propagates the spec-validation error (empty / cyclic / bad parent) or a @@ -294,6 +254,7 @@ impl JobQueue { .sched .append( NodeKind::Dag { + hook: spec.hook, source: spec.source, reason: spec.reason, transient: spec.transient, @@ -373,24 +334,6 @@ impl JobQueue { }; let kind = node.payload.clone(); let agent = node.payload.agent().to_owned(); - let dep_ids: Vec = node - .deps - .iter() - .filter_map(|d| match d { - Dep::Node { id, .. } => Some(*id), - Dep::Resource { .. } => None, - }) - .collect(); - let deps: Vec = dep_ids - .into_iter() - .filter_map(|dep| { - let n = inner.sched.graph().node(dep)?; - Some(DepOutcome { - state: to_wire_state(n.state), - error: n.error.clone(), - }) - }) - .collect(); let Some(container) = inner.dag_of(id) else { continue; }; @@ -405,7 +348,6 @@ impl JobQueue { approval_id: meta.approval_id, inputs: meta.inputs, transient: meta.transient, - deps, }); // `started_at` is stamped on the graph `Node` by the scheduler's // transition to `Running` — no host-side copy needed. @@ -415,12 +357,15 @@ impl JobQueue { /// Mark a claimed node terminal, recording its outcome + (truncated) error. /// The crate releases the node's build slot immediately and cascades the - /// `AfterOk` failure cancellation + subtree lease release. - /// - /// Nothing is returned: a DAG's terminal side effects are its own tail nodes - /// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the - /// scheduler claims and runs like any other node. - pub fn complete_node(&self, _dag_id: u64, node_id: NodeId, result: Result<(), String>) { + /// `AfterOk` failure cancellation + subtree lease release. Returns the DAG's + /// terminal summary **iff** this completion rolled its container terminal — + /// the scheduler runs the DAG's inline hook off it. + pub fn complete_node( + &self, + _dag_id: u64, + node_id: NodeId, + result: Result<(), String>, + ) -> Option { let mut inner = self.lock(); // The failure reason + `finished_at` are stamped onto the graph `Node` // by the scheduler (the reason rides `Outcome::Failed`); no host-side @@ -429,31 +374,27 @@ impl JobQueue { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(truncate_error(&e)), }; + let container = inner.dag_of(node_id); inner.sched.complete(node_id, outcome); + // If this completion rolled the DAG's container up to a terminal state, + // hand its summary back so the scheduler fires the inline hook once. + let terminal = container + .filter(|&c| c != node_id && inner.dag_is_terminal(c)) + .and_then(|c| inner.terminal_dag(c)); drop(inner); self.notify.notify_one(); + terminal } /// Cancel a DAG that hasn't started yet: every work node is still `Pending`, - /// so each is cancelled. `false` once any work node is running or terminal — - /// an in-flight nix build isn't interruptible. - /// - /// **Tail nodes are spared** ([`NodeKind::is_tail`]). They are weak-edged - /// (`AfterAny`), and a `Cancelled` dep satisfies a weak edge, so sparing one - /// leaves it *ready* rather than stranded: the scheduler claims it on the next - /// pass, its [`Claim::deps_state`] reads `Cancelled`, and it resolves the - /// approval as "cancelled before completion". That is what stops a queued - /// approval DAG the operator cancelled from dangling its approval forever — - /// the job the inline hook used to do from outside the graph. - /// - /// Cancelling the tail too would be the bug: `cancel_node` only cascades along - /// `AfterOk` edges and parent links, so nothing else would reach it, and the - /// approval row would simply never be touched. - pub fn cancel(&self, dag_id: u64) -> bool { + /// 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 Some(container) = inner.container(dag_id) else { - return false; - }; + let container = inner.container(dag_id)?; let work = inner.subtree(container); let all_pending = work.iter().all(|&id| { inner @@ -463,29 +404,22 @@ impl JobQueue { .is_some_and(|n| n.state == JobState::Pending) }); if !all_pending { - return false; + return None; } for id in work { - if inner - .sched - .graph() - .node(id) - .is_some_and(|n| n.payload.is_tail()) - { - continue; - } inner.sched.cancel_node(id); } - // Re-run the container's roll-up now that its children are `Cancelled`. - // With a spared tail still `Pending` this is a deliberate no-op — the - // container has a non-terminal child, so `settle_terminal` parks it back - // in `Finishing` and it rolls up for real once the tail finishes. With no - // tail (a power op) every child *is* terminal, so it settles synchronously - // here exactly as before. + // The container was settled to `Finishing` at submit; completing it again + // now re-runs the roll-up with its children all `Cancelled`, driving it to + // a terminal state synchronously within this lock — so the caller reads + // the terminal summary immediately instead of waiting for the scheduler + // loop to observe the cancellation. `dag_rollup` reports `Cancelled` to + // the wire (a container whose children all cancelled). inner.sched.complete(container, Outcome::Done); + let terminal = inner.terminal_dag(container); drop(inner); self.notify.notify_one(); - true + terminal } /// Link a `build_logs` row to a specific `Running` node. @@ -529,6 +463,17 @@ impl JobQueue { inner.dag_first_error(container) } + /// A DAG's terminal roll-up summary, computed on demand from its container. + /// `None` if the DAG id is unknown. Test-only — production reads the summary + /// `complete_node` returns when the container rolls up terminal. + #[cfg(test)] + #[must_use] + pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option { + let inner = self.lock(); + let container = inner.container(dag_id)?; + inner.terminal_dag(container) + } + /// The `(dag_id, agent, kind)` triples for every per-agent lease currently /// held by a DAG that carries a transient pill — the live transient-pill /// set, a pull query over crate resource ownership (replaces the old @@ -621,6 +566,7 @@ impl QueueInner { /// not a stored side-table. fn dag_meta(&self, container: NodeId) -> Option { let NodeKind::Dag { + hook, source, reason, transient, @@ -632,6 +578,7 @@ impl QueueInner { return None; }; Some(DagMeta { + hook: *hook, source: *source, reason: reason.clone(), transient: *transient, @@ -641,6 +588,35 @@ impl QueueInner { }) } + /// Roll-up state over a DAG's work nodes: `Failed` if any failed; else + /// `Running` if any running; else `Queued` if any queued; else `Cancelled` + /// if any cancelled; else `Done`. (Kept eager over the subtree — a failed + /// child shows `Failed` immediately, before the container finishes rolling + /// up — matching the pre-container behaviour.) + fn dag_rollup(&self, container: NodeId) -> State { + let mut any_running = false; + let mut any_queued = false; + let mut any_cancelled = false; + for id in self.subtree(container) { + match self.sched.graph().node(id).map(|n| n.state) { + Some(JobState::Failed) => return State::Failed, + Some(JobState::Running | JobState::Finishing) => any_running = true, + Some(JobState::Pending) => any_queued = true, + Some(JobState::Cancelled) => any_cancelled = true, + Some(JobState::Done) | None => {} + } + } + if any_running { + State::Running + } else if any_queued { + State::Queued + } else if any_cancelled { + State::Cancelled + } else { + State::Done + } + } + /// True when the DAG has settled — its container has rolled up terminal /// (equivalent to every work node being terminal). fn dag_is_terminal(&self, container: NodeId) -> bool { @@ -650,8 +626,22 @@ impl QueueInner { .is_some_and(|n| n.state.is_terminal()) } + /// Distinct agents a DAG's work nodes target, in first-seen order. + fn dag_agents(&self, container: NodeId) -> Vec { + let mut seen: Vec = Vec::new(); + for id in self.subtree(container) { + if let Some(n) = self.sched.graph().node(id) { + let agent = n.payload.agent(); + if !agent.is_empty() && !seen.iter().any(|s| s == agent) { + seen.push(agent.to_owned()); + } + } + } + seen + } + /// First failed work node's error (read off the graph `Node`), for the - /// dashboard's DAG-level error line. + /// terminal roll-up summary the inline hook consumes. fn dag_first_error(&self, container: NodeId) -> Option { for id in self.subtree(container) { if let Some(n) = self.sched.graph().node(id) @@ -664,6 +654,18 @@ impl QueueInner { None } + /// A DAG's terminal roll-up summary — the input to its inline hook. + fn terminal_dag(&self, container: NodeId) -> Option { + let meta = self.dag_meta(container)?; + Some(TerminalDag { + hook: meta.hook, + agents: self.dag_agents(container), + approval_id: meta.approval_id, + state: self.dag_rollup(container), + error: self.dag_first_error(container), + }) + } + /// 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 diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 0d321a36..03ea621f 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,6 +17,27 @@ use serde::Serialize; use crate::coordinator::TransientKind; +/// The inline side effect a settled DAG fires when its container node rolls +/// up terminal (there is no hook *node*). Stated explicitly by the builder in +/// `templates.rs` / `submit.rs` rather than inferred from a DAG-level enum: +/// only the builder knows why it assembled the DAG, so only the builder can +/// say what should happen at the end of it. +/// +/// A cancelled DAG deliberately gets **no** compensating hook. [`super::JobQueue::cancel`] +/// refuses unless every work node is still `Pending`, and a cancel *cascade* +/// rolls up `Failed` (see `dag_rollup`), never `Cancelled` — so on a +/// `Cancelled` DAG no node ever executed and there is nothing to undo. A power +/// op's `SetWanted` head provably never ran, so its intent is still whatever +/// the operator last set it to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum HookKind { + /// Approval-driven DAG (spawn / opaque deploy): resolve the approval row. + ResolveApproval, + /// Rebuild / perm-change: emit one `Rebuilt` manager event per agent. + EmitRebuilt, +} + /// When a dependency edge is considered satisfied. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] @@ -227,31 +248,6 @@ pub enum NodeKind { /// means it survives a `hive-c0re` restart mid-deploy, which an in-memory /// queue does not. DeployTail { agent: String }, - /// Tail node of an approval-carrying DAG (spawn / opaque deploy / config-PR - /// merge): resolve the approval row from how the work actually ended. - /// - /// Weak-edged (`DepWhen::AfterAny`) like [`NodeKind::DeployTail`], so it runs on - /// success, failure **and cancel** alike and decides internally. It reads its - /// dependencies' terminal states off its own [`Claim::deps`] rather than - /// re-deriving them from the world the way `DeployTail` reads git: a node is - /// *told* how the work it follows ended, it does not go back out and ask. - /// - /// Agentless on purpose: the approval row already names its agent, so - /// carrying one here would be a second copy free to drift. Like - /// [`NodeKind::MetaLock`] it reports `""` from [`NodeKind::agent`] and takes - /// no lease — which is also what lets one close a multi-agent DAG. - /// - /// [`Claim::deps`]: super::Claim::deps - ResolveApproval { approval_id: i64 }, - /// Tail node of a rebuild / perm-change: emit this agent's `Rebuilt` manager - /// event — `ok` when its deps are `Done`, `!ok` carrying the failure note when - /// they `Failed`, and **nothing at all** when they `Cancelled` (a cancelled DAG - /// never ran, so there is no rebuild to report). - /// - /// One node **per agent**, unlike the DAG-wide hook it replaces: a multi-agent - /// DAG now reports each agent's own outcome instead of painting every agent with - /// the whole DAG's roll-up. - EmitRebuilt { agent: String }, /// Write the agent's durable power intent (`wanted = Up` when `up`, else /// `Offline`) as a first-class DAG node, at the head of a power-op /// template so the downstream `Reconcile` reads it. Replaces the old @@ -268,11 +264,15 @@ pub enum NodeKind { SetWanted { agent: String, up: bool }, /// The **DAG container** node: one per submitted DAG, carrying the group's /// domain metadata. Every node hangs *under* it (its subtree), so - /// the container's `NodeId` **is** the DAG id and its rolled-up state **is** - /// the DAG state. Pure grouping — lease- and + /// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the + /// DAG state, and it reaching terminal **is** the completion signal that + /// fires the DAG's inline `hook`. Pure grouping — lease- and /// build-slot-exempt; the executor instant-completes it (`Done`) so it /// reaches `Finishing` and its children start. Dag { + /// The side effect to run when this DAG settles, or `None` for a DAG + /// with none (power op, meta-update, boot). + hook: Option, source: Source, reason: String, transient: Option, @@ -307,8 +307,6 @@ impl NodeKind { NodeKind::DeployApply { .. } => "deploy_apply", NodeKind::FinalizeDeploy { .. } => "finalize_deploy", NodeKind::DeployTail { .. } => "deploy_tail", - NodeKind::ResolveApproval { .. } => "resolve_approval", - NodeKind::EmitRebuilt { .. } => "emit_rebuilt", NodeKind::SetWanted { .. } => "set_wanted", NodeKind::Dag { .. } => "dag", } @@ -340,29 +338,11 @@ impl NodeKind { | NodeKind::DeployApply { agent } | NodeKind::FinalizeDeploy { agent } | NodeKind::DeployTail { agent } - | NodeKind::EmitRebuilt { agent } | NodeKind::SetWanted { agent, .. } => agent, - NodeKind::MetaLock { .. } - | NodeKind::Reparent { .. } - | NodeKind::ResolveApproval { .. } - | NodeKind::Dag { .. } => "", + NodeKind::MetaLock { .. } | NodeKind::Reparent { .. } | NodeKind::Dag { .. } => "", } } - /// Whether this is a DAG's **tail** — a node that reports how the rest of the - /// DAG ended rather than doing work of its own. - /// - /// The one place this matters is [`super::JobQueue::cancel`], which spares - /// tails so they still run (and report `Cancelled`) on a cancelled DAG. Note - /// [`NodeKind::DeployTail`] is *not* one: despite the name it does real - /// compensating work, and on a cancelled DAG there is nothing to compensate. - pub fn is_tail(&self) -> bool { - matches!( - self, - NodeKind::ResolveApproval { .. } | NodeKind::EmitRebuilt { .. } - ) - } - /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits /// for the node's duration. pub fn needs_build_slot(&self) -> bool { @@ -466,15 +446,14 @@ pub struct NodeSpec { /// ([`NodeKind::WritePermFile`]), not this generic spec. #[derive(Debug, Clone)] pub struct DagSpec { + /// The inline side effect to fire when this DAG settles. Explicit — the + /// builder assembling the DAG is the only thing that knows its intent. + pub hook: Option, pub source: Source, /// Free-form "why". pub reason: String, - /// The approval row this DAG belongs to, for display + the `approval_id` on - /// every [`Claim`]. The *resolving* of it rides the - /// [`NodeKind::ResolveApproval`] tail node instead — this field does not - /// drive it. - /// - /// [`Claim`]: super::Claim + /// The approval row [`HookKind::ResolveApproval`] resolves. Set together + /// with that hook; carried separately because the hook needs the id. pub approval_id: Option, /// Meta-update only: the inputs to bump. Display copy lives on the DAG. pub inputs: Vec, diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 6191ab6b..87536ef9 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -105,9 +105,10 @@ fn handle_completion(coord: &Arc, done: NodeDone) { .job_queue .append_subgraph(claim.dag_id, subgraph, claim.node_id); } - coord + let terminal = coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(())); + fire_terminal_hook(coord, terminal); } Err(e) => { let msg = format!("{e:#}"); @@ -119,9 +120,10 @@ fn handle_completion(coord: &Arc, done: NodeDone) { error = %msg, "job_queue: node failed" ); - coord + let terminal = coord .job_queue .complete_node(claim.dag_id, claim.node_id, Err(msg)); + fire_terminal_hook(coord, terminal); } } // The next loop iteration re-reconciles the transient pills against the @@ -129,6 +131,19 @@ fn handle_completion(coord: &Arc, done: NodeDone) { coord.emit_rebuild_queue_snapshot(); } +/// Fire a settled DAG's inline terminal hook (approval-resolve / rebuilt-emit / +/// intent-revert) off the container-terminal summary `complete_node` returned — +/// spawned so the async hook doesn't block the scheduler loop. +fn fire_terminal_hook(coord: &Arc, terminal: Option) { + let Some(terminal) = terminal else { + return; + }; + let coord = Arc::clone(coord); + tokio::spawn(async move { + exec::run_terminal_hook(&coord, &terminal).await; + }); +} + /// Reconcile the transient-guard set against live lease ownership: drop pills /// whose lease is no longer held, create one for each newly-held `(dag, agent)`. fn reconcile_transients( diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index e07475d9..3584793c 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -188,9 +188,9 @@ fn concat_subgraphs(chains: Vec>) -> Vec { out } -/// Wrap assembled power-op `nodes` in a `DagSpec`. No tail node: a power op's -/// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once -/// they settle. +/// Wrap assembled power-op `nodes` in a `DagSpec`. No terminal hook: a power +/// op's effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to +/// do once they settle. fn power_dag( transient: TransientKind, source: Source, @@ -198,6 +198,7 @@ fn power_dag( nodes: Vec, ) -> DagSpec { DagSpec { + hook: None, source, reason, approval_id: None, diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index d47be49d..1b85f338 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -30,7 +30,7 @@ use anyhow::{Result, bail}; -use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source}; +use super::model::{DagSpec, Dep, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source}; use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with @@ -43,23 +43,6 @@ pub(crate) fn after_ok(on: u64) -> Vec { }] } -/// Weak edges onto every one of a DAG's other **group-roots** — how a tail node -/// (`ResolveApproval` / `EmitRebuilt`) sees the whole DAG's outcome. -/// -/// Group-roots are the right granularity, not "every node": a root's state *is* -/// its subtree's roll-up, so edging the roots covers every descendant while -/// keeping the tail's dep list small and stable as subtrees grow. `AfterAny` -/// throughout, so the tail runs on success, failure and cancel alike and decides -/// from [`super::Claim::deps_state`]. -pub(crate) fn after_any_all(ons: &[u64]) -> Vec { - ons.iter() - .map(|&on| Dep { - on, - when: DepWhen::AfterAny, - }) - .collect() -} - /// Build one **top-level (group-root)** node — `parent = None`. `kind` carries /// the agent it targets ([`NodeKind`] is the payload directly). Shared with /// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it @@ -190,27 +173,15 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec { /// when `wanted = Offline` (a rebuild of a deliberately-stopped agent /// leaves it stopped). `relock = false` only for meta-update cascade /// children. -/// -/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots -/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the -/// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every -/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so -/// it reaches `Done` even after a failed swap and the tail would report success. pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { - let mut nodes = rebuild_nodes(agent, relock, 0); - nodes.push(node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - }, - after_any_all(&[0, 1, 5]), - )); DagSpec { + hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, inputs: Vec::new(), transient: Some(TransientKind::Rebuilding), - nodes, + nodes: rebuild_nodes(agent, relock, 0), } } @@ -230,18 +201,12 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag /// bookkeeping tail — rollback when a merge landed unfinalized, forge tag /// mirror, PR failure comment (see [`NodeKind::DeployTail`]). /// -/// - `ResolveApproval` (4, **root**, `AfterAny` `DeployWindow`): resolves the -/// approval row. A root rather than another child, so it isn't inside the -/// window's resource subtree — it runs once the window has released the meta -/// window, lease and build slot. One edge suffices here: `DeployWindow` is the -/// DAG's only other group-root, so its roll-up already *is* the whole -/// pipeline's outcome. -/// /// The window still spans the container build, as it must: `prepare_deploy` /// leaves `flake.lock` staged-uncommitted for the build's whole duration. pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { let a = || agent.to_owned(); DagSpec { + hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -259,10 +224,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec when: DepWhen::AfterAny, }], ), - node( - NodeKind::ResolveApproval { approval_id }, - after_any_all(&[0]), - ), ], } } @@ -280,6 +241,7 @@ pub fn reconcile_only( transient: Option, ) -> DagSpec { DagSpec { + hook: None, source, reason, approval_id: None, @@ -302,11 +264,10 @@ pub fn reconcile_only( /// `Create` (child) owns the agent lease; `WriteDropin` + `Reconcile` /// (children of `Create`) borrow it. A failure cancel-cascades the rest — /// unlike rebuild there's no recovery-reconcile (nothing to converge if the -/// container was never created). Closed by a `ResolveApproval` tail root edged -/// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up -/// already carries the whole cascade. +/// container was never created). pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { + hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -319,10 +280,6 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { child(0, NodeKind::Create { agent: a() }, Vec::new()), child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()), child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)), - node( - NodeKind::ResolveApproval { approval_id }, - after_any_all(&[0]), - ), ] }, } @@ -330,9 +287,7 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { /// Perm change: commit the JSON file(s), then the rebuild subgraph so /// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes -/// effect in the container. Group-roots are `WritePermFile`(0) plus the rebuild -/// subgraph's `MetaSync`(1) / `Prebuild`(2) / `Reconcile`(6), so the -/// `EmitRebuilt` tail edges all four. +/// effect in the container. pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { let mut nodes = vec![node( NodeKind::WritePermFile { @@ -342,13 +297,8 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay Vec::new(), )]; nodes.extend(rebuild_nodes(agent, true, 1)); - nodes.push(node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - }, - after_any_all(&[0, 1, 2, 6]), - )); DagSpec { + hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, @@ -374,30 +324,22 @@ pub fn meta_update( reason: String, approval_id: Option, ) -> DagSpec { - let mut nodes = vec![node( - NodeKind::MetaLock { - sweep: false, - fanout: None, - }, - Vec::new(), - )]; - // The bump itself has no side effect, so an operator-driven one ends at the - // `MetaLock`; an approval-driven one still has its row to resolve and gets a - // tail edged onto that single group-root — whose roll-up covers the rebuild - // subgraphs `MetaLock` grows into itself. - if let Some(approval_id) = approval_id { - nodes.push(node( - NodeKind::ResolveApproval { approval_id }, - after_any_all(&[0]), - )); - } DagSpec { + // The bump itself has no side effect; an approval-driven one still has + // its row to resolve. + hook: approval_id.map(|_| HookKind::ResolveApproval), source, reason, approval_id, inputs, transient: Some(TransientKind::Rebuilding), - nodes, + nodes: vec![node( + NodeKind::MetaLock { + sweep: false, + fanout: None, + }, + Vec::new(), + )], } } @@ -409,13 +351,14 @@ pub fn meta_update( /// (dashboard tree, ``/`` sentinel routing, permission /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one -/// off of) and near-instant. No tail node: the write is the whole effect. +/// off of) and near-instant. No terminal hook: the write is the whole effect. pub fn reparent( moves: Vec<(hive_types::Ident, Option)>, source: Source, reason: String, ) -> DagSpec { DagSpec { + hook: None, source, reason, approval_id: None, diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 40251fd7..d3b7b5ce 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -48,39 +48,6 @@ fn claim_one(q: &JobQueue) -> Claim { claims.pop().expect("one claim") } -/// Claim an approval DAG's `ResolveApproval` tail, assert which approval it -/// carries and what it will report to that row, then complete it. Replaces the -/// old `terminal_summary()` assertions: the outcome is no longer a struct handed -/// to a hook, it's what this node reads off its own deps. -fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: State) { - let tail = claim_one(q); - assert!( - matches!(tail.kind, NodeKind::ResolveApproval { approval_id: got } if got == approval_id), - "expected the ResolveApproval tail for #{approval_id}, got {:?}", - tail.kind - ); - assert_eq!( - tail.deps_state(), - expect, - "outcome the tail reports to approval #{approval_id}" - ); - q.complete_node(dag_id, tail.node_id, Ok(())); -} - -/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim a rebuild / -/// perm-change DAG's tail, assert the `Rebuilt` event it will emit, complete it. -fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect: State) { - let tail = claim_one(q); - assert_eq!(tail.kind.as_str(), "emit_rebuilt"); - assert_eq!(tail.agent, agent, "`Rebuilt` is emitted per agent"); - assert_eq!( - tail.deps_state(), - expect, - "ok-ness of the emitted `Rebuilt`" - ); - q.complete_node(dag_id, tail.node_id, Ok(())); -} - fn state_of(q: &JobQueue, dag_id: u64) -> State { // A fully-`Done` DAG drops out of the snapshot (its nodes are all // excluded) — absence is the completion signal, so map it to `Done`. @@ -229,7 +196,6 @@ fn rebuild_chain_claims_in_dep_order() { ); q.complete_node(id, c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -349,8 +315,8 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { assert_eq!(second.kind.as_str(), "reconcile"); q.complete_node(restart, second.node_id, Ok(())); // Restart's work is terminal → its lease releases, so stop's now-unblocked - // Reconcile becomes ready (a power op has no tail node, so nothing of - // restart's remains claimable). + // 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"); @@ -403,8 +369,8 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .clone(); q.complete_node(stop, reconcile.node_id, Ok(())); // stop's Reconcile done → its lease frees, so rebuild's StopForUpdate - // unblocks. (stop's DAG rolls up terminal; a power op has no tail node, so - // nothing of stop's is left in the claim set.) + // 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(); let sfu = after .iter() @@ -625,6 +591,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // the emitter and its LOCAL 0-based deps are rebased onto the DAG. let q = JobQueue::new(4); let spec = DagSpec { + hook: None, source: Source::AutoUpdate, reason: "sweep".to_owned(), approval_id: None, @@ -736,73 +703,6 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() { ); } -// ---- dep outcomes on the claim ---- - -/// Every claimed node reports how each node it depends on finished, and those -/// deps are always already terminal — that is what a node's edges being -/// satisfied *means*. A tail node reads its `deps` instead of going back to the -/// world to find out how the work below it went. -#[test] -fn claim_carries_terminal_dep_outcomes() { - let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); - let mut saw_a_dep = false; - loop { - let mut claims = q.claim_ready(); - let Some(claim) = claims.pop() else { break }; - assert!(claims.is_empty(), "one build slot ⇒ one claim at a time"); - for dep in &claim.deps { - saw_a_dep = true; - assert!( - dep.state.is_terminal(), - "{} was claimed with a non-terminal dep ({:?}) — a node's edges \ - being satisfied is exactly the claim that its deps have finished", - claim.kind.as_str(), - dep.state - ); - assert_eq!( - dep.state, - State::Done, - "on the happy path every dep of {} finished Done", - claim.kind.as_str() - ); - assert_eq!(dep.error, None, "a Done dep carries no error"); - } - q.complete_node(id, claim.node_id, Ok(())); - } - assert!(saw_a_dep, "the rebuild DAG has at least one dependent node"); - assert_eq!(state_of(&q, id), State::Done); -} - -/// The failure direction, which is the whole point of carrying outcomes at all: -/// `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed prebuild -/// cancel-cascades `StopForUpdate`/`Swap`/`PostSwap` and `Reconcile` still runs -/// — and its claim hands it the failure, including the reason, rather than -/// leaving the executor to go and re-derive it from the world. -#[test] -fn claim_dep_outcome_reports_a_failed_dep_with_its_error() { - let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); - let meta_sync = claim_one(&q); - q.complete_node(id, meta_sync.node_id, Ok(())); - let prebuild = claim_one(&q); - assert_eq!(prebuild.kind.as_str(), "prebuild"); - q.complete_node(id, prebuild.node_id, Err("nix build exploded".to_owned())); - let reconcile = claim_one(&q); - assert_eq!(reconcile.kind.as_str(), "reconcile"); - assert_eq!( - reconcile - .deps - .iter() - .map(|d| (d.state, d.error.as_deref())) - .collect::>(), - vec![(State::Failed, Some("nix build exploded"))], - "reconcile's claim carries the failed prebuild and its reason" - ); - assert_eq!(reconcile.deps_state(), State::Failed); - assert_eq!(reconcile.deps_error(), Some("nix build exploded")); -} - // ---- failure: cancel-downstream + AfterAny ---- #[test] @@ -911,7 +811,6 @@ fn swap_ok_runs_post_swap_before_reconcile() { let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); q.complete_node(id, reconcile.node_id, Ok(())); - settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -933,22 +832,10 @@ 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), "fully-queued dag cancels"); - // The operator sees `Cancelled` the moment the cancel returns — the spared - // tail is still `Pending`, and a DAG must not read `Queued` back to the - // operator who just cancelled it (the dashboard renders this roll-up from - // the snapshot `post_rebuild_queue_cancel` emits synchronously). - assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap"); - // Every work node is `Cancelled`, but the tail is spared so it can still - // report the cancellation — so it is the one thing left to claim. - let tail = claim_one(&q); - assert_eq!(tail.kind.as_str(), "emit_rebuilt"); - assert_eq!( - tail.deps_state(), - State::Cancelled, - "the spared tail sees its deps cancelled, so it emits no `Rebuilt`" - ); - q.complete_node(id, tail.node_id, Ok(())); + // 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); assert!(q.claim_ready().is_empty()); } @@ -958,24 +845,23 @@ 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); } -/// A cancelled power op must run **no** compensating node — not even one that +/// A cancelled power op must fire **no** compensating hook — not even one that /// carries a `SetWanted` head. /// -/// Now structural rather than a property of a hook enum: a power op emits no -/// tail node at all, so once its work nodes cancel there is simply nothing left -/// to claim. `cancel` also refuses unless every work node is still `Pending` -/// (`cancel_refuses_running_dag`), so a `Cancelled` DAG provably never executed -/// a node: its `SetWanted` never ran and the agent's intent still reads whatever +/// `cancel` refuses unless every work node is still `Pending` +/// (`cancel_refuses_running_dag`) and a cancel *cascade* rolls up `Failed` +/// rather than `Cancelled`, so a `Cancelled` DAG provably never executed a +/// node: its `SetWanted` never ran and the agent's intent still reads whatever /// the operator last set. A "revert" instead writes the agent's *observed* /// state, which for a down-but-`wanted = Up` agent (crashed, or caught /// mid-bounce) flips the intent to `Offline` and leaves it /// deliberately-stopped as far as reconcile and crash-watch are concerned. #[test] -fn cancelled_power_op_runs_no_compensating_node() { +fn cancelled_power_op_fires_no_hook() { for graceful in [false, true] { for running in [false, true] { let targets = vec![("agent-a".to_owned(), running)]; @@ -1010,12 +896,12 @@ fn cancelled_power_op_runs_no_compensating_node() { ); let q = JobQueue::new(1); let id = submit(&q, spec); - assert!(q.cancel(id), "cancelled while queued"); - assert_eq!(state_of(&q, id), State::Cancelled); - assert!( - q.claim_ready().is_empty(), + let summary = q.cancel(id).expect("cancelled while queued"); + assert_eq!(summary.state, State::Cancelled); + assert_eq!( + summary.hook, None, "cancelled {name} (graceful={graceful}, running={running}) must \ - leave nothing to run — a power op emits no tail node" + fire no hook — no node of it ever ran" ); } } @@ -1034,10 +920,13 @@ fn dag_settles_terminal_and_releases_lease_after_work() { q.complete_node(id, stop.node_id, Ok(())); let rec = claim_one(&q); assert_eq!(rec.kind.as_str(), "reconcile"); - // Completing the last work node rolls the container up terminal. A power op - // has no tail node, so nothing is left to claim. - q.complete_node(id, rec.node_id, Ok(())); - assert!(q.claim_ready().is_empty(), "no tail node to claim"); + // 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); + 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. @@ -1049,44 +938,31 @@ fn dag_settles_terminal_and_releases_lease_after_work() { assert_eq!(c.dag_id, next); } -/// A DAG cancelled while fully queued must still **run its tail**, or a queued -/// approval DAG cancelled by the operator would dangle its approval forever. -/// -/// This is the load-bearing case for sparing tails in [`JobQueue::cancel`]: the -/// work nodes all cancel, but `ResolveApproval` is weak-edged, so a `Cancelled` -/// dep satisfies its edge and it becomes claimable instead of being cancelled -/// along with everything else. It reads `Cancelled` off its own deps and resolves -/// the approval as "cancelled before completion". +/// A DAG cancelled while fully queued must still surface a terminal +/// roll-up for the scheduler's hooks — otherwise a queued approval +/// DAG cancelled by the operator would dangle its approval forever. #[test] -fn cancelled_dag_still_runs_its_approval_tail() { +fn cancelled_dag_finalizes_with_terminal_rollup() { let q = JobQueue::new(1); let id = submit( &q, templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), ); - assert!(q.cancel(id), "fully-queued dag cancels"); - let tail = claim_one(&q); - assert_eq!(tail.dag_id, id); - assert_eq!(tail.kind.as_str(), "resolve_approval"); - assert!( - matches!(tail.kind, NodeKind::ResolveApproval { approval_id: 7 }), - "the tail carries the approval to resolve, got {:?}", - tail.kind - ); - assert_eq!( - tail.deps_state(), - State::Cancelled, - "so the executor resolves the approval as cancelled, not as a failure" - ); - assert_eq!(tail.deps_error(), None, "a cancelled dep carries no error"); - q.complete_node(id, tail.node_id, Ok(())); - assert_eq!(state_of(&q, id), State::Cancelled); - // Unrelated later activity doesn't disturb the settled DAG. + // 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)); + // 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")); let c = claim_one(&q); assert_eq!(c.dag_id, other); q.complete_node(other, c.node_id, Err("boom".to_owned())); - assert_eq!(state_of(&q, id), State::Cancelled); + assert_eq!( + q.terminal_summary(id).map(|t| t.state), + Some(State::Cancelled) + ); } // ---- approval deploy subtree ---- @@ -1130,12 +1006,13 @@ fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 7, State::Failed); + let summary = q.terminal_summary(id).expect("dag terminal"); assert_eq!( - state_of(&q, id), + summary.state, State::Failed, "an Ok tail must not launder a failed deploy into a success" ); + assert_eq!(summary.approval_id, Some(7)); } /// The deploy's happy path: `DeployApply` does not build. It grows the ordinary @@ -1203,8 +1080,9 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { assert!(matches!(tail.kind, NodeKind::DeployTail { .. })); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 11, State::Done); - assert_eq!(state_of(&q, id), State::Done); + let summary = q.terminal_summary(id).expect("dag terminal"); + assert_eq!(summary.state, State::Done); + assert_eq!(summary.approval_id, Some(11)); } /// A failure *inside* the grafted rebuild is the failure mode the subgraph @@ -1254,14 +1132,12 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 13, State::Failed); - assert_eq!(state_of(&q, id), State::Failed); + let summary = q.terminal_summary(id).expect("dag terminal"); + assert_eq!(summary.state, State::Failed); assert_eq!( q.first_error(id).as_deref(), Some("profile swap failed"), - "the tail annotates failed/ with this — and it is also what - `exec::failure_reason` falls back to, since the tail's own dep is a - group root that rolled up Failed and so carries no error itself" + "the tail annotates failed/ with this" ); } @@ -1292,8 +1168,9 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 9, State::Failed); - assert_eq!(state_of(&q, id), State::Failed); + let summary = q.terminal_summary(id).expect("dag terminal"); + assert_eq!(summary.state, State::Failed); + assert_eq!(summary.approval_id, Some(9)); } // ---- build logs, history ---- @@ -1342,7 +1219,8 @@ fn history_evicts_oldest_terminals_past_flat_cap() { // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG // drops off the wire entirely, but a `Failed` one is retained (+ // history-capped) so the operator can still triage it. Completing the - // node rolls the container up terminal. + // 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, Err("boom".to_owned())); ids.push(id); } @@ -1432,8 +1310,8 @@ fn spawn_shape_provision_create_dropin_reconcile() { assert_eq!(c.approval_id, Some(7)); q.complete_node(id, c.node_id, Ok(())); } - settle_approval_tail(&q, id, 7, State::Done); - assert_eq!(state_of(&q, id), State::Done); + let report_terminal = state_of(&q, id); + assert_eq!(report_terminal, State::Done); } #[test] @@ -1464,7 +1342,6 @@ fn perm_change_shape_prefixes_rebuild_chain() { assert_eq!(c.kind.as_str(), expected); q.complete_node(id, c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 90d3e555..0ba9460a 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -344,7 +344,8 @@ fn submit_boot_tree( let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they - // land; the boot DAG as a whole has no terminal side effect, so no tail. + // land; the boot DAG as a whole has no terminal side effect. + hook: None, source: Source::AutoUpdate, reason, approval_id: None, diff --git a/hive-host-sock/src/jobs.rs b/hive-host-sock/src/jobs.rs index 5fa5d24b..1f41ea87 100644 --- a/hive-host-sock/src/jobs.rs +++ b/hive-host-sock/src/jobs.rs @@ -170,21 +170,10 @@ impl DagView { /// Roll-up state derived from the node set — the shared derivation every /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's /// JS render and the host agree: `Failed` if any node failed, else - /// **`Cancelled` if any cancelled**, else `Running` if any running, else - /// `Queued` if any queued, else `Done`. `Done` nodes are excluded + /// `Running` if any running, else `Queued` if any queued, else + /// `Cancelled` if any cancelled, else `Done`. `Done` nodes are excluded /// from the wire, so a DAG that is *entirely* done isn't sent at all — /// its absence from the snapshot is what signals completion. - /// - /// `Cancelled` outranks both `Running` and `Queued` because a cancelled DAG - /// still has its weak-edged tail node to run (it reports the cancellation), - /// so `Queued`-then-`Running` would flicker back at the operator who just - /// cancelled it and read as "the cancel didn't take". Outside that window - /// the states barely co-occur: a cancel *cascade* originates at a `Failed` - /// node, which returns early above. - /// - /// This ordering matches `frontend/packages/dashboard/src/builds.js`'s - /// `rollupState`, which has always ranked cancelled second — the two had - /// silently disagreed, and this is the side that was wrong. #[must_use] pub fn rollup_state(&self) -> State { let mut any_running = false; @@ -199,12 +188,12 @@ impl DagView { State::Done => {} } } - if any_cancelled { - State::Cancelled - } else if any_running { + if any_running { State::Running } else if any_queued { State::Queued + } else if any_cancelled { + State::Cancelled } else { State::Done }