ba71e45486
The push side modelled a store per peer hive: a --peer argument, a swarm.peers.<domain>.snapshotStorePort option, and a swarm_peers module whose entire job was answering "which peer". A swarm has exactly one store, so none of that had anything to select between. The receiver already proved it. It keys destination directories by agent, not by sending hive, precisely so an agent that migrates keeps one unbroken incremental chain -- which only makes sense if every hive pushes to the same place. Per-hive stores would split the chain in two, the case that keying exists to prevent. So the destination moves to services.hyperhive.swarm.snapshotStore, rendered into HYPERHIVE_SNAPSHOT_STORE, and swarm_peers is deleted rather than adapted. address has no default because it is a deployment fact this host cannot derive; port defaults because it is a convention both ends read from the same option docs. An unset or empty address fails naming the option instead of connecting somewhere arbitrary, and a test asserts the message suggests no value.
661 lines
29 KiB
Rust
661 lines
29 KiB
Rust
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context as _, Result};
|
|
use clap::{Parser, Subcommand};
|
|
|
|
// `hive-c0re` is bin-only: this binary owns the whole daemon module
|
|
// tree. The operator CLI moved to the standalone `hivectl` crate (it
|
|
// talks to the daemon over the host admin socket), so there's no longer
|
|
// a library shared between two binaries — the modules that used to live
|
|
// in `src/lib.rs` are declared here directly.
|
|
//
|
|
// Cohesive clusters live in directory submodules (`stores`, `stats`,
|
|
// `agent_config`, `workers`); each child is re-exported at the crate
|
|
// root so `crate::broker::…` style paths keep resolving unchanged.
|
|
mod actions;
|
|
mod agent_config;
|
|
mod container_view;
|
|
mod coordinator;
|
|
mod dashboard;
|
|
mod dashboard_events;
|
|
mod forge;
|
|
mod gateway_nginx;
|
|
mod job_queue;
|
|
mod lifecycle;
|
|
mod loose_ends;
|
|
mod matrix;
|
|
mod meta;
|
|
mod migrate;
|
|
mod paths;
|
|
mod priv_client;
|
|
mod questions;
|
|
mod server;
|
|
mod snapshot_push;
|
|
mod socket_server;
|
|
mod stats;
|
|
mod stores;
|
|
mod webhook_secret;
|
|
mod workers;
|
|
|
|
pub(crate) use agent_config::{capabilities, limits, resource_limits, tool_groups, topology};
|
|
pub(crate) use stats::{
|
|
container_stats, hive_stats, host_stats, otel_metrics, sweep_health, warnings,
|
|
};
|
|
pub(crate) use stores::{
|
|
approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
|
|
};
|
|
pub(crate) use workers::{
|
|
agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker,
|
|
};
|
|
|
|
use coordinator::{Coordinator, HiveEnv, ServeConfig};
|
|
|
|
#[derive(Parser)]
|
|
#[command(name = "hive-c0re", about = "hyperhive coordinator daemon")]
|
|
struct Cli {
|
|
/// Path to the host admin socket.
|
|
#[arg(long, global = true, default_value = crate::paths::HOST_SOCKET)]
|
|
socket: PathBuf,
|
|
|
|
#[command(subcommand)]
|
|
cmd: Cmd,
|
|
}
|
|
|
|
#[derive(Subcommand)]
|
|
enum Cmd {
|
|
/// Run the coordinator daemon.
|
|
Serve {
|
|
/// Path to a JSON config file holding the host-level daemon config
|
|
/// (the [`ServeConfig`](crate::coordinator::ServeConfig) shape:
|
|
/// the container-injected `HiveEnv` fields + the hive-c0re-local
|
|
/// `model_prices` table). Used as the base; any per-flag override
|
|
/// below wins over the file. Absent → start from the built-in
|
|
/// defaults. The NixOS module passes a generated config here so the
|
|
/// `ExecStart` stays short instead of carrying every setting — and
|
|
/// the context-window / price-table JSON blobs — as flags.
|
|
#[arg(long)]
|
|
config: Option<PathBuf>,
|
|
/// Path to the sqlite message store.
|
|
#[arg(long, default_value = crate::paths::BROKER_DB)]
|
|
db: PathBuf,
|
|
/// Override: URL of the hyperhive flake. Inlined into each
|
|
/// per-agent `flake.nix` as the `hyperhive` input.
|
|
#[arg(long)]
|
|
hyperhive_flake: Option<String>,
|
|
/// Override: store-path URL of the nixpkgs to wire into the meta
|
|
/// flake as `inputs.nixpkgs.url`. Empty = legacy
|
|
/// `follows = "hyperhive/nixpkgs"` fallback.
|
|
#[arg(long)]
|
|
nixpkgs_flake: Option<String>,
|
|
/// Override: dashboard HTTP port.
|
|
#[arg(long)]
|
|
dashboard_port: Option<u16>,
|
|
/// Override: operator pronouns (free text). Threaded into each
|
|
/// container's harness via `HIVE_OPERATOR_PRONOUNS`.
|
|
#[arg(long)]
|
|
operator_pronouns: Option<String>,
|
|
/// Override: per-model context-window sizes, as a JSON object
|
|
/// mapping model-family short name to token count. Threaded into
|
|
/// each container as `HIVE_CONTEXT_WINDOW_TOKENS_<KEY_UPPER>` env
|
|
/// vars. Set via the `services.hive-c0re.contextWindowTokens`
|
|
/// NixOS option.
|
|
#[arg(long)]
|
|
context_window_tokens: Option<String>,
|
|
/// Override: systemd `CPUQuota=` applied to every agent container
|
|
/// via a drop-in (e.g. `"200%"` = 2 cores). Set via
|
|
/// `services.hyperhive.agentCpuQuota`.
|
|
#[arg(long)]
|
|
agent_cpu_quota: Option<String>,
|
|
/// Override: systemd `MemoryMax=` applied to every agent
|
|
/// container. Set via `services.hyperhive.agentMemoryMax`.
|
|
#[arg(long)]
|
|
agent_memory_max: Option<String>,
|
|
/// Override: systemd `CPUWeight=` applied to every agent container
|
|
/// via the same drop-in — a cgroup v2 relative share under
|
|
/// contention (1..=10000), not a cap. Set via
|
|
/// `services.hyperhive.agentCpuWeight`.
|
|
#[arg(long)]
|
|
agent_cpu_weight: Option<u32>,
|
|
/// Override: systemd `IOWeight=` applied to every agent container,
|
|
/// the block-IO counterpart of `--agent-cpu-weight`. Set via
|
|
/// `services.hyperhive.agentIoWeight`.
|
|
#[arg(long)]
|
|
agent_io_weight: Option<u32>,
|
|
/// Override: per-model USD prices (per million tokens) for the
|
|
/// hive-wide ST4TS cost estimate, as a JSON object mapping a
|
|
/// model-family short name to `{input, output, cache_read,
|
|
/// cache_write}`. Models not covered fall back to the built-in
|
|
/// estimate. Set via the `services.hyperhive.modelPrices` NixOS
|
|
/// option.
|
|
#[arg(long)]
|
|
model_prices: Option<String>,
|
|
/// Override: number of concurrent nix-heavy job-queue nodes
|
|
/// (prebuild / profile-swap / create / meta lock). Set via the
|
|
/// `services.hyperhive.c0re.buildSlots` NixOS option.
|
|
#[arg(long)]
|
|
build_slots: Option<usize>,
|
|
},
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
tracing_subscriber::fmt()
|
|
.with_env_filter(
|
|
tracing_subscriber::EnvFilter::try_from_default_env()
|
|
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
|
|
)
|
|
.init();
|
|
|
|
let cli = Cli::parse();
|
|
match cli.cmd {
|
|
Cmd::Serve {
|
|
config,
|
|
db,
|
|
hyperhive_flake,
|
|
nixpkgs_flake,
|
|
dashboard_port,
|
|
operator_pronouns,
|
|
context_window_tokens,
|
|
agent_cpu_quota,
|
|
agent_memory_max,
|
|
agent_cpu_weight,
|
|
agent_io_weight,
|
|
model_prices,
|
|
build_slots,
|
|
} => {
|
|
// Base config from the --config file (or the built-in
|
|
// defaults), then apply any per-flag overrides — config
|
|
// file is the base, explicit flags win.
|
|
let mut sc = match &config {
|
|
Some(p) => {
|
|
let s = std::fs::read_to_string(p)
|
|
.with_context(|| format!("read --config {}", p.display()))?;
|
|
serde_json::from_str::<ServeConfig>(&s)
|
|
.with_context(|| format!("parse --config {}", p.display()))?
|
|
}
|
|
None => ServeConfig::default(),
|
|
};
|
|
if let Some(v) = hyperhive_flake {
|
|
sc.env.hyperhive_flake = v;
|
|
}
|
|
if let Some(v) = nixpkgs_flake {
|
|
sc.env.nixpkgs_flake = v;
|
|
}
|
|
if let Some(v) = dashboard_port {
|
|
sc.env.dashboard_port = v;
|
|
}
|
|
if let Some(v) = operator_pronouns {
|
|
sc.env.operator_pronouns = v;
|
|
}
|
|
if let Some(v) = context_window_tokens {
|
|
sc.env.context_window_tokens =
|
|
serde_json::from_str(&v).context("--context-window-tokens: invalid JSON")?;
|
|
}
|
|
if let Some(v) = agent_cpu_quota {
|
|
sc.env.agent_cpu_quota = v;
|
|
}
|
|
if let Some(v) = agent_memory_max {
|
|
sc.env.agent_memory_max = v;
|
|
}
|
|
// Passing the flag sets a weight; omitting it keeps whatever the
|
|
// config file says (including `null` = don't emit the setting).
|
|
// There's deliberately no flag spelling for "clear it" — that's
|
|
// what the nix option's `null` is for.
|
|
if let Some(v) = agent_cpu_weight {
|
|
sc.env.agent_cpu_weight = Some(v);
|
|
}
|
|
if let Some(v) = agent_io_weight {
|
|
sc.env.agent_io_weight = Some(v);
|
|
}
|
|
if let Some(v) = model_prices {
|
|
sc.model_prices =
|
|
serde_json::from_str(&v).context("--model-prices: invalid JSON")?;
|
|
}
|
|
if let Some(v) = build_slots {
|
|
sc.build_slots = v;
|
|
}
|
|
cmd_serve(sc.env, sc.model_prices, sc.build_slots, db, &cli.socket).await
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Banner message for a failing matrix `ensure_all` sweep, shared by both
|
|
/// the initial and periodic `record_err` call sites in `cmd_serve` so the
|
|
/// wording can't drift between them.
|
|
fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String {
|
|
let age = ctx.since_last_ok.map_or_else(
|
|
|| "no success this session".to_owned(),
|
|
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
|
|
);
|
|
format!(
|
|
"matrix user/space sweep failing ({} consecutive, {age}) \
|
|
— some agents may be missing matrix accounts, space membership, \
|
|
or chat-room invites",
|
|
ctx.consecutive
|
|
)
|
|
}
|
|
|
|
/// Start the coordinator daemon: open the broker, run migrations, spawn
|
|
/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts,
|
|
/// dashboard), then serve the admin socket until a signal arrives.
|
|
#[allow(
|
|
clippy::too_many_lines,
|
|
reason = "startup orchestration: open the broker, run migrations, then spawn \
|
|
the full set of background services (auto-update, vacuums, \
|
|
crash-watch, schedulers, dashboard) before the serve loop — the \
|
|
length is inherent to booting the daemon, not the arg list"
|
|
)]
|
|
async fn cmd_serve(
|
|
env: HiveEnv,
|
|
model_prices: crate::hive_stats::PriceTable,
|
|
build_slots: usize,
|
|
db: std::path::PathBuf,
|
|
socket: &std::path::Path,
|
|
) -> Result<()> {
|
|
// Move any host-side state still at the legacy flat layout into its
|
|
// subdir (`db/`, `forge/`, `matrix/`, `run/`) BEFORE opening the
|
|
// broker db — the broker + build-logs dbs are among the relocated
|
|
// files. Idempotent; a no-op once migrated.
|
|
crate::paths::relocate_legacy_state();
|
|
// `dashboard_port` is consumed into the Coordinator below; capture the
|
|
// Copy value first for the dashboard + knowledge-webhook tasks.
|
|
let dashboard_port = env.dashboard_port;
|
|
let coord = Arc::new(Coordinator::open(&db, env, model_prices, build_slots)?);
|
|
socket_server::start_manager(coord.clone())?;
|
|
// Idempotent pre-flight: rewrite pre-meta-layout applied
|
|
// repos, ensure proposed repos carry the `applied`
|
|
// remote, bootstrap the meta repo, repoint containers at
|
|
// `meta#<name>` (one-shot, guarded by a marker file).
|
|
// Runs before manager auto-spawn so the new manager is
|
|
// built against meta from the first attempt.
|
|
if let Err(e) = migrate::run(&coord).await {
|
|
tracing::warn!(error = ?e, "startup migration failed");
|
|
}
|
|
// Auto-create the root agent container if it isn't there yet. Block
|
|
// on this — without root the system has no manager harness.
|
|
// Failures are logged but allowed: a broken auto-spawn shouldn't
|
|
// make the dashboard unreachable for debugging.
|
|
if let Err(e) = auto_update::ensure_root_agent(&coord).await {
|
|
tracing::warn!(error = ?e, "auto-spawn root agent failed");
|
|
}
|
|
// Sync /etc/tmpfiles.d/hyperhive-agents.conf so agent runtime dirs are
|
|
// pre-declared for the next boot. Best-effort background task — a failure
|
|
// here must not block hive-c0re startup. See lifecycle::sync_tmpfiles.
|
|
tokio::spawn(crate::lifecycle::sync_tmpfiles());
|
|
// Auto-update in the background — don't block service start.
|
|
// Sub-agent rebuilds can take tens of seconds; we want the admin
|
|
// socket up immediately.
|
|
let update_coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = auto_update::run(update_coord).await {
|
|
tracing::warn!(error = ?e, "auto-update task failed");
|
|
}
|
|
});
|
|
// Forge user sweep: ensure every existing container has a
|
|
// forgejo user + access token. No-op when the hive-forge
|
|
// container isn't running. Backgrounded — touches the
|
|
// forge state dir via `nixos-container run` which is slow.
|
|
tokio::spawn(async move {
|
|
forge::ensure_all().await;
|
|
});
|
|
// Webhook HMAC secret: load from state dir or generate on first run.
|
|
// Used by both the webhook handlers (verification) and the Forgejo
|
|
// hook registrations (so Forgejo signs deliveries with the same key).
|
|
let webhook_secret: Option<String> = match crate::webhook_secret::load_or_generate() {
|
|
Ok(s) => Some(s),
|
|
Err(e) => {
|
|
tracing::error!(
|
|
error = ?e,
|
|
"webhook secret load/generate failed; /webhook/* endpoints disabled and hooks not registered"
|
|
);
|
|
None
|
|
}
|
|
};
|
|
// Webhook setup: ensure Forgejo webhooks are registered for both
|
|
// `internal/knowledge` (push → git pull) and the `agent-configs` org
|
|
// (pull_request → queue MergeConfigPr approval). Both run after
|
|
// forge::ensure_all so the core token + repos + org are present.
|
|
// URLs use the public hive domain (HYPERHIVE_HIVE_DOMAIN) so Forgejo
|
|
// delivers through the gateway, bypassing the SSRF loopback guard.
|
|
// No-op when the core token or domain are absent, or when the HMAC
|
|
// secret is unavailable (load failure).
|
|
let webhook_secret_reg = webhook_secret.clone();
|
|
tokio::spawn(async move {
|
|
let Some(webhook_secret_reg) = webhook_secret_reg else {
|
|
tracing::debug!("webhook secret unavailable; skipping hook registration");
|
|
return;
|
|
};
|
|
let Some(token) = forge::core_token() else {
|
|
return;
|
|
};
|
|
let domain = std::env::var("HYPERHIVE_HIVE_DOMAIN")
|
|
.ok()
|
|
.filter(|v| !v.is_empty());
|
|
let Some(domain) = domain else {
|
|
tracing::debug!("HYPERHIVE_HIVE_DOMAIN unset; skipping webhook registration");
|
|
return;
|
|
};
|
|
if let Err(e) = knowledge::ensure_webhook(&token, &domain, &webhook_secret_reg).await {
|
|
tracing::warn!(error = ?e, "knowledge: ensure_webhook failed");
|
|
}
|
|
if let Err(e) = forge::ensure_config_pr_webhook(&token, &domain, &webhook_secret_reg).await
|
|
{
|
|
tracing::warn!(error = ?e, "forge: ensure_config_pr_webhook failed");
|
|
}
|
|
});
|
|
// Config-PR polling fallback: scan agent-configs org every 5 minutes
|
|
// for open PRs that have no pending MergeConfigPr approval. Catches
|
|
// anything the webhook missed (c0re was down when PR opened, delivery
|
|
// failed, etc.). First sweep fires immediately on startup.
|
|
let poll_coord = coord.clone();
|
|
let mut poll_shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
let interval = std::time::Duration::from_mins(5);
|
|
loop {
|
|
if let Some(token) = forge::core_token() {
|
|
let result = Box::pin(forge::config_pr_poll::poll_open_config_prs(
|
|
&token,
|
|
&poll_coord,
|
|
))
|
|
.await;
|
|
if let Err(e) = result {
|
|
tracing::debug!(error = ?e, "config-pr poll: sweep failed (forge may be absent)");
|
|
}
|
|
}
|
|
tokio::select! {
|
|
() = tokio::time::sleep(interval) => {}
|
|
_ = poll_shutdown.changed() => {
|
|
tracing::info!("config-pr poll: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Knowledge periodic pull: hourly fallback in case the webhook is
|
|
// missed (e.g. hive-c0re was down during a push). First fires at
|
|
// startup (immediate pull after the clone is already present).
|
|
let mut knowledge_shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
// Initial pull — reconcile any commits that landed while c0re
|
|
// was offline. Not fed to the health tracker: a startup miss is
|
|
// expected (the clone may not exist yet) and is logged at debug.
|
|
if let Err(e) = knowledge::pull().await {
|
|
tracing::debug!(error = ?e, "knowledge: startup pull skipped (no clone yet?)");
|
|
}
|
|
// Persistent-failure → banner. An hourly sweep that keeps failing for
|
|
// several hours means the operator's `/knowledge` is drifting; raise a
|
|
// warn banner after 3 consecutive misses so a one-off network blip
|
|
// self-heals on the next tick without ever bannering. Cleared on the
|
|
// next successful pull.
|
|
let mut health = sweep_health::SweepHealth::new("knowledge_pull", "warn", 3);
|
|
let interval = std::time::Duration::from_hours(1);
|
|
loop {
|
|
tokio::select! {
|
|
() = tokio::time::sleep(interval) => {
|
|
match knowledge::pull().await {
|
|
Ok(()) => health.record_ok(),
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "knowledge: periodic pull failed");
|
|
let err = format!("{e:#}");
|
|
health.record_err(|ctx| {
|
|
let age = ctx.since_last_ok.map_or_else(
|
|
|| "no success this session".to_owned(),
|
|
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
|
|
);
|
|
format!(
|
|
"knowledge repo pull failing ({} consecutive, {age}) \
|
|
— /knowledge is stale until it recovers: {err}",
|
|
ctx.consecutive
|
|
)
|
|
});
|
|
}
|
|
}
|
|
}
|
|
_ = knowledge_shutdown.changed() => {
|
|
tracing::info!("knowledge pull: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Disk-pressure watch: raise a `disk_pressure` banner warning while the
|
|
// host nix store is over threshold and clear it when back under, via the
|
|
// push-based warnings registry. Replaces the old per-`/api/state`
|
|
// `statvfs`. ~60s cadence; first tick fires immediately. The held guard
|
|
// lives for the task's lifetime — dropping it (on shutdown) clears the
|
|
// banner.
|
|
let mut disk_shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
let mut disk_guard: Option<warnings::WarningGuard> = None;
|
|
let interval = std::time::Duration::from_mins(1);
|
|
loop {
|
|
host_stats::refresh_disk_warning(&mut disk_guard);
|
|
tokio::select! {
|
|
() = tokio::time::sleep(interval) => {}
|
|
_ = disk_shutdown.changed() => {
|
|
tracing::info!("disk-pressure watch: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Matrix user sweep: same shape — ensure every container has
|
|
// an account on the local matrix-tuwunel homeserver with an
|
|
// access_token persisted to `<state>/matrix-token`. No-op when
|
|
// the hive-matrix container isn't running. Backgrounded because
|
|
// UIAA is a two-roundtrip dance per agent.
|
|
//
|
|
// Runs once at startup AND periodically every 30 minutes so that
|
|
// token files deleted by `hive-matrix-daemon` (stale-token
|
|
// recovery — `M_UNKNOWN_TOKEN`) get re-provisioned without
|
|
// requiring a hive-c0re restart.
|
|
let mut matrix_shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
let interval = std::time::Duration::from_mins(30);
|
|
// Debounced banner: a lone bad sweep (homeserver mid-restart, a
|
|
// transient HTTP blip) shouldn't flap the dashboard, but a sweep
|
|
// that's been failing for hours (missing agent invites, a broken
|
|
// admin token) should surface. Cleared the moment a sweep is clean.
|
|
let mut health = sweep_health::SweepHealth::new("matrix_ensure_all", "warn", 2);
|
|
if matrix::ensure_all().await {
|
|
health.record_ok();
|
|
} else {
|
|
health.record_err(matrix_sweep_banner);
|
|
}
|
|
loop {
|
|
tokio::select! {
|
|
() = tokio::time::sleep(interval) => {
|
|
if matrix::ensure_all().await {
|
|
health.record_ok();
|
|
} else {
|
|
health.record_err(matrix_sweep_banner);
|
|
}
|
|
}
|
|
_ = matrix_shutdown.changed() => {
|
|
tracing::info!("matrix ensure_all: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Periodic broker vacuum: drop fully-acked messages older
|
|
// than 30 days. Delivered-but-unacked rows (recoverable via
|
|
// requeue_inflight) and undelivered rows are always kept.
|
|
// Runs hourly; first sweep happens immediately.
|
|
let vacuum_coord = coord.clone();
|
|
let mut vacuum_shutdown = coord.shutdown_rx();
|
|
tokio::spawn(async move {
|
|
let interval = std::time::Duration::from_hours(1);
|
|
let keep_secs: i64 = 30 * 24 * 3600;
|
|
loop {
|
|
match vacuum_coord.broker.vacuum_delivered(keep_secs) {
|
|
Ok(0) => {}
|
|
Ok(n) => tracing::info!(removed = n, "broker vacuum"),
|
|
Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
|
|
}
|
|
tokio::select! {
|
|
() = tokio::time::sleep(interval) => {}
|
|
_ = vacuum_shutdown.changed() => {
|
|
tracing::info!("broker vacuum: shutdown signal received");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Per-agent events.sqlite + bash-tasks file cleanup now runs
|
|
// agent-side in the harness (`hive_agent::vacuum`): the files are
|
|
// agent-owned, so host-side deletes hit PermissionDenied / readonly-db
|
|
// under privsep. See issue tracker "perms borked".
|
|
// (turn-stats.sqlite has no vacuum — it's one tiny row per turn,
|
|
// ~hundreds of KB, and the /stats + hive-stats views read it
|
|
// directly; pruning it would just lose trend history for no gain.)
|
|
// Slow per-container disk sampler: a `du` of each agent's state dir
|
|
// + writable rootfs every ~5 min, cached so the 5s container-load
|
|
// poll stays cheap cgroup-only reads. Feeds `disk_bytes` on the LOAD
|
|
// tab. See container_stats::disk_sampler_loop.
|
|
crate::container_stats::spawn_disk_sampler();
|
|
// Per-agent container-resource OTEL export: rides the same
|
|
// cgroup gauges out to the configured OTLP endpoint, reusing the hive
|
|
// `services.hyperhive.otel` config (endpoint + LoadCredential auth).
|
|
// No-op when OTEL isn't configured.
|
|
crate::otel_metrics::spawn_exporter();
|
|
// build_logs.sqlite vacuum: c0re-side (single db). Failures kept
|
|
// 30d, successes 24h — see `build_logs::vacuum` for the rule.
|
|
crate::build_logs::spawn_vacuum(&coord);
|
|
// audit_log.sqlite vacuum: agent-initiated privileged-action trail,
|
|
// 90d retention — see `audit_log::vacuum`.
|
|
crate::audit_log::spawn_vacuum(&coord);
|
|
// Container crash watcher: emits HelperEvent::ContainerCrash
|
|
// when a previously-running container goes away without an
|
|
// operator-initiated transient state.
|
|
crash_watch::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.
|
|
// Also retries any pending gateway nginx reload that failed on
|
|
// the previous tick. write() is idempotent so steady-state cost
|
|
// is one stat per agent per tick.
|
|
// See `docs/gateway.md::Per-agent unix-socket upstream`.
|
|
agent_sockets::spawn_poll();
|
|
// MCP socket listener startup sync: one-shot sweep that re-registers any
|
|
// running agent container whose MCP listener was lost when hive-c0re
|
|
// restarted (Coordinator starts empty; /run/hyperhive/agents/ is tmpfs).
|
|
// After this, listener registration is event-driven: run_create /
|
|
// run_reconcile call register_agent on start; kill/destroy call
|
|
// unregister_agent. No recurring poll needed — c0re owns the listeners.
|
|
mcp_sockets::sync_on_start(coord.clone()).await;
|
|
// Scheduled-prompts worker: drains due scheduled_prompts rows
|
|
// and fans the body out to each active target's inbox. See
|
|
// scheduled_prompts_worker.rs.
|
|
scheduled_prompts_worker::spawn(coord.clone());
|
|
// Job-queue scheduler: drives the global DAG queue (rebuild /
|
|
// meta-update / spawn / power ops). Concurrency comes from the
|
|
// build-slot count + per-agent leases inside the queue, not from
|
|
// multiple workers — cheap nodes (graceful signals, drains,
|
|
// reconciles) overlap nix-heavy ones structurally. Call sites
|
|
// (auto_update, dashboard, manager, approval handler) submit DAGs
|
|
// instead of awaiting lifecycle work inline. See `job_queue/`.
|
|
{
|
|
let q_coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
job_queue::scheduler::run_worker(q_coord).await;
|
|
});
|
|
}
|
|
// Forward every broker event onto the unified dashboard
|
|
// channel with a freshly-stamped seq, so the dashboard SSE
|
|
// sees broker messages + future mutation events on one
|
|
// stream with one monotonic seq. The broker's intra-process
|
|
// channel (used by `recv_blocking_batch`) stays untouched.
|
|
spawn_broker_to_dashboard_forwarder(coord.clone());
|
|
let dash_coord = coord.clone();
|
|
let dash_secret = webhook_secret.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = dashboard::serve(dashboard_port, dash_coord, dash_secret).await {
|
|
tracing::error!(error = ?e, "dashboard failed");
|
|
}
|
|
});
|
|
// Run the admin socket until a signal arrives; then signal
|
|
// all background tasks so they exit cleanly before the
|
|
// process terminates.
|
|
let coord_sig = coord.clone();
|
|
tokio::select! {
|
|
res = server::serve(socket, coord) => { res? }
|
|
_ = tokio::signal::ctrl_c() => {
|
|
tracing::info!("SIGINT received — requesting shutdown");
|
|
coord_sig.request_shutdown();
|
|
}
|
|
() = async {
|
|
let mut sig = tokio::signal::unix::signal(
|
|
tokio::signal::unix::SignalKind::terminate()
|
|
).expect("failed to install SIGTERM handler");
|
|
sig.recv().await;
|
|
} => {
|
|
tracing::info!("SIGTERM received — requesting shutdown");
|
|
coord_sig.request_shutdown();
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Re-emit every broker `MessageEvent` onto the dashboard channel as
|
|
/// a `DashboardEvent::Sent` / `Delivered` with a freshly-stamped seq.
|
|
/// Background task; runs for the life of the process. On a lagged
|
|
/// broker subscription we just keep going — the dashboard channel is
|
|
/// best-effort presentation plumbing, the broker keeps its own sqlite
|
|
/// log for replay.
|
|
fn spawn_broker_to_dashboard_forwarder(coord: Arc<Coordinator>) {
|
|
use broker::MessageEvent;
|
|
use dashboard_events::DashboardEvent;
|
|
let mut rx = coord.broker.subscribe();
|
|
tokio::spawn(async move {
|
|
loop {
|
|
match rx.recv().await {
|
|
Ok(MessageEvent::Sent {
|
|
id,
|
|
from,
|
|
to,
|
|
body,
|
|
at,
|
|
in_reply_to,
|
|
}) => {
|
|
let file_refs = dashboard::scan_validated_paths(&body);
|
|
coord.emit_dashboard_event(DashboardEvent::Sent {
|
|
seq: coord.next_seq(),
|
|
id,
|
|
from,
|
|
to,
|
|
body,
|
|
at: hive_sh4re::wire_time::from_secs(at),
|
|
in_reply_to,
|
|
file_refs,
|
|
});
|
|
}
|
|
Ok(MessageEvent::Delivered {
|
|
id,
|
|
from,
|
|
to,
|
|
body,
|
|
at,
|
|
in_reply_to,
|
|
}) => {
|
|
let file_refs = dashboard::scan_validated_paths(&body);
|
|
coord.emit_dashboard_event(DashboardEvent::Delivered {
|
|
seq: coord.next_seq(),
|
|
id,
|
|
from,
|
|
to,
|
|
body,
|
|
at: hive_sh4re::wire_time::from_secs(at),
|
|
in_reply_to,
|
|
file_refs,
|
|
});
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
|
tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged");
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
|
}
|
|
}
|
|
});
|
|
}
|