656 lines
29 KiB
Rust
656 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 server;
|
|
mod snapshot_push;
|
|
mod socket_server;
|
|
mod stats;
|
|
mod stores;
|
|
mod swarm_notices;
|
|
mod swarm_queue;
|
|
mod swarm_status;
|
|
#[cfg(test)]
|
|
mod test_env;
|
|
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, 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`] 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")),
|
|
)
|
|
// This is a systemd-managed daemon — stdout always goes to journald,
|
|
// never a human terminal, and journald doesn't strip ANSI escapes:
|
|
// they land in victorialogs as raw byte-array spam otherwise.
|
|
.with_ansi(false)
|
|
.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;
|
|
}
|
|
// Network isolation is required and its settings are
|
|
// process-global, so a bad value breaks every container, not
|
|
// one. Validate once here: the unit then fails visibly at
|
|
// start with a single diagnostic naming the bad value,
|
|
// instead of coming up "healthy" and failing each container
|
|
// configure separately — possibly after some already
|
|
// succeeded.
|
|
lifecycle::network_isolation_from_env()
|
|
.context("network isolation settings are required at startup")?;
|
|
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: now a `NodeKind::ForgeSweep` DAG node (see
|
|
// `workers::auto_update::submit_startup_sweep_nodes`), submitted
|
|
// unconditionally on every boot — moved off a bare `tokio::spawn` so it
|
|
// shows as real work on the dashboard.
|
|
// 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: now a `NodeKind::WebhookRegister` DAG node (see
|
|
// `workers::auto_update::submit_startup_sweep_nodes`), registering both
|
|
// `internal/knowledge` (push → git pull) and the `agent-configs` org
|
|
// (pull_request → queue MergeConfigPr approval) hooks. Its executor
|
|
// re-derives the core token / hive domain / HMAC secret itself, mirroring
|
|
// the guard chain that used to live here — see `job_queue::exec::
|
|
// run_webhook_register`.
|
|
// 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). Deliberately does
|
|
// NOT also fire an immediate pull at startup the way this task used
|
|
// to: `auto_update::run`'s `NodeKind::KnowledgePull` DAG node (spawned
|
|
// separately, a few lines up) already does that unconditionally on
|
|
// every boot. The two used to run concurrently with no lock between
|
|
// them, both `git pull --ff-only`-ing the same working tree — a real
|
|
// race, and the likely root cause of the "local changes would be
|
|
// overwritten" wedge this file's `pull()` now defends against
|
|
// (`reset --hard` before every pull). Removing the redundant caller
|
|
// fixes the race at its source instead of just self-healing after it.
|
|
let mut knowledge_shutdown = coord.shutdown_rx();
|
|
let knowledge_coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
// 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(&knowledge_coord).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;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
// Offer this hive's readiness to the swarm, if one is configured.
|
|
// A no-op on a standalone hive (no queue env, logged once) — see
|
|
// swarm_status, which owns the whole task including its own decision
|
|
// not to start.
|
|
swarm_status::spawn(std::sync::Arc::clone(&coord), coord.shutdown_rx());
|
|
// 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(&coord.hyperhive_flake);
|
|
// 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,
|
|
}
|
|
}
|
|
});
|
|
}
|