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

refactor(hive-c0re): group src-root files into submodules

Browse source 
stores/ (sqlite-backed host stores + db helper), stats/, agent_config/,
workers/ — pure git-mv moves; crate-root re-exports keep every
crate::<module> path compiling. flake_check stays at root (synchronous
approval-flow validation, not a background worker)
This commit is contained in:
müde 2026-07-06 22:38:47 +02:00
commit 0e4b5a1120
29 changed files with 68 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

341
hive-c0re/src/workers/agent_sockets.rs Normal file
View file

@ -0,0 +1,341 @@
//! `/var/lib/hyperhive/run/agent-sockets.json` writer. Atomic
//! `<path>.tmp` + `rename()` write so the gateway's nginx worker never
//! reads a partial file. Includes manager and sub-agents so the gateway
//! can route `/agent/<name>/` for all containers with a bound unix
//! socket.
//!
//! Full mechanism — per-agent subdir bind-mount, `hyperhive-socket-bound`
//! marker gate, gateway UDS upstream, 10s poll loop:
//! `docs/gateway.md::Per-agent unix-socket upstream`.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
/// Host-side parent directory holding per-agent socket subdirs. The
/// gateway container bind-mounts this whole tree (read-only) so it
/// can `proxy_pass` to any agent. Each agent's container bind-mounts
/// only its own `<name>/` subdir — agents can only access their own
/// sockets.
pub const AGENT_SOCKET_DIR: &str = "/run/hive-agent";
/// Socket filename inside each per-agent subdir. Fixed so the path
/// derives entirely from `(AGENT_SOCKET_DIR, name)` — no second
/// degree of freedom for callers to get wrong.
pub const SOCKET_FILENAME: &str = "web.sock";
/// Marker file the harness drops next to the socket after a
/// successful `bind_unix`. Presence = "harness has bound the socket,
/// unix upstream is live"; absence = "harness hasn't started yet or
/// hasn't been rebuilt under the new config — keep TCP fallback".
/// Without this gate the gateway would `proxy_pass` to a non-existent
/// socket for an agent that's still starting up after a rebuild.
///
/// Renamed from `.bound` (legacy) to match the `hyperhive-` prefix
/// convention for all harness-written state files. `build_map`
/// checks both names during the transition window so existing containers
/// don't lose gateway routing before their next rebuild.
pub const READY_MARKER: &str = "hyperhive-socket-bound";
const READY_MARKER_LEGACY: &str = ".bound";
#[must_use]
pub fn host_sockets_path() -> PathBuf {
crate::paths::agent_sockets_file()
}
/// Per-agent socket subdir on the host. Lifecycle pre-creates this
/// before container start so the bind-mount source exists; the
/// harness binds the socket inside it as `web.sock`.
#[must_use]
pub fn agent_dir_for(name: &str) -> PathBuf {
Path::new(AGENT_SOCKET_DIR).join(name)
}
/// Compute the deterministic socket path for an agent. Pure function
/// of the agent name so the value matches whatever
/// [`agent_sockets::write`] writes for that agent, and whatever the
/// harness binds via `HIVE_WEB_SOCKET`.
#[must_use]
pub fn socket_path_for(name: &str) -> PathBuf {
agent_dir_for(name).join(SOCKET_FILENAME)
}
/// Compute the agent-socket map for the given logical agent names.
/// Includes manager and sub-agents. Filters by `READY_MARKER`
/// presence: only agents whose harness has actually bound the unix
/// socket appear in the map. Without this, the gateway would
/// `proxy_pass` to a non-existent socket for agents that haven't
/// been rebuilt yet or are mid-restart.
///
/// Accepts either the new `hyperhive-socket-bound` marker or the legacy
/// `.bound` marker so existing containers keep their gateway routing
/// through the transition window (before their next rebuild writes the
/// new marker name).
///
/// `BTreeMap` keeps the JSON output sorted by key so a re-emit
/// without churn produces byte-identical output for idempotent writes.
#[must_use]
pub fn build_map(names: &[String]) -> BTreeMap<String, PathBuf> {
build_map_with(names, |name| {
ready_marker_for(name).exists() || agent_dir_for(name).join(READY_MARKER_LEGACY).exists()
})
}
/// Body of `build_map` with the ready-check parameterised. Tests
/// pass a predicate they control (no real filesystem access).
/// Production callers go through `build_map` which wires the
/// predicate to the on-disk `hyperhive-socket-bound` (or legacy
/// `.bound`) marker check.
fn build_map_with<F>(names: &[String], is_ready: F) -> BTreeMap<String, PathBuf>
where
F: Fn(&str) -> bool,
{
names
.iter()
.filter(|n| is_ready(n))
.map(|n| (n.clone(), socket_path_for(n)))
.collect()
}
/// Path to the `hyperhive-socket-bound` marker file the harness writes
/// after a successful `bind_unix`. Lives next to `web.sock` in the
/// per-agent subdir so it's covered by the same bind-mount and same
/// per-agent isolation as the socket itself.
#[must_use]
pub fn ready_marker_for(name: &str) -> PathBuf {
agent_dir_for(name).join(READY_MARKER)
}
/// Render the map as pretty-printed JSON. Pretty so a human peek at
/// `cat /var/lib/hyperhive/run/agent-sockets.json` shows one row per agent
/// — keeps the file readable without a separate jq step.
fn render(map: &BTreeMap<String, PathBuf>) -> String {
// Serialize as strings (PathBuf → JSON string via the Display
// impl). BTreeMap → serde_json::to_string_pretty preserves key
// order, so the output is deterministic across calls with the
// same agent set.
let stringly: BTreeMap<&String, String> = map
.iter()
.map(|(k, v)| (k, v.display().to_string()))
.collect();
serde_json::to_string_pretty(&stringly)
.expect("BTreeMap<&String, String> is always serialisable")
}
/// Atomically write the JSON for `names` to
/// `/var/lib/hyperhive/run/agent-sockets.json`. Writes via a sibling
/// `<path>.tmp` + rename so a crashing process never leaves a
/// partial file behind that the gateway worker would fail to parse.
///
/// Idempotent — if the rendered content matches what's already on
/// disk, the write + rename are skipped so the file's mtime stays
/// stable and inotify watchers in the gateway (or any future
/// watchers) don't fire spurious reload events.
pub fn write(names: &[String]) -> Result<()> {
let map = build_map(names);
let body = render(&map);
let path = host_sockets_path();
if std::fs::read_to_string(&path).ok().as_deref() == Some(&body) {
return Ok(());
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
}
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path).with_context(|| {
format!(
"rename {} -> {} (atomic publish)",
tmp.display(),
path.display()
)
})?;
Ok(())
}
/// Spawn the marker poll task. Periodically re-runs `write` so the
/// JSON map picks up newly-bound sockets after a rebuild (harness
/// drops a fresh `.bound` marker on start) without needing an explicit
/// hook on container
/// start. `write` is idempotent (skips the rename when content
/// unchanged) so the steady-state cost is one directory stat per
/// agent per poll interval.
///
/// Also calls `gateway_nginx::reload_if_pending` on every tick to
/// retry a gateway nginx reload that may have failed on the previous
/// tick (e.g. gateway container temporarily down). This recovers
/// gateway routing without needing a manual gateway restart.
///
/// Mirrors the spawn-loop shape used by `crash_watch`,
/// `reminder_scheduler`, etc. — the existing background-task
/// convention in `main.rs`.
pub fn spawn_poll() {
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
// First tick fires immediately; that's fine — meta::sync_agents
// also writes on boot, this just catches up the window before
// the next agent restart.
loop {
interval.tick().await;
match crate::lifecycle::agents_for_meta_listing().await {
Ok(agents) => {
let names: Vec<String> = agents.into_iter().map(|a| a.name).collect();
if let Err(e) = write(&names) {
tracing::debug!(error = ?e, "agent_sockets poll write failed");
}
// Regenerate the gateway nginx include whenever
// socket readiness changes — the upstream
// selection (UDS vs TCP) depends on .bound markers
// which change independently of topology. Write is
// idempotent; skips rename when nothing changed.
if let Err(e) = crate::gateway_nginx::write(&names).await {
tracing::debug!(error = ?e, "gateway_nginx poll write failed");
}
// Retry a pending nginx reload that failed on a
// previous tick (no-op if no reload is pending).
crate::gateway_nginx::reload_if_pending().await;
}
Err(e) => {
tracing::debug!(error = ?e, "agent_sockets poll: failed to list agents");
}
}
}
});
}
#[cfg(test)]
mod tests {
use super::*;
use crate::lifecycle::MANAGER_NAME;
#[test]
fn socket_path_for_uses_subdir_layout() {
// Per-agent subdir + fixed socket filename — see module-level
// "Per-agent subdir layout" for why this isn't a flat
// `<name>.sock`. Pin both ends so a future move (e.g. to
// `/run/hyperhive/sockets/`) requires updating both the
// constant and the consumers.
let p = socket_path_for("iris");
assert_eq!(p, Path::new("/run/hive-agent/iris/web.sock"));
}
#[test]
fn agent_dir_for_is_socket_parent() {
// `agent_dir_for` is what lifecycle bind-mounts per agent;
// `socket_path_for` lives inside it. Keep them in lockstep so
// a divergence (e.g. typo in one constant) surfaces here
// rather than as a confusing nspawn bind-source-not-found at
// container start.
let dir = agent_dir_for("iris");
let sock = socket_path_for("iris");
assert_eq!(sock.parent(), Some(dir.as_path()));
}
#[test]
fn build_map_includes_manager() {
let names: Vec<String> = ["iris", MANAGER_NAME, "argus"]
.iter()
.map(|s| (*s).to_owned())
.collect();
let map = build_map_with(&names, |_| true);
assert!(map.contains_key(MANAGER_NAME));
assert!(map.contains_key("iris"));
assert!(map.contains_key("argus"));
}
#[test]
fn build_map_uses_socket_path_for() {
// Map values agree with the helper so callers can use either
// (build_map for the bulk write, socket_path_for for one-off
// lookups) without divergence.
let names = vec!["iris".to_owned()];
let map = build_map_with(&names, |_| true);
assert_eq!(map.get("iris"), Some(&socket_path_for("iris")));
}
#[test]
fn build_map_handles_empty_input() {
let map = build_map_with::<fn(&str) -> bool>(&[], |_| true);
assert!(map.is_empty());
}
#[test]
fn build_map_dedupes_via_btreemap_key_collision() {
// Duplicate inputs collapse via the map; no callsite passes
// dups today, but guarding the invariant here means a future
// bug doesn't surface as a corrupt JSON doc (two `"iris":`
// keys).
let names = vec!["iris".to_owned(), "iris".to_owned()];
let map = build_map_with(&names, |_| true);
assert_eq!(map.len(), 1);
}
#[test]
fn build_map_filters_by_ready_predicate() {
// Only ready agents (with `hyperhive-socket-bound` marker) get
// published. Pin the behaviour so a future refactor that drops
// the filter surfaces here, not as a 502-spew in the gateway.
let names: Vec<String> = ["iris", "argus", "atlas"]
.iter()
.map(|s| (*s).to_owned())
.collect();
// Pretend only `atlas` has flipped + bound — the gate makes
// sure only opted-in agents get a UDS upstream.
let map = build_map_with(&names, |name| name == "atlas");
assert!(map.contains_key("atlas"));
assert!(!map.contains_key("iris"));
assert!(!map.contains_key("argus"));
}
#[test]
fn ready_marker_path_is_sibling_of_socket() {
// Marker lives in the same per-agent subdir as the socket so
// the same bind-mount covers both; harness writes both inside
// the container, host (and gateway via shared bind-mount)
// sees both at the deterministic path.
let marker = ready_marker_for("iris");
let socket = socket_path_for("iris");
assert_eq!(marker.parent(), socket.parent());
assert_eq!(
marker,
Path::new("/run/hive-agent/iris/hyperhive-socket-bound")
);
}
#[test]
fn render_is_pretty_and_sorted() {
let mut map = BTreeMap::new();
map.insert(
"zeta".to_owned(),
PathBuf::from("/run/hive-agent/zeta/web.sock"),
);
map.insert(
"alpha".to_owned(),
PathBuf::from("/run/hive-agent/alpha/web.sock"),
);
let body = render(&map);
// Pretty-print = newlines between keys + indentation.
assert!(body.contains('\n'));
// BTreeMap sorts → alpha before zeta in output.
let alpha_pos = body.find("alpha").expect("alpha in output");
let zeta_pos = body.find("zeta").expect("zeta in output");
assert!(alpha_pos < zeta_pos, "sorted order broken:\n{body}");
}
#[test]
fn render_emits_paths_as_strings() {
// PathBuf-valued map serialises as plain JSON strings (not
// some {"inner": "..."} wrapper). Pin the shape so the
// gateway-side reader can deserialise into String values
// without nested struct logic.
let mut map = BTreeMap::new();
map.insert(
"iris".to_owned(),
PathBuf::from("/run/hive-agent/iris/web.sock"),
);
let body = render(&map);
assert!(body.contains("\"iris\""));
assert!(body.contains("\"/run/hive-agent/iris/web.sock\""));
}
}

320
hive-c0re/src/workers/auto_update.rs Normal file
View file

@ -0,0 +1,320 @@
//! Boot reconcile: on `hive-c0re serve` boot, (a) run the config path
//! for agents whose per-agent rev marker is stale — a `StartupSweep`
//! DAG (meta hyperhive lock bump) fanning out `Rebuild` children for
//! the stale agents whose `wanted` power intent is `Up` — and (b)
//! converge every other drifted agent to its persisted `wanted` via
//! `Reconcile` DAGs. Two rules keep boot-time nix work minimal:
//!
//! 1. **Stale but wanted-offline agents** get no rebuild — their
//! rebuild happens the first time they're started (the start
//! submit path upgrades a stale start to rebuild+start). The sweep
//! parent still runs whenever *any* marker is stale so the meta
//! hyperhive lock is bumped for those later start-upgrades.
//! 2. **Agents whose rev marker matches** the current hyperhive flake
//! path are skipped — nothing changed, no nix work to do.
//!
//! Booting with no config change performs no meta commit — only
//! reconciles. See `docs/coordinator.md::Boot reconcile`.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::Result;
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX, MANAGER_NAME};
/// Marker file recording the hyperhive rev a sub-agent's container was last
/// built against. Sibling of `applied/<name>/` (rather than inside it) to
/// keep it out of the applied repo's git history. Uses a leading dot so a
/// glob over `applied/*` doesn't include it.
pub fn rev_marker_path(name: &str) -> PathBuf {
PathBuf::from(format!("/var/lib/hyperhive/applied/.{name}.hyperhive-rev"))
}
/// Resolve the current rev of `hyperhive_flake`. For a path on disk we
/// canonicalize (following symlinks) so a /etc/hyperhive → /nix/store/...
/// update yields a different string. For anything else we return None.
#[must_use]
pub fn current_flake_rev(hyperhive_flake: &str) -> Option<String> {
let path = Path::new(hyperhive_flake);
if !path.exists() {
return None;
}
std::fs::canonicalize(path)
.ok()
.map(|p| p.display().to_string())
}
/// Returns true when the applied repo has commits that have not yet been
/// deployed (i.e. the applied HEAD differs from the sha currently locked in
/// meta's flake.lock). This is the semantic the dashboard `needs_update` chip
/// conveys: "there is a config change ready to apply via rebuild."
///
/// Async on purpose: this runs per agent inside `container_view::build_all`,
/// which fires on the ~10s dashboard sweep, every `AgentStatus` request, and
/// every `rescan_containers_and_emit` after a lifecycle step. A synchronous
/// `git` fork here blocks a tokio worker for the whole exec — under
/// nix-build disk saturation that's long enough that concurrent sweeps
/// starved the runtime and stalled the per-agent sockets.
pub async fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool {
let applied_head = tokio::process::Command::new("git")
.args([
"-C",
&format!("/var/lib/hyperhive/applied/{name}"),
"rev-parse",
"HEAD",
])
.output()
.await
.ok()
.filter(|o| o.status.success())
.and_then(|o| String::from_utf8(o.stdout).ok())
.map(|s| s.trim().to_owned());
match (applied_head.as_deref(), deployed_sha) {
(Some(head), Some(sha)) => !head.starts_with(sha) && !sha.starts_with(head),
_ => false,
}
}
/// Whether this hive is "ruthless" — running with no root/manager agent at
/// all (no ruth). When true, hive-c0re skips the root-agent create/start
/// sweep entirely. Controlled by the host option
/// `services.hyperhive.ruthless`, threaded in via the `HYPERHIVE_RUTHLESS`
/// env var. Defaults to `false` when the var is unset (back-compat: the
/// root agent was always auto-managed before this opt-out existed); only
/// an explicit `true` / `1` / `yes` enables ruthless mode.
fn ruthless() -> bool {
match std::env::var("HYPERHIVE_RUTHLESS") {
Ok(v) => matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes"),
Err(_) => false,
}
}
/// Auto-create the manager container on startup if it isn't already there.
/// hive-c0re manages the manager end-to-end: operators no longer declare
/// `containers.h-ruth` in their host NixOS config. Bypasses the approval
/// queue — the root/manager is auto-managed by default. Operators who
/// don't want a root agent at all set `services.hyperhive.ruthless = true`,
/// which short-circuits this whole function. Idempotent.
pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
if ruthless() {
tracing::info!(
"ruthless mode (services.hyperhive.ruthless = true) - skipping root agent create/start"
);
return Ok(());
}
let existing = lifecycle::list().await.unwrap_or_default();
let current_rev = current_flake_rev(&coord.hyperhive_flake);
if existing
.iter()
.any(|c| c.strip_prefix(AGENT_PREFIX) == Some(MANAGER_NAME))
{
// Container exists already. If it predates the unified lifecycle
// (no applied flake on disk) we must rebuild — otherwise it's
// running whatever the host-declarative config was at create
// time, with a wrong systemd unit and port.
let applied_flake = Coordinator::agent_applied_dir(MANAGER_NAME).join("flake.nix");
if !applied_flake.exists() && current_rev.is_some() {
tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate"
);
if let Err(e) = coord.job_queue.submit(crate::job_queue::templates::rebuild(
MANAGER_NAME,
crate::job_queue::Source::AutoUpdate,
"manager migration: no applied flake".to_owned(),
None,
true,
)) {
tracing::warn!(error = ?e, "manager migration rebuild submit failed");
}
} else {
tracing::debug!("manager container already present");
}
// hive-c0re auto-manages the root/manager by default, so a
// present-but-stopped root (e.g. a first-start failure on a fresh
// install) is brought back up here: the startup sweep's rebuild only
// restarts a container that was already running, so without this it
// stays down until a manual `nixos-container start`. The sub-agent
// `was_running` guard is intentionally left untouched. (Operators
// opt out of this whole auto-management with
// `services.hyperhive.ruthless = true`, gated at the top of
// this function.)
if !lifecycle::is_running(MANAGER_NAME).await {
tracing::info!("manager container present but not running — starting");
if let Err(e) = coord.power.set(MANAGER_NAME, crate::power::Wanted::Up) {
tracing::warn!(error = ?e, "agent_power: set manager wanted=up failed");
}
if let Err(e) = lifecycle::start(MANAGER_NAME).await {
tracing::warn!(error = ?e, "manager start failed");
}
}
return Ok(());
}
tracing::info!("manager container missing — spawning");
let runtime = coord.ensure_runtime(MANAGER_NAME)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(MANAGER_NAME, runtime);
lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?;
if let Err(e) = coord.power.set(MANAGER_NAME, crate::power::Wanted::Up) {
tracing::warn!(error = ?e, "agent_power: set manager wanted=up failed");
}
if let Some(rev) = current_rev {
let _ = std::fs::write(rev_marker_path(MANAGER_NAME), &rev);
}
Ok(())
}
/// Sort `names` in-place so parents precede their children in the topology.
/// Uses BFS from root agents (depth 0). Agents absent from `topo` sort last,
/// alphabetically within their tier. Stable within each depth tier.
pub fn topology_sort(
names: &mut [String],
topo: &std::collections::BTreeMap<String, Option<String>>,
) {
use std::collections::{HashMap, VecDeque};
// Build depth map using owned clones so the borrow on `names` is released
// before the sort_by mutable borrow.
let name_set: Vec<String> = names.to_vec();
let mut depth: HashMap<String, usize> = HashMap::new();
let mut queue: VecDeque<String> = VecDeque::new();
// Seed roots: entries with no parent, or names not present in topo at all.
for name in &name_set {
if topo.get(name).is_none_or(Option::is_none) {
depth.insert(name.clone(), 0);
queue.push_back(name.clone());
}
}
// BFS to assign depths to children.
while let Some(parent) = queue.pop_front() {
let d = depth[&parent] + 1;
for name in &name_set {
let is_child = topo.get(name).and_then(|p| p.as_deref()) == Some(parent.as_str());
if is_child && !depth.contains_key(name) {
depth.insert(name.clone(), d);
queue.push_back(name.clone());
}
}
}
names.sort_by(|a, b| {
let da = depth.get(a).copied().unwrap_or(usize::MAX);
let db = depth.get(b).copied().unwrap_or(usize::MAX);
da.cmp(&db).then(a.cmp(b))
});
}
/// Boot reconcile (see the module doc): classify every agent by rev
/// freshness + persisted `wanted` intent, submit one `StartupSweep`
/// DAG (hyperhive lock bump → fan-out rebuilds for stale wanted-up
/// agents) when anything is stale, and `Reconcile` DAGs for agents
/// whose observed power state drifted from `wanted`. Returns Ok even
/// if some submissions failed.
pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
let containers = match lifecycle::list().await {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = ?e, "boot reconcile: nixos-container list failed");
return Ok(());
}
};
let current_rev = current_flake_rev(&coord.hyperhive_flake);
// Resolve container names to logical agent names, then sort by
// topology depth so parents are always rebuilt before their
// children. Root agents (depth 0) go first; agents absent from
// the topology file sort last (stable, alphabetical within tier).
let mut logical_names: Vec<String> = containers
.iter()
.filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
.collect();
let topo = crate::topology::read();
topology_sort(&mut logical_names, &topo);
// Classify. `get_or_seed` doubles as the one-time migration: an
// agent without an `agent_power` row is seeded from its observed
// state (running ⇒ Up), after which the DB is authoritative.
let mut any_stale = false;
let mut fanout: Vec<String> = Vec::new(); // stale ∧ wanted=Up → sweep rebuild
let mut drifted: Vec<String> = Vec::new(); // fresh ∧ wanted≠observed → reconcile
let mut n_deferred = 0usize;
let mut n_skipped = 0usize;
for name in &logical_names {
let running = lifecycle::is_running(name).await;
let wanted = match coord.power.get_or_seed(name, running) {
Ok(w) => w,
Err(e) => {
tracing::warn!(%name, error = ?e, "agent_power read failed — assuming observed");
crate::power::Wanted::from_running(running)
}
};
let fresh = current_rev.as_ref().is_some_and(|rev| {
std::fs::read_to_string(rev_marker_path(name))
.is_ok_and(|stored| stored == rev.as_str())
});
if fresh {
n_skipped += 1;
} else {
any_stale = true;
if wanted == crate::power::Wanted::Up {
// Rebuild against the post-bump lock; the DAG's tail
// Reconcile brings the agent (back) up — covering both
// the running-stale and stopped-but-wanted-up cases.
fanout.push(name.clone());
continue;
}
// Stale but wanted offline: no boot-time nix work — the
// start submit path upgrades a stale start to a rebuild.
n_deferred += 1;
tracing::debug!(%name, "boot reconcile: stale but offline — deferring rebuild to on-start");
}
if crate::power::reconcile_action(wanted, running) != crate::power::ReconcileAction::Noop {
drifted.push(name.clone());
}
}
tracing::info!(
total = containers.len(),
rebuilds = fanout.len(),
reconciles = drifted.len(),
deferred = n_deferred,
up_to_date = n_skipped,
"boot reconcile"
);
// Sweep parent whenever ANY marker is stale — even when every
// stale agent is wanted-offline: the hyperhive lock bump must land
// now so their later start-upgrade rebuilds build against it.
// No stale agents ⇒ no sweep ⇒ no meta commit on a no-change boot.
if any_stale {
let reason = format!(
"startup sweep: {} rebuild(s), {} deferred (offline), {} up-to-date",
fanout.len(),
n_deferred,
n_skipped,
);
if let Err(e) = coord
.job_queue
.submit(crate::job_queue::templates::startup_sweep(reason, fanout))
{
tracing::warn!(error = ?e, "boot reconcile: sweep submit failed");
}
}
for name in drifted {
if let Err(e) = coord
.job_queue
.submit(crate::job_queue::templates::reconcile_only(
crate::job_queue::Template::Reconcile,
&name,
crate::job_queue::Source::AutoUpdate,
"boot reconcile".to_owned(),
None,
))
{
tracing::warn!(%name, error = ?e, "boot reconcile: submit failed");
}
}
coord.emit_rebuild_queue_snapshot();
Ok(())
}

210
hive-c0re/src/workers/crash_watch.rs Normal file
View file

@ -0,0 +1,210 @@
//! Per-container crash and login-state watcher. Polls every managed
//! container on a 10s interval. Fires `ContainerCrash`, `LoggedIn`,
//! and `NeedsLogin` helper events. Event semantics and the
//! `RECENT_TRANSIENT_GRACE` window: `docs/approvals.md::Helper events`.
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use crate::container_view::claude_has_session;
use crate::coordinator::{Coordinator, TransientKind};
use crate::lifecycle::{self, AGENT_PREFIX};
const POLL_INTERVAL: Duration = Duration::from_secs(10);
/// How long an operator-initiated transient stays "recently cleared"
/// for the purpose of suppressing crash events. Three full
/// `POLL_INTERVAL`s gives the post-lifecycle path comfortable
/// breathing room — the watcher will have polled at least twice
/// inside the window even with worst-case timer skew.
const RECENT_TRANSIENT_GRACE: Duration = Duration::from_secs(30);
pub fn spawn(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
let mut prev_running: HashSet<String> = HashSet::new();
let mut prev_logged_in: HashSet<String> = HashSet::new();
let mut prev_sub_agents: HashSet<String> = HashSet::new();
let mut seeded = false;
loop {
let raw = lifecycle::list().await.unwrap_or_default();
let mut current_running = HashSet::new();
let mut current_logged_in = HashSet::new();
let mut sub_agents: Vec<String> = Vec::new();
for c in &raw {
let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
continue;
};
let logical = logical.to_owned();
sub_agents.push(logical.clone());
if lifecycle::is_running(&logical).await {
current_running.insert(logical.clone());
}
if claude_has_session(&Coordinator::agent_claude_dir(&logical)) {
current_logged_in.insert(logical.clone());
}
}
if seeded {
emit_crash_transitions(&coord, &prev_running, &current_running);
emit_login_transitions(
&coord,
&prev_logged_in,
&current_logged_in,
&sub_agents,
&prev_sub_agents,
);
}
// Periodic container rescan — catches state flips that
// happen outside our mutation surface (operator runs
// `nixos-container stop` over ssh, agent logs in via its
// own web UI, etc.) so the dashboard converges within one
// POLL_INTERVAL. Idempotent + cheap when nothing changed.
coord.rescan_containers_and_emit().await;
prev_running = current_running;
prev_logged_in = current_logged_in;
prev_sub_agents = sub_agents.into_iter().collect();
seeded = true;
tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("crash watcher: shutdown signal received");
break;
}
}
}
});
}
fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current: &HashSet<String>) {
let transients = coord.transient_snapshot();
// Operator actions whose RAII guard already cleared but only just;
// suppresses the race where `lifecycle::kill` returns + drops the
// guard between two crash-watch polls.
let recent = coord.recent_transient_within(RECENT_TRANSIENT_GRACE);
for stopped in prev.difference(current) {
let active = transients.get(stopped).map(|st| st.kind);
let recently_cleared = recent.get(stopped).copied();
if is_deliberate_stop(active, recently_cleared) {
continue;
}
tracing::warn!(agent = %stopped, "container crash detected");
coord.record_crash(stopped);
coord.notify_manager(&hive_sh4re::HelperEvent::ContainerCrash {
agent: stopped.clone(),
note: Some("container stopped without an operator action".into()),
});
}
}
/// Pure classifier: did the operator stop / restart / destroy /
/// rebuild this container, or did it crash? Splits the matcher out
/// so it has a focused unit test without needing a Coordinator
/// fixture. `active` is the currently-set transient (if any),
/// `recently_cleared` is one whose RAII guard dropped within the
/// grace window.
fn is_deliberate_stop(
active: Option<TransientKind>,
recently_cleared: Option<TransientKind>,
) -> bool {
let is_op_kind = |kind: TransientKind| {
matches!(
kind,
TransientKind::Stopping
| TransientKind::Restarting
| TransientKind::Destroying
| TransientKind::Rebuilding
)
};
active.is_some_and(is_op_kind) || recently_cleared.is_some_and(is_op_kind)
}
fn emit_login_transitions(
coord: &Coordinator,
prev: &HashSet<String>,
current: &HashSet<String>,
sub_agents: &[String],
prev_sub_agents: &HashSet<String>,
) {
for agent in current.difference(prev) {
tracing::info!(%agent, "agent logged in");
coord.notify_manager(&hive_sh4re::HelperEvent::LoggedIn {
agent: agent.clone(),
});
}
// Detect transitions into "needs login": an agent that was previously
// logged-in goes unsigned (credentials deleted), OR a brand-new agent
// appears without a session.
//
// prev_needs uses prev_sub_agents (the agent set from the last tick) so
// that a newly-spawned agent — which does not appear in prev_sub_agents —
// is absent from prev_needs even though it's not in prev_logged_in.
// Without this, new agents land in both prev_needs and current_needs and
// the set difference is empty, silently dropping the event.
let prev_needs: HashSet<&str> = prev_sub_agents
.iter()
.map(String::as_str)
.filter(|n| !prev.contains(*n))
.collect();
let current_needs: HashSet<&str> = sub_agents
.iter()
.map(String::as_str)
.filter(|n| !current.contains(*n))
.collect();
for agent in current_needs.difference(&prev_needs) {
tracing::info!(%agent, "agent needs login");
coord.notify_manager(&hive_sh4re::HelperEvent::NeedsLogin {
agent: (*agent).to_owned(),
});
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn deliberate_when_active_transient_is_operator_kind() {
for kind in [
TransientKind::Stopping,
TransientKind::Restarting,
TransientKind::Destroying,
TransientKind::Rebuilding,
] {
assert!(is_deliberate_stop(Some(kind), None), "{kind:?}");
}
}
#[test]
fn deliberate_when_recent_transient_is_operator_kind() {
// Race repros: lifecycle action completes + drops the guard
// between two polls. recent_transient catches it.
for kind in [
TransientKind::Stopping,
TransientKind::Restarting,
TransientKind::Destroying,
TransientKind::Rebuilding,
] {
assert!(is_deliberate_stop(None, Some(kind)), "{kind:?}");
}
}
#[test]
fn not_deliberate_with_no_transient_at_all() {
// The real-crash case — fires the ContainerCrash event.
assert!(!is_deliberate_stop(None, None));
}
#[test]
fn not_deliberate_when_only_spawning_starting() {
// Spawning/Starting are never paired with a "stopped" transition
// — they're starts. If we see one alongside a stop, it's
// unrelated (e.g. just-started container died), still a crash.
for kind in [TransientKind::Spawning, TransientKind::Starting] {
assert!(!is_deliberate_stop(Some(kind), None), "{kind:?} active");
assert!(!is_deliberate_stop(None, Some(kind)), "{kind:?} recent");
}
}
}

234
hive-c0re/src/workers/knowledge.rs Normal file
View file

@ -0,0 +1,234 @@
//! Hive-wide knowledge repository management.
//!
//! `internal/knowledge` on the forge is cloned to
//! [`LOCAL_DIR`] and bind-mounted read-only into every agent container
//! at `/knowledge`. Agents read documents from it directly and
//! contribute by forking the repo and opening PRs — they never write
//! to the bind-mounted path inside the container.
//!
//! hive-c0re maintains the local clone. A Forgejo webhook notifies it
//! on push to main so agents always see an up-to-date snapshot. The
//! webhook is auto-created by [`ensure_webhook`] at startup. A
//! periodic pull in `main.rs` provides a fallback cadence.
use anyhow::{Context, Result};
pub const ORG: &str = "internal";
pub const REPO: &str = "knowledge";
/// Host-side path for the local clone. Also referenced from
/// `lifecycle.rs` (bind-mount source) and the dashboard webhook handler.
pub const LOCAL_DIR: &str = "/var/lib/hyperhive/knowledge";
/// In-container mount point for the knowledge repo. Bind-mounted
/// read-only from [`LOCAL_DIR`] into every agent container.
pub const CONTAINER_MOUNT: &str = "/knowledge";
/// Default README pushed to a freshly created `internal/knowledge` repo.
/// Short explanation + empty table-of-contents with an HTML comment instructing contributors
/// to add entries when they create new files.
const README_CONTENT: &str = "\
# knowledge
Hive-wide reference documents: conventions, runbooks, and anything that \
every agent should know.
## How to contribute
1. Fork this repo into your own namespace on the forge.
2. Create a branch, add or update a document.
3. Open a pull request — the operator reviews and merges.
4. Every agent container updates automatically on merge.
Do **not** push directly to `main` — agents have read-only access.
## Contents
<!-- Add an entry here each time you create a new document:
- [Title](path/to/file.md) - one-line description
-->
";
/// Clone `internal/knowledge` to [`LOCAL_DIR`] if it is not already a git
/// repository. `core_token` authenticates the HTTPS clone so private repos
/// work. Idempotent — skips if `LOCAL_DIR/.git` exists.
///
/// When the upstream repo is empty (freshly created), seeds it with a
/// README.md before returning so the local clone is always non-empty and
/// agents see a useful starting document.
///
/// Called once at hive-c0re startup after `forge::ensure_all`.
pub async fn ensure_local_clone(core_token: &str) -> Result<()> {
let git_dir = std::path::Path::new(LOCAL_DIR).join(".git");
if git_dir.exists() {
tracing::debug!("knowledge: local clone already present at {LOCAL_DIR}");
return Ok(());
}
std::fs::create_dir_all(LOCAL_DIR).context("create knowledge local dir")?;
// Embed credentials in the URL — safe for localhost-only forge.
let url = format!("http://core:{core_token}@localhost:3000/{ORG}/{REPO}.git");
let out = tokio::process::Command::new("git")
.args(["clone", &url, LOCAL_DIR])
.output()
.await
.context("git clone knowledge repo")?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git clone {ORG}/{REPO} failed: {stderr}");
}
tracing::info!("knowledge: cloned {ORG}/{REPO} to {LOCAL_DIR}");
// If the repo is brand new (no commits), seed it with a README.
let head_out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "rev-parse", "HEAD"])
.output()
.await
.context("git rev-parse HEAD")?;
if !head_out.status.success() {
seed_readme(core_token).await?;
}
Ok(())
}
/// Write the initial README.md, commit, and push to `internal/knowledge`.
/// Called only when the upstream repo is empty.
async fn seed_readme(core_token: &str) -> Result<()> {
let readme = std::path::Path::new(LOCAL_DIR).join("README.md");
std::fs::write(&readme, README_CONTENT).context("write README.md")?;
// Set a minimal git identity for the seed commit.
for (k, v) in [("user.email", "core@hive"), ("user.name", "hive-c0re")] {
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "config", k, v])
.output()
.await
.with_context(|| format!("git config {k}"))?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git config {k} failed: {stderr}");
}
}
for args in [
vec!["add", "README.md"],
vec!["commit", "-m", "init: seed README"],
] {
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR].iter().chain(args.iter()))
.output()
.await
.with_context(|| format!("git {args:?}"))?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git {args:?} failed: {stderr}");
}
}
let url = format!("http://core:{core_token}@localhost:3000/{ORG}/{REPO}.git");
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "push", &url, "HEAD:main"])
.output()
.await
.context("git push knowledge README")?;
if out.status.success() {
tracing::info!("knowledge: seeded README.md and pushed to {ORG}/{REPO}");
Ok(())
} else {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git push {ORG}/{REPO} failed: {stderr}")
}
}
/// Ensure a Forgejo push webhook for `internal/knowledge` exists and
/// points at hive-c0re's `/webhook/knowledge` endpoint. Idempotent —
/// lists existing hooks first and skips creation when one is already
/// targeting the correct URL. `dashboard_port` is the TCP port
/// hive-c0re's dashboard listens on (default 7000); the webhook URL
/// is `http://127.0.0.1:<port>/webhook/knowledge`.
///
/// Called at startup alongside [`ensure_local_clone`]. No-op when the
/// core token is absent (forge not yet provisioned).
pub async fn ensure_webhook(core_token: &str, dashboard_port: u16) -> Result<()> {
let target_url = format!("http://127.0.0.1:{dashboard_port}/webhook/knowledge");
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
.context("build reqwest client for webhook setup")?;
// List existing hooks — skip creation if ours is already there.
let list_url = format!(
"{}/api/v1/repos/{ORG}/{REPO}/hooks",
crate::forge::FORGE_HTTP
);
let resp = client
.get(&list_url)
.header("Authorization", format!("token {core_token}"))
.send()
.await
.with_context(|| format!("GET {list_url}"))?;
if resp.status().is_success() {
let hooks: Vec<serde_json::Value> = resp.json().await.unwrap_or_default();
let already_exists = hooks.iter().any(|h| {
h.get("config")
.and_then(|c| c.get("url"))
.and_then(|u| u.as_str())
== Some(&target_url)
});
if already_exists {
tracing::debug!(%target_url, "knowledge: push webhook already configured");
return Ok(());
}
}
// Create the webhook.
let create_url = format!(
"{}/api/v1/repos/{ORG}/{REPO}/hooks",
crate::forge::FORGE_HTTP
);
let body = serde_json::json!({
"type": "forgejo",
"config": {
"url": target_url,
"content_type": "json"
},
"events": ["push"],
"active": true
});
let resp = client
.post(&create_url)
.header("Authorization", format!("token {core_token}"))
.json(&body)
.send()
.await
.with_context(|| format!("POST {create_url}"))?;
let status = resp.status();
if status.is_success() {
tracing::info!(%target_url, "knowledge: push webhook created");
Ok(())
} else {
let body = resp.text().await.unwrap_or_default();
anyhow::bail!("create webhook for {ORG}/{REPO} failed ({status}): {body}")
}
}
/// Pull the latest changes in the local clone. Called from the webhook
/// handler on every push to `internal/knowledge` main, and periodically
/// from `main.rs` as a fallback. Uses `--ff-only` so a force-push to
/// the knowledge repo never wedges the local copy silently.
pub async fn pull() -> Result<()> {
// Sanity: if the clone is missing (e.g. storage was wiped), refuse
// to pull and let the caller decide whether to re-clone.
let git_dir = std::path::Path::new(LOCAL_DIR).join(".git");
if !git_dir.exists() {
anyhow::bail!("knowledge: {LOCAL_DIR}/.git not found — clone first");
}
let out = tokio::process::Command::new("git")
.args(["-C", LOCAL_DIR, "pull", "--ff-only"])
.output()
.await
.context("git pull knowledge")?;
if out.status.success() {
let stdout = String::from_utf8_lossy(&out.stdout).trim().to_owned();
tracing::info!(result = %stdout, "knowledge: pull succeeded");
Ok(())
} else {
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
anyhow::bail!("git pull {ORG}/{REPO} failed: {stderr}")
}
}

12
hive-c0re/src/workers/mod.rs Normal file
View file

@ -0,0 +1,12 @@
//! Background tasks and periodic sweeps: crash/login watcher, the
//! reminder and scheduled-prompt delivery loops, boot-time auto-update
//! reconcile, the agent-sockets.json writer loop, and knowledge-repo
//! sync. Each submodule is re-exported at the crate root, so
//! `crate::crash_watch::…` etc. keep working unchanged.
pub mod agent_sockets;
pub mod auto_update;
pub mod crash_watch;
pub mod knowledge;
pub mod reminder_scheduler;
pub mod scheduled_prompts_worker;

274
hive-c0re/src/workers/reminder_scheduler.rs Normal file
View file

@ -0,0 +1,274 @@
//! Background loop that drains due reminders from the broker and
//! delivers them as inbox messages. 5s poll cadence, shutdown-aware.
//! File-path semantics (path translation, traversal + symlink defense,
//! pointer delivery): `docs/approvals.md::Reminder delivery`.
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use crate::coordinator::Coordinator;
/// Per-tick cap on reminders delivered. Anything over this stays due
/// in the table and gets picked up on the next tick — keeps a
/// 10k-deep backlog from flooding the broker (or hogging the broker
/// mutex) in one shot. 100/tick × 5s tick = sustained throughput cap
/// of ~20 reminders/sec; bump together if the loose-ends tracker
/// starts firing higher rates.
const REMINDER_BATCH_LIMIT: u64 = 100;
/// Poll interval. Trade-off between latency on a freshly due reminder
/// and CPU spent on empty sweeps; 5s matches the original inline
/// scheduler.
const POLL_INTERVAL: Duration = Duration::from_secs(5);
pub fn spawn(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {
tick(&coord);
tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("reminder scheduler: shutdown signal received");
break;
}
}
}
});
}
fn tick(coord: &Arc<Coordinator>) {
let due = match coord.broker.get_due_reminders(REMINDER_BATCH_LIMIT) {
Ok(rows) => rows,
Err(e) => {
tracing::warn!(error = ?e, "failed to query due reminders");
return;
}
};
if due.is_empty() {
return;
}
// Resolve body strings (file-path writes / inline) before entering
// the batch transaction so the DB lock is held as briefly as possible.
let items: Vec<(i64, String, String)> = due
.iter()
.map(|(agent, id, message, file_path)| {
let body = prepare_body(agent, message, file_path.as_deref());
(*id, agent.clone(), body)
})
.collect();
// Single-transaction batch: one DB lock acquisition for N reminders
// instead of N sequential lock/unlock cycles.
let results = coord.broker.deliver_reminders_batch(&items);
let any_delivered = results.iter().any(Result::is_ok);
for ((id, agent, _body), result) in items.iter().zip(results.iter()) {
if let Err(e) = result {
let reason = format!("{e:#}");
tracing::warn!(
reminder_id = id,
%agent,
error = %reason,
"failed to deliver reminder"
);
// Persist the failure so the dashboard can surface it.
if let Err(persist_err) = coord.broker.record_reminder_failure(*id, &reason) {
tracing::warn!(
reminder_id = id,
error = ?persist_err,
"failed to persist reminder failure"
);
}
}
}
// Emit after the batch so the dashboard's pending-reminders list
// updates when deliveries land (removes delivered rows).
if any_delivered {
coord.emit_reminders_snapshot();
}
}
/// Build the inbox body for a due reminder. When `file_path` is None
/// the body is the original message verbatim. When set, we attempt to
/// persist the message body to the requested file and return a short
/// pointer string instead. Failures (bad prefix, symlink escape,
/// write error, missing parent) fall back to inline delivery with a
/// noted warning so the reminder still fires.
fn prepare_body(agent: &str, message: &str, file_path: Option<&str>) -> String {
let Some(req_path) = file_path else {
return message.to_owned();
};
let host_path = match resolve_host_path(agent, req_path) {
Ok(p) => p,
Err(reason) => {
tracing::warn!(%agent, %req_path, %reason, "reminder file_path rejected; delivering inline");
return inline_fallback(req_path, &format!("rejected: {reason}"), message);
}
};
match write_payload(agent, &host_path, message) {
Ok(()) => {
let bytes = message.len();
// debug! not info! — under load this would dominate the log.
tracing::debug!(%agent, path = %host_path.display(), bytes, "reminder body written to file");
format!(
"reminder body persisted to `{req_path}` ({bytes} bytes); read with your filesystem tools"
)
}
Err(reason) => {
tracing::warn!(%agent, path = %host_path.display(), %reason, "reminder file_path write failed; delivering inline");
inline_fallback(req_path, &reason, message)
}
}
}
fn inline_fallback(req_path: &str, reason: &str, message: &str) -> String {
format!("[reminder file_path '{req_path}' {reason}; delivering body inline]\n\n{message}")
}
/// Persist `message` to `host_path` with the symlink-escape defenses
/// described in the module docs. Returns `Ok(())` on success, or a
/// human-readable reason string on any failure (caller logs +
/// inline-falls-back). `pub` because `socket_server::handle_remind`
/// reuses it for the at-remind-time auto-file path.
pub fn write_payload(agent: &str, host_path: &Path, message: &str) -> Result<(), String> {
let Some(parent) = host_path.parent() else {
return Err("internal: host path has no parent".to_owned());
};
std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?;
// Resolve symlinks in the parent chain, then re-verify the
// canonical form still lives under the agent's host state root —
// catches `ln -s /etc state/escape` style attacks.
let parent_canonical = parent
.canonicalize()
.map_err(|e| format!("parent canonicalize failed: {e}"))?;
let agent_root = Coordinator::agent_notes_dir(agent)
.canonicalize()
.map_err(|e| format!("agent state root canonicalize failed: {e}"))?;
if !parent_canonical.starts_with(&agent_root) {
return Err(format!(
"symlink escape: canonical parent `{}` outside agent root `{}`",
parent_canonical.display(),
agent_root.display()
));
}
let basename = host_path
.file_name()
.ok_or_else(|| "missing basename".to_owned())?;
let target = parent_canonical.join(basename);
// O_NOFOLLOW on the final component refuses to open if the
// basename is itself an existing symlink. Combined with the
// canonicalize-parent check above, no symlink anywhere in the
// path can redirect the write.
let mut file = std::fs::OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.custom_flags(libc::O_NOFOLLOW)
.open(&target)
.map_err(|e| format!("open failed: {e}"))?;
file.write_all(message.as_bytes())
.map_err(|e| format!("write failed: {e}"))?;
Ok(())
}
/// Container-visible state prefix the caller's `file_path` must live
/// under. Every agent sees its state at `/agents/<name>/state/`
/// (see `lifecycle::set_nspawn_flags`). Auto-file paths use the same
/// prefix so the round-trip is symmetric.
#[must_use]
pub fn container_state_prefix(agent: &str) -> String {
format!("/agents/{agent}/state/")
}
/// Map an agent-visible container path to the matching host path,
/// validating that it lives under the agent's own state subtree, has
/// a non-empty relative tail, and doesn't try to traverse out via
/// `..`. Returns the host `PathBuf` on success, or a human-readable
/// reason string on rejection. `pub` so `socket_server::handle_remind`
/// can reuse it for the at-remind-time auto-file path.
pub fn resolve_host_path(agent: &str, req_path: &str) -> Result<PathBuf, String> {
let prefix = container_state_prefix(agent);
let Some(rel) = req_path.strip_prefix(&prefix) else {
return Err(format!(
"must be absolute and under `{prefix}` (got `{req_path}`)"
));
};
if rel.is_empty() {
return Err("file_path must include a filename, not just the state dir".to_owned());
}
let rel_path = Path::new(rel);
for comp in rel_path.components() {
match comp {
std::path::Component::Normal(_) => {}
other => {
return Err(format!(
"path component `{other:?}` not allowed (no traversal / absolute / root)"
));
}
}
}
Ok(Coordinator::agent_notes_dir(agent).join(rel_path))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rejects_paths_outside_agent_state() {
assert!(resolve_host_path("foo", "/etc/passwd").is_err());
assert!(resolve_host_path("foo", "/agents/bar/state/x.md").is_err());
assert!(resolve_host_path("foo", "relative.md").is_err());
}
#[test]
fn rejects_traversal() {
assert!(resolve_host_path("foo", "/agents/foo/state/../../etc/passwd").is_err());
assert!(resolve_host_path("foo", "/agents/foo/state/./x.md").is_err());
}
#[test]
fn rejects_empty_relative_tail() {
// Trailing slash → empty tail. Used to fall through to
// create_dir_all + write-to-dir → confusing inline fallback;
// explicit reject gives a cleaner log.
let err = resolve_host_path("foo", "/agents/foo/state/").unwrap_err();
assert!(err.contains("must include a filename"), "got: {err}");
}
#[test]
fn accepts_well_formed_path() {
let p = resolve_host_path("foo", "/agents/foo/state/reminders/123.md").unwrap();
assert_eq!(
p,
PathBuf::from("/var/lib/hyperhive/agents/foo/state/reminders/123.md")
);
}
#[test]
fn manager_uses_container_name_prefix() {
// Manager's container view of its state is at `/agents/ruth/state/`.
assert_eq!(container_state_prefix("ruth"), "/agents/ruth/state/");
let p = resolve_host_path("ruth", "/agents/ruth/state/reminders/x.md").unwrap();
assert_eq!(
p,
PathBuf::from("/var/lib/hyperhive/agents/ruth/state/reminders/x.md")
);
assert!(resolve_host_path("ruth", "/state/x.md").is_err());
}
#[test]
fn prepare_body_passthrough_when_no_file_path() {
let s = prepare_body("foo", "hello world", None);
assert_eq!(s, "hello world");
}
#[test]
fn prepare_body_falls_back_inline_on_bad_path() {
let s = prepare_body("foo", "payload", Some("/etc/passwd"));
assert!(s.starts_with("[reminder file_path '/etc/passwd' rejected:"));
assert!(s.contains("payload"));
}
}

407
hive-c0re/src/workers/scheduled_prompts_worker.rs Normal file
View file

@ -0,0 +1,407 @@
//! Background loop that drains due `scheduled_prompts` rows and fans
//! the body to each active target. 5s poll cadence, shutdown-aware.
//! Catch-up clamp, missing-target handling, and broker-error retry
//! semantics: `docs/approvals.md::Scheduled prompt worker`.
use std::sync::Arc;
use std::time::Duration;
use hive_sh4re::Message;
use crate::coordinator::Coordinator;
use crate::scheduled_prompts::Schedule;
use hive_sh4re::wire_time::now_unix;
/// Per-tick cap. Each schedule fires once per tick at most;
/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
/// matching `reminder_scheduler::REMINDER_BATCH_LIMIT`. Bump
/// together if real-world rates push past this.
const SCHEDULE_BATCH_LIMIT: u64 = 100;
/// Poll interval. Same 5s as the reminder scheduler — picking
/// up freshly-due rows within at most one tick keeps the
/// dashboard's "next fire in ..." countdown honest without
/// burning CPU on empty sweeps.
const POLL_INTERVAL: Duration = Duration::from_secs(5);
/// Reap cancelled schedules older than this from the table so
/// the dashboard list view doesn't accrue tombstones forever.
/// Cancelled rows live long enough that the operator can still
/// see what they cancelled in the recent past.
const CANCELLED_REAP_AGE: Duration = Duration::from_hours(1);
pub fn spawn(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {
tick(&coord);
tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("scheduled_prompts worker: shutdown signal received");
break;
}
}
}
});
}
fn tick(coord: &Arc<Coordinator>) {
let now = now_unix();
let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
Ok(rows) => rows,
Err(e) => {
tracing::warn!(error = ?e, "scheduled_prompts: query due rows failed");
return;
}
};
if due.is_empty() {
// Periodic reaper still gets a chance even on empty ticks.
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
}
return;
}
for schedule in due {
fire_schedule(coord, &schedule, now);
}
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
}
// Emit after all fires + reaps so the dashboard reflects updated
// last_fired_at_unix, next_fire_at_unix, and any reaped one-shots.
coord.emit_schedules_snapshot();
}
/// Fan out one schedule's body to every active target. Records
/// per-target `last_result`; advances or reaps the parent row at
/// the end depending on whether `interval_seconds` is set.
fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents(coord);
for target_row in &schedule.targets {
if target_row.cancelled_at_unix.is_some() {
continue;
}
let target = &target_row.target;
// `operator` is a valid recipient (mara c4) — operator
// delivery uses the regular broker path; the dashboard
// mirrors `to == operator` into its own pane.
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
let reason = format!("no such agent: {target}");
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule.id, target, now, &reason)
{
tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed");
}
notify_operator_missing_target(coord, schedule, target);
continue;
}
// Skip delivery if there is already an unread message from
// "scheduled" waiting in this target's inbox. Prevents the same
// scheduled prompt from stacking up when an agent is slow or
// briefly offline, while still allowing distinct scheduled
// messages (different body) to enqueue independently.
match coord
.broker
.has_pending_with_body(target, "scheduled", &schedule.body)
{
Ok(true) => {
tracing::debug!(
schedule = schedule.id,
%target,
"scheduled_prompts: skipping — same body already pending for target"
);
let _ = coord.scheduled_prompts.record_target_result(
schedule.id,
target,
now,
"skipped: already pending",
);
continue;
}
Err(e) => {
tracing::warn!(schedule = schedule.id, %target, error = ?e, "has_pending_with_body failed");
}
Ok(false) => {}
}
let msg = Message {
from: "scheduled".to_owned(),
to: target.clone(),
body: schedule.body.clone(),
in_reply_to: None,
};
let result = coord.broker.send(&msg);
let result_str = match &result {
Ok(()) => "ok".to_owned(),
Err(e) => format!("broker send failed: {e:#}"),
};
if let Err(e) = result {
tracing::warn!(
schedule = schedule.id,
%target,
error = ?e,
"scheduled_prompts: broker send failed (will retry on next interval)"
);
}
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule.id, target, now, &result_str)
{
tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed");
}
}
// Advance or reap. One-shots delete; recurring re-arm with
// catch-up clamp.
if schedule.interval_seconds.is_some() {
match coord.scheduled_prompts.rearm(schedule.id, now) {
Ok(0) => {}
Ok(skipped) => {
tracing::info!(
schedule = schedule.id,
skipped,
"scheduled_prompts: caught up missed cycles"
);
}
Err(e) => {
tracing::warn!(error = ?e, schedule = schedule.id, "rearm failed");
}
}
} else if let Err(e) = coord.scheduled_prompts.delete(schedule.id) {
tracing::warn!(error = ?e, schedule = schedule.id, "delete one-shot failed");
}
}
/// Snapshot of live container names for the missing-target check.
/// Always seeds the manager name (which is always reachable);
/// adds every live nspawn container that matches the `h-` prefix.
/// On `lifecycle::list` failure the set stays at just the manager
/// — fail-CLOSED, meaning every non-operator/non-manager target
/// looks missing this tick and gets the same treatment as a
/// genuinely-destroyed agent: operator advisory + per-target
/// `last_result` annotation + skipped delivery. Recurring
/// schedules recover automatically on the next tick (the lifecycle
/// listing usually works); one-shots that land on this window
/// lose their single delivery. Logged at `warn`, not propagated.
fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> {
// `lifecycle::list` is async; the worker tick is sync. Use the
// blocking variant via a small `tokio::runtime::Handle::block_on`
// wrapper. The worker runs in its own tokio task so this is
// safe (we're not in a `current_thread` runtime).
use std::collections::HashSet;
let mut out: HashSet<String> = HashSet::new();
// Manager is always a scheduled-prompt target (fail-safe: include
// it even if `list()` fails so prompts to the manager never silently drop).
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
let containers = tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(crate::lifecycle::list())
});
match containers {
Ok(list) => {
for raw in list {
if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
out.insert(name.to_owned());
}
}
}
Err(e) => {
tracing::warn!(error = ?e, "scheduled_prompts: container listing failed");
}
}
out
}
/// Send the operator a one-line advisory when a schedule fires
/// against an agent that no longer exists. Best-effort — failure
/// to send just gets logged; the schedule continues firing.
fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, target: &str) {
let body = format!(
"scheduled prompt #{id} fired but target `{target}` is not a live agent. \
body was:\n\n{body}",
id = schedule.id,
target = target,
body = schedule.body
);
let msg = Message {
from: "scheduled".to_owned(),
to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
body,
in_reply_to: None,
};
if let Err(e) = coord.broker.send(&msg) {
tracing::warn!(error = ?e, schedule = schedule.id, %target, "operator advisory send failed");
}
}
/// Per-target outcome counts for one `fire_now` invocation.
/// Returned to the operator so the dashboard can render
/// "fired to N (M failed, K missing)" without a follow-up GET.
#[derive(Debug, Clone, serde::Serialize)]
pub struct FireNowReport {
/// Targets the broker accepted the message for.
pub ok: u32,
/// Targets where broker.send returned an error.
pub failed: u32,
/// Targets that didn't resolve to a known agent (and got the
/// operator-advisory treatment).
pub missing: u32,
/// Whether the one-shot was consumed by this manual fire.
/// `true` only when the schedule was a one-shot (recurring
/// schedules never auto-cancel on manual fire — they keep
/// their cadence).
pub one_shot_consumed: bool,
/// Whether this manual fire re-armed a recurring schedule's timer
/// (`next_fire_at = now + interval`). `true` only when the caller
/// passed `reset_timer` AND the schedule is recurring.
pub timer_reset: bool,
}
/// Manual / out-of-band fire of a scheduled prompt ("fire now"
/// dashboard button). Mirrors the per-target fan-out of `fire_schedule`
/// but skips the rearm step entirely — manual fires don't disturb
/// a recurring schedule's rhythm. For one-shots, a manual fire
/// **consumes** the schedule (operator intent: "send this now,
/// the scheduled time was wrong"); recurring schedules keep their
/// `next_fire_at_unix` unchanged.
///
/// `last_result` is annotated with the `manual fire:` prefix so
/// the dashboard's per-target last-result column can distinguish
/// scheduled fires from operator-initiated ones at a glance.
///
/// `reset_timer` re-arms a *recurring* schedule's countdown from now
/// (`next_fire_at = now + interval`) after the fan-out — the dashboard
/// fire-now dialog's "reset timer" checkbox. It's a no-op for one-shots
/// (still consumed) and when `false` (today's default: cadence intact).
///
/// Returns Err if the schedule is missing, cancelled, or fully
/// drained of active targets — the dashboard can surface those
/// as plain 4xxs instead of pretending to fire a phantom row.
pub async fn fire_now(
coord: &std::sync::Arc<Coordinator>,
schedule_id: i64,
reset_timer: bool,
) -> anyhow::Result<FireNowReport> {
let now = now_unix();
let schedule = coord
.scheduled_prompts
.get(schedule_id)?
.ok_or_else(|| anyhow::anyhow!("schedule {schedule_id} not found"))?;
if schedule.cancelled_at_unix.is_some() {
anyhow::bail!("schedule {schedule_id} is already cancelled");
}
if !schedule
.targets
.iter()
.any(|t| t.cancelled_at_unix.is_none())
{
anyhow::bail!("schedule {schedule_id} has no active targets");
}
let known = known_agents_async().await;
let mut report = FireNowReport {
ok: 0,
failed: 0,
missing: 0,
one_shot_consumed: false,
timer_reset: false,
};
for target_row in &schedule.targets {
if target_row.cancelled_at_unix.is_some() {
continue;
}
let target = &target_row.target;
if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
let reason = format!("manual fire: no such agent: {target}");
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule_id, target, now, &reason)
{
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
}
notify_operator_missing_target(coord, &schedule, target);
report.missing += 1;
continue;
}
let msg = Message {
from: "scheduled".to_owned(),
to: target.clone(),
body: schedule.body.clone(),
in_reply_to: None,
};
let result = coord.broker.send(&msg);
let result_str = match &result {
Ok(()) => "manual fire: ok".to_owned(),
Err(e) => format!("manual fire: broker send failed: {e:#}"),
};
if result.is_ok() {
report.ok += 1;
} else {
report.failed += 1;
tracing::warn!(
schedule = schedule_id,
%target,
error = ?result.as_ref().err(),
"fire_now: broker send failed (no retry — manual fires don't loop)"
);
}
if let Err(e) =
coord
.scheduled_prompts
.record_target_result(schedule_id, target, now, &result_str)
{
tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
}
}
match schedule.interval_seconds {
None => {
// One-shot is consumed by the manual fire. (reset_timer is
// moot here — there's no recurring cadence to re-arm.)
if let Err(e) = coord.scheduled_prompts.cancel_all(schedule_id) {
tracing::warn!(error = ?e, schedule = schedule_id, "cancel_all after one-shot manual fire failed");
} else {
report.one_shot_consumed = true;
}
}
Some(interval) if reset_timer => {
// Recurring + operator asked to reset: re-arm the countdown
// from now (now + interval), not along the existing cadence.
let next = now.saturating_add(i64::try_from(interval).unwrap_or(i64::MAX));
if let Err(e) = coord.scheduled_prompts.set_next_fire(schedule_id, next) {
tracing::warn!(error = ?e, schedule = schedule_id, "set_next_fire after manual fire reset failed");
} else {
report.timer_reset = true;
}
}
// Recurring without reset: cadence stays intact (additive fire).
Some(_) => {}
}
Ok(report)
}
/// Async variant of `known_agents` for `fire_now`. Same logic +
/// same fail-closed degradation; the difference is just that the
/// dashboard handler is genuinely async so we `await` the
/// `lifecycle::list` directly instead of going through the
/// `block_in_place` shim.
async fn known_agents_async() -> std::collections::HashSet<String> {
use std::collections::HashSet;
let mut out: HashSet<String> = HashSet::new();
out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
match crate::lifecycle::list().await {
Ok(list) => {
for raw in list {
if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
out.insert(name.to_owned());
}
}
}
Err(e) => {
tracing::warn!(error = ?e, "fire_now: container listing failed");
}
}
out
}