From eae0e875cf5cf83b81029041593e900c896dcda9 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 2 Jun 2026 16:01:43 +0200 Subject: [PATCH] feat(#1086): serialize perm changes through rebuild queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit add QueueKind::PermChange — dashboard tool-group and capability handlers no longer write the shared JSON files inline. instead they enqueue a PermChange entry; the FIFO worker applies the file write then calls rebuild_agent so the updated env var takes effect. concurrent batch-apply actions for different agents previously raced on tool-groups.json / capabilities.json (last write wins, earlier change silently dropped). serialising through the queue prevents this. dedup check extended with perm-type discriminant so tool-groups and capabilities changes for the same agent are kept as distinct entries and never collapse into one slot. --- docs/coordinator.md | 1 + hive-c0re/src/actions.rs | 3 + hive-c0re/src/dashboard.rs | 31 +++++---- hive-c0re/src/rebuild_queue.rs | 115 +++++++++++++++++++++++++++++---- hive-c0re/src/tool_groups.rs | 2 +- 5 files changed, 126 insertions(+), 26 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index e1173709..09126c86 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -39,6 +39,7 @@ somewhere." | `Spawn` | First-deploy of a new agent (approval-driven). Same serialisation as `Rebuild` from the operator's POV. | | `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. | | `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, manager `restart` MCP tool. | +| `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. | **Intentionally not queued** (sub-second ops): `start`, `stop`, `kill`. diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 9412f942..85c05c7b 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -54,6 +54,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { None, Vec::new(), Some(id), + None, ); coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -72,6 +73,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { None, inputs.clone(), Some(id), + None, ); // Pre-enqueue cascade rebuilds in topological order so // agents depending on updated inputs are rebuilt after the @@ -99,6 +101,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { None, Vec::new(), Some(id), + None, ); coord.emit_rebuild_queue_snapshot(); Ok(()) diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 0218229f..69af5662 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -2550,16 +2550,21 @@ async fn post_tool_groups( if let Some(reject) = guard_agent_name(&state, &logical).await { return reject; } - if let Err(e) = crate::tool_groups::set_groups(&logical, &body.groups) { - return error_response(&format!("set tool-groups for {logical}: {e}")); + // Validate group names before queuing — fail fast so the operator + // sees the error immediately rather than waiting for the worker. + if let Err(e) = crate::tool_groups::validate_groups(&body.groups) { + return error_response(&format!("invalid tool-groups for {logical}: {e}")); } - // Trigger a rebuild so the new HIVE_TOOL_GROUPS env var takes effect. - state.coord.rebuild_queue.enqueue( - crate::rebuild_queue::QueueKind::Rebuild, + // Enqueue a PermChange so the JSON file write is serialised through + // the FIFO worker. Prevents concurrent batch-apply actions for + // different agents from racing on the shared tool-groups.json. + state.coord.rebuild_queue.enqueue_with_perm( logical.clone(), crate::rebuild_queue::QueueSource::Manual, "tool-group change via permissions UI".to_owned(), - None, + crate::rebuild_queue::PermPayload::ToolGroups { + groups: body.groups.clone(), + }, ); state.coord.emit_rebuild_queue_snapshot(); tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard"); @@ -2614,16 +2619,16 @@ async fn post_capabilities( return error_response(&format!("unknown capability: {cap}")); } } - if let Err(e) = crate::capabilities::set_caps(&logical, &body.caps) { - return error_response(&format!("set capabilities for {logical}: {e}")); - } - // Trigger a rebuild so the new HIVE_CAPABILITIES env var takes effect. - state.coord.rebuild_queue.enqueue( - crate::rebuild_queue::QueueKind::Rebuild, + // Enqueue a PermChange so the JSON file write is serialised through + // the FIFO worker. Prevents concurrent batch-apply actions for + // different agents from racing on the shared capabilities.json. + state.coord.rebuild_queue.enqueue_with_perm( logical.clone(), crate::rebuild_queue::QueueSource::Manual, "capability change via dashboard".to_owned(), - None, + crate::rebuild_queue::PermPayload::Capabilities { + caps: body.caps.clone(), + }, ); state.coord.emit_rebuild_queue_snapshot(); tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard"); diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 1e7e78fa..25cd044c 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -7,7 +7,8 @@ use std::collections::VecDeque; use std::sync::Mutex; -use serde::Serialize; +use anyhow::Context as _; +use serde::{Deserialize, Serialize}; use tokio::sync::Notify; /// What the queue can run. Each variant maps to a specific worker @@ -36,6 +37,11 @@ pub enum QueueKind { /// Queued so it serialises against in-flight rebuilds for the same /// agent — prevents a restart racing a rebuild mid-flight. Restart, + /// Write a tool-group or capability change to the shared JSON file, + /// then rebuild the agent so the new env var takes effect. + /// Serialised through the queue so concurrent dashboard batch-apply + /// actions for different agents never race on the shared JSON file. + PermChange, } impl QueueKind { @@ -47,10 +53,24 @@ impl QueueKind { QueueKind::Destroy => "destroy", QueueKind::StartupSweep => "startup_sweep", QueueKind::Restart => "restart", + QueueKind::PermChange => "perm_change", } } } +/// Kind-specific payload for `QueueKind::PermChange` entries. +/// Carries the desired new value so the worker can apply the file +/// write (serialised, in FIFO order) without racing concurrent HTTP +/// handlers writing to the same shared JSON file. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum PermPayload { + /// Set the tool groups for one agent (`tool-groups.json`). + ToolGroups { groups: Vec }, + /// Set the capabilities for one agent (`capabilities.json`). + Capabilities { caps: Vec }, +} + /// Where the enqueue request originated. Drives the "why" chip on the /// dashboard and lets the UI group cascade entries under their parent /// without parsing the reason text. @@ -181,6 +201,11 @@ pub struct QueueEntry { /// pipeline; the kind-specific worker is the source of truth. #[serde(default, skip_serializing_if = "Option::is_none")] pub step: Option, + /// `PermChange`-only payload: the desired new permission value to + /// apply. Absent (`None`) on all other entry kinds — omitted from + /// the wire in those cases. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub perm_payload: Option, } /// How many terminal-state entries (`Done` / `Failed` / `Cancelled`) @@ -252,7 +277,7 @@ impl RebuildQueue { reason: String, parent_id: Option, ) -> u64 { - self.enqueue_full(kind, agent, source, reason, parent_id, Vec::new(), None) + self.enqueue_full(kind, agent, source, reason, parent_id, Vec::new(), None, None) } /// Same as `enqueue` but carries an `inputs` payload — used by @@ -269,19 +294,40 @@ impl RebuildQueue { parent_id: Option, inputs: Vec, ) -> u64 { - self.enqueue_full(kind, agent, source, reason, parent_id, inputs, None) + self.enqueue_full(kind, agent, source, reason, parent_id, inputs, None, None) + } + + /// Enqueue a `PermChange` entry for `agent`. The worker applies the + /// JSON file write (serialised through FIFO) then rebuilds the + /// container so the updated env var takes effect. + pub fn enqueue_with_perm( + &self, + agent: String, + source: QueueSource, + reason: String, + payload: PermPayload, + ) -> u64 { + self.enqueue_full( + QueueKind::PermChange, + agent, + source, + reason, + None, + Vec::new(), + None, + Some(payload), + ) } /// 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 call it directly with the source row's id so the + /// at submit time. Existing `enqueue` / `enqueue_with_inputs` / + /// `enqueue_with_perm` delegate to this; the approval-driven POST + /// handlers 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. + // 9 args: the queue entry has 6 independent submit-time fields plus + // three kind-specific payload fields (inputs, approval_id, perm_payload). + // A builder struct would obscure the call sites; the shorter wrappers + // already cover all common cases. #[allow(clippy::too_many_arguments)] pub fn enqueue_full( &self, @@ -292,6 +338,7 @@ impl RebuildQueue { parent_id: Option, inputs: Vec, approval_id: Option, + perm_payload: Option, ) -> u64 { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); // Dedup against a pending entry with the same (kind, agent) — @@ -301,14 +348,29 @@ impl RebuildQueue { // agent never collapse into one queue slot. Rebuild (and Spawn / // Destroy) entries also require parent_id to match so a // MetaUpdate cascade rebuild is never swallowed by an unrelated - // queued rebuild (e.g. from the startup sweep). + // queued rebuild (e.g. from the startup sweep). PermChange + // entries additionally check the perm type discriminant — a + // tool-groups change and a capabilities change for the same + // agent are distinct operations and must not collapse into one. for entry in &mut inner.entries { + let perm_type_matches = match (&entry.perm_payload, &perm_payload) { + (Some(PermPayload::ToolGroups { .. }), Some(PermPayload::ToolGroups { .. })) => { + true + } + ( + Some(PermPayload::Capabilities { .. }), + Some(PermPayload::Capabilities { .. }), + ) => true, + (None, None) => true, + _ => false, + }; if entry.state == QueueState::Queued && entry.kind == kind && entry.agent == agent && (kind != QueueKind::MetaUpdate || entry.inputs == inputs) && entry.approval_id == approval_id && entry.parent_id == parent_id + && perm_type_matches { if !entry.reason.contains(&reason) { use std::fmt::Write as _; @@ -334,6 +396,7 @@ impl RebuildQueue { inputs, approval_id, step: None, + perm_payload, }; inner.entries.push_back(entry); // Wake the worker. `notify_one` is a no-op when there's no @@ -605,6 +668,34 @@ async fn dispatch( coord.rescan_containers_and_emit().await; Ok(()) } + (QueueKind::PermChange, _) => { + let name = &entry.agent; + // Apply the file write first — serialised here so concurrent + // dashboard batch-apply actions never race on the shared JSON. + coord.set_queue_step(Some(entry.id), "writing perm file"); + match &entry.perm_payload { + Some(PermPayload::ToolGroups { groups }) => { + crate::tool_groups::set_groups(name, groups) + .with_context(|| format!("set tool-groups for {name}"))?; + } + Some(PermPayload::Capabilities { caps }) => { + crate::capabilities::set_caps(name, caps) + .map_err(|e| anyhow::anyhow!("set capabilities for {name}: {e}"))?; + } + None => { + anyhow::bail!( + "PermChange entry id={} agent={} is missing perm_payload", + entry.id, + entry.agent, + ); + } + } + // Now rebuild so the updated HIVE_TOOL_GROUPS / HIVE_CAPABILITIES + // env var takes effect in the container. + let current_rev = + crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); + crate::auto_update::rebuild_agent(coord, name, ¤t_rev, Some(entry.id)).await + } } } diff --git a/hive-c0re/src/tool_groups.rs b/hive-c0re/src/tool_groups.rs index 11df0e4c..b85d7bfd 100644 --- a/hive-c0re/src/tool_groups.rs +++ b/hive-c0re/src/tool_groups.rs @@ -71,7 +71,7 @@ fn write(map: &BTreeMap>) -> std::io::Result<()> { /// Validate a slice of group name strings against `ToolGroup::ALL`. /// Returns `Ok(())` when all names are known, or `Err` listing the /// unrecognised names so callers can surface a useful error message. -fn validate_groups(groups: &[String]) -> anyhow::Result<()> { +pub fn validate_groups(groups: &[String]) -> anyhow::Result<()> { let valid: std::collections::BTreeSet<&str> = hive_sh4re::ToolGroup::ALL.iter().map(|g| g.as_str()).collect(); let unknown: Vec<&str> = groups