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

c0re: route approval execution through rebuild_queue (closes #436)

Browse source
This commit is contained in:
damocles 2026-05-26 00:13:10 +02:00 • committed by Mara
commit a4789760ed
2 changed files with 313 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<Coordinator>, id: i64) -> Result<()> {
let approval = coord.approvals.mark_approved(id)?;
tracing::info!(
@ -30,38 +34,80 @@ pub async fn approve(coord: Arc<Coordinator>, 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<String> =
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<Coordinator>,
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<Coordinator>,
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<Coordinator>,
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();
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<Coordinator>, 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,
&notes_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<hive_sh4re::Approval> {
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<Coordinator>, agent: &str) {
crate::dashboard::emit_tombstones_snapshot(coord).await;
}
async fn approve_init_config(
coord: Arc<Coordinator>,
/// 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<Coordinator>,
approval: hive_sh4re::Approval,
) -> Result<()> {
// Inputs stored as JSON in commit_ref by the manager's submit path.
let inputs: Vec<String> = 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<Coordinator>,
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,
&notes_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(

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

@ -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<String>,
/// 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<i64>,
}
/// How many terminal-state entries (`Done` / `Failed` / `Cancelled`)
@ -240,7 +254,7 @@ impl RebuildQueue {
reason: String,
parent_id: Option<u64>,
) -> 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<u64>,
inputs: Vec<String>,
) -> 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<u64>,
inputs: Vec<String>,
approval_id: Option<i64>,
) -> 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<crate::coordinator::Coordinator>)
/// 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<crate::coordinator::Coordinator>,
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, &current_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();