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

remove the 1NFR4 dashboard panel and the now-writer-less audit log

Browse source
This commit is contained in:
damocles 2026-08-30 23:54:24 +02:00 • committed by mara
commit 22adfd1451
21 changed files with 61 additions and 983 deletions
Download patch file Download diff file Expand all files Collapse all files

23
hive-c0re/src/coordinator.rs
View file

@ -42,9 +42,6 @@ pub struct Coordinator {
/// `get_full` for the per-card chip + side-panel viewer. See
/// `build_logs.rs` for retention.
pub build_logs: Arc<crate::build_logs::BuildLogs>,
/// Audit trail of agent-initiated privileged actions (infra restart,
/// …). See `audit_log.rs`. Same dir as `build_logs`.
pub audit_log: Arc<crate::audit_log::AuditLog>,
/// URL of the hyperhive flake (no fragment). Inlined into per-agent
/// `flake.nix` files as `inputs.hyperhive.url`.
pub hyperhive_flake: String,
@ -480,13 +477,6 @@ impl Coordinator {
// to thread an `Arc<BuildLogs>` through every public entry
// point in the lifecycle surface.
crate::build_logs::install(build_logs.clone());
// Audit log shares the same db dir; install its process-wide
// handle so privileged-action recording sites (e.g.
// `dashboard::infra_containers::post_infra_container`) write
// without threading an `Arc<AuditLog>` through the surface.
let audit_log =
Arc::new(crate::audit_log::AuditLog::open(build_logs_dir).context("open audit_log")?);
crate::audit_log::install(audit_log.clone());
let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?);
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
let (shutdown_tx, _) = watch::channel(false);
@ -495,7 +485,6 @@ impl Coordinator {
approvals: Arc::new(approvals),
scheduled_prompts: Arc::new(scheduled_prompts),
build_logs,
audit_log,
hyperhive_flake,
hyperhive_docs_flake,
nixpkgs_flake,
@ -751,18 +740,6 @@ impl Coordinator {
self.meta_updates_active.load(Ordering::SeqCst) > 0
}
/// Emit `AuditEntryAdded` immediately after a privileged-action row
/// is recorded, so the dashboard audit view live-appends it off
/// `/dashboard/stream`. Pass the [`AuditEntry`](crate::audit_log::AuditEntry)
/// returned by `audit_log::record` so the streamed event is the same
/// canonical row that was stored.
pub fn emit_audit_entry(&self, entry: crate::audit_log::AuditEntry) {
self.emit_dashboard_event(DashboardEvent::AuditEntryAdded {
seq: self.next_seq(),
entry,
});
}
/// Emit `ApprovalAdded` immediately after the row is inserted in
/// sqlite.
pub fn emit_approval_added(&self, ev: ApprovalAdded<'_>) {

70
hive-c0re/src/dashboard/infra_containers.rs
View file

@ -1,70 +0,0 @@
//! Dashboard endpoint for operator-driven infra lifecycle (start / stop on
//! `hive-ci`, `hive-forge`, `hive-gateway`, `hive-matrix`). Operator-only —
//! there is no agent-facing equivalent for either action.
//! Already fully operator-authenticated by the time a request reaches here,
//! so no capability check is needed, just the audit trail.
use axum::{
extract::{Path as AxumPath, State},
http::StatusCode,
response::{IntoResponse, Response},
};
use hive_priv_sock::{InfraAction, InfraContainer};
use super::{AppState, error_response};
/// Start / stop a hive infrastructure container from the dashboard.
///
/// `name` parses into [`InfraContainer`] (the allowlist; unrecognised
/// names 400), `action` into `start` / `stop`. Every attempt lands in the
/// audit log (actor `"operator"`, action `start_infra` / `stop_infra`) and
/// streams as an `AuditEntryAdded` event.
#[utoipa::path(
post,
path = "/api/infra-container/{name}/{action}",
params(
("name" = String, Path, description = "infra service name (hive-ci/hive-forge/hive-gateway/hive-matrix)"),
("action" = String, Path, description = "start | stop"),
),
responses(
(status = 200, description = "action completed", body = String),
(status = 500, description = "unknown container/action, or the systemd action failed"),
),
tag = "infra_containers"
)]
pub(super) async fn post_infra_container(
State(state): State<AppState>,
AxumPath((name, action)): AxumPath<(String, String)>,
) -> Response {
let Ok(container) = name.parse::<InfraContainer>() else {
return error_response(&format!("unknown infra container: {name}"));
};
let (infra_action, action_label) = match action.as_str() {
"start" => (InfraAction::Start, "start_infra"),
"stop" => (InfraAction::Stop, "stop_infra"),
other => {
return error_response(&format!("unknown action: {other} (want start|stop)"));
}
};
let target = container.name();
tracing::info!(%target, %action, "dashboard: infra container action");
let result = crate::priv_client::control_infra_container(container, infra_action).await;
let outcome = if result.is_ok() {
crate::audit_log::AuditOutcome::Ok
} else {
crate::audit_log::AuditOutcome::Err
};
let detail = result.as_ref().err().map(|e| format!("{e:#}"));
if let Some(entry) =
state
.coord
.audit_log
.record("operator", action_label, target, outcome, detail.as_deref())
{
state.coord.emit_audit_entry(entry);
}
match result {
Ok(()) => (StatusCode::OK, "ok").into_response(),
Err(e) => error_response(&format!("{target}: {e:#}")),
}
}

38
hive-c0re/src/dashboard/misc_api.rs
View file

@ -1,7 +1,6 @@
//! Remaining single-endpoint dashboard handlers: the operator inbox
//! (`Y3R C4LL`) + mark-all-read, operator compose (`op-send`),
//! spawn-request, hive-wide turn stats, container resources, and the
//! audit log.
//! spawn-request, hive-wide turn stats, and container resources.
use axum::{
extract::{Form, Path as AxumPath, State},
@ -12,7 +11,6 @@ use serde::{Deserialize, Serialize};
use utoipa::{IntoParams, ToSchema};
use super::{AppState, Ident, error_response, scan_validated_paths};
use crate::audit_log::AuditEntry;
use crate::container_stats::ContainerResource;
use crate::hive_stats::HiveStats;
@ -129,40 +127,6 @@ pub(super) async fn api_container_resources() -> Response {
axum::Json(crate::container_stats::gather().await).into_response()
}
#[derive(Serialize, ToSchema)]
pub(super) struct AuditLogBody {
entries: Vec<AuditEntry>,
total: i64,
}
/// Most-recent agent-initiated privileged-action
/// audit entries, newest first (server-clamped to 500).
///
/// Backs the operator dashboard's audit view. `total` lets the UI show
/// "latest 500 of N" rather than silently capping. `ts_unix` is in
/// **seconds**.
#[utoipa::path(
get,
path = "/api/audit-log",
responses(
(status = 200, description = "recent audit entries + total count", body = AuditLogBody),
(status = 500, description = "sqlite read failed"),
),
tag = "misc_api"
)]
pub(super) async fn api_audit_log(State(state): State<AppState>) -> Response {
const LIMIT: usize = 500;
let entries = match state.coord.audit_log.list_recent(LIMIT) {
Ok(rows) => rows,
Err(e) => return error_response(&format!("audit-log: {e:#}")),
};
let total = match state.coord.audit_log.count_total() {
Ok(n) => n,
Err(e) => return error_response(&format!("audit-log count: {e:#}")),
};
axum::Json(AuditLogBody { entries, total }).into_response()
}
#[derive(Serialize, ToSchema)]
pub(super) struct MarkAllReadBody {
marked: u64,

8
hive-c0re/src/dashboard/mod.rs
View file

@ -38,11 +38,10 @@ use crate::lifecycle;
(name = "approvals", description = "approve/deny pending approval rows"),
(name = "build_logs", description = "build log headers, full rows, and raw text downloads"),
(name = "extra_forges", description = "external (non-internal) forge account provisioning"),
(name = "infra_containers", description = "start/stop of hive infrastructure containers"),
(name = "lifecycle_ops", description = "agent container lifecycle: rebuild/restart/start/stop/pause/limits"),
(name = "matrix_accounts", description = "matrix + github account provisioning for agents"),
(name = "meta_inputs", description = "bulk flake-input update for the meta flake"),
(name = "misc_api", description = "operator inbox, compose, spawn-request, hive stats, audit log"),
(name = "misc_api", description = "operator inbox, compose, spawn-request, hive stats"),
(name = "permissions", description = "tool-group + capability assignment for agents"),
(name = "schedules", description = "scheduled-prompt + rebuild-queue CRUD"),
(name = "state_files", description = "proxied reads of allow-listed per-agent state files"),
@ -63,7 +62,6 @@ mod extra_forges;
// server reach it as `crate::dashboard::Ident`.
pub(crate) use hive_types::Ident;
mod health;
mod infra_containers;
mod journal;
mod lifecycle_ops;
mod matrix_accounts;
@ -154,7 +152,6 @@ pub async fn serve(
.routes(routes!(misc_api::api_operator_inbox))
.routes(routes!(misc_api::api_stats_hive))
.routes(routes!(misc_api::api_container_resources))
.routes(routes!(misc_api::api_audit_log))
.routes(routes!(misc_api::post_mark_all_read))
.routes(routes!(misc_api::post_request_spawn))
.routes(routes!(misc_api::post_op_send))
@ -195,7 +192,6 @@ pub async fn serve(
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(lifecycle_ops::post_resource_limits))
.routes(routes!(lifecycle_ops::post_update_all))
.routes(routes!(infra_containers::post_infra_container))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(meta_inputs::post_meta_update))
.routes(routes!(build_logs::get_build_log_stream))
@ -397,7 +393,6 @@ mod router_build_probe {
.routes(routes!(misc_api::api_operator_inbox))
.routes(routes!(misc_api::api_stats_hive))
.routes(routes!(misc_api::api_container_resources))
.routes(routes!(misc_api::api_audit_log))
.routes(routes!(misc_api::post_mark_all_read))
.routes(routes!(misc_api::post_request_spawn))
.routes(routes!(misc_api::post_op_send))
@ -438,7 +433,6 @@ mod router_build_probe {
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(lifecycle_ops::post_resource_limits))
.routes(routes!(lifecycle_ops::post_update_all))
.routes(routes!(infra_containers::post_infra_container))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(meta_inputs::post_meta_update))
.routes(routes!(build_logs::get_build_log_stream))

30
hive-c0re/src/dashboard/state_snapshot.rs
View file

@ -117,33 +117,6 @@ pub(super) struct StateSnapshot {
/// `host_stats::server_warnings`; the frontend renders this list
/// generically, so new warning kinds need no frontend change.
server_warnings: Vec<crate::host_stats::ServerWarning>,
/// Live running/stopped status for the four hive infra containers
/// (`hive-ci`, `hive-forge`, `hive-gateway`, `hive-matrix`). Feeds the
/// C0R3 page's 1NFR4 sub-tab, the operator-only surface for starting
/// and stopping them.
infra_containers: Vec<InfraContainerView>,
}
/// One row for the C0R3 page's 1NFR4 sub-tab.
#[derive(Serialize)]
struct InfraContainerView {
/// Container / systemd-unit name (e.g. `"hive-ci"`).
name: &'static str,
running: bool,
}
/// Live running/stopped status for all four hive infra containers.
/// Extracted out of [`api_state`] to keep it under clippy's
/// `too_many_lines` limit.
async fn infra_container_views() -> Vec<InfraContainerView> {
let mut infra_containers = Vec::with_capacity(hive_priv_sock::InfraContainer::ALL.len());
for container in hive_priv_sock::InfraContainer::ALL {
infra_containers.push(InfraContainerView {
name: container.name(),
running: crate::lifecycle::infra_is_running(container).await,
});
}
infra_containers
}
#[derive(Serialize)]
@ -328,8 +301,6 @@ pub(super) async fn api_state(
w
};
let infra_containers = infra_container_views().await;
axum::Json(StateSnapshot {
seq,
hostname,
@ -368,7 +339,6 @@ pub(super) async fn api_state(
.ok()
.filter(|s| !s.is_empty()),
server_warnings,
infra_containers,
})
}

54
hive-c0re/src/dashboard_events.rs
View file

@ -13,17 +13,6 @@ use chrono::{DateTime, Utc};
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum DashboardEvent {
/// A new agent-initiated privileged action was recorded in the audit
/// log. The audit view (`/audit.html`) prepends `entry` live off
/// `/dashboard/stream` instead of polling. The `AuditEntry` fields
/// are flattened alongside the `kind` tag + `seq`, so the wire shape
/// matches one row of the `/api/audit-log` `entries` array exactly
/// (`{kind, seq, id, ts_unix, agent, action, target, outcome, detail}`).
AuditEntryAdded {
seq: u64,
#[serde(flatten)]
entry: crate::audit_log::AuditEntry,
},
/// Broker `Sent` event mirrored onto the dashboard channel.
/// `file_refs` carries every path-shaped token in `body` that
/// hive-c0re verified is a regular file under the allow-listed
@ -270,7 +259,6 @@ impl DashboardEvent {
DashboardEvent::SchedulesChanged { .. } => "schedules_changed",
DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed",
DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed",
DashboardEvent::AuditEntryAdded { .. } => "audit_entry_added",
}
}
}
@ -385,18 +373,6 @@ mod tests {
agents: Vec::new(),
effective: std::collections::BTreeMap::new(),
},
DashboardEvent::AuditEntryAdded {
seq: 1,
entry: crate::audit_log::AuditEntry {
id: 1,
ts_unix: hive_sh4re::wire_time::from_secs(0),
agent: "operator".into(),
action: "stop_infra".into(),
target: "hive-ci".into(),
outcome: "ok".into(),
detail: None,
},
},
];
for ev in samples {
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
@ -407,34 +383,4 @@ mod tests {
assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}");
}
}
/// The flattened `AuditEntry` fields must sit alongside `kind`/`seq`
/// at the top level (not nested under `entry`) so the wire shape
/// matches one `/api/audit-log` row — the audit view prepends it
/// directly.
#[test]
fn audit_entry_added_flattens_to_top_level() {
let ev = DashboardEvent::AuditEntryAdded {
seq: 7,
entry: crate::audit_log::AuditEntry {
id: 42,
ts_unix: hive_sh4re::wire_time::from_secs(1_700_000_000),
agent: "operator".into(),
action: "stop_infra".into(),
target: "hive-gateway".into(),
outcome: "err".into(),
detail: Some("systemctl stop failed".into()),
},
};
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
assert_eq!(v["kind"], "audit_entry_added");
assert_eq!(v["seq"], 7);
assert_eq!(v["id"], 42);
assert_eq!(v["agent"], "operator");
assert_eq!(v["target"], "hive-gateway");
assert_eq!(v["outcome"], "err");
assert_eq!(v["detail"], "systemctl stop failed");
// Not nested — there must be no `entry` sub-object.
assert!(v.get("entry").is_none());
}
}

15
hive-c0re/src/lifecycle/mod.rs
View file

@ -680,21 +680,6 @@ pub async fn is_running(name: &str) -> bool {
.is_ok_and(|s| s.success())
}
/// True when a hive infrastructure service's systemd unit is active.
/// Sibling of [`is_running`] for sub-agents, but infra names (`hive-ci`, …)
/// have no `h-` prefix to strip and are not all containers, so the unit
/// comes from the variant itself rather than from [`container_name`]. Used
/// by the dashboard C0R3 page's 1NFR4 sub-tab to show each one's live
/// status dot.
pub async fn infra_is_running(container: hive_priv_sock::InfraContainer) -> bool {
let unit = container.service_unit();
Command::new("systemctl")
.args(["is-active", "--quiet", &unit])
.status()
.await
.is_ok_and(|s| s.success())
}
/// Fully tear down a sub-agent's container: stop + remove via `nixos-container
/// destroy`, then clean our own systemd drop-in. Leaves it to the caller to
/// wipe `/var/lib/hyperhive/...` state and the per-agent runtime dir.

5
hive-c0re/src/main.rs
View file

@ -46,7 +46,7 @@ pub(crate) use agent_config::{capabilities, limits, resource_limits, tool_groups
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 stores::{approvals, broker, build_logs, db, power, scheduled_prompts};
pub(crate) use workers::{
agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker,
};
@ -518,9 +518,6 @@ async fn cmd_serve(
// 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.

329
hive-c0re/src/stores/audit_log.rs
View file

@ -1,329 +0,0 @@
//! Sqlite-backed audit trail of privileged actions worth a durable,
//! operator-visible who/what/when record beyond hive-priv's low-level
//! journal trace — currently the dashboard's operator-driven infra
//! container start/stop (`dashboard::infra_containers::post_infra_container`).
//!
//! Deliberately narrow: the bulk of `PrivRequest` traffic (token writes,
//! nspawn-flag edits) fires constantly during normal lifecycle and is
//! hive-c0re's own bookkeeping, not a privileged action worth a standalone
//! record — logging all of it would drown the signal the operator
//! actually wants. Nothing agent-initiated lands here today; the module
//! stays generic for whatever privileged action needs this record next.
//!
//! Same process-singleton handle pattern as `build_logs`: installed once
//! at `Coordinator::open`, and fetched by recording sites so they don't
//! have to thread an `Arc<AuditLog>` through every call path. Recording is
//! best-effort: a sqlite blip must never fail the underlying privileged
//! action.
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use rusqlite::{Connection, params};
use serde::Serialize;
use utoipa::ToSchema;
/// Process-singleton handle, set once at coordinator startup. Mirrors
/// `build_logs::GLOBAL` — lets recording sites write without threading an
/// `Arc<AuditLog>` through every entry point.
static GLOBAL: OnceLock<Arc<AuditLog>> = OnceLock::new();
/// Install the process-wide `AuditLog` handle. Idempotent: a second call
/// silently keeps the first handle.
pub fn install(handle: Arc<AuditLog>) {
let _ = GLOBAL.set(handle);
}
/// Retain audit rows for 90 days. Longer than build-log retention — this
/// is a security/accountability record, not debug noise; the operator may
/// want to review "who restarted what" well after the fact.
const KEEP_SECS: i64 = 90 * 24 * 3600;
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts_unix INTEGER NOT NULL,
agent TEXT NOT NULL,
action TEXT NOT NULL,
target TEXT NOT NULL,
outcome TEXT NOT NULL,
detail TEXT
);
CREATE INDEX IF NOT EXISTS idx_audit_log_ts ON audit_log (ts_unix DESC);
";
/// Outcome of a recorded privileged action. Stored as the literal string
/// in the `outcome` column.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AuditOutcome {
/// The privileged action succeeded.
Ok,
/// The privileged action was attempted but failed (e.g. the
/// underlying systemctl call errored). Denied-by-capability attempts
/// are recorded too — see the recording site.
Err,
}
impl AuditOutcome {
fn as_str(self) -> &'static str {
match self {
Self::Ok => "ok",
Self::Err => "err",
}
}
}
/// One audit row as returned to the dashboard.
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct AuditEntry {
pub id: i64,
pub ts_unix: DateTime<Utc>,
/// Actor who took the action (e.g. `"operator"`, or an agent name for
/// a future agent-initiated entry).
pub agent: String,
/// What was done (e.g. `stop_infra`).
pub action: String,
/// What it acted on (e.g. `hive-ci`).
pub target: String,
/// `"ok"` | `"err"`.
pub outcome: String,
/// Optional free-text detail (e.g. the error message on failure).
pub detail: Option<String>,
}
/// Sqlite-backed audit-log store. `Arc<AuditLog>`-friendly: all methods
/// take `&self`, an internal `Mutex<Connection>` serializes access.
pub struct AuditLog {
conn: Mutex<Connection>,
}
impl AuditLog {
/// Open (creating if absent) the `audit_log.sqlite` store under
/// `db_dir` and apply the schema. `db_dir` is shared with
/// `build_logs` (the broker db's parent directory).
///
/// # Errors
/// Returns an error if the directory can't be created, the sqlite
/// file can't be opened, or applying the schema fails.
pub fn open(db_dir: &Path) -> Result<Self> {
let path = db_dir.join("audit_log.sqlite");
let conn = crate::db::open(&path, "audit_log")?;
conn.execute_batch(SCHEMA)
.context("apply audit_log schema")?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Record one privileged action. Best-effort: a sqlite error is logged
/// but never returned, so a transient blip never fails the underlying
/// privileged action (the action already happened — losing its audit
/// row is strictly less bad than failing the action retroactively).
///
/// Returns the inserted [`AuditEntry`] (with its assigned id +
/// timestamp) on success, or `None` if the insert failed. The
/// returned row is the canonical record — callers that also push a
/// live event (e.g. the dashboard stream) emit *this* rather than
/// re-deriving the fields, so the stored row and the streamed event
/// can't drift.
#[must_use]
pub fn record(
&self,
agent: &str,
action: &str,
target: &str,
outcome: AuditOutcome,
detail: Option<&str>,
) -> Option<AuditEntry> {
let now = Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
match conn.execute(
"INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![now, agent, action, target, outcome.as_str(), detail],
) {
Ok(_) => Some(AuditEntry {
id: conn.last_insert_rowid(),
ts_unix: hive_sh4re::wire_time::from_secs(now),
agent: agent.to_owned(),
action: action.to_owned(),
target: target.to_owned(),
outcome: outcome.as_str().to_owned(),
detail: detail.map(str::to_owned),
}),
Err(e) => {
tracing::warn!(
%agent, %action, %target,
error = ?e,
"audit_log: record failed (dropping entry)"
);
None
}
}
}
/// Return the most recent `limit` rows, newest first. Limit is
/// hard-clamped to 500 to bound the worst-case payload.
///
/// # Errors
/// Returns an error if the query fails to prepare or a row fails to
/// deserialize.
pub fn list_recent(&self, limit: usize) -> Result<Vec<AuditEntry>> {
let limit = limit.min(500);
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, ts_unix, agent, action, target, outcome, detail
FROM audit_log
ORDER BY ts_unix DESC, id DESC
LIMIT ?1",
)?;
let rows = stmt.query_map(params![i64::try_from(limit).unwrap_or(500)], row_to_entry)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
/// Total row count, regardless of the `list_recent` clamp. Lets the
/// dashboard show "latest N of TOTAL" instead of silently capping.
///
/// # Errors
/// Returns an error if the `COUNT(*)` query fails.
pub fn count_total(&self) -> Result<i64> {
let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row("SELECT COUNT(*) FROM audit_log", [], |r| r.get(0))?;
Ok(n)
}
/// Drop rows older than the retention window. Returns the number of
/// rows deleted. Called from the hourly vacuum loop.
///
/// # Errors
/// Returns an error if the `DELETE` query fails.
pub fn vacuum(&self) -> Result<u64> {
let cutoff = Utc::now().timestamp() - KEEP_SECS;
let conn = self.conn.lock().unwrap();
let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?;
Ok(u64::try_from(removed).unwrap_or(0))
}
}
/// Spawn the hourly retention sweep. Mirrors `build_logs::spawn_vacuum`
/// in cadence + shutdown handling.
pub fn spawn_vacuum(coord: &Arc<crate::coordinator::Coordinator>) {
use std::time::Duration;
let audit = coord.audit_log.clone();
let mut shutdown = coord.shutdown_rx();
let interval = Duration::from_hours(1);
tokio::spawn(async move {
loop {
match audit.vacuum() {
Ok(0) => {}
Ok(n) => tracing::info!(removed = n, "audit_log vacuum"),
Err(e) => tracing::warn!(error = ?e, "audit_log vacuum failed"),
}
tokio::select! {
() = tokio::time::sleep(interval) => {}
_ = shutdown.changed() => {
tracing::info!("audit_log vacuum: shutdown signal received");
break;
}
}
}
});
}
fn row_to_entry(r: &rusqlite::Row) -> rusqlite::Result<AuditEntry> {
Ok(AuditEntry {
id: r.get(0)?,
ts_unix: hive_sh4re::wire_time::from_secs(r.get(1)?),
agent: r.get(2)?,
action: r.get(3)?,
target: r.get(4)?,
outcome: r.get(5)?,
detail: r.get(6)?,
})
}
#[cfg(test)]
mod tests {
use super::*;
fn tmpdb() -> (tempfile::TempDir, AuditLog) {
let dir = tempfile::tempdir().expect("tempdir");
let db = AuditLog::open(dir.path()).expect("open");
(dir, db)
}
#[test]
fn record_and_list_newest_first() {
let (_d, db) = tmpdb();
// record() returns the canonical inserted row (id + ts assigned).
let entry = db
.record("operator", "stop_infra", "hive-ci", AuditOutcome::Ok, None)
.expect("record returns the inserted entry");
assert!(entry.id > 0);
assert_eq!(entry.target, "hive-ci");
assert_eq!(entry.outcome, "ok");
assert!(entry.detail.is_none());
let _ = db.record(
"operator",
"stop_infra",
"hive-gateway",
AuditOutcome::Err,
Some("systemctl failed"),
);
let rows = db.list_recent(10).expect("list");
assert_eq!(rows.len(), 2);
// Newest first: the gateway/err row was inserted last.
assert_eq!(rows[0].target, "hive-gateway");
assert_eq!(rows[0].outcome, "err");
assert_eq!(rows[0].detail.as_deref(), Some("systemctl failed"));
assert_eq!(rows[1].target, "hive-ci");
assert_eq!(rows[1].outcome, "ok");
assert!(rows[1].detail.is_none());
assert_eq!(rows[0].agent, "operator");
assert_eq!(rows[0].action, "stop_infra");
assert_eq!(db.count_total().expect("count"), 2);
}
#[test]
fn list_clamps_to_500() {
let (_d, db) = tmpdb();
let _ = db.record("a", "x", "t", AuditOutcome::Ok, None);
let rows = db.list_recent(999_999).expect("list");
assert!(rows.len() <= 500);
}
#[test]
fn vacuum_drops_only_old_rows() {
let (_d, db) = tmpdb();
let _ = db.record("operator", "stop_infra", "hive-ci", AuditOutcome::Ok, None);
// Backdate it past the retention window.
{
let conn = db.conn.lock().unwrap();
conn.execute(
"UPDATE audit_log SET ts_unix = ?1",
params![Utc::now().timestamp() - KEEP_SECS - 60],
)
.unwrap();
}
let _ = db.record(
"operator",
"stop_infra",
"hive-forge",
AuditOutcome::Ok,
None,
);
let removed = db.vacuum().expect("vacuum");
assert_eq!(removed, 1, "only the backdated row should be reaped");
let rows = db.list_recent(10).expect("list");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].target, "hive-forge");
}
}

7
hive-c0re/src/stores/mod.rs
View file

@ -1,10 +1,9 @@
//! Sqlite-backed host-side stores (broker, approval / schedule queues,
//! build logs, audit trail, power intent) plus the shared connection
//! open/migration helper (`db`). Each submodule is re-exported at the
//! crate root, so `crate::broker::…` etc. keep working unchanged.
//! build logs, power intent) plus the shared connection open/migration
//! helper (`db`). Each submodule is re-exported at the crate root, so
//! `crate::broker::…` etc. keep working unchanged.
pub mod approvals;
pub mod audit_log;
pub mod broker;
pub mod build_logs;
pub mod db;