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

refactor(#1474): replace too_many_arguments allows on pub fns with param structs

Browse source
This commit is contained in:
damocles 2026-06-09 09:19:40 +02:00 • committed by mara
commit 3130e56cfb
9 changed files with 407 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

14
hive-ag3nt/src/bin/hive.rs
View file

@ -533,17 +533,17 @@ async fn handle_turn<S: Surface>(
let ended_at = serve_common::now_unix(); let ended_at = serve_common::now_unix();
let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) = S::post_turn_counts(socket).await; let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
let row = serve_common::build_row( let row = serve_common::build_row(serve_common::TurnRowArgs {
started_at, started_at,
ended_at, ended_at,
duration_ms, duration_ms,
model_at_start, model: model_at_start,
from.clone(), wake_from: from.clone(),
&outcome, outcome: &outcome,
bus, bus,
open_threads, open_threads_count: open_threads,
open_reminders, open_reminders_count: open_reminders,
); });
stats.record(&row); stats.record(&row);
} }
let pending = S::inbox_unread(socket).await; let pending = S::inbox_unread(socket).await;

42
hive-ag3nt/src/serve_common.rs
View file

@ -36,26 +36,36 @@ pub fn now_unix() -> i64 {
.unwrap_or(0) .unwrap_or(0)
} }
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
/// columns; `outcome` and `bus` borrow for the duration of the call.
pub struct TurnRowArgs<'a> {
pub started_at: i64,
pub ended_at: i64,
pub duration_ms: i64,
pub model: String,
pub wake_from: String,
pub outcome: &'a TurnOutcome,
pub bus: &'a Bus,
pub open_threads_count: Option<u64>,
pub open_reminders_count: Option<u64>,
}
/// Assemble a `TurnStatRow` from the harness's per-turn state. Used by both /// Assemble a `TurnStatRow` from the harness's per-turn state. Used by both
/// the agent and manager serve loops — the shape is identical, only the /// the agent and manager serve loops — the shape is identical, only the
/// post-turn count fetch helpers differ (and those stay in each binary). /// post-turn count fetch helpers differ (and those stay in each binary).
#[must_use] #[must_use]
#[allow( pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow {
clippy::too_many_arguments, let TurnRowArgs {
reason = "args mirror the turn-stats row columns 1:1; a builder struct used \ started_at,
only here would just relabel the same fields" ended_at,
)] duration_ms,
pub fn build_row( model,
started_at: i64, wake_from,
ended_at: i64, outcome,
duration_ms: i64, bus,
model: String, open_threads_count,
wake_from: String, open_reminders_count,
outcome: &TurnOutcome, } = args;
bus: &Bus,
open_threads_count: Option<u64>,
open_reminders_count: Option<u64>,
) -> TurnStatRow {
// Prefer the API-resolved model id (e.g. `claude-opus-4-8`) captured // Prefer the API-resolved model id (e.g. `claude-opus-4-8`) captured
// from this turn's assistant events over the requested `--model` // from this turn's assistant events over the requested `--model`
// name/alias, so the model-mix + cost rollup label the concrete // name/alias, so the model-mix + cost rollup label the concrete

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

@ -46,17 +46,19 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
} }
ApprovalKind::ApplyCommit => { ApprovalKind::ApplyCommit => {
coord.rebuild_queue.enqueue_full( coord
crate::rebuild_queue::QueueKind::Rebuild, .rebuild_queue
approval.agent.clone(), .enqueue_full(crate::rebuild_queue::FullEnqueue {
crate::rebuild_queue::QueueSource::Approval, kind: crate::rebuild_queue::QueueKind::Rebuild,
format!("approval #{id} apply commit"), agent: approval.agent.clone(),
None, source: crate::rebuild_queue::QueueSource::Approval,
Vec::new(), reason: format!("approval #{id} apply commit"),
Some(id), parent_id: None,
None, inputs: Vec::new(),
Vec::new(), approval_id: Some(id),
); perm_payload: None,
depends_on: Vec::new(),
});
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
} }
@ -66,17 +68,19 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// dashboard can show *which* inputs are about to bump. // dashboard can show *which* inputs are about to bump.
let inputs: Vec<String> = let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default(); serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let parent_id = coord.rebuild_queue.enqueue_full( let parent_id = coord
crate::rebuild_queue::QueueKind::MetaUpdate, .rebuild_queue
approval.agent.clone(), .enqueue_full(crate::rebuild_queue::FullEnqueue {
crate::rebuild_queue::QueueSource::Approval, kind: crate::rebuild_queue::QueueKind::MetaUpdate,
format!("approval #{id} meta input update"), agent: approval.agent.clone(),
None, source: crate::rebuild_queue::QueueSource::Approval,
inputs.clone(), reason: format!("approval #{id} meta input update"),
Some(id), parent_id: None,
None, inputs: inputs.clone(),
Vec::new(), approval_id: Some(id),
); perm_payload: None,
depends_on: Vec::new(),
});
// Pre-enqueue cascade rebuilds in topological order so // Pre-enqueue cascade rebuilds in topological order so
// agents depending on updated inputs are rebuilt after the // agents depending on updated inputs are rebuilt after the
// lock bump, matching the dashboard post_meta_update path. // lock bump, matching the dashboard post_meta_update path.
@ -95,17 +99,19 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
Ok(()) Ok(())
} }
ApprovalKind::Spawn => { ApprovalKind::Spawn => {
coord.rebuild_queue.enqueue_full( coord
crate::rebuild_queue::QueueKind::Spawn, .rebuild_queue
approval.agent.clone(), .enqueue_full(crate::rebuild_queue::FullEnqueue {
crate::rebuild_queue::QueueSource::Approval, kind: crate::rebuild_queue::QueueKind::Spawn,
format!("approval #{id} spawn"), agent: approval.agent.clone(),
None, source: crate::rebuild_queue::QueueSource::Approval,
Vec::new(), reason: format!("approval #{id} spawn"),
Some(id), parent_id: None,
None, inputs: Vec::new(),
Vec::new(), approval_id: Some(id),
); perm_payload: None,
depends_on: Vec::new(),
});
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
} }
@ -363,15 +369,15 @@ fn finish_approval(
.as_deref() .as_deref()
.map(|s| s[..s.len().min(12)].to_owned()); .map(|s| s[..s.len().min(12)].to_owned());
let status_str = if ok { "approved" } else { "failed" }; let status_str = if ok { "approved" } else { "failed" };
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
approval.id, id: approval.id,
&approval.agent, agent: &approval.agent,
approval_kind, approval_kind,
sha_short, sha_short,
status_str, status: status_str,
note.clone(), note: note.clone(),
approval.description.clone(), description: approval.description.clone(),
); });
// For spawn/rebuild/init_config approvals, also surface the underlying // For spawn/rebuild/init_config approvals, also surface the underlying
// action so the manager knows whether the lifecycle step succeeded. // action so the manager knows whether the lifecycle step succeeded.
// The ApprovalResolved event already carries the same `ok` signal but // The ApprovalResolved event already carries the same `ok` signal but
@ -756,15 +762,15 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()
sha, sha,
tag, tag,
}); });
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id, id,
&agent_owned, agent: &agent_owned,
approval_kind, approval_kind,
sha_short, sha_short,
"denied", status: "denied",
note.map(String::from), note: note.map(String::from),
description, description,
); });
} }
Ok(()) Ok(())
} }

51
hive-c0re/src/agent_server.rs
View file

@ -298,7 +298,19 @@ pub(crate) async fn dispatch_shared(
since, since,
until, until,
} => { } => {
dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await dispatch_host_journal(
agent,
HostJournalArgs {
unit,
container,
lines,
priority,
grep,
since,
until,
},
)
.await
} }
// Not a shared variant. // Not a shared variant.
_ => return None, _ => return None,
@ -561,21 +573,28 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
/// ///
/// The manager is not exempt - grant `read_host_journal` in /// The manager is not exempt - grant `read_host_journal` in
/// `meta/capabilities.json` to enable it for any agent including the manager. /// `meta/capabilities.json` to enable it for any agent including the manager.
#[allow( /// Field-named journal-query knobs for [`dispatch_host_journal`].
clippy::too_many_arguments, /// Borrows straight from the matched `GetHostJournal` request variant.
reason = "args mirror the GetHostJournal wire variant 1:1 at this single \ pub struct HostJournalArgs<'a> {
call site; a params struct would just relabel the same fields" pub unit: &'a Option<String>,
)] pub container: &'a Option<String>,
pub async fn dispatch_host_journal( pub lines: &'a Option<u32>,
agent: &str, pub priority: &'a Option<hive_sh4re::JournalPriority>,
unit: &Option<String>, pub grep: &'a Option<String>,
container: &Option<String>, pub since: &'a Option<String>,
lines: &Option<u32>, pub until: &'a Option<String>,
priority: &Option<hive_sh4re::JournalPriority>, }
grep: &Option<String>,
since: &Option<String>, pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> AgentResponse {
until: &Option<String>, let HostJournalArgs {
) -> AgentResponse { unit,
container,
lines,
priority,
grep,
since,
until,
} = args;
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) { if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) {
return AgentResponse::Err { return AgentResponse::Err {
message: "agent does not have the read_host_journal capability".to_owned(), message: "agent does not have the read_host_journal capability".to_owned(),

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

@ -349,6 +349,33 @@ impl TransientKind {
} }
} }
/// Field-named payload for [`Coordinator::emit_approval_resolved`].
/// Mirrors the `ApprovalResolved` dashboard-event fields. `agent`
/// borrows from the caller; `approval_kind` / `status` are
/// compile-time constants.
pub struct ApprovalResolved<'a> {
pub id: i64,
pub agent: &'a str,
pub approval_kind: &'static str,
pub sha_short: Option<String>,
pub status: &'static str,
pub note: Option<String>,
pub description: Option<String>,
}
/// Field-named payload for [`Coordinator::emit_question_added`].
/// Mirrors the `QuestionAdded` dashboard-event fields; all references
/// share the caller's lifetime.
pub struct QuestionAdded<'a> {
pub id: i64,
pub asker: &'a str,
pub question: &'a str,
pub options: &'a [String],
pub multi: bool,
pub deadline_at: Option<i64>,
pub target: Option<&'a str>,
}
impl Coordinator { impl Coordinator {
pub fn open( pub fn open(
db_path: &Path, db_path: &Path,
@ -652,21 +679,19 @@ impl Coordinator {
/// already have an authoritative timestamp from the db update, /// already have an authoritative timestamp from the db update,
/// the tiny skew between "row updated" and "event emitted" is /// the tiny skew between "row updated" and "event emitted" is
/// presentation-only and doesn't matter to clients. /// presentation-only and doesn't matter to clients.
#[allow( ///
clippy::too_many_arguments, /// Takes [`ApprovalResolved`] rather than a positional arg list so
reason = "args mirror the approval-resolved event payload fields; \ /// the seven fields are named at every call site.
bundling them into a struct used only here adds no clarity" pub fn emit_approval_resolved(&self, ev: ApprovalResolved<'_>) {
)] let ApprovalResolved {
pub fn emit_approval_resolved( id,
&self, agent,
id: i64, approval_kind,
agent: &str, sha_short,
approval_kind: &'static str, status,
sha_short: Option<String>, note,
status: &'static str, description,
note: Option<String>, } = ev;
description: Option<String>,
) {
let resolved_at = std::time::SystemTime::now() let resolved_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.ok() .ok()
@ -689,21 +714,16 @@ impl Coordinator {
/// both operator-targeted (`target = None`) and peer-to-peer /// both operator-targeted (`target = None`) and peer-to-peer
/// (`target = Some(agent)`) threads — the dashboard surfaces /// (`target = Some(agent)`) threads — the dashboard surfaces
/// both, distinguishing visually + offering operator override. /// both, distinguishing visually + offering operator override.
#[allow( pub fn emit_question_added(&self, ev: &QuestionAdded<'_>) {
clippy::too_many_arguments, let &QuestionAdded {
reason = "args mirror the question-added event payload fields; \ id,
bundling them into a struct used only here adds no clarity" asker,
)] question,
pub fn emit_question_added( options,
&self, multi,
id: i64, deadline_at,
asker: &str, target,
question: &str, } = ev;
options: &[String],
multi: bool,
deadline_at: Option<i64>,
target: Option<&str>,
) {
let asked_at = std::time::SystemTime::now() let asked_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH) .duration_since(std::time::UNIX_EPOCH)
.ok() .ok()

16
hive-c0re/src/dashboard/approvals.rs
View file

@ -82,15 +82,15 @@ pub(super) fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<A
.fetched_sha .fetched_sha
.as_deref() .as_deref()
.map(|s| s[..s.len().min(12)].to_owned()); .map(|s| s[..s.len().min(12)].to_owned());
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
a.id, id: a.id,
&a.agent, agent: &a.agent,
"apply_commit", approval_kind: "apply_commit",
sha_short, sha_short,
"failed", status: "failed",
Some(note.to_owned()), note: Some(note.to_owned()),
a.description.clone(), description: a.description.clone(),
); });
false false
} }
}) })

99
hive-c0re/src/manager_server.rs
View file

@ -210,12 +210,14 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
coord, coord,
MANAGER_AGENT, MANAGER_AGENT,
*id, *id,
body.clone(), EditSchedulePatch {
description.clone(), body: body.clone(),
*interval_seconds, description: description.clone(),
*next_fire_at_unix, interval_seconds: *interval_seconds,
targets_add.clone(), next_fire_at_unix: *next_fire_at_unix,
targets_remove.clone(), targets_add: targets_add.clone(),
targets_remove: targets_remove.clone(),
},
), ),
ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() { ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
Ok(schedules) => ManagerResponse::Schedules { Ok(schedules) => ManagerResponse::Schedules {
@ -425,15 +427,15 @@ pub(crate) async fn submit_apply_commit(
// explanation of why the approval can't be approved. // explanation of why the approval can't be approved.
let note = format!("{e:#}"); let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note); let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id, id,
agent, agent,
"apply_commit", approval_kind: "apply_commit",
None, sha_short: None,
"failed", status: "failed",
Some(note), note: Some(note),
description.map(str::to_owned), description: description.map(str::to_owned),
); });
return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}")); return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}"));
} }
}; };
@ -455,29 +457,29 @@ pub(crate) async fn submit_apply_commit(
if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await { if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await {
let note = format!("{e:#}"); let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note); let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id, id,
agent, agent,
"apply_commit", approval_kind: "apply_commit",
Some(sha_short.clone()), sha_short: Some(sha_short.clone()),
"failed", status: "failed",
Some(note), note: Some(note),
description.map(str::to_owned), description: description.map(str::to_owned),
); });
return Err(anyhow::anyhow!("flake lock-sync check: {e:#}")); return Err(anyhow::anyhow!("flake lock-sync check: {e:#}"));
} }
if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await { if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await {
let note = format!("{e:#}"); let note = format!("{e:#}");
let _ = coord.approvals.mark_failed(id, &note); let _ = coord.approvals.mark_failed(id, &note);
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id, id,
agent, agent,
"apply_commit", approval_kind: "apply_commit",
Some(sha_short.clone()), sha_short: Some(sha_short.clone()),
"failed", status: "failed",
Some(note), note: Some(note),
description.map(str::to_owned), description: description.map(str::to_owned),
); });
return Err(anyhow::anyhow!("flake dedup check: {e:#}")); return Err(anyhow::anyhow!("flake dedup check: {e:#}"));
} }
// Mirror the freshly-planted proposal/<id> tag to the forge. // Mirror the freshly-planted proposal/<id> tag to the forge.
@ -660,6 +662,24 @@ async fn handle_fire_schedule_now(
} }
} }
/// Field-named PATCH payload for [`handle_edit_schedule`]. Every
/// field is "leave alone" when `None`; the double-`Option` fields
/// additionally distinguish clear (`Some(None)`) from set
/// (`Some(Some(v))`).
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics: outer None = \
leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
struct EditSchedulePatch {
body: Option<String>,
description: Option<Option<String>>,
interval_seconds: Option<Option<u64>>,
next_fire_at_unix: Option<i64>,
targets_add: Option<Vec<String>>,
targets_remove: Option<Vec<String>>,
}
/// Authorize + dispatch a `EditSchedule` patch. Same ownership /// Authorize + dispatch a `EditSchedule` patch. Same ownership
/// rules as `CancelSchedule` — the manager can edit /// rules as `CancelSchedule` — the manager can edit
/// schedules it owns + any owned by an agent in its subtree. /// schedules it owns + any owned by an agent in its subtree.
@ -668,27 +688,20 @@ async fn handle_fire_schedule_now(
/// zero-interval validation. Returns `Ok` on a clean update; /// zero-interval validation. Returns `Ok` on a clean update;
/// `Err` with the underlying message on any auth / validation /// `Err` with the underlying message on any auth / validation
/// failure so the dashboard can surface it verbatim. /// failure so the dashboard can surface it verbatim.
#[allow(
clippy::too_many_arguments,
reason = "args mirror the edit-schedule PATCH fields 1:1; bundling them into \
a struct used only here adds no clarity"
)]
#[allow(
clippy::option_option,
reason = "double-Option carries three-state PATCH semantics: outer None = \
leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
fn handle_edit_schedule( fn handle_edit_schedule(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
requester: &str, requester: &str,
schedule_id: i64, schedule_id: i64,
body: Option<String>, patch: EditSchedulePatch,
description: Option<Option<String>>,
interval_seconds: Option<Option<u64>>,
next_fire_at_unix: Option<i64>,
targets_add: Option<Vec<String>>,
targets_remove: Option<Vec<String>>,
) -> ManagerResponse { ) -> ManagerResponse {
let EditSchedulePatch {
body,
description,
interval_seconds,
next_fire_at_unix,
targets_add,
targets_remove,
} = patch;
let schedule = match coord.scheduled_prompts.get(schedule_id) { let schedule = match coord.scheduled_prompts.get(schedule_id) {
Ok(Some(s)) => s, Ok(Some(s)) => s,
Ok(None) => { Ok(None) => {

26
hive-c0re/src/questions.rs
View file

@ -90,7 +90,15 @@ pub fn handle_ask(
} }
// Always fire on the dashboard channel — both operator-targeted // Always fire on the dashboard channel — both operator-targeted
// and peer threads now surface in the dashboard's questions pane. // and peer threads now surface in the dashboard's questions pane.
coord.emit_question_added(id, asker, question, options, multi, deadline_at, target); coord.emit_question_added(&crate::coordinator::QuestionAdded {
id,
asker,
question,
options,
multi,
deadline_at,
target,
});
if let Some(t) = ttl { if let Some(t) = ttl {
spawn_question_watchdog(coord, id, t); spawn_question_watchdog(coord, id, t);
} }
@ -195,15 +203,15 @@ pub fn handle_cancel_loose_end(
.fetched_sha .fetched_sha
.as_deref() .as_deref()
.map(|s| s[..s.len().min(12)].to_owned()); .map(|s| s[..s.len().min(12)].to_owned());
coord.emit_approval_resolved( coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
approval.id, id: approval.id,
&approval.agent, agent: &approval.agent,
kind_to_str(approval.kind), approval_kind: kind_to_str(approval.kind),
sha_short, sha_short,
"cancelled", status: "cancelled",
approval.note, note: approval.note,
approval.description, description: approval.description,
); });
Ok(()) Ok(())
} }
} }

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

@ -259,6 +259,22 @@ impl Default for RebuildQueue {
} }
} }
/// Full-shape submit spec for [`RebuildQueue::enqueue_full`] — every
/// `QueueEntry` field settable at submit time. The thinner `enqueue`
/// / `enqueue_with_inputs` / `enqueue_with_perm` wrappers build this
/// for the common cases.
pub struct FullEnqueue {
pub kind: QueueKind,
pub agent: String,
pub source: QueueSource,
pub reason: String,
pub parent_id: Option<u64>,
pub inputs: Vec<String>,
pub approval_id: Option<i64>,
pub perm_payload: Option<PermPayload>,
pub depends_on: Vec<u64>,
}
impl RebuildQueue { impl RebuildQueue {
pub fn new() -> Self { pub fn new() -> Self {
Self::default() Self::default()
@ -294,17 +310,17 @@ impl RebuildQueue {
reason: String, reason: String,
parent_id: Option<u64>, parent_id: Option<u64>,
) -> u64 { ) -> u64 {
self.enqueue_full( self.enqueue_full(FullEnqueue {
kind, kind,
agent, agent,
source, source,
reason, reason,
parent_id, parent_id,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
) })
} }
/// Same as `enqueue` but carries an `inputs` payload — used by /// Same as `enqueue` but carries an `inputs` payload — used by
@ -321,17 +337,17 @@ impl RebuildQueue {
parent_id: Option<u64>, parent_id: Option<u64>,
inputs: Vec<String>, inputs: Vec<String>,
) -> u64 { ) -> u64 {
self.enqueue_full( self.enqueue_full(FullEnqueue {
kind, kind,
agent, agent,
source, source,
reason, reason,
parent_id, parent_id,
inputs, inputs,
None, approval_id: None,
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
) })
} }
/// Enqueue a `PermChange` entry for `agent`. The worker applies the /// Enqueue a `PermChange` entry for `agent`. The worker applies the
@ -344,17 +360,17 @@ impl RebuildQueue {
reason: String, reason: String,
payload: PermPayload, payload: PermPayload,
) -> u64 { ) -> u64 {
self.enqueue_full( self.enqueue_full(FullEnqueue {
QueueKind::PermChange, kind: QueueKind::PermChange,
agent, agent,
source, source,
reason, reason,
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
Some(payload), perm_payload: Some(payload),
Vec::new(), depends_on: Vec::new(),
) })
} }
/// Full-shape enqueue — every `QueueEntry` field that's settable /// Full-shape enqueue — every `QueueEntry` field that's settable
@ -362,27 +378,18 @@ impl RebuildQueue {
/// `enqueue_with_perm` delegate to this; the approval-driven POST /// `enqueue_with_perm` delegate to this; the approval-driven POST
/// handlers call it directly with the source row's id so the /// handlers call it directly with the source row's id so the
/// worker can re-fetch the kind-specific payload. /// worker can re-fetch the kind-specific payload.
// 10 args: the queue entry has 6 independent submit-time fields plus pub fn enqueue_full(&self, spec: FullEnqueue) -> u64 {
// four kind-specific payload fields (inputs, approval_id, perm_payload, let FullEnqueue {
// depends_on). A builder struct would obscure the call sites; the kind,
// shorter wrappers already cover all common cases. agent,
#[allow( source,
clippy::too_many_arguments, reason,
reason = "args mirror the queue-entry fields the row is built from; the \ parent_id,
thinner enqueue helpers wrap this for the common cases" inputs,
)] approval_id,
pub fn enqueue_full( perm_payload,
&self, depends_on,
kind: QueueKind, } = spec;
agent: String,
source: QueueSource,
reason: String,
parent_id: Option<u64>,
inputs: Vec<String>,
approval_id: Option<i64>,
perm_payload: Option<PermPayload>,
depends_on: Vec<u64>,
) -> u64 {
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
// Dedup against a pending entry with the same (kind, agent) — // Dedup against a pending entry with the same (kind, agent) —
// and, for MetaUpdate, the same `inputs` list (see method // and, for MetaUpdate, the same `inputs` list (see method
@ -1213,17 +1220,17 @@ mod tests {
#[test] #[test]
fn approval_entries_keep_approval_id() { fn approval_entries_keep_approval_id() {
let q = RebuildQueue::new(); let q = RebuildQueue::new();
let id = q.enqueue_full( let id = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"agent-a".to_owned(), agent: "agent-a".to_owned(),
QueueSource::Approval, source: QueueSource::Approval,
"approval #42 apply commit".to_owned(), reason: "approval #42 apply commit".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
Some(42), approval_id: Some(42),
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
); });
let snap = q.snapshot(); let snap = q.snapshot();
let entry = snap.iter().find(|e| e.id == id).expect("entry present"); let entry = snap.iter().find(|e| e.id == id).expect("entry present");
assert_eq!(entry.approval_id, Some(42)); assert_eq!(entry.approval_id, Some(42));
@ -1237,43 +1244,43 @@ mod tests {
// approve click is a separate piece of work even when the // approve click is a separate piece of work even when the
// (kind, agent) pair matches. // (kind, agent) pair matches.
let q = RebuildQueue::new(); let q = RebuildQueue::new();
let a = q.enqueue_full( let a = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"agent-a".to_owned(), agent: "agent-a".to_owned(),
QueueSource::Approval, source: QueueSource::Approval,
"approval #1".to_owned(), reason: "approval #1".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
Some(1), approval_id: Some(1),
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
); });
let b = q.enqueue_full( let b = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"agent-a".to_owned(), agent: "agent-a".to_owned(),
QueueSource::Approval, source: QueueSource::Approval,
"approval #2".to_owned(), reason: "approval #2".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
Some(2), approval_id: Some(2),
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
); });
assert_ne!(a, b); assert_ne!(a, b);
assert_eq!(q.snapshot().len(), 2); assert_eq!(q.snapshot().len(), 2);
// Same approval_id submitted twice DOES dedup (rapid double- // Same approval_id submitted twice DOES dedup (rapid double-
// click on the dashboard's approve button is a single op). // click on the dashboard's approve button is a single op).
let c = q.enqueue_full( let c = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"agent-a".to_owned(), agent: "agent-a".to_owned(),
QueueSource::Approval, source: QueueSource::Approval,
"approval #1 (duplicate)".to_owned(), reason: "approval #1 (duplicate)".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
Some(1), approval_id: Some(1),
None, perm_payload: None,
Vec::new(), depends_on: Vec::new(),
); });
assert_eq!(a, c); assert_eq!(a, c);
assert_eq!(q.snapshot().len(), 2); assert_eq!(q.snapshot().len(), 2);
} }
@ -1459,17 +1466,17 @@ mod tests {
"first".to_owned(), "first".to_owned(),
None, None,
); );
let b = q.enqueue_full( let b = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"agent-b".to_owned(), agent: "agent-b".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"second (blocked on a)".to_owned(), reason: "second (blocked on a)".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![a], depends_on: vec![a],
); });
// B depends on A — take_next should give A first. // B depends on A — take_next should give A first.
let first = q.take_next().expect("a is ready"); let first = q.take_next().expect("a is ready");
assert_eq!(first.id, a); assert_eq!(first.id, a);
@ -1529,17 +1536,17 @@ mod tests {
"dep must be evicted from history" "dep must be evicted from history"
); );
// An entry that depends on the (evicted) dep must be immediately runnable. // An entry that depends on the (evicted) dep must be immediately runnable.
let downstream = q.enqueue_full( let downstream = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"downstream".to_owned(), agent: "downstream".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"downstream (dep evicted = resolved)".to_owned(), reason: "downstream (dep evicted = resolved)".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![dep], depends_on: vec![dep],
); });
let got = q.take_next().expect("downstream runnable when dep evicted"); let got = q.take_next().expect("downstream runnable when dep evicted");
assert_eq!(got.id, downstream); assert_eq!(got.id, downstream);
} }
@ -1563,42 +1570,42 @@ mod tests {
"d2".to_owned(), "d2".to_owned(),
None, None,
); );
let a = q.enqueue_full( let a = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"target".to_owned(), agent: "target".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"r".to_owned(), reason: "r".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![dep1], depends_on: vec![dep1],
); });
// Same kind+agent but different depends_on — must NOT dedup. // Same kind+agent but different depends_on — must NOT dedup.
let b = q.enqueue_full( let b = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"target".to_owned(), agent: "target".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"r".to_owned(), reason: "r".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![dep2], depends_on: vec![dep2],
); });
assert_ne!(a, b, "different depends_on must produce distinct entries"); assert_ne!(a, b, "different depends_on must produce distinct entries");
// Same depends_on as a — must dedup. // Same depends_on as a — must dedup.
let c = q.enqueue_full( let c = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"target".to_owned(), agent: "target".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"r again".to_owned(), reason: "r again".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![dep1], depends_on: vec![dep1],
); });
assert_eq!(a, c, "identical depends_on must dedup"); assert_eq!(a, c, "identical depends_on must dedup");
} }
@ -1615,17 +1622,17 @@ mod tests {
"a".to_owned(), "a".to_owned(),
None, None,
); );
let b = q.enqueue_full( let b = q.enqueue_full(FullEnqueue {
QueueKind::Rebuild, kind: QueueKind::Rebuild,
"b".to_owned(), agent: "b".to_owned(),
QueueSource::Manual, source: QueueSource::Manual,
"b (blocked on a)".to_owned(), reason: "b (blocked on a)".to_owned(),
None, parent_id: None,
Vec::new(), inputs: Vec::new(),
None, approval_id: None,
None, perm_payload: None,
vec![a], depends_on: vec![a],
); });
q.take_next(); // pop a, mark Running q.take_next(); // pop a, mark Running
q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned())); q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned()));
let got = q.take_next().expect("b runnable after a failed"); let got = q.take_next().expect("b runnable after a failed");