Fimeg/souveraine
Watch
1
0
Fork
You've already forked souveraine
 0
Code Releases Activity

instances(registry): per-process agent_instances table + uptime ticker

Browse source 
Schema (idempotent migration in server/db.rs::init_database):
- agent_instances table (agent_id, instance_id, pid, hostname,
  started_at, last_seen_at), primary key (agent_id, instance_id)
- lifetime_active_seconds INTEGER column on agents (added via
  add_column_if_missing — SQLite has no IF NOT EXISTS on columns,
  so PRAGMA table_info is checked first)
- idx_instances_agent index on (agent_id, last_seen_at DESC)

AgentInventory wires the lifecycle:
- register_instance(instance_id): prunes rows older than 5 min, then
  INSERTs (or upserts last_seen_at) for every known agent
- heartbeat_instance(instance_id, tick_seconds): UPDATEs last_seen_at
  and increments lifetime_active_seconds by tick_seconds for every
  agent this instance has rows for
- instance_count(agent_id): COUNT of live rows (>=5min recent)
- lifetime_active_seconds(agent_id): cumulative active seconds

SouveraineServer::new generates a process-lifetime instance_id (UUID),
calls register_instance once, then spawns a 30s heartbeat loop. Both
local CLI mode and full server mode get the same registration path
(the manager-mockup "N Instances" badge will count rows; uptime % is
int(100 * lifetime_active_seconds / age_seconds), capped at 99 in UI).

Verified end-to-end via `souveraine agents`: both Annie and Souveraine
rows land with matching instance_id, current pid, hostname.

Build clean. 111/111 tests pass.
This commit is contained in:
Fimeg 2026-05-12 13:59:42 -04:00
commit ba576f63f4
3 changed files with 171 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

93
src/server/agent_inventory.rs
View file

@ -6,6 +6,18 @@ use std::path::PathBuf;
use std::sync::Arc;
use uuid::Uuid;
fn hostname_or_unknown() -> String {
std::env::var("HOSTNAME")
.ok()
.or_else(|| {
std::fs::read_to_string("/etc/hostname")
.ok()
.map(|s| s.trim().to_string())
})
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "unknown".to_string())
}
pub struct AgentInventory {
/// Server-managed dir for `agent.json` + `conversations/`.
/// Layout: `~/.souveraine/server/agents/{id}/`.
@ -81,6 +93,87 @@ impl AgentInventory {
crate::core::identity::SeedId::load_or_generate(&self.seed_dir(agent_id))
}
/// Register an instance row for every known agent on this process.
/// Idempotent: re-running with the same `instance_id` overwrites the
/// last_seen_at. Stale rows (>5 min since last_seen) are pruned first
/// so the manager view doesn't surface dead processes.
pub async fn register_instance(&self, instance_id: &str) -> anyhow::Result<()> {
let pid = std::process::id() as i64;
let hostname = hostname_or_unknown();
// Prune dead instances first.
sqlx::query("DELETE FROM agent_instances WHERE last_seen_at < datetime('now', '-5 minutes')")
.execute(&self.db)
.await?;
let agents = self.list(None).await?;
for a in agents {
sqlx::query(
"INSERT INTO agent_instances (agent_id, instance_id, pid, hostname, started_at, last_seen_at)
VALUES (?1, ?2, ?3, ?4, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT(agent_id, instance_id) DO UPDATE SET
last_seen_at = CURRENT_TIMESTAMP,
pid = excluded.pid,
hostname = excluded.hostname"
)
.bind(&a.id)
.bind(instance_id)
.bind(pid)
.bind(&hostname)
.execute(&self.db)
.await?;
}
Ok(())
}
/// Heartbeat the instance rows for this process: bump last_seen_at and
/// add `tick_seconds` to each agent's lifetime_active_seconds. Called
/// from a 30s loop in SouveraineServer.
pub async fn heartbeat_instance(&self, instance_id: &str, tick_seconds: i64) -> anyhow::Result<()> {
// Bump last_seen for every row owned by this instance.
let updated = sqlx::query(
"UPDATE agent_instances SET last_seen_at = CURRENT_TIMESTAMP WHERE instance_id = ?1"
)
.bind(instance_id)
.execute(&self.db)
.await?;
if updated.rows_affected() == 0 {
return Ok(());
}
// Increment lifetime_active_seconds for every agent this instance is alive on.
sqlx::query(
"UPDATE agents
SET lifetime_active_seconds = lifetime_active_seconds + ?1
WHERE id IN (SELECT agent_id FROM agent_instances WHERE instance_id = ?2)"
)
.bind(tick_seconds)
.bind(instance_id)
.execute(&self.db)
.await?;
Ok(())
}
/// How many running instances does this agent currently have?
pub async fn instance_count(&self, agent_id: &str) -> anyhow::Result<i64> {
let (count,): (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM agent_instances WHERE agent_id = ?1
AND last_seen_at >= datetime('now', '-5 minutes')"
)
.bind(agent_id)
.fetch_one(&self.db)
.await?;
Ok(count)
}
/// Total active seconds (lifetime) for an agent, used to compute uptime %.
pub async fn lifetime_active_seconds(&self, agent_id: &str) -> anyhow::Result<i64> {
let (secs,): (i64,) = sqlx::query_as(
"SELECT COALESCE(lifetime_active_seconds, 0) FROM agents WHERE id = ?1"
)
.bind(agent_id)
.fetch_one(&self.db)
.await?;
Ok(secs)
}
pub async fn list(&self, filters: Option<String>) -> anyhow::Result<Vec<AgentSummary>> {
let query = if let Some(filter) = filters {
sqlx::query_as::<_, AgentSummaryRow>(

49
src/server/db.rs
View file

@ -73,5 +73,54 @@ pub async fn init_database(db_path: &Path) -> anyhow::Result<SqlitePool> {
.execute(&pool)
.await?;
// Instance registry — one row per running souveraine process per agent.
// last_seen_at is heartbeated; rows stale > 5 min are pruned on startup.
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS agent_instances (
agent_id TEXT NOT NULL,
instance_id TEXT NOT NULL,
pid INTEGER NOT NULL,
hostname TEXT NOT NULL,
started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_seen_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (agent_id, instance_id)
);
"#,
)
.execute(&pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_instances_agent ON agent_instances(agent_id, last_seen_at DESC)")
.execute(&pool)
.await?;
// lifetime_active_seconds: cumulative time this agent has had at least
// one running instance. Incremented by the heartbeat tick. Drives the
// uptime % on the manager card (capped at 99 in the UI).
add_column_if_missing(&pool, "agents", "lifetime_active_seconds", "INTEGER NOT NULL DEFAULT 0").await?;
Ok(pool)
}
/// SQLite-friendly idempotent column add. PRAGMA table_info is checked first
/// because SQLite has no `ALTER TABLE ... ADD COLUMN IF NOT EXISTS`.
async fn add_column_if_missing(
pool: &SqlitePool,
table: &str,
column: &str,
column_decl: &str,
) -> anyhow::Result<()> {
let rows: Vec<(i64, String)> = sqlx::query_as(
&format!("SELECT cid, name FROM pragma_table_info('{table}')"),
)
.fetch_all(pool)
.await?;
if rows.iter().any(|(_, name)| name == column) {
return Ok(());
}
sqlx::query(&format!("ALTER TABLE {table} ADD COLUMN {column} {column_decl}"))
.execute(pool)
.await?;
Ok(())
}

29
src/server/mod.rs
View file

@ -72,6 +72,35 @@ impl SouveraineServer {
}
}
// Instance registry: one row per (agent, process), heartbeated every
// 30s. Stale rows (>5min) pruned on registration. The instance_id is
// a fresh UUID this process keeps for its entire lifetime, so the
// heartbeat hits the right rows on every tick.
let instance_id = uuid::Uuid::new_v4().to_string();
if let Err(e) = agents.register_instance(&instance_id).await {
tracing::warn!("Instance registration failed (continuing): {}", e);
}
{
const TICK_SECONDS: i64 = 30;
let agents_for_tick = agents.clone();
let id_for_tick = instance_id.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(
std::time::Duration::from_secs(TICK_SECONDS as u64),
);
// Skip the immediate first tick so we wait a full interval
// before bumping lifetime_active_seconds (already registered
// at started_at).
interval.tick().await;
loop {
interval.tick().await;
if let Err(e) = agents_for_tick.heartbeat_instance(&id_for_tick, TICK_SECONDS).await {
tracing::warn!("instance heartbeat failed: {}", e);
}
}
});
}
let sessions = Arc::new(SessionManager::with_persistence(data_dir.join("agents")));
let primary = &config.bifrost.primary_model;