diff --git a/docs/coordinator.md b/docs/coordinator.md index 9e3ac7c8..316326f4 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -20,7 +20,7 @@ fast-lane follow-up, the meta-update cascade pre-enqueue — are all just DAG The **DAG** is the unit of cancel / approval-resolution and the dashboard group; the **node** is the unit of scheduling / execution / -build-log / step label. Deps are intra-DAG edges only (`AfterOk` by default: +build-log. Deps are intra-DAG edges only (`AfterOk` by default: the dep must succeed, a failed/cancelled dep cancels the dependent — cancel-downstream). Cross-DAG ordering comes from the per-agent lease, never from edges between DAGs. Submit-time validation (petgraph `toposort`) @@ -167,7 +167,7 @@ without a row are seeded from observed state on first touch (running ⇒ The admin-socket responses carry the submitted DAG ids; `hivectl` polls `HostRequest::QueueDag` (~1s) and prints a progress line per DAG — roll-up -glyph, template, agent, node chain with the running node's step label — so +glyph, template, agent, node chain — so CLI verbs block until their jobs finish (`--no-wait` opts out; failures exit non-zero). Nodes appended in-DAG (a `MetaLock` growing per-agent rebuild subgraphs, a `Reconcile` fanning its `Start`/`Stop`) join the same DAG, so @@ -248,11 +248,12 @@ cancelled-while-queued, which fails the approval instead of dangling it). `DagView` carries the entry-level fields (`id`, `kind` = template string, roll-up `state`, `source`, `reason`, timestamps, `inputs`, `approval_id`) plus `nodes: [NodeView…]` — per-node `agent`, `kind`, `deps`, -`state`, `step`, `build_log_id`, timestamps, `error`. There is **no +`state`, `build_log_id`, timestamps, `error`. There is **no DAG-level `agent`** (agent is per-node, so a DAG can span agents); consumers -derive a DAG's agent(s) from its nodes. Step labels and build logs are -**per-node**; the dashboard renders the node chain on each queue card and -keys the live-log panel off the running node. +derive a DAG's agent(s) from its nodes. The node kind *is* the phase label — +there is no separate sub-step string; build logs are **per-node**. The +dashboard renders the node chain on each queue card and keys the live-log +panel off the running node. --- diff --git a/docs/web-ui/dashboard.md b/docs/web-ui/dashboard.md index efb6c51b..7556155b 100644 --- a/docs/web-ui/dashboard.md +++ b/docs/web-ui/dashboard.md @@ -220,12 +220,10 @@ op for the same agent collapses into the existing entry. All timing labels stay live: running entries tick elapsed seconds every second; queued and terminal ("done N ago" / "failed N ago") labels tick every 30s so keyed rows never show stale timestamps as they persist across -`rebuild_queue_changed` snapshots. When the worker has annotated the -current phase a cyan `↳ ` sub-line appears under the main row -showing the in-flight step name (e.g. `↳ meta prepare_deploy` → -`↳ nixos-container update` → `↳ finalize deploy`). Terminal -transitions clear `step` on the backend so Done / Failed rows don't -render stale labels. +`rebuild_queue_changed` snapshots. The in-flight phase is the running +**node's kind** in the node chain — there is no separate sub-step +label (the old cyan `↳ ` sub-line went away with the DAG queue: +each phase is its own node now). Queued entries carry a `✗` cancel button on the right edge; running / done / failed / cancelled entries don't show it — the backend refuses cancellation for non-`Queued` rows anyway @@ -242,7 +240,7 @@ It's keyed to the running entry's `build_log_id` and opens one `EventSource` to `GET /api/build-logs/id/{id}/stream` (the same stream the BUILD L0GS tab uses; the stream replays accumulated output on connect). It lives in its own container outside `#rebuild-queue-section` -so the queue's per-row re-render (rows rebuild as the `step` advances) +so the queue's per-row re-render (rows rebuild as nodes advance) never tears down the open stream; it hides when nothing is building and each row keeps its `logs →` link out to the full build log history. @@ -912,7 +910,7 @@ agent is stale. Banner pulses on each broker SSE event `queued` / `running` entries, a compact amber banner sits above the container list: `◐ build queue — N running · M queued — view queue →` (the link goes to the BU1LDS page's R3BU1LD QU3U3). It replaces the -old per-transient spinner list; the actual running step for each +old per-transient spinner list; the actual running node for each agent is already shown on its card (transient + in-flight-queue badges), so the top of the tab only needs the at-a-glance summary. diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index e9a1df93..7b442178 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -211,17 +211,12 @@ fn deploy_ctx(coord: &Coordinator, approval_id: i64) -> Result { /// fails, or if the eval-verify of the reviewed commit fails. Every one of /// these leaves the forge and `main` untouched, so the node is safely /// retryable. -pub async fn run_deploy_merge_verify( - coord: &Arc, - queue_entry_id: Option, - approval_id: i64, -) -> Result<()> { +pub async fn run_deploy_merge_verify(coord: &Arc, approval_id: i64) -> Result<()> { let ctx = deploy_ctx(coord, approval_id)?; let pr = ctx.pr; let reviewed = ctx.reviewed.as_str(); // 1. Drift gate: the live PR head must still equal what was reviewed. - coord.set_queue_step(queue_entry_id, "verify PR head"); let head = crate::forge::pr_head_sha(&ctx.repo, pr) .await .map_err(|e| anyhow::anyhow!("read PR #{pr} head: {e}"))?; @@ -232,13 +227,11 @@ pub async fn run_deploy_merge_verify( } // 2. Fetch the reviewed head into applied so ff/verify/deploy resolve it. - coord.set_queue_step(queue_entry_id, "fetch PR head"); crate::forge::fetch_pr_head_into_applied(&ctx.repo, pr) .await .map_err(|e| anyhow::anyhow!("fetch PR #{pr} head into applied: {e}"))?; // 3. Eval-verify BEFORE the irreversible merge (bad nix fails fast here). - coord.set_queue_step(queue_entry_id, "verify proposal (eval)"); crate::meta::verify_commit(ctx.approval.agent.as_str(), &ctx.applied_dir, reviewed) .await .map_err(|e| anyhow::anyhow!("verify merge head {reviewed}: {e:#}"))?; @@ -264,11 +257,7 @@ pub async fn run_deploy_merge_verify( /// fast-forwarding `applied/main` / `meta::prepare_deploy` fails. From the merge /// onward a failure is *not* retryable on its own — [`run_deploy_tail`] runs /// `AfterAny` to compensate. -pub async fn run_deploy_apply( - coord: &Arc, - queue_entry_id: Option, - approval_id: i64, -) -> Result<()> { +pub async fn run_deploy_apply(coord: &Arc, approval_id: i64) -> Result<()> { let ctx = deploy_ctx(coord, approval_id)?; let agent = ctx.approval.agent.as_str(); let pr = ctx.pr; @@ -285,7 +274,6 @@ pub async fn run_deploy_apply( // both advances `main` to the reviewed head and marks the PR merged — no // direct push to the protected branch. A failure here means `main` was NOT // advanced, so it's fatal: we must not deploy a head the forge didn't merge. - coord.set_queue_step(queue_entry_id, "fast-forward-merge PR"); match crate::forge::merge_config_pr_ff(&ctx.repo, pr, &ctx.reviewed).await { Ok(()) => {} Err(crate::forge::ForgeMergeError::HeadDrift { expected, actual }) => bail!( @@ -294,14 +282,7 @@ pub async fn run_deploy_apply( Err(e) => bail!("ff-merge PR #{pr}: {e}"), } - prepare_applied_target( - coord, - agent, - &ctx.applied_dir, - &ctx.reviewed, - queue_entry_id, - ) - .await + prepare_applied_target(agent, &ctx.applied_dir, &ctx.reviewed).await } /// `DeployTail` node body — compensation + bookkeeping, `AfterAny` the apply @@ -371,7 +352,6 @@ pub async fn run_deploy_tail( } } - coord.set_queue_step(queue_entry_id, "roll back applied/main"); if let Err(e) = lifecycle::git_update_ref(&applied_dir, "refs/heads/main", &prev_main).await { @@ -389,7 +369,6 @@ pub async fn run_deploy_tail( } } - coord.set_queue_step(queue_entry_id, "forge push"); if let Err(e) = crate::forge::push_config(agent).await { tracing::warn!(%agent, error = ?e, "forge: push_config after merge failed"); } @@ -789,13 +768,10 @@ fn finish_approval( /// (a merge is never a first spawn), so there's no `sync_agents` step — the /// operator `Spawn` flow owns first-time meta registration. async fn prepare_applied_target( - coord: &Arc, agent: &str, applied_dir: &std::path::Path, target: &str, - queue_entry_id: Option, ) -> Result<()> { - coord.set_queue_step(queue_entry_id, "fast-forward applied/main"); // Fast-forward applied/main to target + sync the working tree. Meta input // pins `?ref=main`, so this is what makes nix re-lock to the target commit // on the prepare_deploy step below. @@ -806,7 +782,6 @@ async fn prepare_applied_target( .await .map_err(|e| anyhow::anyhow!("read-tree to main: {e:#}"))?; - coord.set_queue_step(queue_entry_id, "meta prepare_deploy"); // Phase 1 of the meta two-phase deploy: relock without committing. The // staged lock then stays uncommitted across the whole appended rebuild — // which is why the `MetaWindow` is held by the deploy root, not by a node. @@ -835,27 +810,20 @@ async fn prepare_applied_target( /// `meta::finalize_deploy` is deliberately *not* an error: the container is /// already running the new config by then, and the staged `flake.lock` it /// couldn't commit is something the operator can land by hand. -pub async fn run_finalize_deploy( - coord: &Arc, - queue_entry_id: Option, - approval_id: i64, -) -> Result<()> { +pub async fn run_finalize_deploy(coord: &Arc, approval_id: i64) -> Result<()> { let ctx = deploy_ctx(coord, approval_id)?; let agent = ctx.approval.agent.as_str(); let target = ctx.reviewed.as_str(); - coord.set_queue_step(queue_entry_id, "drop rollback ref"); lifecycle::git_delete_ref(&ctx.applied_dir, &rollback_ref(approval_id)) .await .map_err(|e| anyhow::anyhow!("drop rollback ref for approval {approval_id}: {e:#}"))?; - coord.set_queue_step(queue_entry_id, "plant deployed tag"); let tag = format!("deployed/{approval_id}"); lifecycle::git_tag(&ctx.applied_dir, &tag, target) .await .map_err(|e| anyhow::anyhow!("plant {tag}: {e:#}"))?; - coord.set_queue_step(queue_entry_id, "meta finalize_deploy"); if let Err(e) = crate::meta::finalize_deploy(agent, target, &tag).await { tracing::warn!(%agent, approval_id, error = ?e, "meta finalize_deploy failed"); } diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index f9848e16..d44fe4cc 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -666,23 +666,6 @@ impl Coordinator { }); } - /// Update the `step` label on the currently-running node of DAG - /// `id` and (if it actually changed) re-emit the queue snapshot so - /// the dashboard renders the new phase. DAG-id-only surface for the - /// approval-deploy bodies in `actions.rs`, which don't know their node id. - /// The lookup is still exact for them: a deploy DAG is a strictly - /// sequential chain whose resource-holding root sits in `Finishing` while - /// the phases run, so at most one node is ever `Running`. Queue executors - /// that do know their node id use the precise per-node sink in - /// `job_queue::exec` instead. No-op when `id` is `None` (callers not - /// running from the queue) or when nothing is `Running`. - pub fn set_queue_step(self: &Arc, id: Option, step: &str) { - let Some(id) = id else { return }; - if self.job_queue.set_step_running(id, step) { - self.emit_rebuild_queue_snapshot(); - } - } - /// Subscribe to the shutdown watch channel. Background tasks call /// this at spawn time and break their loop when the receiver /// transitions to `true` (via `Coordinator::request_shutdown`). diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 6d038de2..2b8f55e3 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -43,7 +43,7 @@ pub struct NodeOutput { pub append_subgraph: Vec>, } -/// Step-label + build-log sink for one claimed node. +/// Build-log sink for one claimed node. struct Ctx<'a> { coord: &'a Arc, dag_id: u64, @@ -51,16 +51,6 @@ struct Ctx<'a> { } impl Ctx<'_> { - fn step(&self, step: &str) { - if self - .coord - .job_queue - .set_step(self.dag_id, self.node_id, step) - { - self.coord.emit_rebuild_queue_snapshot(); - } - } - fn build_log(&self, log_id: i64) { if self .coord @@ -85,20 +75,20 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await, NodeKind::Prebuild { .. } => run_prebuild(claim, &ctx).await, NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await, - NodeKind::PostSwap { .. } => run_post_swap(coord, claim, &ctx).await, - NodeKind::Provision { .. } => run_provision(coord, claim, &ctx).await, - NodeKind::Create { .. } => run_create(claim, &ctx).await, + NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await, + NodeKind::Provision { .. } => run_provision(coord, claim).await, + NodeKind::Create { .. } => run_create(claim).await, NodeKind::MetaLock { sweep, fanout } => { - run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await + run_meta_lock(coord, claim, *sweep, fanout.clone()).await } NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await, - NodeKind::Start { .. } => run_start(coord, claim, &ctx).await, - NodeKind::Stop { .. } => run_stop(coord, claim, &ctx).await, - NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim, &ctx).await, - NodeKind::Signal { .. } => Ok(run_signal(coord, claim, &ctx)), - NodeKind::Drain { .. } => run_drain(coord, claim, &ctx).await, + NodeKind::Start { .. } => run_start(coord, claim).await, + NodeKind::Stop { .. } => run_stop(coord, claim).await, + NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim).await, + NodeKind::Signal { .. } => Ok(run_signal(coord, claim)), + NodeKind::Drain { .. } => run_drain(coord, claim).await, NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await, - NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim, &ctx).await, + NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim).await, NodeKind::DeployWindow { .. } => run_deploy_window(claim), NodeKind::MergeVerify { .. } => run_merge_verify(coord, claim).await, NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await, @@ -239,12 +229,9 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result { // stopped agent has no uptime to preserve, so skip the (expensive) // eval and let the downstream `Swap` build inline. if crate::lifecycle::is_running(name).await { - ctx.step("nix build"); let flake_ref = format!("{}#{name}", crate::paths::meta_root().display()); crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id)) .await?; - } else { - ctx.step("skipped prebuild (container down)"); } Ok(NodeOutput::default()) } @@ -261,11 +248,10 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); - let result = - crate::lifecycle::swap_update(name, &hive, &paths, &|step| ctx.step(step), &|log_id| { - ctx.build_log(log_id); - }) - .await; + let result = crate::lifecycle::swap_update(name, &hive, &paths, &|log_id| { + ctx.build_log(log_id); + }) + .await; // On success the Ok-only bookkeeping tail (rev marker, forge/matrix // sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node, // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded @@ -284,11 +270,7 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res /// means the profile swap succeeded. Store/forge/matrix work only — no nix /// build (build-slot-exempt); the agent lease taken at `Swap` is still held /// (the whole chain up to `Reconcile` is one agent's subgraph). -async fn run_post_swap( - coord: &Arc, - claim: &Claim, - ctx: &Ctx<'_>, -) -> Result { +async fn run_post_swap(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake) && let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev) @@ -298,7 +280,6 @@ async fn run_post_swap( // The `Rebuilt` manager event fires exactly once per DAG from the // terminal hook — emitting ok here and letting a failed tail `Reconcile` // add a contradictory !ok would double-report the same rebuild. - ctx.step("forge sync"); // Full forge + matrix sync on every successful rebuild so the rebuild // path is equivalent to the startup sweep: tokens, config-repo mirror, // meta access all recover without a hive-c0re restart. @@ -317,16 +298,11 @@ async fn run_post_swap( /// subvolume, and the meta `sync_agents` registration. Runs under the /// deploy window (`NodeKind::needs_meta_window`) so its commit can't /// land inside another node's staged deploy window. -async fn run_provision( - coord: &Arc, - claim: &Claim, - ctx: &Ctx<'_>, -) -> Result { +async fn run_provision(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); - ctx.step("provisioning"); crate::lifecycle::provision_container(name, &hive, &paths).await?; Ok(NodeOutput::default()) } @@ -337,8 +313,7 @@ async fn run_provision( /// dir creation and MCP listener registration are deferred to the tail /// `Reconcile` (`converge_start_preamble` + `register_agent`) so this /// node stays purely "create", not "create + start". -async fn run_create(claim: &Claim, ctx: &Ctx<'_>) -> Result { - ctx.step("nixos-container create"); +async fn run_create(claim: &Claim) -> Result { crate::lifecycle::create_only(&claim.agent).await?; Ok(NodeOutput::default()) } @@ -350,12 +325,10 @@ async fn run_create(claim: &Claim, ctx: &Ctx<'_>) -> Result { async fn run_meta_lock( coord: &Arc, claim: &Claim, - ctx: &Ctx<'_>, sweep: bool, fanout: Option>, ) -> Result { if sweep { - ctx.step("nix flake update hyperhive"); if let Err(e) = crate::meta::lock_update_hyperhive().await { tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed"); } @@ -371,7 +344,6 @@ async fn run_meta_lock( return Ok(NodeOutput { append_subgraph }); } let _progress = coord.meta_update_guard(); - ctx.step("nix flake update"); crate::meta::lock_update(&claim.inputs).await?; // Lock file changed — meta-inputs panel re-renders. crate::dashboard::emit_meta_inputs_snapshot(coord); @@ -424,7 +396,7 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result, claim: &Claim, ctx: &Ctx<'_>) -> Result { +async fn run_start(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; // Node-local transient only when the DAG holds none (the // boot-reconcile template); a rebuild/spawn/etc. DAG's lease-window @@ -441,7 +413,6 @@ async fn run_start(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Re let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?; - ctx.step("nixos-container start"); crate::lifecycle::start_with_fallback(token).await?; // Bind the MCP listener immediately after starting the container. // The preamble created the runtime dir; the container is now coming @@ -456,13 +427,12 @@ async fn run_start(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Re /// Mechanical container stop — the sub-step a `Reconcile` planner fans /// out when it observes `wanted = Offline` and the container up. -async fn run_stop(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result { +async fn run_stop(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; let _guard = claim .transient .is_none() .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping)); - ctx.step("nixos-container stop"); crate::lifecycle::kill(name).await?; coord.unregister_agent(name); coord.notify_manager(&hive_sh4re::HelperEvent::Killed { @@ -474,11 +444,7 @@ async fn run_stop(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res /// Mechanical stop for the profile swap. Never *changes* `wanted`; /// noop when already stopped. -async fn run_stop_for_update( - coord: &Arc, - claim: &Claim, - ctx: &Ctx<'_>, -) -> Result { +async fn run_stop_for_update(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; if crate::lifecycle::is_running(name).await { // Seed a missing agent_power row from the PRE-stop observation @@ -488,7 +454,6 @@ async fn run_stop_for_update( if let Err(e) = coord.power.get_or_seed(name, true) { tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed"); } - ctx.step("nixos-container stop"); crate::lifecycle::kill(name).await?; coord.rescan_containers_and_emit().await; } @@ -504,12 +469,10 @@ async fn run_stop_for_update( /// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at /// the top of its loop — a paused agent has no turn in flight, so there /// is nothing to checkpoint. -fn run_signal(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> NodeOutput { +fn run_signal(coord: &Arc, claim: &Claim) -> NodeOutput { if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) { - ctx.step("graceful stop: agent paused, nothing to drain"); return NodeOutput::default(); } - ctx.step("graceful stop: signalling agent"); coord.mark_graceful_stop(&claim.agent); coord.kick_agent(&claim.agent, "graceful stop requested"); NodeOutput::default() @@ -518,9 +481,8 @@ fn run_signal(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> NodeOut /// Await the harness clearing the fence (`GracefulStopComplete`) or /// the timeout — either way the downstream `Reconcile` proceeds with /// the actual stop. -async fn run_drain(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result { +async fn run_drain(coord: &Arc, claim: &Claim) -> Result { let name = &claim.agent; - ctx.step("graceful stop: draining"); let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT; while coord.is_graceful_stop_pending(name) { if std::time::Instant::now() >= deadline { @@ -549,18 +511,13 @@ async fn run_write_dropin(coord: &Arc, claim: &Claim) -> Result, - claim: &Claim, - ctx: &Ctx<'_>, -) -> Result { +async fn run_write_perm_file(coord: &Arc, claim: &Claim) -> Result { use super::model::PermPayload; let name = &claim.agent; // The perm file payload rides the node itself (the only consumer). let NodeKind::WritePermFile { payload, .. } = &claim.kind else { anyhow::bail!("run_write_perm_file on a non-WritePermFile node"); }; - ctx.step("writing + committing perm file"); // Runs under the deploy window (`NodeKind::needs_meta_window`): a // perm commit landing inside another node's staged prepare→finalize // window would sweep the staged deploy lock into its commit (the @@ -622,7 +579,7 @@ fn run_deploy_window(claim: &Claim) -> Result { /// failure here cancel-cascades the rest of the subtree with the forge and the /// applied repo exactly as they were. async fn run_merge_verify(coord: &Arc, claim: &Claim) -> Result { - crate::actions::run_deploy_merge_verify(coord, Some(claim.dag_id), deploy_approval_id(claim)?) + crate::actions::run_deploy_merge_verify(coord, deploy_approval_id(claim)?) .await .map(|()| NodeOutput::default()) } @@ -637,7 +594,7 @@ async fn run_merge_verify(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { - crate::actions::run_deploy_apply(coord, Some(claim.dag_id), deploy_approval_id(claim)?).await?; + crate::actions::run_deploy_apply(coord, deploy_approval_id(claim)?).await?; Ok(NodeOutput { append_subgraph: vec![super::templates::deploy_rebuild_nodes(claim.kind.agent())], }) @@ -647,7 +604,7 @@ async fn run_deploy_apply(coord: &Arc, claim: &Claim) -> Result` tag, commit /// the staged lock. async fn run_finalize_deploy(coord: &Arc, claim: &Claim) -> Result { - crate::actions::run_finalize_deploy(coord, Some(claim.dag_id), deploy_approval_id(claim)?) + crate::actions::run_finalize_deploy(coord, deploy_approval_id(claim)?) .await .map(|()| NodeOutput::default()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index eee3dbc7..877ca89a 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -99,11 +99,10 @@ pub struct TerminalDag { /// Per-node runtime metadata the crate graph doesn't carry. Lifecycle /// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node` -/// itself now, so only the two host-side extras remain: the live sub-step -/// label and the build-log row link (the client fetches the log by node id). +/// itself now, so only the build-log row link remains host-side (the +/// client fetches the log by node id). #[derive(Debug, Default, Clone)] struct NodeRuntime { - step: Option, build_log_id: Option, } @@ -129,8 +128,8 @@ struct DagMeta { /// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG. struct QueueInner { sched: Scheduler, - /// Per-node runtime metadata (build-log id, step, timestamps, error) — - /// mutable after insert, so it can't ride the immutable node payload. + /// Per-node runtime metadata (the build-log id) — mutable after + /// insert, so it can't ride the immutable node payload. node_rt: HashMap, } @@ -405,14 +404,11 @@ impl JobQueue { 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 - // copy. We only clear the live sub-step label here. + // copy, so there is nothing to clear here. let outcome = match result { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(truncate_error(&e)), }; - if let Some(rt) = inner.node_rt.get_mut(&node_id) { - rt.step = None; - } 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, @@ -461,35 +457,6 @@ impl JobQueue { terminal } - /// Set the step label on a `Running` node. Returns `true` when it changed. - pub fn set_step(&self, dag_id: u64, node_id: NodeId, step: &str) -> bool { - let mut inner = self.lock(); - if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) { - return false; - } - let rt = inner.node_rt.entry(node_id).or_default(); - if rt.step.as_deref() == Some(step) { - return false; - } - rt.step = Some(step.to_owned()); - true - } - - /// Set the step label on the DAG's currently-running node — the DAG-id-only - /// compatibility surface for the opaque approval pipeline. - pub fn set_step_running(&self, dag_id: u64, step: &str) -> bool { - let mut inner = self.lock(); - let Some(node_id) = inner.running_node_of(dag_id) else { - return false; - }; - let rt = inner.node_rt.entry(node_id).or_default(); - if rt.step.as_deref() == Some(step) { - return false; - } - rt.step = Some(step.to_owned()); - true - } - /// Link a `build_logs` row to a specific `Running` node. pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { let mut inner = self.lock(); @@ -670,15 +637,6 @@ impl QueueInner { }) } - /// The DAG's currently-running work node, if any (the opaque approval - /// pipeline's single-node DAGs make this exact). - fn running_node_of(&self, dag_id: u64) -> Option { - let container = self.container(dag_id)?; - self.subtree(container) - .into_iter() - .find(|&id| self.node_running(id)) - } - /// 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 diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 4323ab35..d393018d 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -7,7 +7,7 @@ //! //! Two levels: the **DAG** is the unit of cancel / approval-resolution //! and the dashboard group; the **node** is the unit of scheduling / -//! execution / build-log / step label, and carries its own `agent` (a +//! execution / build-log, and carries its own `agent` (a //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 32b2fdec..1327515a 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1134,29 +1134,7 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { assert_eq!(summary.approval_id, Some(9)); } -// ---- steps, build logs, history ---- - -#[test] -fn set_step_only_on_running_and_signals_change() { - let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); - assert!( - !q.set_step_running(id, "too early"), - "no running node yet → refused" - ); - let c = claim_one(&q); - assert!(q.set_step(id, c.node_id, "nix build")); - assert!( - !q.set_step(id, c.node_id, "nix build"), - "same label → false" - ); - assert!(q.set_step(id, c.node_id, "next phase")); - assert!(q.set_step_running(id, "via running lookup")); - // `step` is host-side only now (off the wire); completion clears it - // internally, but there's no wire field to observe — the return-value - // contract above (running-gating + change signalling) is the behaviour. - q.complete_node(id, c.node_id, Ok(())); -} +// ---- build logs, history ---- #[test] fn set_build_log_id_links_running_node() { diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index f5818f23..c06b0985 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -273,11 +273,9 @@ pub async fn swap_update( name: &str, hive: &HiveEnv, paths: &AgentPaths, - on_step: &(dyn Fn(&str) + Send + Sync), on_build_log_id: &(dyn Fn(i64) + Send + Sync), ) -> Result<()> { write_dropins(name, hive, paths).await?; - on_step("nixos-container update"); priv_run_inner("update", name, Some(on_build_log_id)).await }