From a4789760eddab600c2df0cbde634dbfb79fc0083 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 26 May 2026 00:13:10 +0200 Subject: [PATCH] c0re: route approval execution through rebuild_queue (closes #436) --- hive-c0re/src/actions.rs | 293 +++++++++++++++++++++------------ hive-c0re/src/rebuild_queue.rs | 143 ++++++++++++++-- 2 files changed, 314 insertions(+), 122 deletions(-) diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 36fa764a..fa0b965a 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -11,18 +11,22 @@ use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT}; use crate::coordinator::{Coordinator, TransientKind}; use crate::lifecycle::{self, MANAGER_NAME}; -/// Approve a pending request and run the underlying action. Dispatches on the -/// approval kind: -/// - `ApplyCommit`: read agent.nix at the approval's commit from the proposed -/// repo, copy into the applied repo, commit there, rebuild the container. -/// Synchronous — returns once the rebuild completes. -/// - `Spawn`: create + start a brand-new sub-agent container. Runs in a -/// background task so the operator's approve click returns immediately; -/// the dashboard surfaces a transient `Spawning` state until the container -/// is up. On failure, the approval is marked failed. +/// Approve a pending request. Marks the approval row durably, then +/// either runs the work inline (`InitConfig`, sub-second git ops) or +/// enqueues it into `rebuild_queue` so the dashboard POST returns +/// immediately while the long-running pipeline runs off-thread +/// (closes #436 — operator no longer eats a 30-90s spinner on +/// `ApplyCommit`). /// -/// In all cases an `ApprovalResolved` helper event lands in the manager's -/// inbox when the work resolves. +/// Dispatch: +/// - `ApplyCommit` → `QueueKind::Rebuild` (~30-90s wall time) +/// - `UpdateMetaInputs` → `QueueKind::MetaUpdate` (~3-15s) +/// - `Spawn` → `QueueKind::Spawn` (~30-90s) +/// - `InitConfig` → inline (<1s; queue card would be noise) +/// +/// The queue worker re-fetches the approval row on dispatch, runs +/// the kind-specific pipeline, and fires `ApprovalResolved` / +/// `Spawned` / `Rebuilt` / `ConfigReady` via `finish_approval`. pub async fn approve(coord: Arc, id: i64) -> Result<()> { let approval = coord.approvals.mark_approved(id)?; tracing::info!( @@ -30,38 +34,80 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { %approval.agent, kind = ?approval.kind, %approval.commit_ref, - "approval: running action", + "approval: dispatching", ); - let agent_dir = coord.ensure_runtime(&approval.agent)?; - let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent); - 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); match approval.kind { - ApprovalKind::ApplyCommit => { - approve_apply_commit(coord, approval, agent_dir, applied_dir, claude_dir, notes_dir).await - } ApprovalKind::InitConfig => { - approve_init_config(coord, approval, proposed_dir, claude_dir, notes_dir).await + // Sub-second git seed + forge-remote wire. Routing through + // the queue would surface a queue card that's gone before + // the operator's eyes refocus. Run inline. + let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent); + let claude_dir = Coordinator::agent_claude_dir(&approval.agent); + let notes_dir = Coordinator::agent_notes_dir(&approval.agent); + run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await + } + ApprovalKind::ApplyCommit => { + coord.rebuild_queue.enqueue_full( + crate::rebuild_queue::QueueKind::Rebuild, + approval.agent.clone(), + crate::rebuild_queue::QueueSource::Approval, + format!("approval #{id} apply commit"), + None, + Vec::new(), + Some(id), + ); + coord.emit_rebuild_queue_snapshot(); + Ok(()) + } + ApprovalKind::UpdateMetaInputs => { + // Inputs JSON-encoded into commit_ref by the manager's + // submit path — surface them on the queue entry so the + // dashboard can show *which* inputs are about to bump. + let inputs: Vec = + serde_json::from_str(&approval.commit_ref).unwrap_or_default(); + coord.rebuild_queue.enqueue_full( + crate::rebuild_queue::QueueKind::MetaUpdate, + approval.agent.clone(), + crate::rebuild_queue::QueueSource::Approval, + format!("approval #{id} meta input update"), + None, + inputs, + Some(id), + ); + coord.emit_rebuild_queue_snapshot(); + Ok(()) } - ApprovalKind::UpdateMetaInputs => approve_update_meta_inputs(coord, approval).await, ApprovalKind::Spawn => { - approve_spawn(&coord, &approval, agent_dir, proposed_dir, applied_dir, claude_dir, notes_dir); + coord.rebuild_queue.enqueue_full( + crate::rebuild_queue::QueueKind::Spawn, + approval.agent.clone(), + crate::rebuild_queue::QueueSource::Approval, + format!("approval #{id} spawn"), + None, + Vec::new(), + Some(id), + ); + coord.emit_rebuild_queue_snapshot(); Ok(()) } } } -async fn approve_apply_commit( - coord: Arc, - approval: hive_sh4re::Approval, - agent_dir: std::path::PathBuf, - applied_dir: std::path::PathBuf, - claude_dir: std::path::PathBuf, - notes_dir: std::path::PathBuf, +/// Worker entry point for `ApprovalKind::ApplyCommit` queue entries. +/// Re-fetches the approval row, runs the commit pipeline (same +/// shape as the pre-#436 inline path), and fires `ApprovalResolved` +/// + the lifecycle event (`Rebuilt` / `Spawned` for first-spawn). +pub async fn run_approval_apply_commit( + coord: &Arc, + approval_id: i64, ) -> Result<()> { + let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?; + let agent_dir = coord.ensure_runtime(&approval.agent)?; + 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); let (result, terminal_tag, is_first_spawn) = run_apply_commit( - &coord, + coord, &approval, &agent_dir, &applied_dir, @@ -73,9 +119,105 @@ async fn approve_apply_commit( tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed"); } if is_first_spawn && result.is_ok() { - forge_after_first_spawn(&coord, &approval.agent).await; + forge_after_first_spawn(coord, &approval.agent).await; } - finish_approval(&coord, &approval, result, terminal_tag, is_first_spawn) + // `finish_approval` returns the original `result` so the queue + // worker sees Ok/Err and marks the queue entry accordingly. The + // approval row + helper events have already been fanned out. + finish_approval(coord, &approval, result, terminal_tag, is_first_spawn) +} + +/// Worker entry point for `ApprovalKind::UpdateMetaInputs` queue +/// entries. Inputs come from the approval row's `commit_ref` field +/// (JSON-encoded by the manager submit path), not the queue entry's +/// `inputs` — the queue copy is for dashboard display only. +pub async fn run_approval_update_meta_inputs( + coord: &Arc, + 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(); + let result = crate::meta::lock_update(&inputs).await; + finish_approval(coord, &approval, result, None, false) +} + +/// Worker entry point for `ApprovalKind::Spawn` queue entries. +/// Differs from `run_approval_apply_commit` only in routing through +/// `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<()> { + 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); + 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); + // Transient guard keeps the per-container "Spawning" pill lit while + // 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); + let result = lifecycle::spawn( + &approval.agent, + &coord.hyperhive_flake, + &agent_dir, + &proposed_dir, + &applied_dir, + &claude_dir, + ¬es_dir, + coord.dashboard_port, + &coord.operator_pronouns, + &coord.context_window_tokens, + ) + .await; + if result.is_ok() { + 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"); + } + 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"); + } + if let Err(e) = crate::forge::push_config(&approval.agent).await { + tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after spawn failed"); + } + if let Some(core_token) = crate::forge::core_token() + && let Err(e) = crate::forge::meta_read_access(&approval.agent, &core_token).await + { + tracing::warn!(agent = %approval.agent, error = ?e, "forge: meta_read_access after spawn failed"); + } + if let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await { + tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after spawn failed"); + } + } + let final_result = finish_approval(coord, &approval, result, None, false); + coord.rescan_containers_and_emit().await; + crate::dashboard::emit_tombstones_snapshot(coord).await; + final_result +} + +/// Re-fetch an approval row from sqlite for a queue-worker dispatch. +/// Bails if the row is gone (deny race), if its kind doesn't match, +/// or if the lookup itself fails. The kind check is defensive — the +/// queue's `dispatch` already routes by `QueueKind`, but the approval +/// kind is the authoritative source of truth and a mismatch points +/// at a deeper bug we'd want to surface. +fn fetch_approval_for_worker( + coord: &Coordinator, + approval_id: i64, + expected_kind: ApprovalKind, +) -> Result { + let approval = coord + .approvals + .get(approval_id) + .map_err(|e| anyhow::anyhow!("read approval {approval_id}: {e:#}"))? + .ok_or_else(|| anyhow::anyhow!("approval {approval_id} no longer exists"))?; + if approval.kind != expected_kind { + bail!( + "approval {approval_id} kind mismatch: queue expected {expected_kind:?}, row is {actual:?}", + actual = approval.kind + ); + } + Ok(approval) } /// Forge bookkeeping run once after the very first container spawn: @@ -100,14 +242,16 @@ async fn forge_after_first_spawn(coord: &Arc, agent: &str) { crate::dashboard::emit_tombstones_snapshot(coord).await; } -async fn approve_init_config( - coord: Arc, +/// Inline (non-queued) handler for `ApprovalKind::InitConfig`. Just +/// seeds the proposed git repo + the per-agent dirs — sub-second +/// work that doesn't justify a queue card. +async fn run_approval_init_config( + coord: &Coordinator, approval: hive_sh4re::Approval, proposed_dir: std::path::PathBuf, claude_dir: std::path::PathBuf, notes_dir: std::path::PathBuf, ) -> Result<()> { - // Seed the proposed config repo — just git operations, no nixos-container. let result: Result<()> = async { lifecycle::setup_proposed(&proposed_dir, &approval.agent).await?; lifecycle::ensure_claude_dir(&claude_dir)?; @@ -116,78 +260,11 @@ async fn approve_init_config( } .await; if result.is_ok() - && let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await { - tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed"); - } - finish_approval(&coord, &approval, result, None, false) -} - -async fn approve_update_meta_inputs( - coord: Arc, - approval: hive_sh4re::Approval, -) -> Result<()> { - // Inputs stored as JSON in commit_ref by the manager's submit path. - let inputs: Vec = serde_json::from_str(&approval.commit_ref).unwrap_or_default(); - let result = crate::meta::lock_update(&inputs).await; - finish_approval(&coord, &approval, result, None, false) -} - -fn approve_spawn( - coord: &Arc, - approval: &hive_sh4re::Approval, - agent_dir: std::path::PathBuf, - proposed_dir: std::path::PathBuf, - applied_dir: std::path::PathBuf, - claude_dir: std::path::PathBuf, - notes_dir: std::path::PathBuf, -) { - // Run spawn in the background so approve POST returns immediately. - // Guard created synchronously so the spinner appears the moment - // the operator clicks approve; auto-clears when the task drops it. - let coord_bg = Arc::clone(coord); - let approval_bg = approval.clone(); - let guard = coord_bg.transient_guard(&approval_bg.agent, TransientKind::Spawning); - tokio::spawn(async move { - let guard = guard; - let agent_bg = approval_bg.agent.clone(); - let result = lifecycle::spawn( - &approval_bg.agent, - &coord_bg.hyperhive_flake, - &agent_dir, - &proposed_dir, - &applied_dir, - &claude_dir, - ¬es_dir, - coord_bg.dashboard_port, - &coord_bg.operator_pronouns, - &coord_bg.context_window_tokens, - ) - .await; - drop(guard); - if result.is_ok() { - if let Err(e) = crate::forge::ensure_user_for(&agent_bg).await { - tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_user after spawn failed"); - } - if let Err(e) = crate::forge::ensure_config_repo(&agent_bg).await { - tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_config_repo after spawn failed"); - } - if let Err(e) = crate::forge::push_config(&agent_bg).await { - tracing::warn!(agent = %agent_bg, error = ?e, "forge: push_config after spawn failed"); - } - if let Some(core_token) = crate::forge::core_token() - && let Err(e) = crate::forge::meta_read_access(&agent_bg, &core_token).await { - tracing::warn!(agent = %agent_bg, error = ?e, "forge: meta_read_access after spawn failed"); - } - if let Err(e) = crate::forge::ensure_meta_remote(&agent_bg).await { - tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_meta_remote after spawn failed"); - } - } - if let Err(e) = finish_approval(&coord_bg, &approval_bg, result, None, false) { - tracing::warn!(agent = %agent_bg, error = ?e, "spawn approval failed"); - } - coord_bg.rescan_containers_and_emit().await; - crate::dashboard::emit_tombstones_snapshot(&coord_bg).await; - }); + && let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await + { + tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed"); + } + finish_approval(coord, &approval, result, None, false) } fn finish_approval( diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index bec96646..68210017 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -50,7 +50,7 @@ //! current run started). use std::collections::VecDeque; -use std::sync::{Arc, Mutex}; +use std::sync::Mutex; use serde::Serialize; use tokio::sync::Notify; @@ -103,6 +103,11 @@ pub enum QueueSource { /// Crash recovery path (future use — currently no auto-rebuild on /// crash, but the variant exists for the imminent feature). CrashRecover, + /// Operator approved a pending `Approval` row on the dashboard. + /// `QueueEntry.approval_id` points back at the source row so the + /// worker can fetch the kind-specific payload (commit_ref, inputs, + /// description) before dispatching. + Approval, } impl QueueSource { @@ -112,6 +117,7 @@ impl QueueSource { QueueSource::MetaUpdate => "meta_update", QueueSource::AutoUpdate => "auto_update", QueueSource::CrashRecover => "crash_recover", + QueueSource::Approval => "approval", } } } @@ -175,6 +181,14 @@ pub struct QueueEntry { /// serialised) when the entry kind doesn't have meaningful inputs. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub inputs: Vec, + /// Source approval row id when this entry was created by an + /// operator-approve POST (`source == Approval`). The worker uses + /// it to re-fetch the kind-specific payload (commit_ref / inputs / + /// description / fetched_sha) and to fire `ApprovalResolved` on + /// completion. `None` for non-approval entries — preserved on + /// the wire that way too. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub approval_id: Option, } /// How many terminal-state entries (`Done` / `Failed` / `Cancelled`) @@ -240,7 +254,7 @@ impl RebuildQueue { reason: String, parent_id: Option, ) -> u64 { - self.enqueue_with_inputs(kind, agent, source, reason, parent_id, Vec::new()) + self.enqueue_full(kind, agent, source, reason, parent_id, Vec::new(), None) } /// Same as `enqueue` but carries an `inputs` payload — used by @@ -256,16 +270,43 @@ impl RebuildQueue { reason: String, parent_id: Option, inputs: Vec, + ) -> u64 { + self.enqueue_full(kind, agent, source, reason, parent_id, inputs, None) + } + + /// Full-shape enqueue — every `QueueEntry` field that's settable + /// at submit time. Existing `enqueue` / `enqueue_with_inputs` + /// delegate to this with `approval_id: None`; the approval-driven + /// POST handlers (#436) call it directly with the source row's id + /// so the worker can re-fetch the kind-specific payload. + // 8/7 args: the queue entry has 6 independent submit-time fields plus + // the inputs/approval_id pair specific to MetaUpdate and approval + // entries. A builder struct would obscure the call sites; the + // shorter `enqueue` / `enqueue_with_inputs` wrappers already cover + // the common cases. + #[allow(clippy::too_many_arguments)] + pub fn enqueue_full( + &self, + kind: QueueKind, + agent: String, + source: QueueSource, + reason: String, + parent_id: Option, + inputs: Vec, + approval_id: Option, ) -> u64 { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); // Dedup against a pending entry with the same (kind, agent) — // and, for MetaUpdate, the same `inputs` list (see method - // docstring + #365 for why). + // docstring + #365 for why). Approval-driven entries also + // require the approval_id to match so two distinct approvals + // for the same agent never collapse into one queue slot. for entry in inner.entries.iter_mut() { if entry.state == QueueState::Queued && entry.kind == kind && entry.agent == agent && (kind != QueueKind::MetaUpdate || entry.inputs == inputs) + && entry.approval_id == approval_id { if !entry.reason.contains(&reason) { entry.reason.push_str(&format!("\nalso requested by: {reason}")); @@ -288,6 +329,7 @@ impl RebuildQueue { finished_at: None, error: None, inputs, + approval_id, }; inner.entries.push_back(entry); // Wake the worker. `notify_one` is a no-op when there's no @@ -462,31 +504,44 @@ pub async fn run_worker(coord: std::sync::Arc) /// Run a single queue entry to completion. Kind-dispatched; failures /// bubble up to the worker which marks the entry `Failed`. +/// +/// Approval-driven entries (`approval_id.is_some()`) route through +/// `actions::run_approval_*` which carry the kind-specific commit +/// pipeline + the `ApprovalResolved` event fan-out. Non-approval +/// entries hit the original auto/manual rebuild paths. async fn dispatch( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { - match entry.kind { - QueueKind::Rebuild => { + match (entry.kind, entry.approval_id) { + (QueueKind::Rebuild, Some(approval_id)) => { + crate::actions::run_approval_apply_commit(coord, 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 } - QueueKind::MetaUpdate => run_meta_update(coord, entry).await, - QueueKind::Spawn => { - // First-deploy spawns route through `actions::approve_spawn` - // / `actions::approve_apply_commit` today; they enqueue a - // Spawn entry only to claim the queue slot, the actual - // spawn work runs inside those handlers before completion. - // Keeping this arm a no-op so we don't double-run. + (QueueKind::MetaUpdate, Some(approval_id)) => { + crate::actions::run_approval_update_meta_inputs(coord, 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 + } + (QueueKind::Spawn, None) => { + // No non-approval Spawn caller today. The variant exists so + // operator-triggered `RequestSpawn` (deprecated) and the + // future direct-spawn admin path can route through here + // without a wire change. tracing::debug!( id = entry.id, agent = %entry.agent, - "rebuild_queue: Spawn entry is a queue claim; actual work elsewhere" + "rebuild_queue: Spawn entry without approval_id is a no-op" ); Ok(()) } - QueueKind::Destroy => { + (QueueKind::Destroy, _) => { // Reserved for future `destroy --purge` integration. anyhow::bail!("Destroy kind not yet implemented in rebuild_queue worker"); } @@ -893,6 +948,66 @@ mod tests { assert_eq!(find(c).state, QueueState::Queued); } + #[test] + fn approval_entries_keep_approval_id() { + let q = RebuildQueue::new(); + let id = q.enqueue_full( + QueueKind::Rebuild, + "agent-a".to_owned(), + QueueSource::Approval, + "approval #42 apply commit".to_owned(), + None, + Vec::new(), + Some(42), + ); + let snap = q.snapshot(); + let entry = snap.iter().find(|e| e.id == id).expect("entry present"); + assert_eq!(entry.approval_id, Some(42)); + assert_eq!(entry.source, QueueSource::Approval); + } + + #[test] + fn approval_entries_dedup_only_on_matching_id() { + // Two pending approval-driven entries for the same agent but + // DIFFERENT approval ids must NOT collapse — each operator + // approve click is a separate piece of work even when the + // (kind, agent) pair matches. + let q = RebuildQueue::new(); + let a = q.enqueue_full( + QueueKind::Rebuild, + "agent-a".to_owned(), + QueueSource::Approval, + "approval #1".to_owned(), + None, + Vec::new(), + Some(1), + ); + let b = q.enqueue_full( + QueueKind::Rebuild, + "agent-a".to_owned(), + QueueSource::Approval, + "approval #2".to_owned(), + None, + Vec::new(), + Some(2), + ); + assert_ne!(a, b); + assert_eq!(q.snapshot().len(), 2); + // Same approval_id submitted twice DOES dedup (rapid double- + // click on the dashboard's approve button is a single op). + let c = q.enqueue_full( + QueueKind::Rebuild, + "agent-a".to_owned(), + QueueSource::Approval, + "approval #1 (duplicate)".to_owned(), + None, + Vec::new(), + Some(1), + ); + assert_eq!(a, c); + assert_eq!(q.snapshot().len(), 2); + } + #[test] fn cancel_children_skips_running_and_terminal() { let q = RebuildQueue::new();