pub const METRICS_SCHEMA: &str = "1.2";
#[derive(Debug, Clone, Copy)]
pub enum RunOutcome {
Completed,
Failed,
Killed,
}
pub fn record_run_started() {
#[cfg(feature = "metrics")]
imp::REGISTRY.runs_started.fetch_add(1, Ordering::Relaxed);
}
pub fn record_run(outcome: RunOutcome) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_run(outcome);
#[cfg(not(feature = "metrics"))]
let _ = outcome;
}
pub fn record_run_status(status: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_run_status(status);
#[cfg(not(feature = "metrics"))]
let _ = status;
}
pub fn record_reaction() {
#[cfg(feature = "metrics")]
imp::REGISTRY.reactions.fetch_add(1, Ordering::Relaxed);
}
pub fn record_tokens(input: u64, output: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_tokens(input, output);
#[cfg(not(feature = "metrics"))]
let _ = (input, output);
}
pub fn record_restart_tripped() {
#[cfg(feature = "metrics")]
imp::REGISTRY
.restarts_tripped
.fetch_add(1, Ordering::Relaxed);
}
pub fn record_loop_step() {
#[cfg(feature = "metrics")]
imp::REGISTRY.loop_steps.fetch_add(1, Ordering::Relaxed);
}
pub fn record_refusal(reason: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_refusal(reason);
#[cfg(not(feature = "metrics"))]
let _ = reason;
}
pub fn record_limit_exceeded(limit: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_limit_exceeded(limit);
#[cfg(not(feature = "metrics"))]
let _ = limit;
}
pub fn record_subagent_spawned() {
#[cfg(feature = "metrics")]
imp::REGISTRY
.subagents_spawned
.fetch_add(1, Ordering::Relaxed);
}
pub fn record_subagent_exited(status: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_subagent_exited(status);
#[cfg(not(feature = "metrics"))]
let _ = status;
}
pub fn record_subagent_restart(reason: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_subagent_restart(reason);
#[cfg(not(feature = "metrics"))]
let _ = reason;
}
pub fn record_subagent_stuck_kill(signal: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_subagent_stuck_kill(signal);
#[cfg(not(feature = "metrics"))]
let _ = signal;
}
pub fn record_intel_call() {
#[cfg(feature = "metrics")]
imp::REGISTRY.intel_calls.fetch_add(1, Ordering::Relaxed);
}
pub fn record_intel_error(reason: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_intel_error(reason);
#[cfg(not(feature = "metrics"))]
let _ = reason;
}
pub fn record_mcp_connect_failure(server: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_mcp_connect_failure(server);
#[cfg(not(feature = "metrics"))]
let _ = server;
}
pub fn record_drain(phase: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_drain(phase);
#[cfg(not(feature = "metrics"))]
let _ = phase;
}
pub fn record_supervisor_restart() {
#[cfg(feature = "metrics")]
imp::REGISTRY
.supervisor_restarts
.fetch_add(1, Ordering::Relaxed);
}
pub fn record_reactor_stall() {
#[cfg(feature = "metrics")]
imp::REGISTRY.reactor_stalls.fetch_add(1, Ordering::Relaxed);
}
pub fn set_intel_up(up: bool) {
#[cfg(feature = "metrics")]
imp::REGISTRY
.intel_up
.store(u64::from(up), Ordering::Relaxed);
#[cfg(not(feature = "metrics"))]
let _ = up;
}
pub fn set_paused(on: bool) {
#[cfg(feature = "metrics")]
imp::REGISTRY.paused.store(u64::from(on), Ordering::Relaxed);
#[cfg(not(feature = "metrics"))]
let _ = on;
}
pub fn set_intel_all_down(on: bool) {
#[cfg(feature = "metrics")]
imp::REGISTRY
.intel_all_down
.store(u64::from(on), Ordering::Relaxed);
#[cfg(not(feature = "metrics"))]
let _ = on;
}
pub fn set_tree_shape(active: u64, depth: u64, breadth: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.set_tree_shape(active, depth, breadth);
#[cfg(not(feature = "metrics"))]
let _ = (active, depth, breadth);
}
pub fn set_reactive_backlog(pending: u64, inflight: u64, subscriptions: u64, lag_ms: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.set_reactive_backlog(pending, inflight, subscriptions, lag_ms);
#[cfg(not(feature = "metrics"))]
let _ = (pending, inflight, subscriptions, lag_ms);
}
pub fn record_config_reload(result: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_config_reload(result);
#[cfg(not(feature = "metrics"))]
let _ = result;
}
pub fn record_turn(kind: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_turn(kind);
#[cfg(not(feature = "metrics"))]
let _ = kind;
}
pub fn record_step(status: &str) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_step(status);
#[cfg(not(feature = "metrics"))]
let _ = status;
}
pub fn record_store_op(result: &str, latency_ms: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.record_store_op(result, latency_ms);
#[cfg(not(feature = "metrics"))]
let _ = (result, latency_ms);
}
pub fn set_inbox_pending(n: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.set_inbox_pending(n);
#[cfg(not(feature = "metrics"))]
let _ = n;
}
pub fn set_context_tokens(n: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY.set_context_tokens(n);
#[cfg(not(feature = "metrics"))]
let _ = n;
}
pub fn set_config_generation(generation: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY
.config_generation
.store(generation, Ordering::Relaxed);
#[cfg(not(feature = "metrics"))]
let _ = generation;
}
pub fn set_budget_tokens_remaining(remaining: u64) {
#[cfg(feature = "metrics")]
imp::REGISTRY
.budget_tokens_remaining
.store(remaining, Ordering::Relaxed);
#[cfg(not(feature = "metrics"))]
let _ = remaining;
}
pub fn set_pressure(level: u64, disk_free: Option<u64>) {
#[cfg(feature = "metrics")]
{
imp::REGISTRY.pressure_level.store(level, Ordering::Relaxed);
imp::REGISTRY
.disk_free_bytes
.store(disk_free.unwrap_or(u64::MAX), Ordering::Relaxed);
}
#[cfg(not(feature = "metrics"))]
let _ = (level, disk_free);
}
pub fn set_work_backlog(runs_active: u64, turns_queued: u64) {
#[cfg(feature = "metrics")]
{
imp::REGISTRY
.runs_active
.store(runs_active, Ordering::Relaxed);
imp::REGISTRY
.turns_queued
.store(turns_queued, Ordering::Relaxed);
}
#[cfg(not(feature = "metrics"))]
let _ = (runs_active, turns_queued);
}
#[cfg(feature = "metrics")]
pub fn render_prometheus() -> String {
let mut s = imp::REGISTRY.render();
s.push_str(&imp::memory_gauges(crate::supervisor::cgroup::snapshot()));
s
}
#[cfg(feature = "metrics")]
use std::sync::atomic::Ordering;
#[cfg(feature = "metrics")]
mod imp {
use super::RunOutcome;
use std::fmt::Write;
use std::sync::atomic::{AtomicU64, Ordering};
pub(super) static REGISTRY: Registry = Registry::new();
const STATUS_LABELS: &[&str] = &[
"completed",
"refused",
"exhausted_steps",
"exhausted_tokens",
"deadline",
"stalled",
"loop_detected",
"cancelled",
"crashed",
"other",
];
const REFUSAL_REASONS: &[&str] = &["trifecta", "rate", "budget", "depth", "mcp", "other"];
const LIMIT_LABELS: &[&str] = &[
"steps",
"tokens",
"deadline",
"depth",
"tree_tokens",
"tokens_lifetime",
"restart_storm",
"spawn_rate",
"other",
];
const RESTART_REASONS: &[&str] = &["crashed", "stuck", "rate", "other"];
const SIGNAL_LABELS: &[&str] = &["term", "kill", "other"];
const INTEL_ERROR_REASONS: &[&str] = &["unreachable", "auth", "timeout", "5xx", "other"];
const DRAIN_PHASES: &[&str] = &["started", "completed", "forced", "other"];
const TOKEN_TYPES: &[&str] = &["in", "out"];
const RELOAD_RESULTS: &[&str] = &["applied", "rejected", "other"];
const TURN_KINDS: &[&str] = &["root", "preflight", "compaction", "knowledge", "other"];
const STEP_STATUS: &[&str] = &["done", "failed", "skipped", "other"];
const STORE_RESULTS: &[&str] = &["ok", "conflict", "error", "other"];
struct LabelCounter<const N: usize> {
slots: [AtomicU64; N],
}
impl<const N: usize> LabelCounter<N> {
const fn new() -> Self {
LabelCounter {
slots: [const { AtomicU64::new(0) }; N],
}
}
fn inc(&self, domain: &[&str], value: &str) {
let idx = domain.iter().position(|&l| l == value).unwrap_or(N - 1);
self.slots[idx].fetch_add(1, Ordering::Relaxed);
}
}
pub(super) struct Registry {
pub(super) runs_started: AtomicU64,
runs_completed: AtomicU64,
runs_failed: AtomicU64,
runs_killed: AtomicU64,
pub(super) reactions: AtomicU64,
tokens_input: AtomicU64,
tokens_output: AtomicU64,
pub(super) restarts_tripped: AtomicU64,
runs_total: LabelCounter<{ STATUS_LABELS.len() }>,
pub(super) loop_steps: AtomicU64,
refusals: LabelCounter<{ REFUSAL_REASONS.len() }>,
limit_exceeded: LabelCounter<{ LIMIT_LABELS.len() }>,
active_subagents: AtomicU64,
tree_depth: AtomicU64,
tree_breadth: AtomicU64,
pub(super) subagents_spawned: AtomicU64,
subagents_exited: LabelCounter<{ STATUS_LABELS.len() }>,
subagent_restarts: LabelCounter<{ RESTART_REASONS.len() }>,
subagent_stuck_kills: LabelCounter<{ SIGNAL_LABELS.len() }>,
pub(super) intel_calls: AtomicU64,
pub(super) intel_up: AtomicU64,
pub(super) intel_all_down: AtomicU64,
intel_errors: LabelCounter<{ INTEL_ERROR_REASONS.len() }>,
pub(super) paused: AtomicU64,
mcp_connect_failures: LabelCounter<{ MCP_SERVER_SLOTS }>,
drains: LabelCounter<{ DRAIN_PHASES.len() }>,
pub(super) supervisor_restarts: AtomicU64,
pub(super) reactor_stalls: AtomicU64,
tokens_typed: LabelCounter<{ TOKEN_TYPES.len() }>,
pending_events: AtomicU64,
inflight_reactions: AtomicU64,
subscriptions_active: AtomicU64,
reaction_lag_ms: AtomicU64,
config_reloads: LabelCounter<{ RELOAD_RESULTS.len() }>,
pub(super) config_generation: AtomicU64,
pub(super) budget_tokens_remaining: AtomicU64,
pub(super) pressure_level: AtomicU64,
pub(super) disk_free_bytes: AtomicU64,
pub(super) runs_active: AtomicU64,
pub(super) turns_queued: AtomicU64,
turns_total: LabelCounter<{ TURN_KINDS.len() }>,
steps_total: LabelCounter<{ STEP_STATUS.len() }>,
store_ops: LabelCounter<{ STORE_RESULTS.len() }>,
store_latency_ms_sum: AtomicU64,
pub(super) inbox_pending: AtomicU64,
pub(super) context_tokens: AtomicU64,
}
impl Registry {
const fn new() -> Registry {
Registry {
runs_started: AtomicU64::new(0),
runs_completed: AtomicU64::new(0),
runs_failed: AtomicU64::new(0),
runs_killed: AtomicU64::new(0),
reactions: AtomicU64::new(0),
tokens_input: AtomicU64::new(0),
tokens_output: AtomicU64::new(0),
restarts_tripped: AtomicU64::new(0),
runs_total: LabelCounter::new(),
loop_steps: AtomicU64::new(0),
refusals: LabelCounter::new(),
limit_exceeded: LabelCounter::new(),
active_subagents: AtomicU64::new(0),
tree_depth: AtomicU64::new(0),
tree_breadth: AtomicU64::new(0),
subagents_spawned: AtomicU64::new(0),
subagents_exited: LabelCounter::new(),
subagent_restarts: LabelCounter::new(),
subagent_stuck_kills: LabelCounter::new(),
intel_calls: AtomicU64::new(0),
intel_up: AtomicU64::new(0),
intel_all_down: AtomicU64::new(0),
intel_errors: LabelCounter::new(),
paused: AtomicU64::new(0),
mcp_connect_failures: LabelCounter::new(),
drains: LabelCounter::new(),
supervisor_restarts: AtomicU64::new(0),
reactor_stalls: AtomicU64::new(0),
tokens_typed: LabelCounter::new(),
pending_events: AtomicU64::new(0),
inflight_reactions: AtomicU64::new(0),
subscriptions_active: AtomicU64::new(0),
reaction_lag_ms: AtomicU64::new(0),
config_reloads: LabelCounter::new(),
config_generation: AtomicU64::new(0),
budget_tokens_remaining: AtomicU64::new(0),
pressure_level: AtomicU64::new(0),
disk_free_bytes: AtomicU64::new(u64::MAX),
runs_active: AtomicU64::new(0),
turns_queued: AtomicU64::new(0),
turns_total: LabelCounter::new(),
steps_total: LabelCounter::new(),
store_ops: LabelCounter::new(),
store_latency_ms_sum: AtomicU64::new(0),
inbox_pending: AtomicU64::new(0),
context_tokens: AtomicU64::new(0),
}
}
pub(super) fn record_run(&self, outcome: RunOutcome) {
let c = match outcome {
RunOutcome::Completed => &self.runs_completed,
RunOutcome::Failed => &self.runs_failed,
RunOutcome::Killed => &self.runs_killed,
};
c.fetch_add(1, Ordering::Relaxed);
let status = match outcome {
RunOutcome::Completed => "completed",
RunOutcome::Failed => "crashed",
RunOutcome::Killed => "cancelled",
};
self.runs_total.inc(STATUS_LABELS, status);
}
pub(super) fn record_run_status(&self, status: &str) {
self.runs_total.inc(STATUS_LABELS, status);
}
pub(super) fn record_tokens(&self, input: u64, output: u64) {
self.tokens_input.fetch_add(input, Ordering::Relaxed);
self.tokens_output.fetch_add(output, Ordering::Relaxed);
self.tokens_typed.slots[0].fetch_add(input, Ordering::Relaxed);
self.tokens_typed.slots[1].fetch_add(output, Ordering::Relaxed);
}
pub(super) fn record_refusal(&self, reason: &str) {
self.refusals.inc(REFUSAL_REASONS, reason);
}
pub(super) fn record_limit_exceeded(&self, limit: &str) {
self.limit_exceeded.inc(LIMIT_LABELS, limit);
}
pub(super) fn record_subagent_exited(&self, status: &str) {
self.subagents_exited.inc(STATUS_LABELS, status);
}
pub(super) fn record_subagent_restart(&self, reason: &str) {
self.subagent_restarts.inc(RESTART_REASONS, reason);
}
pub(super) fn record_subagent_stuck_kill(&self, signal: &str) {
self.subagent_stuck_kills.inc(SIGNAL_LABELS, signal);
}
pub(super) fn record_intel_error(&self, reason: &str) {
self.intel_errors.inc(INTEL_ERROR_REASONS, reason);
}
pub(super) fn record_mcp_connect_failure(&self, server: &str) {
mcp_servers().record_failure(&self.mcp_connect_failures, server);
}
pub(super) fn record_drain(&self, phase: &str) {
self.drains.inc(DRAIN_PHASES, phase);
}
pub(super) fn record_config_reload(&self, result: &str) {
self.config_reloads.inc(RELOAD_RESULTS, result);
}
pub(super) fn record_turn(&self, kind: &str) {
self.turns_total.inc(TURN_KINDS, kind);
}
pub(super) fn record_step(&self, status: &str) {
self.steps_total.inc(STEP_STATUS, status);
}
pub(super) fn record_store_op(&self, result: &str, latency_ms: u64) {
self.store_ops.inc(STORE_RESULTS, result);
self.store_latency_ms_sum
.fetch_add(latency_ms, Ordering::Relaxed);
}
pub(super) fn set_inbox_pending(&self, n: u64) {
self.inbox_pending.store(n, Ordering::Relaxed);
}
pub(super) fn set_context_tokens(&self, n: u64) {
self.context_tokens.store(n, Ordering::Relaxed);
}
pub(super) fn set_tree_shape(&self, active: u64, depth: u64, breadth: u64) {
self.active_subagents.store(active, Ordering::Relaxed);
self.tree_depth.store(depth, Ordering::Relaxed);
self.tree_breadth.store(breadth, Ordering::Relaxed);
}
pub(super) fn set_reactive_backlog(
&self,
pending: u64,
inflight: u64,
subscriptions: u64,
lag_ms: u64,
) {
self.pending_events.store(pending, Ordering::Relaxed);
self.inflight_reactions.store(inflight, Ordering::Relaxed);
self.subscriptions_active
.store(subscriptions, Ordering::Relaxed);
self.reaction_lag_ms.store(lag_ms, Ordering::Relaxed);
}
pub(super) fn render(&self) -> String {
let mut s = String::new();
let g = |a: &AtomicU64| a.load(Ordering::Relaxed);
gauge(&mut s, "agent_up", "1 while the process is alive", 1);
let ready = u64::from(
!crate::signals::draining()
&& !crate::signals::lame_duck()
&& !crate::signals::intel_all_down(),
);
gauge(
&mut s,
"agent_ready",
"1 when ready to accept work (not draining / lame-ducked / intel-all-down)",
ready,
);
gauge(
&mut s,
"agent_paused",
"1 while the agentic tree is paused at turn boundaries",
g(&self.paused),
);
labelled_counter(
&mut s,
"agent_runs_total",
"Runs by terminal status.",
"status",
STATUS_LABELS,
&self.runs_total,
);
counter(
&mut s,
"agent_loop_steps_total",
"Agentic loop steps (process-local; emitted in the child loop, so the supervisor scrape reflects its own process only — there is no cross-process rollup).",
g(&self.loop_steps),
);
labelled_counter(
&mut s,
"agent_tokens_total",
"Model tokens by direction (the frozen `model` label is deferred in metrics_schema 1.0 — the AgentMsg::Usage hook carries no model id; never faked).",
"type",
TOKEN_TYPES,
&self.tokens_typed,
);
counter(
&mut s,
"agent_intel_calls_total",
"Intelligence calls made (process-local — the LLM client runs in the child; the frozen `model` label is deferred in metrics_schema 1.0, never faked).",
g(&self.intel_calls),
);
labelled_counter(
&mut s,
"agent_refusals_total",
"Refusals/guard trips by reason (process-local; tripped in the child loop, so the supervisor scrape reflects its own process only).",
"reason",
REFUSAL_REASONS,
&self.refusals,
);
labelled_counter(
&mut s,
"agent_limit_exceeded_total",
"Hard-bound trips by limit (the `tree_tokens` leg is supervisor-live; the steps/tokens/deadline/depth legs trip in the child loop and are process-local).",
"limit",
LIMIT_LABELS,
&self.limit_exceeded,
);
gauge(
&mut s,
"agent_active_subagents",
"Subagents currently alive in the tree.",
g(&self.active_subagents),
);
gauge(
&mut s,
"agent_tree_depth",
"Current max subagent-tree depth.",
g(&self.tree_depth),
);
gauge(
&mut s,
"agent_tree_breadth",
"Current max siblings at any tree node.",
g(&self.tree_breadth),
);
counter(
&mut s,
"agent_subagents_spawned_total",
"Subagents spawned.",
g(&self.subagents_spawned),
);
labelled_counter(
&mut s,
"agent_subagents_exited_total",
"Subagents exited by terminal status.",
"status",
STATUS_LABELS,
&self.subagents_exited,
);
labelled_counter(
&mut s,
"agent_subagent_restarts_total",
"Subagent restarts by reason.",
"reason",
RESTART_REASONS,
&self.subagent_restarts,
);
labelled_counter(
&mut s,
"agent_subagent_stuck_kills_total",
"Wedged-subagent kills by signal.",
"signal",
SIGNAL_LABELS,
&self.subagent_stuck_kills,
);
gauge(
&mut s,
"agent_intel_up",
"1 when the intelligence endpoint is reachable.",
g(&self.intel_up),
);
gauge(
&mut s,
"agent_intel_all_down",
"1 while all intelligence endpoints are down.",
g(&self.intel_all_down),
);
labelled_counter(
&mut s,
"agent_intel_errors_total",
"Intelligence-endpoint errors by reason.",
"reason",
INTEL_ERROR_REASONS,
&self.intel_errors,
);
mcp_servers().render_connect_failures(&mut s, &self.mcp_connect_failures);
reserved(
&mut s,
"agent_tool_calls_total",
"counter",
"Tool calls by server/tool/ok — reserved in metrics_schema 1.0; the tool-call boundary runs in the child loop, so a supervisor scrape can't reflect it (derive from tool.result log lines).",
);
reserved(
&mut s,
"agent_tool_call_duration_ms",
"histogram",
"Tool-call latency — reserved in metrics_schema 1.0; histogram exposition not implemented (use the tool.result dur_ms field).",
);
reserved(
&mut s,
"agent_intel_call_duration_ms",
"histogram",
"Intelligence-call latency — reserved in metrics_schema 1.0; histogram exposition not implemented (use the intel.result dur_ms field).",
);
reserved(
&mut s,
"agent_run_duration_ms",
"histogram",
"Run latency by terminal status — reserved in metrics_schema 1.0; histogram exposition not implemented (derive from run start→terminal log lines).",
);
labelled_counter(
&mut s,
"agent_drains_total",
"Drain phase transitions.",
"phase",
DRAIN_PHASES,
&self.drains,
);
counter(
&mut s,
"agent_restarts_total",
"Supervisor process restarts observed — reserved in metrics_schema 1.0; no in-process restart/reconcile emit site.",
g(&self.supervisor_restarts),
);
counter(
&mut s,
"agent_reactor_stalls_total",
"Wedged-reactor liveness trips — reserved in metrics_schema 1.0; the live signal is the /healthz 503, there is no one-shot in-process emit site.",
g(&self.reactor_stalls),
);
labelled_counter(
&mut s,
"agent_config_reload_total",
"Hot reloads by result.",
"result",
RELOAD_RESULTS,
&self.config_reloads,
);
gauge(
&mut s,
"agent_config_generation",
"Successfully-applied config reloads (the live generation).",
g(&self.config_generation),
);
gauge(
&mut s,
"agent_budget_tokens_remaining",
"Tokens left before the per-instance lifetime budget; 0 when unbounded or exhausted.",
g(&self.budget_tokens_remaining),
);
gauge(
&mut s,
"agent_pressure_level",
"Resource-pressure level: 0 ok, 1 warn, 2 shedding (admission stopped, in-flight drains).",
g(&self.pressure_level),
);
let free = g(&self.disk_free_bytes);
if free != u64::MAX {
gauge(
&mut s,
"agent_disk_free_bytes",
"Free bytes on the file store's filesystem (absent without a file store).",
free,
);
}
gauge(
&mut s,
"agent_runs_active",
"Workflow runs in a non-terminal state.",
g(&self.runs_active),
);
gauge(
&mut s,
"agent_turns_queued",
"Conversation turns queued for a dispatch slot.",
g(&self.turns_queued),
);
gauge(
&mut s,
"agent_pending_events",
"Reactive events received but not yet routed.",
g(&self.pending_events),
);
gauge(
&mut s,
"agent_inflight_reactions",
"Reactions currently executing.",
g(&self.inflight_reactions),
);
gauge(
&mut s,
"agent_subscriptions_active",
"Reconciled declared subscriptions.",
g(&self.subscriptions_active),
);
gauge(
&mut s,
"agent_reaction_lag_ms",
"Age of the oldest un-routed pending event (ms).",
g(&self.reaction_lag_ms),
);
counter(
&mut s,
"agent_runs_started_total",
"Supervised runs started",
g(&self.runs_started),
);
counter(
&mut s,
"agent_runs_completed_total",
"Supervised runs that completed",
g(&self.runs_completed),
);
counter(
&mut s,
"agent_runs_failed_total",
"Supervised runs that failed on infra",
g(&self.runs_failed),
);
counter(
&mut s,
"agent_runs_killed_total",
"Supervised runs torn down by the supervisor",
g(&self.runs_killed),
);
counter(
&mut s,
"agent_reactions_total",
"Reactive triggers fired",
g(&self.reactions),
);
counter(
&mut s,
"agent_tokens_input_total",
"Input tokens reported by direct children",
g(&self.tokens_input),
);
counter(
&mut s,
"agent_tokens_output_total",
"Output tokens reported by direct children",
g(&self.tokens_output),
);
counter(
&mut s,
"agent_restarts_tripped_total",
"Restart-governor breaker trips",
g(&self.restarts_tripped),
);
labelled_counter(
&mut s,
"agent_turns_total",
"Turn-worker runs by context kind.",
"kind",
TURN_KINDS,
&self.turns_total,
);
labelled_counter(
&mut s,
"agent_steps_total",
"Workflow steps by terminal status.",
"status",
STEP_STATUS,
&self.steps_total,
);
labelled_counter(
&mut s,
"agent_store_ops_total",
"Remote-store ops by result.",
"result",
STORE_RESULTS,
&self.store_ops,
);
counter(
&mut s,
"agent_store_latency_ms_sum",
"Cumulative remote-store op latency (ms); divide by agent_store_ops_total for the mean.",
g(&self.store_latency_ms_sum),
);
gauge(
&mut s,
"agent_inbox_pending",
"Durable inbox events awaiting processing.",
g(&self.inbox_pending),
);
gauge(
&mut s,
"agent_context_tokens",
"Estimated token size of the largest live conversation context.",
g(&self.context_tokens),
);
s
}
}
const MCP_SERVER_SLOTS: usize = 16;
struct McpServerTable {
names: std::sync::Mutex<Vec<String>>,
}
impl McpServerTable {
const fn new() -> Self {
McpServerTable {
names: std::sync::Mutex::new(Vec::new()),
}
}
fn index(&self, server: &str) -> usize {
let mut names = match self.names.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
if let Some(i) = names.iter().position(|n| n == server) {
return i;
}
if names.len() < MCP_SERVER_SLOTS - 1 {
names.push(server.to_string());
return names.len() - 1;
}
MCP_SERVER_SLOTS - 1
}
fn record_failure(&self, ctr: &LabelCounter<MCP_SERVER_SLOTS>, server: &str) {
let idx = self.index(server);
ctr.slots[idx].fetch_add(1, Ordering::Relaxed);
}
fn render_connect_failures(&self, s: &mut String, ctr: &LabelCounter<MCP_SERVER_SLOTS>) {
let names = match self.names.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
let name = "agent_mcp_connect_failures_total";
let _ = writeln!(s, "# HELP {name} MCP connect failures by server.");
let _ = writeln!(s, "# TYPE {name} counter");
for (i, server) in names.iter().enumerate() {
let v = ctr.slots[i].load(Ordering::Relaxed);
let _ = writeln!(s, "{name}{{server={:?}}} {v}", server.as_str());
}
let other = ctr.slots[MCP_SERVER_SLOTS - 1].load(Ordering::Relaxed);
if other != 0 {
let _ = writeln!(s, "{name}{{server=\"other\"}} {other}");
}
}
}
fn mcp_servers() -> &'static McpServerTable {
static TABLE: McpServerTable = McpServerTable::new();
&TABLE
}
fn counter(s: &mut String, name: &str, help: &str, value: u64) {
let _ = writeln!(s, "# HELP {name} {help}");
let _ = writeln!(s, "# TYPE {name} counter");
let _ = writeln!(s, "{name} {value}");
}
fn reserved(s: &mut String, name: &str, kind: &str, help: &str) {
let _ = writeln!(s, "# HELP {name} {help}");
let _ = writeln!(s, "# TYPE {name} {kind}");
}
fn gauge(s: &mut String, name: &str, help: &str, value: u64) {
let _ = writeln!(s, "# HELP {name} {help}");
let _ = writeln!(s, "# TYPE {name} gauge");
let _ = writeln!(s, "{name} {value}");
}
fn labelled_counter<const N: usize>(
s: &mut String,
name: &str,
help: &str,
label: &str,
domain: &[&str],
ctr: &LabelCounter<N>,
) {
let _ = writeln!(s, "# HELP {name} {help}");
let _ = writeln!(s, "# TYPE {name} counter");
for (i, value) in domain.iter().enumerate() {
let v = ctr.slots[i].load(Ordering::Relaxed);
let _ = writeln!(s, "{name}{{{label}={value:?}}} {v}");
}
}
pub(super) fn memory_gauges(mem: crate::supervisor::cgroup::MemorySnapshot) -> String {
let mut s = String::new();
if let Some(v) = mem.max {
gauge(
&mut s,
"agent_memory_max_bytes",
"cgroup v2 memory.max hard limit (bytes)",
v,
);
}
if let Some(v) = mem.current {
gauge(
&mut s,
"agent_memory_current_bytes",
"cgroup v2 memory.current usage (bytes)",
v,
);
}
s
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn render_is_valid_prometheus_text() {
let r = Registry::new();
r.runs_started.fetch_add(3, Ordering::Relaxed);
r.record_run(RunOutcome::Completed);
r.record_run(RunOutcome::Failed);
r.record_tokens(100, 50);
let out = r.render();
assert!(out.contains("# TYPE agent_runs_started_total counter"));
assert!(out.contains("agent_runs_started_total 3"));
assert!(out.contains("agent_runs_completed_total 1"));
assert!(out.contains("agent_runs_failed_total 1"));
assert!(out.contains("agent_tokens_input_total 100"));
assert!(out.contains("agent_tokens_output_total 50"));
}
#[test]
fn pressure_gauges_emit_and_disk_free_is_absent_until_known() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_pressure_level gauge"));
assert!(out.contains("agent_pressure_level 0"));
assert!(out.contains("agent_runs_active 0"));
assert!(out.contains("agent_turns_queued 0"));
assert!(!out.contains("agent_disk_free_bytes"));
r.pressure_level.store(2, Ordering::Relaxed);
r.disk_free_bytes.store(123_456, Ordering::Relaxed);
r.runs_active.store(3, Ordering::Relaxed);
r.turns_queued.store(7, Ordering::Relaxed);
let out = r.render();
assert!(out.contains("agent_pressure_level 2"));
assert!(out.contains("agent_disk_free_bytes 123456"));
assert!(out.contains("agent_runs_active 3"));
assert!(out.contains("agent_turns_queued 7"));
}
#[test]
fn frozen_schema_emits_up_and_ready_gauges() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_up gauge"));
assert!(out.contains("agent_up 1"));
assert!(out.contains("# TYPE agent_ready gauge"));
assert!(out.contains("agent_ready "));
}
#[test]
fn paused_gauge_renders_zero_then_one() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_paused gauge"));
assert!(out.contains("agent_paused 0"));
r.paused.store(1, Ordering::Relaxed);
assert!(r.render().contains("agent_paused 1"));
}
#[test]
fn intel_all_down_gauge_renders_zero_then_one() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_intel_all_down gauge"));
assert!(out.contains("agent_intel_all_down 0"));
r.intel_all_down.store(1, Ordering::Relaxed);
assert!(r.render().contains("agent_intel_all_down 1"));
}
#[test]
fn runs_total_uses_the_closed_status_domain() {
let r = Registry::new();
r.record_run_status("completed");
r.record_run_status("refused");
r.record_run_status("refused");
r.record_run_status("totally_made_up");
let out = r.render();
assert!(out.contains("agent_runs_total{status=\"completed\"} 1"));
assert!(out.contains("agent_runs_total{status=\"refused\"} 2"));
assert!(out.contains("agent_runs_total{status=\"other\"} 1"));
assert!(out.contains("agent_runs_total{status=\"loop_detected\"} 0"));
assert_eq!(out.matches("# TYPE agent_runs_total counter").count(), 1);
}
#[test]
fn typed_tokens_track_direction() {
let r = Registry::new();
r.record_tokens(880, 40);
r.record_tokens(120, 10);
let out = r.render();
assert!(out.contains("agent_tokens_total{type=\"in\"} 1000"));
assert!(out.contains("agent_tokens_total{type=\"out\"} 50"));
}
#[test]
fn refusals_and_limits_use_closed_domains() {
let r = Registry::new();
r.record_refusal("trifecta");
r.record_refusal("depth");
r.record_refusal("depth");
r.record_limit_exceeded("spawn_rate");
let out = r.render();
assert!(out.contains("agent_refusals_total{reason=\"trifecta\"} 1"));
assert!(out.contains("agent_refusals_total{reason=\"depth\"} 2"));
assert!(out.contains("agent_limit_exceeded_total{limit=\"spawn_rate\"} 1"));
r.record_refusal("nope");
assert!(
r.render()
.contains("agent_refusals_total{reason=\"other\"} 1")
);
}
#[test]
fn tree_and_backlog_gauges_are_settable() {
let r = Registry::new();
r.set_tree_shape(4, 2, 3);
r.set_reactive_backlog(7, 1, 9, 250);
let out = r.render();
assert!(out.contains("agent_active_subagents 4"));
assert!(out.contains("agent_tree_depth 2"));
assert!(out.contains("agent_tree_breadth 3"));
assert!(out.contains("agent_pending_events 7"));
assert!(out.contains("agent_inflight_reactions 1"));
assert!(out.contains("agent_subscriptions_active 9"));
assert!(out.contains("agent_reaction_lag_ms 250"));
}
#[test]
fn mcp_connect_failures_label_by_server_and_fold_overflow() {
let r = Registry::new();
r.record_mcp_connect_failure("github");
r.record_mcp_connect_failure("github");
r.record_mcp_connect_failure("filesystem");
let out = r.render();
assert!(out.contains("agent_mcp_connect_failures_total{server=\"github\"} 2"));
assert!(out.contains("agent_mcp_connect_failures_total{server=\"filesystem\"} 1"));
}
#[test]
fn drains_phase_distinguishes_clean_from_forced() {
let r = Registry::new();
r.record_drain("started");
r.record_drain("completed");
r.record_drain("forced");
let out = r.render();
assert!(out.contains("agent_drains_total{phase=\"completed\"} 1"));
assert!(out.contains("agent_drains_total{phase=\"forced\"} 1"));
}
#[test]
fn config_reload_total_renders_both_label_values_and_generation() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_config_reload_total counter"));
assert!(out.contains("agent_config_reload_total{result=\"applied\"} 0"));
assert!(out.contains("agent_config_reload_total{result=\"rejected\"} 0"));
assert!(out.contains("# TYPE agent_config_generation gauge"));
assert!(out.contains("agent_config_generation 0"));
r.record_config_reload("applied");
r.record_config_reload("rejected");
r.record_config_reload("rejected");
r.record_config_reload("totally_made_up");
r.config_generation.store(1, Ordering::Relaxed);
let out = r.render();
assert!(out.contains("agent_config_reload_total{result=\"applied\"} 1"));
assert!(out.contains("agent_config_reload_total{result=\"rejected\"} 2"));
assert!(out.contains("agent_config_reload_total{result=\"other\"} 1"));
assert!(out.contains("agent_config_generation 1"));
assert_eq!(
out.matches("# TYPE agent_config_reload_total counter")
.count(),
1
);
}
#[test]
fn budget_gauge_and_lifetime_limit_render() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_budget_tokens_remaining gauge"));
assert!(out.contains("agent_budget_tokens_remaining 0"));
r.budget_tokens_remaining.store(1500, Ordering::Relaxed);
r.record_limit_exceeded("tokens_lifetime");
let out = r.render();
assert!(out.contains("agent_budget_tokens_remaining 1500"));
assert!(out.contains("agent_limit_exceeded_total{limit=\"tokens_lifetime\"} 1"));
}
#[test]
fn no_unbounded_identifier_labels_leak() {
let r = Registry::new();
r.record_run_status("completed");
r.record_tokens(1, 1);
r.record_refusal("trifecta");
r.record_mcp_connect_failure("github");
let out = r.render();
for forbidden in [
"run_id=",
"agent_id=",
"agent_path=",
"call_id=",
"session_id=",
"uri=",
] {
assert!(
!out.contains(forbidden),
"leaked unbounded label: {forbidden}"
);
}
}
#[test]
fn memory_gauges_emit_only_present_fields() {
use crate::supervisor::cgroup::MemorySnapshot;
let g = memory_gauges(MemorySnapshot {
max: Some(1024),
current: Some(512),
high: None,
});
assert!(g.contains("# TYPE agent_memory_max_bytes gauge"));
assert!(g.contains("agent_memory_max_bytes 1024"));
assert!(g.contains("agent_memory_current_bytes 512"));
assert_eq!(g.matches(" gauge\n").count(), 2);
assert!(memory_gauges(MemorySnapshot::default()).is_empty());
}
#[test]
fn frozen_schema_4_3_series_all_present_emitted_or_reserved() {
let r = Registry::new();
let out = r.render();
for name in [
"agent_up",
"agent_ready",
"agent_runs_total",
"agent_run_duration_ms", "agent_loop_steps_total",
"agent_tokens_total",
"agent_intel_calls_total",
"agent_intel_call_duration_ms", "agent_refusals_total",
"agent_limit_exceeded_total",
"agent_active_subagents",
"agent_tree_depth",
"agent_tree_breadth",
"agent_subagents_spawned_total",
"agent_subagents_exited_total",
"agent_subagent_restarts_total",
"agent_subagent_stuck_kills_total",
"agent_intel_up",
"agent_intel_errors_total",
"agent_mcp_connect_failures_total",
"agent_tool_calls_total",
"agent_tool_call_duration_ms", "agent_drains_total",
"agent_restarts_total", "agent_reactor_stalls_total", "agent_pending_events",
"agent_inflight_reactions",
"agent_subscriptions_active",
"agent_reaction_lag_ms",
] {
assert!(
out.contains(&format!("# TYPE {name} ")),
"frozen series missing from render: {name}"
);
}
for reserved in [
"agent_run_duration_ms",
"agent_intel_call_duration_ms",
"agent_tool_call_duration_ms",
"agent_tool_calls_total",
] {
assert!(
out.contains(&format!("# TYPE {reserved} ")),
"reserved series marker missing: {reserved}"
);
for line in out.lines() {
if line.starts_with('#') {
continue;
}
assert!(
!line.starts_with(reserved),
"reserved series {reserved} must not emit a sample line: {line:?}"
);
}
}
assert!(out.contains("reserved in metrics_schema 1.0"));
}
#[test]
fn wired_supervisor_counters_increment() {
let r = Registry::new();
r.subagents_spawned.fetch_add(1, Ordering::Relaxed);
r.record_subagent_exited("completed");
r.record_subagent_exited("cancelled");
r.record_subagent_stuck_kill("term");
r.record_subagent_stuck_kill("kill");
r.record_drain("started");
r.record_drain("completed");
r.record_drain("forced");
r.record_subagent_restart("crashed");
r.record_mcp_connect_failure("github");
r.record_limit_exceeded("tree_tokens");
let out = r.render();
assert!(out.contains("agent_subagents_spawned_total 1"));
assert!(out.contains("agent_subagents_exited_total{status=\"completed\"} 1"));
assert!(out.contains("agent_subagents_exited_total{status=\"cancelled\"} 1"));
assert!(out.contains("agent_subagent_stuck_kills_total{signal=\"term\"} 1"));
assert!(out.contains("agent_subagent_stuck_kills_total{signal=\"kill\"} 1"));
assert!(out.contains("agent_drains_total{phase=\"started\"} 1"));
assert!(out.contains("agent_drains_total{phase=\"completed\"} 1"));
assert!(out.contains("agent_drains_total{phase=\"forced\"} 1"));
assert!(out.contains("agent_subagent_restarts_total{reason=\"crashed\"} 1"));
assert!(out.contains("agent_mcp_connect_failures_total{server=\"github\"} 1"));
assert!(out.contains("agent_limit_exceeded_total{limit=\"tree_tokens\"} 1"));
}
#[test]
fn reserved_no_emit_counters_render_zero() {
let r = Registry::new();
let out = r.render();
assert!(out.contains("# TYPE agent_restarts_total counter"));
assert!(out.contains("agent_restarts_total 0"));
assert!(out.contains("# TYPE agent_reactor_stalls_total counter"));
assert!(out.contains("agent_reactor_stalls_total 0"));
assert!(out.matches("reserved in metrics_schema 1.0").count() >= 2);
}
}
}