hive-c0re: reconcile declared agents periodically, with backoff

hive-c0re/src/job_queue/mod.rs

@@ -224,6 +224,36 @@ impl JobQueue {
         Ok(Some(named))
     }
 
+    /// Queue a lone `Reconcile` for `agent` unless any queued or running node
+    /// already targets it.
+    ///
+    /// `None` means the agent is busy: an update, create, destroy, power op or
+    /// earlier reconcile is live, and the work in flight converges it (or
+    /// decides not to). The check and the insert share one lock, so a caller
+    /// racing another insert cannot queue behind a node it did not see. The
+    /// shape is the down-agent restart: a `Reconcile` on the agent's lease,
+    /// which reads the agent's power intent when it runs.
+    ///
+    /// # Errors
+    /// Propagates a graph-insert error.
+    pub fn insert_reconcile_if_idle(&self, agent: &str) -> anyhow::Result<Option<Vec<NodeId>>> {
+        let mut inner = self.lock();
+        let busy = inner
+            .graph()
+            .nodes()
+            .any(|n| !n.state.is_terminal() && n.payload.agent() == agent);
+        if busy {
+            return Ok(None);
+        }
+        let targets = [(agent.to_owned(), false)];
+        let named = inner
+            .insert_job(None, |b| power::restart_nodes(b, &targets, false))
+            .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
+        drop(inner);
+        self.notify.notify_one();
+        Ok(Some(named))
+    }
+
     /// The scheduler itself, for `hive_jobq`'s run-loop seam
     /// (`Scheduler::claim_next`), which takes exactly this type.
     ///

hive-c0re/src/job_queue/tests.rs

@@ -1556,6 +1556,62 @@ fn a_rebuild_is_live_for_its_own_agent_only() {
     assert!(!q.rebuild_live("bob"));
 }
 
+/// The periodic reconcile tick queues one lone `Reconcile` on the agent's lease
+/// and nothing for an agent any live node already targets — whichever kind
+/// of work it is — while other agents and finished work do not block it.
+#[test]
+fn an_idle_agent_gets_one_reconcile_and_a_busy_one_none() {
+    let q = JobQueue::new(1);
+    let first = q
+        .insert_reconcile_if_idle("alice")
+        .expect("insert")
+        .expect("control: an idle agent gets a reconcile");
+    assert_eq!(
+        declared_shape(&q)
+            .iter()
+            .filter(|d| d.kind == "reconcile")
+            .count(),
+        1,
+        "exactly one Reconcile node"
+    );
+    assert_eq!(
+        declared_resources_of_kind(&q, "reconcile"),
+        vec![vec![Resource::Agent("alice".to_owned())]],
+        "it holds the agent's lease"
+    );
+
+    assert!(
+        q.insert_reconcile_if_idle("alice")
+            .expect("insert")
+            .is_none(),
+        "the live reconcile blocks another"
+    );
+    assert!(
+        q.insert_reconcile_if_idle("bob").expect("insert").is_some(),
+        "control: another agent is independent"
+    );
+
+    // Work that is not a reconcile blocks too.
+    insert(&q, |b| {
+        let _ = rebuild(b, "carol");
+    });
+    assert!(
+        q.insert_reconcile_if_idle("carol")
+            .expect("insert")
+            .is_none(),
+        "a rebuild in flight blocks it"
+    );
+
+    // A finished reconcile does not.
+    assert!(q.cancel(first[0].get()), "cancel the queued reconcile");
+    assert!(
+        q.insert_reconcile_if_idle("alice")
+            .expect("insert")
+            .is_some(),
+        "a terminal node does not count as busy"
+    );
+}
+
 /// Agents whose running closure differs from the declaration are rebuilt in one
 /// sweep. Each rebuild's window holds a build slot, so `buildSlots` caps how
 /// many swap at once, however many agents the sweep rebuilds.

hive-c0re/src/main.rs

@@ -342,6 +342,9 @@ async fn cmd_serve(
     // previously-running container goes away without an operator-initiated
     // transient state — a swarm-level grafana alert matches that line.
     crash_watch::spawn(coord.clone());
+    // Periodic reconcile: queues a start for an agent that is declared up and
+    // found down. The crash alert above keeps firing; this only converges.
+    workers::reconcile_tick::spawn(coord.clone());
     // Agent-sockets marker poll: re-fires `agent_sockets::write`
     // and `gateway_nginx::write` every 10s so the JSON and nginx
     // config pick up newly-bound `.bound` markers after a rebuild.

hive-c0re/src/stats/otel_metrics.rs

@@ -27,6 +27,7 @@
 //! `agent` attribute**, and that absence is the point: it is what makes them
 //! selectable as `{hive!="", agent=""}`.
 
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::{Duration, Instant};
 
@@ -58,6 +59,14 @@ static PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
 /// doc says exporter-start rather than claiming a precision it does not have.
 static STARTED: OnceLock<Instant> = OnceLock::new();
 
+/// Restarts the periodic reconcile tick has queued since this process started.
+static RECONCILE_RESTARTS: AtomicU64 = AtomicU64::new(0);
+
+/// Count one restart queued by the periodic reconcile tick.
+pub fn count_reconcile_restart() {
+    RECONCILE_RESTARTS.fetch_add(1, Ordering::Relaxed);
+}
+
 /// Spawn the container-resource OTEL exporter if OTEL is configured
 /// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty). No-op otherwise. Call once at
 /// startup.
@@ -497,6 +506,17 @@ fn register_hive_instruments(meter: &opentelemetry::metrics::Meter) {
         })
         .build();
 
+    // Restarts the periodic reconcile tick queued for an agent that was wanted
+    // up and found down. Hive-scoped: the agent is in the log line, and a
+    // per-agent label would grow a series for every agent that ever failed.
+    meter
+        .u64_observable_counter("hyperhive.hive.reconcile_tick.restarts")
+        .with_unit("{restart}")
+        .with_callback(|obs| {
+            obs.observe(RECONCILE_RESTARTS.load(Ordering::Relaxed), HIVE_ATTRS);
+        })
+        .build();
+
     // Host-wide stall totals from `/proc/pressure`, by `resource` (`cpu`,
     // `memory`, `io`) and `kind` (`some`, `full`). Counters in microseconds,
     // so a rate over them is the stalled fraction of wall time. No sample on a

hive-c0re/src/workers/mod.rs

@@ -1,5 +1,5 @@
 //! Background tasks and periodic sweeps: crash/login watcher, boot-time
-//! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener
+//! auto-update reconcile, the periodic declared ⇒ running reconcile tick, the agent-sockets.json writer loop, the MCP socket listener
 //! reconcile loop, knowledge-repo sync, the swarm wanted-state pull, and the
 //! swarm resource-cap and permissions reads, and the squid users file.
 //! Each submodule is re-exported
@@ -11,6 +11,7 @@ pub mod crash_watch;
 pub mod knowledge;
 pub mod mcp_sockets;
 pub mod permissions_import;
+pub mod reconcile_tick;
 pub mod squid_allow;
 pub mod squid_deny;
 pub mod squid_users;

hive-c0re/src/workers/reconcile_tick.rs

@@ -0,0 +1,371 @@
+//! Periodic declared ⇒ running convergence: an agent whose power intent is
+//! `Up` and whose container is down gets a `Reconcile` queued, once a
+//! [`TICK_INTERVAL`], without anyone having to ask.
+//!
+//! The tick queues the same `NodeKind::Reconcile` the boot sweep, a watch
+//! attach and a declaration write end in, so what it starts, and how, is
+//! `run_reconcile`'s decision. It only ever converges *toward a start*: an
+//! agent wanted offline is left alone, and a running one is never touched.
+//! `crash_watch` keeps raising its alert; this worker does not replace it.
+//!
+//! # What the tick leaves alone
+//!
+//! - an agent with no known intent, or one wanted offline (the operator's stop
+//!   stands);
+//! - a paused agent, and one whose container state could not be read in time;
+//! - an agent any queued or running node already targets — an update, create,
+//!   destroy, power op or earlier reconcile (see
+//!   [`JobQueue::insert_reconcile_if_idle`]).
+//!
+//! # Backoff
+//!
+//! An agent that is still down on the tick after a restart was queued for it
+//! has failed to come up. Restarts of one agent are spaced `1, 2, 4, …` ticks
+//! apart, capped at [`MAX_BACKOFF`]. Seeing the agent running, or no longer
+//! wanted up, forgets its history.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::coordinator::Coordinator;
+use crate::lifecycle::{self, AGENT_PREFIX};
+use crate::power::Wanted;
+
+/// How often declared intent is compared with the observed containers. The
+/// first tick fires one interval after boot: the boot sweep owns startup.
+const TICK_INTERVAL: Duration = Duration::from_mins(1);
+
+/// The longest wait between two restarts of one agent.
+const MAX_BACKOFF: Duration = Duration::from_mins(15);
+
+/// What the tick can see of one agent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct Observation {
+    /// The power intent this process holds; `None` before anything set or
+    /// seeded one.
+    wanted: Option<Wanted>,
+    /// Whether the container's unit is active; `None` when `systemctl` did not
+    /// answer in time.
+    running: Option<bool>,
+    paused: bool,
+    /// An update has the agent down and not yet back up.
+    updating: bool,
+}
+
+/// Why the tick did nothing for an agent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Leave {
+    /// Running, or wanted offline, or no intent: nothing to converge.
+    Converged,
+    /// Paused, updating, or the container state is unreadable.
+    Held,
+    /// A restart was queued recently and the agent has not come up; waiting
+    /// out the backoff.
+    BackingOff,
+}
+
+/// What one tick does for one agent.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Step {
+    Leave(Leave),
+    /// Queue a `Reconcile` (unless the agent turns out to be busy).
+    Restart,
+}
+
+/// Per-agent restart history, in ticks.
+#[derive(Debug, Default)]
+struct Backoffs {
+    agents: HashMap<String, Backoff>,
+}
+
+#[derive(Debug, Default)]
+struct Backoff {
+    /// Restarts queued since the agent was last seen converged.
+    attempts: u32,
+    /// Ticks still to skip before the next restart.
+    skip: u32,
+}
+
+impl Backoffs {
+    /// Forget `agent`: it is running, or no longer wanted up.
+    fn settle(&mut self, agent: &str) {
+        self.agents.remove(agent);
+    }
+
+    /// Whether a restart may be queued for `agent` this tick. A refusal uses
+    /// up one skipped tick.
+    fn ready(&mut self, agent: &str) -> bool {
+        match self.agents.get_mut(agent) {
+            Some(b) if b.skip > 0 => {
+                b.skip -= 1;
+                false
+            }
+            _ => true,
+        }
+    }
+
+    /// Record that a restart was queued; returns how many this makes.
+    fn record_attempt(&mut self, agent: &str) -> u32 {
+        let b = self.agents.entry(agent.to_owned()).or_default();
+        b.attempts = b.attempts.saturating_add(1);
+        b.skip = ticks_between(b.attempts).saturating_sub(1);
+        b.attempts
+    }
+}
+
+/// Ticks between the `attempts`-th restart and the next: `2^(attempts-1)`,
+/// capped at [`MAX_BACKOFF`].
+fn ticks_between(attempts: u32) -> u32 {
+    let cap = u32::try_from(MAX_BACKOFF.as_secs() / TICK_INTERVAL.as_secs()).unwrap_or(u32::MAX);
+    1_u32
+        .checked_shl(attempts.saturating_sub(1))
+        .unwrap_or(u32::MAX)
+        .min(cap)
+}
+
+/// The decision for one agent. Pure but for the backoff counters it advances.
+fn step(backoffs: &mut Backoffs, agent: &str, seen: Observation) -> Step {
+    if seen.running == Some(true) || seen.wanted != Some(Wanted::Up) {
+        backoffs.settle(agent);
+        return Step::Leave(Leave::Converged);
+    }
+    if seen.running.is_none() || seen.paused || seen.updating {
+        return Step::Leave(Leave::Held);
+    }
+    if !backoffs.ready(agent) {
+        return Step::Leave(Leave::BackingOff);
+    }
+    Step::Restart
+}
+
+/// Run the tick until shutdown.
+pub fn spawn(coord: Arc<Coordinator>) {
+    let mut shutdown = coord.shutdown_rx();
+    tokio::spawn(async move {
+        let mut backoffs = Backoffs::default();
+        loop {
+            tokio::select! {
+                () = tokio::time::sleep(TICK_INTERVAL) => {}
+                _ = shutdown.changed() => {
+                    tracing::info!("reconcile tick: shutdown signal received");
+                    return;
+                }
+            }
+            tick(&coord, &mut backoffs).await;
+        }
+    });
+}
+
+/// One pass over every agent container.
+async fn tick(coord: &Arc<Coordinator>, backoffs: &mut Backoffs) {
+    let containers = match lifecycle::list().await {
+        Ok(c) => c,
+        Err(e) => {
+            tracing::warn!(error = ?e, "reconcile tick: container list unreadable — skipping this tick");
+            return;
+        }
+    };
+    for container in &containers {
+        let Some(name) = container.strip_prefix(AGENT_PREFIX) else {
+            continue;
+        };
+        let seen = Observation {
+            wanted: coord.power.get(name),
+            running: lifecycle::running_state(name).await,
+            paused: hive_types::Ident::parse(name).is_ok_and(|id| Coordinator::is_paused(&id)),
+            updating: coord.is_updating(name),
+        };
+        if step(backoffs, name, seen) != Step::Restart {
+            continue;
+        }
+        match coord.job_queue.insert_reconcile_if_idle(name) {
+            Ok(Some(_)) => {
+                let attempt = backoffs.record_attempt(name);
+                tracing::info!(
+                    agent = %name,
+                    observed = "stopped",
+                    wanted = "up",
+                    attempt,
+                    "reconcile tick: restarting agent"
+                );
+                crate::otel_metrics::count_reconcile_restart();
+                coord.emit_rebuild_queue_snapshot();
+            }
+            // Work targeting the agent is in flight; it converges the agent or
+            // decides not to.
+            Ok(None) => {
+                tracing::debug!(agent = %name, "reconcile tick: agent is busy; leaving it");
+            }
+            Err(e) => {
+                tracing::warn!(agent = %name, error = ?e, "reconcile tick: queueing the reconcile failed");
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A wanted-up agent whose container is down, nothing holding it back.
+    const DOWN: Observation = Observation {
+        wanted: Some(Wanted::Up),
+        running: Some(false),
+        paused: false,
+        updating: false,
+    };
+
+    #[test]
+    fn a_wanted_up_stopped_agent_is_restarted() {
+        assert_eq!(step(&mut Backoffs::default(), "a", DOWN), Step::Restart);
+    }
+
+    /// Each deviation from [`DOWN`] that must leave the agent alone, against
+    /// the control above that the same agent is restarted without it.
+    #[test]
+    fn the_tick_leaves_alone_what_is_not_a_plain_down_agent() {
+        let cases = [
+            (
+                "wanted offline",
+                Observation {
+                    wanted: Some(Wanted::Offline),
+                    ..DOWN
+                },
+                Leave::Converged,
+            ),
+            (
+                "no intent yet",
+                Observation {
+                    wanted: None,
+                    ..DOWN
+                },
+                Leave::Converged,
+            ),
+            (
+                "already running",
+                Observation {
+                    running: Some(true),
+                    ..DOWN
+                },
+                Leave::Converged,
+            ),
+            (
+                "paused",
+                Observation {
+                    paused: true,
+                    ..DOWN
+                },
+                Leave::Held,
+            ),
+            (
+                "mid-update",
+                Observation {
+                    updating: true,
+                    ..DOWN
+                },
+                Leave::Held,
+            ),
+            (
+                "container state unreadable",
+                Observation {
+                    running: None,
+                    ..DOWN
+                },
+                Leave::Held,
+            ),
+        ];
+        for (what, seen, why) in cases {
+            assert_eq!(
+                step(&mut Backoffs::default(), "a", seen),
+                Step::Leave(why),
+                "{what}"
+            );
+        }
+    }
+
+    /// Ticks between restarts of an agent that never comes up: 1, 2, 4, 8,
+    /// then the 15-tick cap, and it stays there.
+    #[test]
+    fn restarts_of_a_failing_agent_back_off_to_the_cap() {
+        let mut backoffs = Backoffs::default();
+        let mut gaps = Vec::new();
+        let mut since = 0_u32;
+        for _ in 0..80 {
+            since += 1;
+            if step(&mut backoffs, "a", DOWN) == Step::Restart {
+                backoffs.record_attempt("a");
+                gaps.push(since);
+                since = 0;
+            }
+        }
+        assert_eq!(gaps[..6], [1, 1, 2, 4, 8, 15]);
+        assert!(gaps[5..].iter().all(|g| *g == 15), "{gaps:?}");
+    }
+
+    #[test]
+    fn the_wait_is_a_backoff_not_a_refusal() {
+        let mut backoffs = Backoffs::default();
+        backoffs.record_attempt("a");
+        backoffs.record_attempt("a");
+        assert_eq!(
+            step(&mut backoffs, "a", DOWN),
+            Step::Leave(Leave::BackingOff)
+        );
+        assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart);
+        // Another agent's history is not this one's.
+        assert_eq!(step(&mut backoffs, "b", DOWN), Step::Restart);
+    }
+
+    #[test]
+    fn seeing_the_agent_running_resets_its_backoff() {
+        let mut backoffs = Backoffs::default();
+        for _ in 0..4 {
+            backoffs.record_attempt("a");
+        }
+        assert_eq!(
+            step(&mut backoffs, "a", DOWN),
+            Step::Leave(Leave::BackingOff),
+            "control: the history holds the agent back"
+        );
+        let up = Observation {
+            running: Some(true),
+            ..DOWN
+        };
+        assert_eq!(step(&mut backoffs, "a", up), Step::Leave(Leave::Converged));
+        assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart);
+        assert_eq!(backoffs.record_attempt("a"), 1, "the count starts over");
+    }
+
+    #[test]
+    fn an_operator_stop_resets_the_backoff() {
+        let mut backoffs = Backoffs::default();
+        for _ in 0..4 {
+            backoffs.record_attempt("a");
+        }
+        let stopped = Observation {
+            wanted: Some(Wanted::Offline),
+            ..DOWN
+        };
+        step(&mut backoffs, "a", stopped);
+        assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart);
+    }
+
+    /// A refused restart (the agent turned out busy) does not count as an
+    /// attempt, so it does not lengthen the next wait.
+    #[test]
+    fn only_a_recorded_attempt_lengthens_the_backoff() {
+        let mut backoffs = Backoffs::default();
+        assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart);
+        assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart);
+        assert_eq!(backoffs.record_attempt("a"), 1);
+    }
+
+    #[test]
+    fn the_cap_is_reached_without_overflow() {
+        assert_eq!(ticks_between(1), 1);
+        assert_eq!(ticks_between(4), 8);
+        assert_eq!(ticks_between(5), 15);
+        assert_eq!(ticks_between(u32::MAX), 15);
+    }
+}