fix(#1957): persist hivectl-start running-agents snapshot in the broker kv table

hive-c0re/src/broker.rs

@@ -36,6 +36,11 @@ CREATE TABLE IF NOT EXISTS reminders (
 );
 CREATE INDEX IF NOT EXISTS idx_reminders_due
     ON reminders (agent, due_at) WHERE sent_at IS NULL;
+
+CREATE TABLE IF NOT EXISTS kv (
+    key   TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);
 ";
 
 /// Capacity of the live event channel. Slow subscribers (e.g. an idle browser)
@@ -197,6 +202,52 @@ impl Broker {
         self.events.subscribe()
     }
 
+    /// Set a small persistent key/value pair (upsert). The `kv` table is a
+    /// general single-value store for state that must survive a hive-c0re
+    /// restart but isn't worth a dedicated table — currently just the
+    /// `hivectl start` running-agents snapshot (see
+    /// `Coordinator::set_last_stopped_running`).
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the sqlite upsert fails.
+    pub fn kv_set(&self, key: &str, value: &str) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "INSERT INTO kv (key, value) VALUES (?1, ?2)
+             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+            params![key, value],
+        )?;
+        Ok(())
+    }
+
+    /// Read a persistent key/value pair. `None` when the key is absent.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the sqlite query fails.
+    pub fn kv_get(&self, key: &str) -> Result<Option<String>> {
+        let conn = self.conn.lock().unwrap();
+        let value: Option<String> = conn
+            .query_row("SELECT value FROM kv WHERE key = ?1", params![key], |row| {
+                row.get(0)
+            })
+            .optional()?;
+        Ok(value)
+    }
+
+    /// Delete a persistent key/value pair. Idempotent — deleting an absent
+    /// key is a no-op.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the sqlite delete fails.
+    pub fn kv_delete(&self, key: &str) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute("DELETE FROM kv WHERE key = ?1", params![key])?;
+        Ok(())
+    }
+
     pub fn send(&self, message: &Message) -> Result<()> {
         let conn = self.conn.lock().unwrap();
         let now = now_unix();
@@ -1216,6 +1267,24 @@ mod tests {
         batch.pop()
     }
 
+    #[test]
+    fn kv_set_get_delete_roundtrip() {
+        let tb = open_broker();
+        let b = &tb.broker;
+        // Absent key -> None.
+        assert_eq!(b.kv_get("k").unwrap(), None);
+        // Set then get.
+        b.kv_set("k", "v1").unwrap();
+        assert_eq!(b.kv_get("k").unwrap(), Some("v1".to_owned()));
+        // Upsert overwrites.
+        b.kv_set("k", "v2").unwrap();
+        assert_eq!(b.kv_get("k").unwrap(), Some("v2".to_owned()));
+        // Delete clears; deleting again is a no-op.
+        b.kv_delete("k").unwrap();
+        assert_eq!(b.kv_get("k").unwrap(), None);
+        b.kv_delete("k").unwrap();
+    }
+
     /// Happy path: send → recv → `ack_turn` drains the in-memory list
     /// and marks the row `acked_at IS NOT NULL`. A second recv finds
     /// nothing pending (the row stays in the table for vacuum).

hive-c0re/src/coordinator.rs

@@ -22,6 +22,12 @@ use crate::socket_server::{self, AgentSocket};
 /// fine, the seq dedupe makes a reconnect resync safe.
 const DASHBOARD_CHANNEL: usize = 256;
 
+/// Broker `kv` key under which the broad-stop running-agents snapshot is
+/// persisted, so a `hivectl start` after a hive-c0re restart can still
+/// restore only the previously-running agents. See
+/// `Coordinator::set_last_stopped_running`.
+const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running";
+
 const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents";
 /// Manager-editable per-agent config repos. Bind-mounted RW into the manager
 /// container as `/agents/<name>/`. Hive-c0re only writes to these on first
@@ -137,8 +143,10 @@ pub struct Coordinator {
     /// every configured container, so agents the operator intentionally
     /// left stopped stay stopped. `None` when no broad stop has happened
     /// since the last start (or since daemon boot) — start then falls
-    /// back to "start all". In-daemon memory only (hive-c0re survives
+    /// back to "start all". Mirrored to the broker `kv` table
-    /// `hivectl stop`); host-reboot persistence is a separate follow-up.
+    /// (`last_stopped_running` key) so the snapshot also survives a
+    /// hive-c0re restart between the stop-all and the start: this
+    /// in-memory copy is the fast path, the persisted copy the backstop.
     last_stopped_running: Mutex<Option<Vec<String>>>,
     /// Unified wire-facing event channel feeding the dashboard SSE
     /// stream. Carries broker messages (mirrored from `broker.subscribe`
@@ -1185,16 +1193,44 @@ impl Coordinator {
 
     /// Record the set of agents that were running at a broad-scope
     /// `hivectl stop`, so the next broad-scope `start` restores exactly
-    /// this set. See the `last_stopped_running` field doc.
+    /// this set. See the `last_stopped_running` field doc. Persists a
+    /// copy to the broker `kv` table (best-effort) so the snapshot
+    /// survives a hive-c0re restart; the in-memory copy is the fast path.
     pub fn set_last_stopped_running(&self, agents: Vec<String>) {
+        match serde_json::to_string(&agents) {
+            Ok(json) => {
+                if let Err(e) = self.broker.kv_set(LAST_STOPPED_RUNNING_KEY, &json) {
+                    tracing::warn!(error = ?e, "persist last_stopped_running failed (in-memory copy still set)");
+                }
+            }
+            Err(e) => tracing::warn!(error = ?e, "serialise last_stopped_running failed"),
+        }
         *self.last_stopped_running.lock().unwrap() = Some(agents);
     }
 
     /// Take (and clear) the recorded broad-stop running set, if any. A
     /// broad-scope `start` uses this to restore only the previously
-    /// running agents; `None` means "no record — start all".
+    /// running agents; `None` means "no record — start all". Falls back
+    /// to the persisted `kv` copy when the in-memory snapshot is empty
+    /// (hive-c0re restarted between the stop-all and the start). Clears
+    /// the persisted copy either way — the snapshot is one-shot.
     pub fn take_last_stopped_running(&self) -> Option<Vec<String>> {
-        self.last_stopped_running.lock().unwrap().take()
+        let result = self
+            .last_stopped_running
+            .lock()
+            .unwrap()
+            .take()
+            .or_else(|| {
+                self.broker
+                    .kv_get(LAST_STOPPED_RUNNING_KEY)
+                    .ok()
+                    .flatten()
+                    .and_then(|json| serde_json::from_str::<Vec<String>>(&json).ok())
+            });
+        if let Err(e) = self.broker.kv_delete(LAST_STOPPED_RUNNING_KEY) {
+            tracing::warn!(error = ?e, "clear persisted last_stopped_running failed");
+        }
+        result
     }
 
     /// Set of agents whose transient was cleared within the last