vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

rebuild_queue: per-entry step label + worker phase annotations (#437)

Browse source
This commit is contained in:
damocles 2026-05-26 22:47:59 +02:00 • committed by Mara
commit a286ae777c
4 changed files with 159 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

25
hive-c0re/src/actions.rs
View file

@ -108,6 +108,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
/// + the lifecycle event (`Rebuilt` / `Spawned` for first-spawn).
pub async fn run_approval_apply_commit(
coord: &Arc<Coordinator>,
queue_entry_id: Option<u64>,
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,
&notes_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<Coordinator>,
queue_entry_id: Option<u64>,
approval_id: i64,
) -> Result<()> {
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::UpdateMetaInputs)?;
let inputs: Vec<String> = 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<Coordinator>, approval_id: i64) -> Result<()> {
pub async fn run_approval_spawn(
coord: &Arc<Coordinator>,
queue_entry_id: Option<u64>,
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<Coordinator>, 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<Coordinator>, 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<u64>,
) -> (Result<()>, Option<String>, 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/<id> + 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#<name>.
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");

16
hive-c0re/src/auto_update.rs
View file

@ -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<Coordinator>, 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<Coordinator>,
name: &str,
current_rev: &str,
queue_entry_id: Option<u64>,
) -> 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<Coordinator>, 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<Coordinator>, 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<Coordinator>) -> 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 {

13
hive-c0re/src/coordinator.rs
View file

@ -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<Self>, id: Option<u64>, 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`).

112
hive-c0re/src/rebuild_queue.rs
View file

@ -189,6 +189,17 @@ pub struct QueueEntry {
/// the wire that way too.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub approval_id: Option<i64>,
/// 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<String>,
}
/// 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<String>) {
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<String>) -> 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<QueueEntry> {
@ -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, &current_rev).await
crate::auto_update::rebuild_agent(coord, &entry.agent, &current_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
);
}
}