hyperhive/hive-c0re/src/rebuild_queue.rs

//! Global rebuild queue — serialises all long-running container/meta
//! operations (rebuild, meta-update, first-spawn) through a single
//! background worker. Design rationale, kind taxonomy, dedup rules,
//! cascade parent tracking, and step labels:
//! `docs/coordinator.md::Rebuild queue`.

use std::collections::VecDeque;
use std::sync::Mutex;

use anyhow::Context as _;
use serde::{Deserialize, Serialize};
use tokio::sync::Notify;

/// What the queue can run. Each variant maps to a specific worker
/// execution path; `agent` (in `QueueEntry`) names the target where
/// relevant.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum QueueKind {
    /// Rebuild a single agent's container (`auto_update::rebuild_agent`).
    Rebuild,
    /// Run `nix flake update` on the meta flake. Triggers cascade
    /// `Rebuild` entries (with `parent_id`) once the lock bump lands.
    MetaUpdate,
    /// First-deploy spawn of a new agent (approval-driven).
    Spawn,
    /// Destroy with `--purge` (real fs work). Not yet routed here; the
    /// variant exists so the wire shape doesn't need to change later.
    #[allow(dead_code, reason = "wire shape — routed by a future PR")]
    Destroy,
    /// hive-c0re boot-time sweep: bumps the meta hyperhive lock then
    /// enqueues a `Rebuild` child for every managed container. Completes
    /// after the lock bump; children run as independent queue entries
    /// grouped under this parent's `id`. `agent` = `"hyperhive"`.
    StartupSweep,
    /// Stop + start a container without touching config. Fast op (~5-10s).
    /// 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,
    /// Gracefully stop a container: signal the harness to run one
    /// stop-checkpoint turn (flush durable `/state`), then hand the drain-wait
    /// to a detached watcher (freeing the build lane) which, once the agent
    /// drains or `GRACEFUL_STOP_TIMEOUT` elapses, enqueues a fast-lane `Stop`
    /// for the actual `nixos-container stop`. The build worker only does the
    /// cheap signal, so whole-hive graceful stops overlap every agent's drain.
    GracefulStop,
    /// Start a stopped container (`lifecycle::start`). Routed through the
    /// queue so the dashboard shows a visible queued→running transient — a
    /// direct sub-second start only flashes the badge — and bulk starts
    /// serialise legibly on the queue. Fast op.
    Start,
    /// Hard-stop a container (`lifecycle::kill`), no quiesce. Routed through
    /// the queue for the same visible-progress reason as `Start`; the
    /// quiescing variant is `GracefulStop`. Fast op.
    Stop,
}

impl QueueKind {
    pub fn as_str(self) -> &'static str {
        match self {
            QueueKind::Rebuild => "rebuild",
            QueueKind::MetaUpdate => "meta_update",
            QueueKind::Spawn => "spawn",
            QueueKind::Destroy => "destroy",
            QueueKind::StartupSweep => "startup_sweep",
            QueueKind::Restart => "restart",
            QueueKind::PermChange => "perm_change",
            QueueKind::GracefulStop => "graceful_stop",
            QueueKind::Start => "start",
            QueueKind::Stop => "stop",
        }
    }

    /// Fast-lane kinds: hard `Start` / `Stop`. These run on a separate
    /// serial fast worker concurrently with the build lane (so a stop/start
    /// never waits behind another container's slow build). `GracefulStop`
    /// and `Restart` are deliberately NOT fast — they go through the build
    /// lane (`GracefulStop` does the cheap harness signal then detaches the
    /// drain-wait, enqueueing a fast-lane `Stop` for the real container stop;
    /// `Restart` is a stop+start).
    pub fn is_fast(self) -> bool {
        matches!(self, QueueKind::Start | QueueKind::Stop)
    }
}

/// 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<String> },
    /// Set the capabilities for one agent (`capabilities.json`).
    Capabilities { caps: Vec<String> },
    /// Set both perm-types for one agent in a single entry — the batch
    /// `POST /api/permissions` path. Either field `None` leaves that
    /// file untouched (no write, no commit); the worker commits whichever
    /// are present in one git commit, then rebuilds once. Collapses the
    /// dedup key to `(kind, agent)` so caps + groups for one agent
    /// produce a single rebuild rather than two.
    Combined {
        groups: Option<Vec<String>>,
        caps: Option<Vec<String>>,
    },
}

/// 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.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum QueueSource {
    /// Operator clicked rebuild / update-all / meta-update on the
    /// dashboard, or any other direct human action (CLI, manager tool).
    Manual,
    /// Spawned as a cascade from a `MetaUpdate` entry's lock-bump
    /// fan-out. The `parent_id` on the `QueueEntry` points back at
    /// the originating meta-update.
    MetaUpdate,
    /// `auto_update::run` startup sweep — rebuild every container on
    /// hive-c0re boot. Legacy flat source (no parent); replaced by
    /// `StartupSweep` for the parent entry and child rebuilds once the
    /// queue introduced `parent_id` grouping. Kept for wire compatibility
    /// with entries logged before the migration.
    AutoUpdate,
    /// Direct child of a `StartupSweep` queue entry — one per agent in
    /// the boot-time rebuild sweep. Carries `parent_id` back-link so
    /// the dashboard renders the sweep's per-agent rebuilds nested under
    /// the parent header. The parent entry itself uses `QueueSource::AutoUpdate`
    /// (automated boot action, not operator-driven).
    StartupSweep,
    /// Crash recovery path (future use — currently no auto-rebuild on
    /// crash, but the variant exists for the imminent feature).
    #[allow(dead_code, reason = "wire shape — used by a future 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 {
    pub fn as_str(self) -> &'static str {
        match self {
            QueueSource::Manual => "manual",
            QueueSource::MetaUpdate => "meta_update",
            QueueSource::AutoUpdate => "auto_update",
            QueueSource::StartupSweep => "startup_sweep",
            QueueSource::CrashRecover => "crash_recover",
            QueueSource::Approval => "approval",
        }
    }
}

/// Lifecycle state of an entry. `Done` / `Failed` / `Cancelled` are
/// retained in the queue snapshot for a short tail (`MAX_HISTORY_PER_KIND`)
/// so the dashboard can show "last few" runs alongside live state.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum QueueState {
    Queued,
    Running,
    Done,
    Failed,
    Cancelled,
}

impl QueueState {
    pub fn is_terminal(self) -> bool {
        matches!(
            self,
            QueueState::Done | QueueState::Failed | QueueState::Cancelled
        )
    }
}

/// A single queue entry — what's pending, running, or recently finished.
/// Serialised verbatim onto the dashboard event channel and the
/// `/api/state` snapshot.
#[derive(Debug, Clone, Serialize)]
pub struct QueueEntry {
    /// Monotonic per-process id. Stable for the lifetime of the entry
    /// so SSE upserts land in place rather than churning the list.
    pub id: u64,
    /// Target agent name, or the literal `"hyperhive"` for entries
    /// (`MetaUpdate`) that affect the meta flake rather than a single
    /// agent.
    pub agent: String,
    pub kind: QueueKind,
    pub state: QueueState,
    pub source: QueueSource,
    /// Groups cascade entries under their originating parent. For a
    /// `MetaUpdate` entry this is `None`; for the per-agent rebuilds
    /// the worker enqueues after the lock bump it's `Some(meta_id)`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent_id: Option<u64>,
    /// Human-readable "why" — populated by the enqueuer (`"manual via
    /// dashboard"`, `"meta-update cascade (hyperhive bumped)"`,
    /// `"startup sweep"`). Free-form; dedup appends `(also requested
    /// by …)` lines on repeated enqueues.
    pub reason: String,
    pub enqueued_at: i64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started_at: Option<i64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finished_at: Option<i64>,
    /// Populated when `state == Failed`. Carries the worker's error
    /// string (already truncated to a reasonable length by the caller).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// `MetaUpdate`-only payload: the list of meta flake inputs to run
    /// through `nix flake update`. Empty / absent on `Rebuild` /
    /// `Spawn` / `Destroy` entries; absent on the wire (never
    /// 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>,
    /// Current sub-step inside the running entry. Worker mutates this
    /// as the kind-specific pipeline
    /// advances through phases (e.g. `"plant tags"` →
    /// `"nixos-container update"` → `"finalize deploy"`). `None` while
    /// `Queued` and after terminal — only meaningful with
    /// `state == Running`. Each transition fires a fresh
    /// `RebuildQueueChanged` snapshot so the dashboard can render
    /// the label as a sub-line on the queue card. Free-form per
    /// pipeline; the kind-specific worker is the source of truth.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub step: Option<String>,
    /// `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<PermPayload>,
    /// Entries this entry must wait for before it can run. The worker
    /// skips this entry until every id in the list has reached a
    /// terminal state (`Done` / `Failed` / `Cancelled`) — or no longer
    /// exists in the queue (evicted terminal entries are treated as
    /// resolved, since `trim_history` only evicts terminals). Empty on
    /// most entries; serialised only when non-empty.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub depends_on: Vec<u64>,
    /// Row id of the associated `build_logs` entry (opened by the
    /// lifecycle worker when `nixos-container update` starts). Set
    /// shortly after `state` transitions to `Running`; `None` while
    /// `Queued` or for entries that don't open a build log (`Restart`,
    /// `PermChange` file-write phase, etc.). Links the queue card to
    /// the live-streaming `/api/build-logs/id/{id}/stream` endpoint so
    /// the operator can follow the nix build output in real time.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build_log_id: Option<i64>,
}

/// How many terminal-state entries (`Done` / `Failed` / `Cancelled`)
/// to retain per kind in the snapshot. Older entries get evicted to
/// keep `/api/state` tight; the live event channel is unaffected.
const MAX_HISTORY_PER_KIND: usize = 5;

/// Inner state guarded by a single mutex. Held briefly — every
/// operation is constant-time relative to the queue's depth, and
/// the depths in practice are tiny (single-digit).
#[derive(Debug, Default)]
struct Inner {
    entries: VecDeque<QueueEntry>,
    next_id: u64,
}

/// Global rebuild queue. Lives on `Coordinator` (one per hive-c0re
/// process). The associated `Notify` wakes the worker when something
/// new arrives.
#[derive(Debug)]
pub struct RebuildQueue {
    inner: Mutex<Inner>,
    /// Build-lane worker wakes on this signal. The worker checks the queue
    /// and loops back to `notified().await` when there's nothing to run.
    /// Also nudged by the fast worker when a fast op finishes (a build may
    /// have been deferred behind a `Running` fast op for the same agent).
    pub(crate) notify: Notify,
    /// Fast-lane worker wakes on this signal. Nudged on a fast `Start` /
    /// `Stop` enqueue and by the build worker when a build finishes (a
    /// deferred `Start` may now be runnable).
    pub(crate) fast_notify: Notify,
}

impl Default for RebuildQueue {
    fn default() -> Self {
        Self {
            inner: Mutex::new(Inner::default()),
            notify: Notify::new(),
            fast_notify: Notify::new(),
        }
    }
}

/// 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 {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add an entry to the queue. Returns the entry's id (newly-allocated
    /// or — on dedup — the existing entry's id with the new reason
    /// appended).
    ///
    /// Dedup rule:
    /// - `Rebuild` / `Spawn` / `Destroy`: a `Queued` entry with the same
    ///   `(kind, agent, parent_id)` swallows the new request. `parent_id`
    ///   is part of the key so that a `MetaUpdate` cascade rebuild (with a
    ///   specific `parent_id`) never collapses into a standalone rebuild or
    ///   a cascade from a different `MetaUpdate`. Without this guard a
    ///   cascade rebuild pre-enqueued before the lock bump would be swallowed
    ///   by an existing `Queued` startup-sweep rebuild, causing the agent to
    ///   never rebuild against the post-bump meta.
    /// - `MetaUpdate`: dedup ALSO requires the `inputs` field to match —
    ///   two meta-updates with different input lists are distinct work
    ///   and must queue separately, otherwise the second meta-update
    ///   would silently collapse into the first whenever it was still
    ///   `Queued`, losing the second's input set.
    ///
    /// Running and terminal entries never dedup — operators are free
    /// to re-queue a rebuild that's currently running (something
    /// changed since it started) or re-run one that just finished.
    pub fn enqueue(
        &self,
        kind: QueueKind,
        agent: String,
        source: QueueSource,
        reason: String,
        parent_id: Option<u64>,
    ) -> u64 {
        self.enqueue_full(FullEnqueue {
            kind,
            agent,
            source,
            reason,
            parent_id,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: Vec::new(),
        })
    }

    /// Same as `enqueue` but carries an `inputs` payload — used by
    /// `MetaUpdate` enqueues to tell the worker which meta-flake
    /// inputs to bump. For `MetaUpdate` the `inputs` value is part of
    /// the dedup key (two meta-updates with different inputs are
    /// distinct operations).
    pub fn enqueue_with_inputs(
        &self,
        kind: QueueKind,
        agent: String,
        source: QueueSource,
        reason: String,
        parent_id: Option<u64>,
        inputs: Vec<String>,
    ) -> u64 {
        self.enqueue_full(FullEnqueue {
            kind,
            agent,
            source,
            reason,
            parent_id,
            inputs,
            approval_id: None,
            perm_payload: None,
            depends_on: Vec::new(),
        })
    }

    /// 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(FullEnqueue {
            kind: QueueKind::PermChange,
            agent,
            source,
            reason,
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: Some(payload),
            depends_on: Vec::new(),
        })
    }

    /// Full-shape enqueue — every `QueueEntry` field that's settable
    /// 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.
    pub fn enqueue_full(&self, spec: FullEnqueue) -> u64 {
        let FullEnqueue {
            kind,
            agent,
            source,
            reason,
            parent_id,
            inputs,
            approval_id,
            perm_payload,
            depends_on,
        } = spec;
        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 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. 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). 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 = matches!(
                (&entry.perm_payload, &perm_payload),
                (
                    Some(PermPayload::ToolGroups { .. }),
                    Some(PermPayload::ToolGroups { .. })
                ) | (
                    Some(PermPayload::Capabilities { .. }),
                    Some(PermPayload::Capabilities { .. })
                ) | (
                    Some(PermPayload::Combined { .. }),
                    Some(PermPayload::Combined { .. })
                ) | (None, None)
            );
            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
                && entry.depends_on == depends_on
            {
                if !entry.reason.contains(&reason) {
                    use std::fmt::Write as _;
                    let _ = write!(entry.reason, "\nalso requested by: {reason}");
                }
                return entry.id;
            }
        }
        inner.next_id += 1;
        let id = inner.next_id;
        let entry = QueueEntry {
            id,
            agent,
            kind,
            state: QueueState::Queued,
            source,
            parent_id,
            reason,
            enqueued_at: now_unix(),
            started_at: None,
            finished_at: None,
            error: None,
            inputs,
            approval_id,
            step: None,
            perm_payload,
            depends_on,
            build_log_id: None,
        };
        inner.entries.push_back(entry);
        // Wake the worker for this entry's lane (fast = Start/Stop, build =
        // everything else). `notify_one` is a no-op when there's no waiter;
        // the next `notified().await` returns immediately.
        if kind.is_fast() {
            self.fast_notify.notify_one();
        } else {
            self.notify.notify_one();
        }
        id
    }

    /// Claim the next runnable `Queued` entry for the **build** lane (every
    /// kind except the fast `Start` / `Stop`) and mark it `Running`. See
    /// [`Self::claim`] for the dependency + per-agent rules.
    pub fn take_next_build(&self) -> Option<QueueEntry> {
        self.claim(false)
    }

    /// Claim the next runnable `Queued` entry for the **fast** lane (hard
    /// `Start` / `Stop`) and mark it `Running`. Runs on its own serial
    /// worker concurrently with the build lane. See [`Self::claim`].
    pub fn take_next_fast(&self) -> Option<QueueEntry> {
        self.claim(true)
    }

    /// Pop the next `Queued` entry for one lane whose dependencies are
    /// resolved and which doesn't race the same agent's other-lane work,
    /// and mark it `Running`. Returns the entry (a clone — the original
    /// stays in the queue so live state reflects "this is currently
    /// running"). Returns `None` when nothing in this lane is runnable.
    ///
    /// `want_fast` selects the lane: `true` = fast (`Start` / `Stop`),
    /// `false` = build (everything else). The two lanes run on separate
    /// serial workers, so this is called from both — the lane filter keeps
    /// each worker to its own kinds.
    ///
    /// A dependency is "resolved" when the dep's id is either:
    /// - still in the queue AND in a terminal state (`Done` / `Failed`
    ///   / `Cancelled`), OR
    /// - no longer in the queue (evicted by `trim_history` — only
    ///   terminal entries are ever evicted, so missing == completed).
    ///
    /// Per-agent cross-lane guard (so a fast op never races that agent's
    /// own build, and vice versa):
    /// - a fast op waits while its agent has a build entry `Running`;
    ///   a `Start` additionally waits while its agent has a build entry
    ///   `Queued` (a start of a soon-to-be-rebuilt container is pointless);
    /// - a build op waits while its agent has a fast op `Running`.
    fn claim(&self, want_fast: bool) -> Option<QueueEntry> {
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        // Collect ids that are still in the queue and terminal. Entries
        // absent from the queue are also considered resolved (see above).
        let terminal_ids: std::collections::HashSet<u64> = inner
            .entries
            .iter()
            .filter(|e| e.state.is_terminal())
            .map(|e| e.id)
            .collect();
        // Active (non-terminal) ids: Queued + Running. Named `active_ids`
        // rather than `queued_ids` because Running entries are included;
        // used to distinguish "still in flight" from "evicted (= resolved)".
        let active_ids: std::collections::HashSet<u64> = inner
            .entries
            .iter()
            .filter(|e| !e.state.is_terminal())
            .map(|e| e.id)
            .collect();
        let pos = {
            let entries = &inner.entries;
            entries.iter().position(|e| {
                e.state == QueueState::Queued
                    && e.kind.is_fast() == want_fast
                    && e.depends_on.iter().all(|dep_id| {
                        // Resolved if terminal in queue OR not in queue at all.
                        // Note: circular deps (A depends on B, B depends on A)
                        // silently deadlock — neither entry ever becomes
                        // runnable. Callers must ensure acyclic dep graphs.
                        terminal_ids.contains(dep_id) || !active_ids.contains(dep_id)
                    })
                    && lane_clear(entries, e)
            })
        }?;
        let entry = &mut inner.entries[pos];
        entry.state = QueueState::Running;
        entry.started_at = Some(now_unix());
        Some(entry.clone())
    }

    /// Mark an entry terminal. `error` is populated for `Failed`;
    /// `Done` / `Cancelled` ignore it. Trims the history tail.
    /// Clears `step` — the field is only meaningful while `Running`,
    /// and leaving a stale "in flight" label after a terminal
    /// transition would mislead the dashboard render.
    pub fn finish(&self, id: u64, state: QueueState, error: Option<String>) {
        debug_assert!(
            state.is_terminal(),
            "finish() called with non-terminal {state:?}"
        );
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) {
            entry.state = state;
            entry.finished_at = Some(now_unix());
            entry.error = error.filter(|_| state == QueueState::Failed);
            entry.step = None;
        }
        Self::trim_history(&mut inner);
    }

    /// Set the current sub-step label on a `Running` entry.
    /// Returns `true` when the row was found AND the label changed
    /// (caller should emit a `RebuildQueueChanged` snapshot only on
    /// `true` to avoid noisy duplicate frames). No-op for entries not
    /// in `Running` — the field is conceptually undefined outside
    /// that state.
    pub fn set_step(&self, id: u64, step: impl Into<String>) -> bool {
        let new_step = step.into();
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) else {
            return false;
        };
        if entry.state != QueueState::Running {
            return false;
        }
        if entry.step.as_deref() == Some(new_step.as_str()) {
            return false;
        }
        entry.step = Some(new_step);
        true
    }

    /// Link a `build_logs` row to a `Running` entry. Called by the
    /// lifecycle worker when `nixos-container update` opens a build log
    /// row so the dashboard can surface a "view logs" link while the
    /// build is in flight. Returns `true` when the row was found and
    /// the id was stored; `false` when the entry is no longer in the
    /// queue or is not `Running`.
    pub fn set_build_log_id(&self, id: u64, log_id: i64) -> bool {
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) else {
            return false;
        };
        if entry.state != QueueState::Running {
            return false;
        }
        entry.build_log_id = Some(log_id);
        true
    }

    /// Snapshot the queue for `/api/state` and `RebuildQueueChanged`.
    /// Cheap clone — entries are small (~hundreds of bytes each).
    pub fn snapshot(&self) -> Vec<QueueEntry> {
        let inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        inner.entries.iter().cloned().collect()
    }

    /// Cancel every `Queued` entry whose `parent_id` matches `parent`.
    /// Used when a `MetaUpdate` parent fails its lock bump — the
    /// cascade rebuilds the enqueuer pre-queued no longer apply
    /// (nothing actually changed, so they'd be wasted work). Running
    /// children are left alone — they were started under the parent's
    /// assumption and can't be cleanly aborted from the queue side.
    /// Returns the count of cancelled entries.
    pub fn cancel_children(&self, parent: u64) -> usize {
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        let mut count = 0;
        for entry in &mut inner.entries {
            if entry.parent_id == Some(parent) && entry.state == QueueState::Queued {
                entry.state = QueueState::Cancelled;
                entry.finished_at = Some(now_unix());
                count += 1;
            }
        }
        if count > 0 {
            Self::trim_history(&mut inner);
        }
        count
    }

    /// Cancel a `Queued` entry (no-op for `Running` / terminal — the
    /// in-flight rebuild owns the agent's nix store and can't be
    /// safely interrupted). Returns true when an entry was cancelled.
    pub fn cancel(&self, id: u64) -> bool {
        let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
        if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id)
            && entry.state == QueueState::Queued
        {
            entry.state = QueueState::Cancelled;
            entry.finished_at = Some(now_unix());
            Self::trim_history(&mut inner);
            return true;
        }
        false
    }

    /// Keep only the most recent `MAX_HISTORY_PER_KIND` terminal entries
    /// per kind. Pending + running entries are never evicted.
    fn trim_history(inner: &mut Inner) {
        let mut counts: std::collections::HashMap<QueueKind, usize> =
            std::collections::HashMap::new();
        // Walk newest-first; keep the first MAX_HISTORY_PER_KIND
        // terminals per kind, evict the rest.
        let entries: Vec<QueueEntry> = inner
            .entries
            .iter()
            .rev()
            .filter(|e| {
                if !e.state.is_terminal() {
                    return true;
                }
                let n = counts.entry(e.kind).or_insert(0);
                *n += 1;
                *n <= MAX_HISTORY_PER_KIND
            })
            .cloned()
            .collect();
        inner.entries = entries.into_iter().rev().collect();
    }
}

/// Per-agent cross-lane guard for [`RebuildQueue::claim`]: returns true when
/// entry `e` is safe to start given the same agent's other-lane work in
/// `entries`. A fast op waits for the agent's `Running` build (and a `Start`
/// also for a `Queued` build); a build op waits for the agent's `Running`
/// fast op. Keeps a stop/start from racing that container's own rebuild.
fn lane_clear(entries: &VecDeque<QueueEntry>, e: &QueueEntry) -> bool {
    let agent = e.agent.as_str();
    if e.kind.is_fast() {
        let build_blocking = entries.iter().any(|b| {
            !b.kind.is_fast()
                && b.agent == agent
                && (b.state == QueueState::Running
                    || (e.kind == QueueKind::Start && b.state == QueueState::Queued))
        });
        !build_blocking
    } else {
        !entries
            .iter()
            .any(|f| f.kind.is_fast() && f.agent == agent && f.state == QueueState::Running)
    }
}

/// Background worker that drains the queue. Spawned once at hive-c0re
/// startup from `main.rs`. Loops forever:
/// 1. Pop the next `Queued` entry (`take_next` marks it `Running` and
///    fires a `RebuildQueueChanged` snapshot via the caller).
/// 2. Dispatch by kind — single-agent rebuild, meta-update + cascade,
///    or first-spawn.
/// 3. Mark the entry terminal (`finish`) and emit another snapshot.
/// 4. When the queue is empty, `await` on `notify` until something
///    new lands.
///
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true
/// signal the worker exits after its current entry finishes; pending
/// `Queued` entries are dropped (they'll either be replayed by the
/// startup sweep on next boot or left for an operator to re-queue).
/// Max time the `GracefulStop` drain watcher waits for the harness to run its
/// stop-checkpoint turn + drain before falling back to a hard container stop.
/// Generous — a checkpoint turn can take a while — but bounded so a wedged
/// agent never blocks the stop indefinitely. The wait runs in a detached
/// watcher task (not the build worker), so a whole-hive graceful stop overlaps
/// every agent's drain instead of serialising N × this timeout.
const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);

/// Run one claimed queue entry to completion: snapshot, dispatch, mark
/// terminal, snapshot. Shared by both lane workers.
async fn run_one(coord: &std::sync::Arc<crate::coordinator::Coordinator>, entry: &QueueEntry) {
    coord.emit_rebuild_queue_snapshot();
    tracing::info!(
        id = entry.id,
        kind = entry.kind.as_str(),
        agent = %entry.agent,
        source = entry.source.as_str(),
        "rebuild_queue: running"
    );
    match dispatch(coord, entry).await {
        Ok(()) => {
            coord.rebuild_queue.finish(entry.id, QueueState::Done, None);
            tracing::info!(id = entry.id, "rebuild_queue: done");
        }
        Err(e) => {
            let msg = format!("{e:#}");
            let truncated = if msg.len() > 2_000 {
                format!("{}…", &msg[..2_000])
            } else {
                msg.clone()
            };
            coord
                .rebuild_queue
                .finish(entry.id, QueueState::Failed, Some(truncated));
            tracing::warn!(id = entry.id, error = %msg, "rebuild_queue: failed");
        }
    }
    coord.emit_rebuild_queue_snapshot();
}

/// Build-lane worker: drains every non-fast kind serially. Spawned once at
/// hive-c0re startup from `main.rs`, alongside [`run_fast_worker`] which
/// drains the fast `Start` / `Stop` lane concurrently.
///
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true
/// signal the worker exits after its current entry finishes; pending
/// `Queued` entries are dropped (replayed by the startup sweep on next boot
/// or left for an operator to re-queue).
pub async fn run_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
    let mut shutdown = coord.shutdown_rx();
    loop {
        while let Some(entry) = coord.rebuild_queue.take_next_build() {
            run_one(&coord, &entry).await;
            // A finished build may unblock a fast op that was deferred behind
            // this agent's build — nudge the fast lane to re-check.
            coord.rebuild_queue.fast_notify.notify_one();
        }
        tokio::select! {
            biased;
            res = shutdown.changed() => {
                if res.is_err() || *shutdown.borrow() {
                    tracing::info!("rebuild_queue: build worker exiting on shutdown");
                    return;
                }
            }
            () = coord.rebuild_queue.notify.notified() => {}
        }
    }
}

/// Fast-lane worker: drains hard `Start` / `Stop` serially, concurrently
/// with [`run_worker`], so a stop/start never waits behind another
/// container's slow build. Per-agent ordering vs that agent's own build is
/// enforced in [`RebuildQueue::claim`].
pub async fn run_fast_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
    let mut shutdown = coord.shutdown_rx();
    loop {
        while let Some(entry) = coord.rebuild_queue.take_next_fast() {
            run_one(&coord, &entry).await;
            // A finished fast op may unblock a build deferred behind it.
            coord.rebuild_queue.notify.notify_one();
        }
        tokio::select! {
            biased;
            res = shutdown.changed() => {
                if res.is_err() || *shutdown.borrow() {
                    tracing::info!("rebuild_queue: fast worker exiting on shutdown");
                    return;
                }
            }
            () = coord.rebuild_queue.fast_notify.notified() => {}
        }
    }
}

/// 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.
/// Pick the right approval pipeline for a `Rebuild` queue entry. Both
/// `ApplyCommit` and `MergeConfigPr` approvals enqueue a `Rebuild` entry
/// (both end in a container rebuild); branch on the approval kind. Falls
/// back to the apply-commit path if the row can't be read — it re-fetches
/// + surfaces a clean error itself.
async fn dispatch_rebuild_approval(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
    approval_id: i64,
) -> anyhow::Result<()> {
    let kind = coord
        .approvals
        .get(approval_id)
        .ok()
        .flatten()
        .map(|a| a.kind);
    if kind == Some(hive_sh4re::ApprovalKind::MergeConfigPr) {
        crate::actions::run_approval_merge_config_pr(coord, Some(entry.id), approval_id).await
    } else {
        crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await
    }
}

async fn dispatch(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
) -> anyhow::Result<()> {
    match (entry.kind, entry.approval_id) {
        (QueueKind::Rebuild, Some(approval_id)) => {
            dispatch_rebuild_approval(coord, entry, approval_id).await
        }
        (QueueKind::Rebuild, None) => {
            // A meta-update cascade has just set the meta lock; re-locking
            // in the per-agent rebuild would revert it (the agent's own
            // flake.lock wins). Every other source wants the relock so it
            // advances to applied/<n>/main.
            let relock = entry.source != QueueSource::MetaUpdate;
            rebuild_for_entry(coord, entry, relock).await
        }
        (QueueKind::MetaUpdate, Some(approval_id)) => {
            crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id)
                .await
        }
        (QueueKind::MetaUpdate, None) => run_meta_update(coord, entry).await,
        (QueueKind::Spawn, Some(approval_id)) => {
            crate::actions::run_approval_spawn(coord, Some(entry.id), approval_id).await
        }
        (QueueKind::Spawn, None) => {
            // Unreachable today: every Spawn entry is born from an
            // approval (HostRequest::RequestSpawn → submit_kind →
            // approve → enqueue with approval_id). The manager-side
            // `RequestSpawn` surface that used to bypass approvals
            // was removed; if a future direct-spawn admin path needs
            // to skip the approval ride it should wire its own action
            // call rather than route through here.
            anyhow::bail!(
                "rebuild_queue: Spawn entry id={} agent={} arrived without an approval_id — \
                 nothing should enqueue this shape today",
                entry.id,
                entry.agent,
            )
        }
        (QueueKind::Destroy, _) => {
            // Reserved for future `destroy --purge` integration.
            anyhow::bail!("Destroy kind not yet implemented in rebuild_queue worker");
        }
        (QueueKind::StartupSweep, _) => {
            // Bump meta's hyperhive input before per-agent child rebuilds
            // run so they build against the latest base. Non-fatal on
            // failure — child rebuilds proceed regardless. After the bump
            // (or failure) this entry transitions to Done and the worker
            // drains the pre-enqueued child Rebuild entries.
            coord.set_queue_step(Some(entry.id), "nix flake update hyperhive");
            if let Err(e) = crate::meta::lock_update_hyperhive().await {
                tracing::warn!(error = ?e, "startup_sweep: meta lock_update_hyperhive failed");
            }
            // `finish` clears the step label; no explicit clear needed here.
            Ok(())
        }
        (QueueKind::Restart, _) => {
            let name = &entry.agent;
            let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Restarting);
            coord.set_queue_step(Some(entry.id), "nixos-container restart");
            crate::lifecycle::restart(name).await?;
            coord.kick_agent(name, "container restarted");
            coord.rescan_containers_and_emit().await;
            Ok(())
        }
        (QueueKind::PermChange, _) => {
            let name = &entry.agent;
            // Write + commit the perm file under META_LOCK so the
            // working tree is never left dirty between the file write
            // and the subsequent prepare_deploy git operations.
            coord.set_queue_step(Some(entry.id), "writing + committing perm file");
            match &entry.perm_payload {
                Some(PermPayload::ToolGroups { groups }) => {
                    crate::meta::commit_tool_groups(name, groups)
                        .await
                        .with_context(|| format!("commit tool-groups for {name}"))?;
                    // Emit after the commit so the P3RM1SS10NS tab
                    // reflects the new assignment without the operator
                    // needing to navigate away and back.
                    coord.emit_tool_groups_snapshot();
                }
                Some(PermPayload::Capabilities { caps }) => {
                    crate::meta::commit_capabilities(name, caps)
                        .await
                        .with_context(|| format!("commit capabilities for {name}"))?;
                    coord.emit_capabilities_snapshot();
                }
                Some(PermPayload::Combined { groups, caps }) => {
                    // Batch perm change: commit whichever file(s) are
                    // present in a single git commit, then the rebuild
                    // below runs once — no double-rebuild for an agent
                    // whose caps AND groups both changed.
                    crate::meta::commit_perms(name, groups.as_deref(), caps.as_deref())
                        .await
                        .with_context(|| format!("commit perms for {name}"))?;
                    if groups.is_some() {
                        coord.emit_tool_groups_snapshot();
                    }
                    if caps.is_some() {
                        coord.emit_capabilities_snapshot();
                    }
                }
                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.
            rebuild_for_entry(coord, entry, true).await
        }
        (QueueKind::GracefulStop, _) => {
            run_graceful_stop(coord, entry);
            Ok(())
        }
        (QueueKind::Start, _) => run_start(coord, entry).await,
        (QueueKind::Stop, _) => run_stop(coord, entry).await,
    }
}

/// Queue-side container rebuild for `entry.agent`: resolves the current
/// flake rev and hands off to `rebuild_agent` with the entry's id +
/// source. Passing the source defers the start-after-rebuild to a
/// fast-lane `Start` follow-up (grouped under this entry via
/// `parent_id`), so the build lane is freed for the next entry instead
/// of waiting out the container boot.
async fn rebuild_for_entry(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
    relock: bool,
) -> anyhow::Result<()> {
    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,
        Some(entry.id),
        relock,
        Some(entry.source),
    )
    .await
}

/// Start a stopped container off the queue (`QueueKind::Start`), with a
/// `Starting` transient so the dashboard shows a visible queued→running
/// progression rather than the sub-second flash of a direct start.
/// Uses the cold-start fallback (stop + kill + start retry) so the
/// deferred start-after-rebuild keeps the same activation-error recovery
/// it had when it ran inline on the build lane.
///
/// If the hyperhive flake rev has changed since the container was last built
/// (i.e. the rev marker is stale or missing), the start is upgraded to a full
/// rebuild so the container runs current nix derivations. This is the
/// "deferred stopped container" path from `auto_update::run`.
async fn run_start(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
) -> anyhow::Result<()> {
    let name = &entry.agent;
    // Upgrade to rebuild+start if the rev marker is stale.
    let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
    if let Some(ref rev) = current_rev {
        let stored = std::fs::read_to_string(crate::auto_update::rev_marker_path(name)).ok();
        if stored.as_deref() != Some(rev.as_str()) {
            tracing::info!(%name, "start: rev stale — upgrading to rebuild+start");
            return crate::auto_update::rebuild_agent(
                coord,
                name,
                rev,
                Some(entry.id),
                true,
                Some(entry.source),
            )
            .await;
        }
    }
    let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Starting);
    coord.set_queue_step(Some(entry.id), "nixos-container start");
    crate::lifecycle::start_with_fallback(name).await?;
    coord.kick_agent(name, "container started");
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// Hard-stop a container off the queue (`QueueKind::Stop`) — same teardown as
/// a direct kill (unregister + `Killed` event), but with a `Stopping`
/// transient for visible queue progress. The quiescing variant is
/// `run_graceful_stop`.
async fn run_stop(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
) -> anyhow::Result<()> {
    let name = &entry.agent;
    let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Stopping);
    coord.set_queue_step(Some(entry.id), "nixos-container stop");
    crate::lifecycle::kill(name).await?;
    coord.unregister_agent(name);
    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
        agent: name.clone(),
    });
    coord.rescan_containers_and_emit().await;
    Ok(())
}

/// Run one `GracefulStop` entry: signal the harness to quiesce (it returns
/// `GracefulStop` on its next `Recv`, runs one stop-checkpoint turn to flush
/// durable `/state`, then exits), then hand the drain-wait + container stop to
/// a detached watcher and return — freeing the build lane immediately.
///
/// This is the concurrency split: the build worker only does the cheap signal,
/// so a whole-hive graceful stop signals every agent up front and their
/// checkpoint drains overlap. The watcher waits for this agent's drain
/// (bounded by `GRACEFUL_STOP_TIMEOUT` so a wedged agent can't block forever),
/// then enqueues a fast-lane `Stop` for the actual `nixos-container stop`.
/// Routing the real stop through the fast lane means the container stops
/// serialise there (one stop at a time) while the drains ran in parallel.
fn run_graceful_stop(coord: &std::sync::Arc<crate::coordinator::Coordinator>, entry: &QueueEntry) {
    let name = entry.agent.clone();
    let parent_id = entry.id;
    let source = entry.source;
    // Signal the harness; the kick breaks an idle long-poll so it's seen promptly.
    coord.set_queue_step(Some(entry.id), "graceful stop: signalling agent");
    coord.mark_graceful_stop(&name);
    coord.kick_agent(&name, "graceful stop requested");
    // Detached watcher: wait for the drain (or timeout), then enqueue the
    // container stop on the fast lane. The build entry itself is now Done —
    // the dashboard groups the follow-up `Stop` under it via `parent_id`.
    let coord = std::sync::Arc::clone(coord);
    tokio::spawn(async move {
        // Hold the `Stopping` transient across the drain so the dashboard keeps
        // showing the agent quiescing; dropped before the fast `Stop` is
        // enqueued (its `run_stop` re-establishes the transient) so the two
        // never clobber each other's clear-on-drop.
        let guard = coord.transient_guard(&name, crate::coordinator::TransientKind::Stopping);
        // Wait for the harness to drain (it clears the flag via
        // `GracefulStopComplete`) or fall back to a hard stop after the timeout.
        let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
        while coord.is_graceful_stop_pending(&name) {
            if std::time::Instant::now() >= deadline {
                tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping");
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }
        coord.clear_graceful_stop(&name);
        drop(guard);
        // Enqueue the actual container stop on the fast lane (same teardown as a
        // plain kill — `run_stop`). `parent_id` links it to the graceful entry
        // for dashboard grouping. `enqueue_full` nudges the fast worker itself.
        coord.rebuild_queue.enqueue_full(FullEnqueue {
            kind: QueueKind::Stop,
            agent: name.clone(),
            source,
            reason: format!("container stop after graceful drain of {name}"),
            parent_id: Some(parent_id),
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: Vec::new(),
        });
        coord.emit_rebuild_queue_snapshot();
    });
}

/// Run one `MetaUpdate` entry: bump the meta flake's locks for the
/// requested inputs, then enqueue a cascade of `Rebuild` entries
/// (with `parent_id` set to this entry's id) for every agent affected
/// by the bump. Mirrors the previous `dashboard::run_meta_update`
/// semantics; that path now enqueues into this queue rather than
/// running the bump + rebuild loop inline.
async fn run_meta_update(
    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
    entry: &QueueEntry,
) -> anyhow::Result<()> {
    let _progress = coord.meta_update_guard();
    let inputs = entry.inputs.clone();
    tracing::info!(
        ?inputs,
        parent = entry.id,
        "rebuild_queue: meta-update starting"
    );
    coord.set_queue_step(Some(entry.id), "nix flake update");
    let result = if inputs.is_empty() {
        crate::meta::lock_update(&[]).await
    } else {
        crate::meta::lock_update(&inputs).await
    };
    if let Err(e) = result {
        // Lock bump failed — cancel any pending cascade rebuilds the
        // enqueuer pre-queued for this MetaUpdate. Their parent_id
        // matches this entry; the children no longer make sense (we
        // never bumped the lock that justified them).
        let cancelled = coord.rebuild_queue.cancel_children(entry.id);
        if cancelled > 0 {
            tracing::warn!(
                cancelled,
                parent = entry.id,
                "rebuild_queue: meta-update failed; cancelled cascade rebuilds"
            );
            coord.emit_rebuild_queue_snapshot();
        }
        return Err(e);
    }
    // Lock file changed — meta-inputs panel re-renders. The cascade
    // rebuilds were already enqueued at MetaUpdate submission time,
    // so no further enqueue is needed here.
    crate::dashboard::emit_meta_inputs_snapshot(coord.as_ref());
    Ok(())
}

/// Compute which agents a `nix flake update <inputs>` on the meta
/// flake would affect. Used by callers that pre-enqueue cascade
/// `Rebuild` entries at `MetaUpdate` submission time so the dashboard
/// can render the dependent work alongside its parent before the lock
/// bump actually runs.
///
/// Mirrors `run_meta_update`'s post-bump fan-out logic. Empty `inputs`
/// or any input under `hyperhive` → every container; otherwise just
/// the agents named by `agent-<name>` inputs.
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec<String> {
    let touched_hyperhive = inputs
        .iter()
        .any(|i| i == "hyperhive" || i.starts_with("hyperhive/"));
    let touched_agents: Vec<String> = inputs
        .iter()
        .filter_map(|i| i.strip_prefix("agent-"))
        .map(|rest| rest.split('/').next().unwrap_or(rest).to_owned())
        .collect();
    let mut names = if touched_hyperhive || inputs.is_empty() {
        crate::lifecycle::list()
            .await
            .unwrap_or_default()
            .into_iter()
            .filter_map(|c| {
                c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
                    .map(str::to_owned)
            })
            .collect()
    } else {
        touched_agents
    };
    // Sort parents before children so the sequential queue worker
    // always rebuilds a parent before any of its dependents.
    let topo = crate::topology::read();
    crate::auto_update::topology_sort(&mut names, &topo);
    names
}

/// Current unix timestamp in seconds. `now()` calls are pulled into a
/// helper so tests can swap them out later.
fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .and_then(|d| i64::try_from(d.as_secs()).ok())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn enqueue_and_take_in_order() {
        let q = RebuildQueue::new();
        let a = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "first".to_owned(),
            None,
        );
        let b = q.enqueue(
            QueueKind::Rebuild,
            "agent-b".to_owned(),
            QueueSource::Manual,
            "second".to_owned(),
            None,
        );
        assert_ne!(a, b);
        let next = q.take_next_build().expect("queued");
        assert_eq!(next.id, a);
        assert_eq!(next.state, QueueState::Running);
        let next = q.take_next_build().expect("queued");
        assert_eq!(next.id, b);
        assert!(q.take_next_build().is_none());
    }

    #[test]
    fn dedup_pending_same_kind_and_agent() {
        let q = RebuildQueue::new();
        let a = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "first".to_owned(),
            None,
        );
        let b = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::AutoUpdate,
            "auto sweep".to_owned(),
            None,
        );
        assert_eq!(a, b, "dedup should return existing id");
        let snap = q.snapshot();
        assert_eq!(snap.len(), 1);
        assert!(snap[0].reason.contains("first"));
        assert!(snap[0].reason.contains("auto sweep"));
    }

    #[test]
    fn meta_update_dedup_matches_inputs() {
        // Two MetaUpdate enqueues with identical inputs → dedup.
        let q = RebuildQueue::new();
        let a = q.enqueue_with_inputs(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "first".to_owned(),
            None,
            vec!["nixpkgs".to_owned()],
        );
        let b = q.enqueue_with_inputs(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "duplicate click".to_owned(),
            None,
            vec!["nixpkgs".to_owned()],
        );
        assert_eq!(a, b, "identical-inputs meta-updates should dedup");
        assert_eq!(q.snapshot().len(), 1);
    }

    #[test]
    fn meta_update_dedup_separates_distinct_inputs() {
        // Two MetaUpdate enqueues with DIFFERENT inputs → distinct
        // entries, not deduped.
        let q = RebuildQueue::new();
        let a = q.enqueue_with_inputs(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "bump nixpkgs".to_owned(),
            None,
            vec!["nixpkgs".to_owned()],
        );
        let b = q.enqueue_with_inputs(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "bump bitburner-agent".to_owned(),
            None,
            vec!["agent-bitburner/bitburner-agent".to_owned()],
        );
        assert_ne!(a, b, "different-inputs meta-updates must NOT dedup");
        let snap = q.snapshot();
        assert_eq!(snap.len(), 2);
        // Both inputs lists are preserved.
        let inputs: Vec<&[String]> = snap.iter().map(|e| e.inputs.as_slice()).collect();
        assert!(inputs.iter().any(|i| *i == ["nixpkgs"]));
        assert!(
            inputs
                .iter()
                .any(|i| *i == ["agent-bitburner/bitburner-agent"])
        );
    }

    #[test]
    fn dedup_does_not_apply_across_kinds_or_agents() {
        let q = RebuildQueue::new();
        let a = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        let b = q.enqueue(
            QueueKind::Rebuild,
            "agent-b".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        let c = q.enqueue(
            QueueKind::Spawn,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "s".to_owned(),
            None,
        );
        assert_ne!(a, b);
        assert_ne!(a, c);
        assert_eq!(q.snapshot().len(), 3);
    }

    #[test]
    fn dedup_skips_running_entries() {
        let q = RebuildQueue::new();
        q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "first".to_owned(),
            None,
        );
        let running = q.take_next_build().expect("queued");
        assert_eq!(running.state, QueueState::Running);
        // While the original is running, re-enqueue is legitimate.
        let again = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "config bumped during build".to_owned(),
            None,
        );
        assert_ne!(running.id, again);
        let snap = q.snapshot();
        assert_eq!(snap.len(), 2);
    }

    #[test]
    fn finish_marks_state_and_keeps_history() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        q.take_next_build();
        q.finish(id, QueueState::Done, None);
        let snap = q.snapshot();
        assert_eq!(snap.len(), 1);
        assert_eq!(snap[0].state, QueueState::Done);
        assert!(snap[0].finished_at.is_some());
        assert!(snap[0].error.is_none());
    }

    #[test]
    fn finish_with_failure_records_error() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        q.take_next_build();
        q.finish(id, QueueState::Failed, Some("nix build failed".to_owned()));
        let snap = q.snapshot();
        assert_eq!(snap[0].state, QueueState::Failed);
        assert_eq!(snap[0].error.as_deref(), Some("nix build failed"));
    }

    #[test]
    fn history_evicts_old_terminals_per_kind() {
        let q = RebuildQueue::new();
        for i in 0..(MAX_HISTORY_PER_KIND + 3) {
            let id = q.enqueue(
                QueueKind::Rebuild,
                format!("agent-{i}"),
                QueueSource::Manual,
                "r".to_owned(),
                None,
            );
            q.take_next_build();
            q.finish(id, QueueState::Done, None);
        }
        let snap = q.snapshot();
        assert_eq!(snap.len(), MAX_HISTORY_PER_KIND);
    }

    #[test]
    fn cancel_clears_queued_entry() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        assert!(q.cancel(id));
        let snap = q.snapshot();
        assert_eq!(snap[0].state, QueueState::Cancelled);
        assert!(q.take_next_build().is_none());
    }

    #[test]
    fn cancel_refuses_running_entry() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "r".to_owned(),
            None,
        );
        q.take_next_build();
        assert!(!q.cancel(id));
        let snap = q.snapshot();
        assert_eq!(snap[0].state, QueueState::Running);
    }

    #[test]
    fn parent_id_groups_cascade() {
        let q = RebuildQueue::new();
        let meta = q.enqueue(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "lock bump".to_owned(),
            None,
        );
        let child = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        let snap = q.snapshot();
        let child_entry = snap.iter().find(|e| e.id == child).expect("child queued");
        assert_eq!(child_entry.parent_id, Some(meta));
    }

    #[test]
    fn cancel_children_marks_queued_descendants() {
        let q = RebuildQueue::new();
        let meta = q.enqueue(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "lock bump".to_owned(),
            None,
        );
        let a = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        let b = q.enqueue(
            QueueKind::Rebuild,
            "agent-b".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        // An unrelated queued entry must not be cancelled.
        let c = q.enqueue(
            QueueKind::Rebuild,
            "agent-c".to_owned(),
            QueueSource::Manual,
            "operator queued".to_owned(),
            None,
        );
        let cancelled = q.cancel_children(meta);
        assert_eq!(cancelled, 2);
        let snap = q.snapshot();
        let find = |id: u64| snap.iter().find(|e| e.id == id).expect("present");
        assert_eq!(find(a).state, QueueState::Cancelled);
        assert_eq!(find(b).state, QueueState::Cancelled);
        assert_eq!(find(c).state, QueueState::Queued);
    }

    #[test]
    fn approval_entries_keep_approval_id() {
        let q = RebuildQueue::new();
        let id = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "agent-a".to_owned(),
            source: QueueSource::Approval,
            reason: "approval 42 apply commit".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: Some(42),
            perm_payload: None,
            depends_on: Vec::new(),
        });
        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(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "agent-a".to_owned(),
            source: QueueSource::Approval,
            reason: "approval #1".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: Some(1),
            perm_payload: None,
            depends_on: Vec::new(),
        });
        let b = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "agent-a".to_owned(),
            source: QueueSource::Approval,
            reason: "approval #2".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: Some(2),
            perm_payload: None,
            depends_on: Vec::new(),
        });
        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(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "agent-a".to_owned(),
            source: QueueSource::Approval,
            reason: "approval #1 (duplicate)".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: Some(1),
            perm_payload: None,
            depends_on: Vec::new(),
        });
        assert_eq!(a, c);
        assert_eq!(q.snapshot().len(), 2);
    }

    #[test]
    fn cancel_children_skips_running_and_terminal() {
        let q = RebuildQueue::new();
        let meta = q.enqueue(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "lock bump".to_owned(),
            None,
        );
        // Running child — must NOT be cancelled.
        let running = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        q.take_next_build(); // pops meta, marks it Running
        q.take_next_build(); // pops `running`, marks it Running
        // Terminal child — must NOT be re-cancelled (its state stays Done).
        let done = q.enqueue(
            QueueKind::Rebuild,
            "agent-b".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        q.take_next_build();
        q.finish(done, QueueState::Done, None);
        // Queued child that should be cancelled.
        let queued = q.enqueue(
            QueueKind::Rebuild,
            "agent-c".to_owned(),
            QueueSource::MetaUpdate,
            "cascade".to_owned(),
            Some(meta),
        );
        let n = q.cancel_children(meta);
        assert_eq!(n, 1);
        let snap = q.snapshot();
        let find = |id: u64| snap.iter().find(|e| e.id == id).expect("present");
        assert_eq!(find(running).state, QueueState::Running);
        assert_eq!(find(done).state, QueueState::Done);
        assert_eq!(find(queued).state, QueueState::Cancelled);
    }

    #[test]
    fn set_step_updates_running_entry_and_signals_change() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            "test".to_owned(),
            None,
        );
        // Queued — set_step should refuse (returns false).
        assert!(!q.set_step(id, "plant tags"));
        // Promote to Running.
        let entry = q.take_next_build().expect("queued entry");
        assert_eq!(entry.id, id);
        // First label transition — true.
        assert!(q.set_step(id, "plant tags"));
        assert_eq!(
            q.snapshot()
                .iter()
                .find(|e| e.id == id)
                .and_then(|e| e.step.as_deref()),
            Some("plant tags")
        );
        // Same label again — false (caller can skip the snapshot emit).
        assert!(!q.set_step(id, "plant tags"));
        // Different label — true.
        assert!(q.set_step(id, "nixos-container update"));
        assert_eq!(
            q.snapshot()
                .iter()
                .find(|e| e.id == id)
                .and_then(|e| e.step.as_deref()),
            Some("nixos-container update")
        );
    }

    #[test]
    fn set_step_no_op_on_unknown_id() {
        let q = RebuildQueue::new();
        assert!(!q.set_step(999, "anything"));
    }

    /// A `MetaUpdate` cascade `Rebuild` (with `parent_id` = `Some(meta_id)`) must
    /// NOT dedup into a pre-existing `Queued` `Rebuild` with a different `parent_id`
    /// (e.g. from a startup sweep). Without the `parent_id` dedup guard the
    /// cascade rebuild would be swallowed and the agent would never rebuild
    /// against the post-lock-bump meta.
    #[test]
    fn meta_update_cascade_does_not_dedup_into_startup_sweep_rebuild() {
        let q = RebuildQueue::new();
        // Startup sweep enqueues a Rebuild for alice with its own parent_id.
        let sweep = q.enqueue(
            QueueKind::StartupSweep,
            "hyperhive".to_owned(),
            QueueSource::AutoUpdate,
            "boot sweep".to_owned(),
            None,
        );
        let sweep_rebuild = q.enqueue(
            QueueKind::Rebuild,
            "alice".to_owned(),
            QueueSource::StartupSweep,
            "startup sweep".to_owned(),
            Some(sweep),
        );
        // MetaUpdate cascade pre-enqueues another Rebuild for alice.
        let meta = q.enqueue(
            QueueKind::MetaUpdate,
            "hyperhive".to_owned(),
            QueueSource::Manual,
            "bump nixpkgs".to_owned(),
            None,
        );
        let cascade_rebuild = q.enqueue(
            QueueKind::Rebuild,
            "alice".to_owned(),
            QueueSource::MetaUpdate,
            "meta-update cascade".to_owned(),
            Some(meta),
        );
        // The two Rebuilds have different parent_ids — must NOT dedup.
        assert_ne!(
            sweep_rebuild, cascade_rebuild,
            "cascade rebuild must be distinct from startup-sweep rebuild"
        );
        let snap = q.snapshot();
        let rebuilds: Vec<_> = snap
            .iter()
            .filter(|e| e.kind == QueueKind::Rebuild && e.agent == "alice")
            .collect();
        assert_eq!(
            rebuilds.len(),
            2,
            "both rebuilds must be present in the queue"
        );
    }

    #[test]
    fn finish_clears_step() {
        let q = RebuildQueue::new();
        let id = q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            "test".to_owned(),
            None,
        );
        q.take_next_build();
        assert!(q.set_step(id, "running phase"));
        q.finish(id, QueueState::Done, None);
        assert_eq!(
            q.snapshot()
                .iter()
                .find(|e| e.id == id)
                .and_then(|e| e.step.as_deref()),
            None
        );
    }

    // --- depends_on tests ---

    /// An entry whose dep is not yet terminal must be skipped by
    /// `take_next`; it runs only after the dep finishes.
    #[test]
    fn depends_on_blocks_until_dep_is_terminal() {
        let q = RebuildQueue::new();
        let a = q.enqueue(
            QueueKind::Rebuild,
            "agent-a".to_owned(),
            QueueSource::Manual,
            "first".to_owned(),
            None,
        );
        let b = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "agent-b".to_owned(),
            source: QueueSource::Manual,
            reason: "second (blocked on a)".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![a],
        });
        // B depends on A — take_next should give A first.
        let first = q.take_next_build().expect("a is ready");
        assert_eq!(first.id, a);
        // A is Running, not terminal — B must still be blocked.
        assert!(
            q.take_next_build().is_none(),
            "b must be blocked while a runs"
        );
        // Finish A → B should now be available.
        q.finish(a, QueueState::Done, None);
        let second = q.take_next_build().expect("b unblocked after a done");
        assert_eq!(second.id, b);
    }

    /// An entry whose dep finished and was evicted from history is
    /// treated as resolved (eviction only happens to terminal entries).
    #[test]
    fn depends_on_evicted_dep_counts_as_resolved() {
        let q = RebuildQueue::new();
        // Fill the history cap for Rebuild so old terminals get evicted.
        for i in 0..MAX_HISTORY_PER_KIND {
            let id = q.enqueue(
                QueueKind::Rebuild,
                format!("filler-{i}"),
                QueueSource::Manual,
                "filler".to_owned(),
                None,
            );
            q.take_next_build();
            q.finish(id, QueueState::Done, None);
        }
        // `dep` gets enqueued, run, finished, and evicted by the
        // next history-trimming call.
        let dep = q.enqueue(
            QueueKind::Rebuild,
            "dep-agent".to_owned(),
            QueueSource::Manual,
            "dep".to_owned(),
            None,
        );
        q.take_next_build();
        q.finish(dep, QueueState::Done, None);
        // Push `dep` out of the per-kind history window: `trim_history`
        // keeps the newest MAX_HISTORY_PER_KIND terminals per kind, so it
        // takes that many newer terminals to evict `dep`.
        for i in 0..MAX_HISTORY_PER_KIND {
            let extra = q.enqueue(
                QueueKind::Rebuild,
                format!("extra-{i}"),
                QueueSource::Manual,
                format!("extra-{i}"),
                None,
            );
            q.take_next_build();
            q.finish(extra, QueueState::Done, None);
        }
        // `dep` should now be evicted.
        assert!(
            q.snapshot().iter().all(|e| e.id != dep),
            "dep must be evicted from history"
        );
        // An entry that depends on the (evicted) dep must be immediately runnable.
        let downstream = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "downstream".to_owned(),
            source: QueueSource::Manual,
            reason: "downstream (dep evicted = resolved)".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![dep],
        });
        let got = q
            .take_next_build()
            .expect("downstream runnable when dep evicted");
        assert_eq!(got.id, downstream);
    }

    /// Dedup respects `depends_on`: two otherwise-identical entries with
    /// different dep sets are distinct and must NOT collapse.
    #[test]
    fn depends_on_is_part_of_dedup_key() {
        let q = RebuildQueue::new();
        let dep1 = q.enqueue(
            QueueKind::Rebuild,
            "dep1".to_owned(),
            QueueSource::Manual,
            "d1".to_owned(),
            None,
        );
        let dep2 = q.enqueue(
            QueueKind::Rebuild,
            "dep2".to_owned(),
            QueueSource::Manual,
            "d2".to_owned(),
            None,
        );
        let a = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "target".to_owned(),
            source: QueueSource::Manual,
            reason: "r".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![dep1],
        });
        // Same kind+agent but different depends_on — must NOT dedup.
        let b = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "target".to_owned(),
            source: QueueSource::Manual,
            reason: "r".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![dep2],
        });
        assert_ne!(a, b, "different depends_on must produce distinct entries");
        // Same depends_on as a — must dedup.
        let c = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "target".to_owned(),
            source: QueueSource::Manual,
            reason: "r again".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![dep1],
        });
        assert_eq!(a, c, "identical depends_on must dedup");
    }

    /// An entry with Failed dep is still resolved — the dependent runs
    /// regardless of whether its upstream succeeded or not. Callers that
    /// need to abort on dep failure should cancel the downstream manually.
    #[test]
    fn depends_on_failed_dep_counts_as_resolved() {
        let q = RebuildQueue::new();
        let a = q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            "a".to_owned(),
            None,
        );
        let b = q.enqueue_full(FullEnqueue {
            kind: QueueKind::Rebuild,
            agent: "b".to_owned(),
            source: QueueSource::Manual,
            reason: "b (blocked on a)".to_owned(),
            parent_id: None,
            inputs: Vec::new(),
            approval_id: None,
            perm_payload: None,
            depends_on: vec![a],
        });
        q.take_next_build(); // pop a, mark Running
        q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned()));
        let got = q.take_next_build().expect("b runnable after a failed");
        assert_eq!(got.id, b);
    }

    // ---- fast lane (Start / Stop run on a separate concurrent worker) ----

    #[test]
    fn lanes_claim_only_their_own_kinds() {
        let q = RebuildQueue::new();
        let r = q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        let s = q.enqueue(
            QueueKind::Stop,
            "b".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        let build = q.take_next_build().expect("build entry");
        assert_eq!(build.id, r);
        let fast = q.take_next_fast().expect("fast entry");
        assert_eq!(fast.id, s);
        assert!(q.take_next_build().is_none());
        assert!(q.take_next_fast().is_none());
    }

    #[test]
    fn start_defers_behind_same_agent_queued_build() {
        let q = RebuildQueue::new();
        let b = q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.enqueue(
            QueueKind::Start,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        assert!(
            q.take_next_fast().is_none(),
            "start blocked while same agent has a queued build"
        );
        q.take_next_build().expect("build runs");
        q.finish(b, QueueState::Done, None);
        let started = q
            .take_next_fast()
            .expect("start unblocked after build done");
        assert_eq!(started.kind, QueueKind::Start);
    }

    #[test]
    fn start_for_other_agent_runs_concurrently_with_a_build() {
        let q = RebuildQueue::new();
        q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.enqueue(
            QueueKind::Start,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.enqueue(
            QueueKind::Start,
            "b".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.take_next_build().expect("a's build running");
        let got = q
            .take_next_fast()
            .expect("start for b runs while a's build runs");
        assert_eq!(got.agent, "b");
        assert!(
            q.take_next_fast().is_none(),
            "start for a still blocked by a's running build"
        );
    }

    #[test]
    fn stop_jumps_queued_build_but_waits_running_build() {
        // Stop jumps ahead of a *queued* build for the same agent.
        let q = RebuildQueue::new();
        q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.enqueue(
            QueueKind::Stop,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        let got = q
            .take_next_fast()
            .expect("stop jumps ahead of a's queued build");
        assert_eq!(got.kind, QueueKind::Stop);

        // But a stop waits for a *running* build of the same agent.
        let q2 = RebuildQueue::new();
        let b = q2.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q2.enqueue(
            QueueKind::Stop,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q2.take_next_build().expect("a's build running");
        assert!(
            q2.take_next_fast().is_none(),
            "stop waits for a's running build (no kill mid-rebuild)"
        );
        q2.finish(b, QueueState::Done, None);
        assert!(
            q2.take_next_fast().is_some(),
            "stop runs once a's build is done"
        );
    }

    #[test]
    fn build_defers_behind_same_agent_running_fast_op() {
        let q = RebuildQueue::new();
        q.enqueue(
            QueueKind::Stop,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        q.enqueue(
            QueueKind::Rebuild,
            "a".to_owned(),
            QueueSource::Manual,
            String::new(),
            None,
        );
        let s = q.take_next_fast().expect("stop running");
        assert!(
            q.take_next_build().is_none(),
            "build waits while a's fast op is running"
        );
        q.finish(s.id, QueueState::Done, None);
        assert!(
            q.take_next_build().is_some(),
            "build runs once the fast op is done"
        );
    }
}