diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 1d545efd..76998cf1 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -108,6 +108,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { /// + the lifecycle event (`Rebuilt` / `Spawned` for first-spawn). pub async fn run_approval_apply_commit( coord: &Arc, + queue_entry_id: Option, approval_id: i64, ) -> Result<()> { let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?; @@ -115,6 +116,7 @@ pub async fn run_approval_apply_commit( let applied_dir = Coordinator::agent_applied_dir(&approval.agent); let claude_dir = Coordinator::agent_claude_dir(&approval.agent); let notes_dir = Coordinator::agent_notes_dir(&approval.agent); + coord.set_queue_step(queue_entry_id, "apply commit"); let (result, terminal_tag, is_first_spawn) = run_apply_commit( coord, &approval, @@ -122,12 +124,15 @@ pub async fn run_approval_apply_commit( &applied_dir, &claude_dir, ¬es_dir, + queue_entry_id, ) .await; + coord.set_queue_step(queue_entry_id, "forge push"); if let Err(e) = crate::forge::push_config(&approval.agent).await { tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed"); } if is_first_spawn && result.is_ok() { + coord.set_queue_step(queue_entry_id, "first-spawn forge bootstrap"); forge_after_first_spawn(coord, &approval.agent).await; } // `finish_approval` returns the original `result` so the queue @@ -175,10 +180,12 @@ async fn run_approval_schedule_prompt( /// `inputs` — the queue copy is for dashboard display only. pub async fn run_approval_update_meta_inputs( coord: &Arc, + queue_entry_id: Option, approval_id: i64, ) -> Result<()> { let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::UpdateMetaInputs)?; let inputs: Vec = serde_json::from_str(&approval.commit_ref).unwrap_or_default(); + coord.set_queue_step(queue_entry_id, "nix flake update"); let result = crate::meta::lock_update(&inputs).await; finish_approval(coord, &approval, result, None, false) } @@ -188,7 +195,11 @@ pub async fn run_approval_update_meta_inputs( /// `lifecycle::spawn` (the deprecated direct-spawn path). Synchronous /// in the queue worker — the previous `tokio::spawn` wrapper is gone /// (the queue worker itself is the async task). -pub async fn run_approval_spawn(coord: &Arc, approval_id: i64) -> Result<()> { +pub async fn run_approval_spawn( + coord: &Arc, + queue_entry_id: Option, + approval_id: i64, +) -> Result<()> { let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::Spawn)?; let agent_dir = coord.ensure_runtime(&approval.agent)?; let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent); @@ -199,6 +210,7 @@ pub async fn run_approval_spawn(coord: &Arc, approval_id: i64) -> R // the worker is doing the actual nixos-container create. Auto-clears // on the function's scope exit (success or panic). let _guard = coord.transient_guard(&approval.agent, TransientKind::Spawning); + coord.set_queue_step(queue_entry_id, "lifecycle::spawn"); let result = lifecycle::spawn( &approval.agent, &coord.hyperhive_flake, @@ -213,15 +225,19 @@ pub async fn run_approval_spawn(coord: &Arc, approval_id: i64) -> R ) .await; if result.is_ok() { + coord.set_queue_step(queue_entry_id, "forge user"); if let Err(e) = crate::forge::ensure_user_for(&approval.agent).await { tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_user after spawn failed"); } + coord.set_queue_step(queue_entry_id, "forge config repo"); if let Err(e) = crate::forge::ensure_config_repo(&approval.agent).await { tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_config_repo after spawn failed"); } + coord.set_queue_step(queue_entry_id, "forge push"); if let Err(e) = crate::forge::push_config(&approval.agent).await { tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after spawn failed"); } + coord.set_queue_step(queue_entry_id, "forge meta access"); if let Some(core_token) = crate::forge::core_token() && let Err(e) = crate::forge::meta_read_access(&approval.agent, &core_token).await { @@ -418,6 +434,7 @@ async fn run_apply_commit( applied_dir: &std::path::Path, claude_dir: &std::path::Path, notes_dir: &std::path::Path, + queue_entry_id: Option, ) -> (Result<()>, Option, bool) { let id = approval.id; let proposal_ref = format!("refs/tags/proposal/{id}"); @@ -453,6 +470,7 @@ async fn run_apply_commit( } }; + coord.set_queue_step(queue_entry_id, "plant tags"); if let Err(e) = lifecycle::git_tag(applied_dir, &format!("approved/{id}"), &proposal_ref).await { return ( @@ -470,6 +488,7 @@ async fn run_apply_commit( ); } + coord.set_queue_step(queue_entry_id, "fast-forward applied/main"); // Fast-forward applied/main to proposal/ + sync the working // tree. Meta input pins `?ref=main`, so this is what makes nix // re-lock to the proposal commit on the prepare_deploy step @@ -497,6 +516,7 @@ async fn run_apply_commit( // before prepare_deploy can update its input lock (which won't // exist yet if this is the agent's first deploy). if is_first_spawn { + coord.set_queue_step(queue_entry_id, "meta sync_agents (first spawn)"); let agents = match lifecycle::agents_for_meta_listing_with(&approval.agent).await { Ok(a) => a, Err(e) => { @@ -531,6 +551,7 @@ async fn run_apply_commit( } } + coord.set_queue_step(queue_entry_id, "meta prepare_deploy"); // Phase 1 of the meta two-phase deploy: relock without committing. if let Err(e) = crate::meta::prepare_deploy(&approval.agent).await { let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await; @@ -542,6 +563,7 @@ async fn run_apply_commit( ); } + coord.set_queue_step(queue_entry_id, "nixos-container update"); // Container-level rebuild (or first-time create) against meta#. let build_result = lifecycle::rebuild_no_meta( &approval.agent, @@ -554,6 +576,7 @@ async fn run_apply_commit( match build_result { Ok(()) => { + coord.set_queue_step(queue_entry_id, "finalize deploy"); let tag = format!("deployed/{id}"); if let Err(e) = lifecycle::git_tag(applied_dir, &tag, &proposal_ref).await { tracing::warn!(agent = %approval.agent, %id, error = ?e, "plant deployed tag failed"); diff --git a/hive-c0re/src/auto_update.rs b/hive-c0re/src/auto_update.rs index 1ac2add7..31799898 100644 --- a/hive-c0re/src/auto_update.rs +++ b/hive-c0re/src/auto_update.rs @@ -63,7 +63,17 @@ pub fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool { /// Rebuild one sub-agent and refresh its marker. Used by both the startup /// scanner and the dashboard's manual "update" button so the two paths /// can't diverge. -pub async fn rebuild_agent(coord: &Arc, name: &str, current_rev: &str) -> Result<()> { +/// +/// `queue_entry_id` is `Some(id)` when the rebuild was dispatched from +/// the rebuild_queue worker (lets the function annotate its phase via +/// `coord.set_queue_step`) and `None` when called directly (e.g. the +/// manager-migration nudge in `ensure_manager`). +pub async fn rebuild_agent( + coord: &Arc, + name: &str, + current_rev: &str, + queue_entry_id: Option, +) -> Result<()> { tracing::info!(%name, rev = %current_rev, "rebuild agent"); let agent_dir = coord .ensure_runtime(name) @@ -76,6 +86,7 @@ pub async fn rebuild_agent(coord: &Arc, name: &str, current_rev: &s // lifecycle_action; this catches the auto-update scan + any // other direct caller. let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding); + coord.set_queue_step(queue_entry_id, "nixos-container update"); let result = lifecycle::rebuild( name, &coord.hyperhive_flake, @@ -101,6 +112,7 @@ pub async fn rebuild_agent(coord: &Arc, name: &str, current_rev: &s sha: None, tag: None, }); + coord.set_queue_step(queue_entry_id, "forge sync"); // Run the full forge sync on every successful rebuild so // the rebuild path is equivalent to the hive-c0re startup // sweep: token, config-repo mirror, meta read access, and @@ -156,7 +168,7 @@ pub async fn ensure_manager(coord: &Arc) -> Result<()> { "manager container exists but no applied flake — forcing rebuild to migrate" ); let coord_clone = coord.clone(); - if let Err(e) = rebuild_agent(&coord_clone, MANAGER_NAME, rev.as_str()).await { + if let Err(e) = rebuild_agent(&coord_clone, MANAGER_NAME, rev.as_str(), None).await { tracing::warn!(error = ?e, "manager migration rebuild failed"); } } else { diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 6a7f46c2..4963ea16 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -232,6 +232,19 @@ impl Coordinator { }); } + /// Update the `step` label on a running queue entry and (if it + /// actually changed) re-emit the queue snapshot so the dashboard + /// renders the new phase. Returns `true` when the label was new + /// and an emit fired, mostly for tracing/logging callers; safe to + /// ignore. No-op when `id` is `None` (e.g. callers that aren't + /// running from the queue worker) or when the row isn't `Running`. + pub fn set_queue_step(self: &Arc, id: Option, step: &str) { + let Some(id) = id else { return }; + if self.rebuild_queue.set_step(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/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 68210017..abbcdd08 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -189,6 +189,17 @@ pub struct QueueEntry { /// the wire that way too. #[serde(default, skip_serializing_if = "Option::is_none")] pub approval_id: Option, + /// Current sub-step inside the running entry (#437, option A from + /// the issue). Worker mutates this as the kind-specific pipeline + /// advances through phases (e.g. `"plant tags"` → + /// `"nixos-container update"` → `"finalize deploy"`). `None` while + /// `Queued` and after terminal — only meaningful with + /// `state == Running`. Each transition fires a fresh + /// `RebuildQueueChanged` snapshot so the dashboard can render + /// the label as a sub-line on the queue card. Free-form per + /// pipeline; the kind-specific worker is the source of truth. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub step: Option, } /// How many terminal-state entries (`Done` / `Failed` / `Cancelled`) @@ -330,6 +341,7 @@ impl RebuildQueue { error: None, inputs, approval_id, + step: None, }; inner.entries.push_back(entry); // Wake the worker. `notify_one` is a no-op when there's no @@ -356,6 +368,9 @@ impl RebuildQueue { /// Mark an entry terminal. `error` is populated for `Failed`; /// `Done` / `Cancelled` ignore it. Trims the history tail. + /// Clears `step` — the field is only meaningful while `Running`, + /// and leaving a stale "in flight" label after a terminal + /// transition would mislead the dashboard render. pub fn finish(&self, id: u64, state: QueueState, error: Option) { debug_assert!(state.is_terminal(), "finish() called with non-terminal {state:?}"); let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); @@ -363,10 +378,33 @@ impl RebuildQueue { entry.state = state; entry.finished_at = Some(now_unix()); entry.error = error.filter(|_| state == QueueState::Failed); + entry.step = None; } Self::trim_history(&mut inner); } + /// Set the current sub-step label on a `Running` entry (#437). + /// Returns `true` when the row was found AND the label changed + /// (caller should emit a `RebuildQueueChanged` snapshot only on + /// `true` to avoid noisy duplicate frames). No-op for entries not + /// in `Running` — the field is conceptually undefined outside + /// that state. + pub fn set_step(&self, id: u64, step: impl Into) -> bool { + let new_step = step.into(); + let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); + let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) else { + return false; + }; + if entry.state != QueueState::Running { + return false; + } + if entry.step.as_deref() == Some(new_step.as_str()) { + return false; + } + entry.step = Some(new_step); + true + } + /// Snapshot the queue for `/api/state` and `RebuildQueueChanged`. /// Cheap clone — entries are small (~hundreds of bytes each). pub fn snapshot(&self) -> Vec { @@ -515,19 +553,19 @@ async fn dispatch( ) -> anyhow::Result<()> { match (entry.kind, entry.approval_id) { (QueueKind::Rebuild, Some(approval_id)) => { - crate::actions::run_approval_apply_commit(coord, approval_id).await + crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await } (QueueKind::Rebuild, None) => { let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake) .unwrap_or_default(); - crate::auto_update::rebuild_agent(coord, &entry.agent, ¤t_rev).await + crate::auto_update::rebuild_agent(coord, &entry.agent, ¤t_rev, Some(entry.id)).await } (QueueKind::MetaUpdate, Some(approval_id)) => { - crate::actions::run_approval_update_meta_inputs(coord, approval_id).await + crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id).await } (QueueKind::MetaUpdate, None) => run_meta_update(coord, entry).await, (QueueKind::Spawn, Some(approval_id)) => { - crate::actions::run_approval_spawn(coord, approval_id).await + crate::actions::run_approval_spawn(coord, Some(entry.id), approval_id).await } (QueueKind::Spawn, None) => { // No non-approval Spawn caller today. The variant exists so @@ -561,6 +599,7 @@ async fn run_meta_update( let _progress = coord.meta_update_guard(); let inputs = entry.inputs.clone(); tracing::info!(?inputs, parent = entry.id, "rebuild_queue: meta-update starting"); + coord.set_queue_step(Some(entry.id), "nix flake update"); let result = if inputs.is_empty() { crate::meta::lock_update(&[]).await } else { @@ -1054,4 +1093,69 @@ mod tests { assert_eq!(find(done).state, QueueState::Done); assert_eq!(find(queued).state, QueueState::Cancelled); } + + #[test] + fn set_step_updates_running_entry_and_signals_change() { + let q = RebuildQueue::new(); + let id = q.enqueue( + QueueKind::Rebuild, + "a".to_owned(), + QueueSource::Manual, + "test".to_owned(), + None, + ); + // Queued — set_step should refuse (returns false). + assert!(!q.set_step(id, "plant tags")); + // Promote to Running. + let entry = q.take_next().expect("queued entry"); + assert_eq!(entry.id, id); + // First label transition — true. + assert!(q.set_step(id, "plant tags")); + assert_eq!( + q.snapshot() + .iter() + .find(|e| e.id == id) + .and_then(|e| e.step.as_deref()), + Some("plant tags") + ); + // Same label again — false (caller can skip the snapshot emit). + assert!(!q.set_step(id, "plant tags")); + // Different label — true. + assert!(q.set_step(id, "nixos-container update")); + assert_eq!( + q.snapshot() + .iter() + .find(|e| e.id == id) + .and_then(|e| e.step.as_deref()), + Some("nixos-container update") + ); + } + + #[test] + fn set_step_no_op_on_unknown_id() { + let q = RebuildQueue::new(); + assert!(!q.set_step(999, "anything")); + } + + #[test] + fn finish_clears_step() { + let q = RebuildQueue::new(); + let id = q.enqueue( + QueueKind::Rebuild, + "a".to_owned(), + QueueSource::Manual, + "test".to_owned(), + None, + ); + q.take_next(); + assert!(q.set_step(id, "running phase")); + q.finish(id, QueueState::Done, None); + assert_eq!( + q.snapshot() + .iter() + .find(|e| e.id == id) + .and_then(|e| e.step.as_deref()), + None + ); + } }