diff --git a/frontend/packages/dashboard/src/dashboard.css b/frontend/packages/dashboard/src/dashboard.css index c0f7fc14..5ce00126 100644 --- a/frontend/packages/dashboard/src/dashboard.css +++ b/frontend/packages/dashboard/src/dashboard.css @@ -766,7 +766,6 @@ code { .rqe-source-meta_update { color: var(--purple); border-color: var(--purple); } .rqe-source-auto_update { color: var(--muted); } .rqe-source-crash_recover { color: var(--amber); border-color: var(--amber); } -.rqe-source-approval { color: var(--green); border-color: var(--green); } .rqe-when { color: var(--muted); font-size: 0.85em; } .rqe-reason { color: var(--muted); font-size: 0.85em; flex: 1 1 auto; } .rqe-error { diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index fa0b965a..36fa764a 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -11,22 +11,18 @@ use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT}; use crate::coordinator::{Coordinator, TransientKind}; use crate::lifecycle::{self, MANAGER_NAME}; -/// 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`). +/// 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. /// -/// 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`. +/// In all cases an `ApprovalResolved` helper event lands in the manager's +/// inbox when the work resolves. pub async fn approve(coord: Arc, id: i64) -> Result<()> { let approval = coord.approvals.mark_approved(id)?; tracing::info!( @@ -34,80 +30,38 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { %approval.agent, kind = ?approval.kind, %approval.commit_ref, - "approval: dispatching", + "approval: running action", ); + 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::InitConfig => { - // 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(()) + approve_apply_commit(coord, approval, agent_dir, applied_dir, claude_dir, notes_dir).await } - 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::InitConfig => { + approve_init_config(coord, approval, proposed_dir, claude_dir, notes_dir).await } + ApprovalKind::UpdateMetaInputs => approve_update_meta_inputs(coord, approval).await, ApprovalKind::Spawn => { - 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(); + approve_spawn(&coord, &approval, agent_dir, proposed_dir, applied_dir, claude_dir, notes_dir); Ok(()) } } } -/// 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, +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, ) -> 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, @@ -119,105 +73,9 @@ pub async fn run_approval_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` 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) + finish_approval(&coord, &approval, result, terminal_tag, is_first_spawn) } /// Forge bookkeeping run once after the very first container spawn: @@ -242,16 +100,14 @@ async fn forge_after_first_spawn(coord: &Arc, agent: &str) { crate::dashboard::emit_tombstones_snapshot(coord).await; } -/// 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, +async fn approve_init_config( + coord: Arc, 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)?; @@ -260,11 +116,78 @@ async fn run_approval_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) + && 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; + }); } fn finish_approval( diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 68210017..bec96646 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::Mutex; +use std::sync::{Arc, Mutex}; use serde::Serialize; use tokio::sync::Notify; @@ -103,11 +103,6 @@ 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 { @@ -117,7 +112,6 @@ impl QueueSource { QueueSource::MetaUpdate => "meta_update", QueueSource::AutoUpdate => "auto_update", QueueSource::CrashRecover => "crash_recover", - QueueSource::Approval => "approval", } } } @@ -181,14 +175,6 @@ 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`) @@ -254,7 +240,7 @@ impl RebuildQueue { reason: String, parent_id: Option, ) -> u64 { - self.enqueue_full(kind, agent, source, reason, parent_id, Vec::new(), None) + self.enqueue_with_inputs(kind, agent, source, reason, parent_id, Vec::new()) } /// Same as `enqueue` but carries an `inputs` payload — used by @@ -270,43 +256,16 @@ 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). Approval-driven entries also - // require the approval_id to match so two distinct approvals - // for the same agent never collapse into one queue slot. + // docstring + #365 for why). 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}")); @@ -329,7 +288,6 @@ 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 @@ -504,44 +462,31 @@ 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, entry.approval_id) { - (QueueKind::Rebuild, Some(approval_id)) => { - crate::actions::run_approval_apply_commit(coord, approval_id).await - } - (QueueKind::Rebuild, None) => { + match entry.kind { + QueueKind::Rebuild => { 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, 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. + 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. tracing::debug!( id = entry.id, agent = %entry.agent, - "rebuild_queue: Spawn entry without approval_id is a no-op" + "rebuild_queue: Spawn entry is a queue claim; actual work elsewhere" ); Ok(()) } - (QueueKind::Destroy, _) => { + QueueKind::Destroy => { // Reserved for future `destroy --purge` integration. anyhow::bail!("Destroy kind not yet implemented in rebuild_queue worker"); } @@ -948,66 +893,6 @@ 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();