agentd/obs/

metrics.rs

// SPDX-License-Identifier: AGPL-3.0-only
//! Process-local counters → Prometheus text. [feature: metrics]
//!
//! Off the default path: the public `record_*` fns are **no-ops unless built
//! with `--features metrics`**, so call sites stay clean and the default build
//! pays nothing (metrics are otherwise derivable from the JSON-lines event
//! stream — that is the default story). With the feature, a tiny dependency-free
//! atomic registry backs an opt-in HTTP `/metrics` scrape surface (`obs::serve`).
//!
//! Counters are **per supervisor process**. The long-lived root daemon's surface
//! reflects the runs it supervises — every one-shot, reaction, and scheduled fire
//! flows through `supervise_once` — plus the tokens its *direct* children report
//! up the control channel. Nested subagents keep their own (process-local)
//! counters, still visible in their logs; cross-process metric rollup is a
//! deliberate non-goal (the same process boundary the tree token ceiling draws).
//!
//! ## The frozen `metrics_schema` contract
//!
//! The metric **names** and label **keys** below are a versioned public API
//! ([`METRICS_SCHEMA`]) that a control plane (agentctl) authors dashboards,
//! alerts and scalers against. Exposition is hand-written Prometheus 0.0.4 text
//! — no `prometheus`/`metrics` crate. The enumerated set *is* the contract: it is
//! additive within a major, and removing or renaming a metric or a label key
//! bumps the major.
//!
//! **Cardinality is binding:** `/metrics` is unauthenticated and may be bound on
//! all interfaces. Labels carry **bounded** values only (`status`, `model`,
//! `type`, `server`, `tool`, `reason`, `limit`, `signal`, `phase`, `ok`);
//! **never** `run_id` / `agent_id` / `agent_path` / `call_id` / a resource URI —
//! those are unbounded and live in logs and traces only. A control plane that
//! needs per-run granularity reads the run report or the event stream, never a
//! metric. This module therefore stores label-bearing series as small
//! **fixed-domain** atomic arrays (the closed label set is known at compile
//! time), so the cardinality is structurally bounded.
//!
//! Telemetry never crashes the agent: every fn here is a plain atomic add/store
//! that cannot fail; `render` only ever reads.

/// Frozen metrics-schema version. Surfaced in the manifest at
/// `surfaces.metrics_schema`; the integrator wires that surface — this const is
/// the single source of truth for the value. Additive series and label values
/// bump the minor; a removed or renamed metric or label key bumps the major.
///
/// Minors are additive, so a consumer written against an earlier minor still
/// parses a later render. Over 1.0, minor 1.1 carries the
/// `agent_budget_tokens_remaining` gauge and the `tokens_lifetime` value of the
/// `agent_limit_exceeded_total{limit}` domain; 1.2 carries the resource-pressure
/// set on top of that — `agent_pressure_level` (0 ok / 1 warn / 2 shed),
/// `agent_disk_free_bytes` (file-store filesystem headroom; absent without a
/// file store), `agent_runs_active` and `agent_turns_queued`.
pub const METRICS_SCHEMA: &str = "1.2";

/// Terminal disposition of one supervised run.
#[derive(Debug, Clone, Copy)]
pub enum RunOutcome {
    Completed,
    Failed,
    Killed,
}

/// A supervised run began (`supervise_once` entry).
pub fn record_run_started() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.runs_started.fetch_add(1, Ordering::Relaxed);
}

/// A supervised run reached a terminal disposition.
///
/// Also increments the frozen `agent_runs_total{status}` under a **coarse**
/// status projection of the three `RunOutcome` variants this call site carries
/// (`completed` / `crashed` / `cancelled`). The precise terminal-status string is
/// available at the loop boundary but not at this supervisor hook, so a caller
/// holding a `TerminalStatus` should use [`record_run_status`] instead to get the
/// full closed-vocabulary label domain.
pub fn record_run(outcome: RunOutcome) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_run(outcome);
    #[cfg(not(feature = "metrics"))]
    let _ = outcome;
}

/// A supervised run reached a terminal status — the **frozen**, precise form.
///
/// `status` MUST be a closed-vocabulary terminal-status string
/// ([`crate::agentloop::stop::TerminalStatus::as_str`]); an out-of-vocabulary
/// value is bucketed under `other` so the label domain stays closed and the
/// cardinality bounded. This is the precise driver for `agent_runs_total{status}`
/// and is called where the `TerminalStatus` is known — not from a supervisor
/// hook, which only holds the coarse `RunOutcome`.
pub fn record_run_status(status: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_run_status(status);
    #[cfg(not(feature = "metrics"))]
    let _ = status;
}

/// A reactive trigger fired (one reaction).
pub fn record_reaction() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.reactions.fetch_add(1, Ordering::Relaxed);
}

/// Tokens reported up by a direct child (`AgentMsg::Usage`).
///
/// Feeds both the bare `agent_tokens_{input,output}_total` and the frozen
/// `agent_tokens_total{type}`. The frozen schema also reserves a `model` label;
/// the `AgentMsg::Usage` control-channel message this call site rides does not
/// carry the model, so the label is left absent rather than faked — populating
/// it needs a call site at the intelligence boundary.
pub fn record_tokens(input: u64, output: u64) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_tokens(input, output);
    #[cfg(not(feature = "metrics"))]
    let _ = (input, output);
}

/// The restart governor's circuit breaker tripped.
pub fn record_restart_tripped() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY
        .restarts_tripped
        .fetch_add(1, Ordering::Relaxed);
}

/// One loop step executed (`loop.step`). Drives `agent_loop_steps_total`.
///
/// **Process-local / unwired:** `loop.step` is emitted inside the re-exec'd child
/// agentic loop, a different process from the supervisor that `/metrics` scrapes,
/// so calling this would only bump the child's own registry — there is no
/// cross-process rollup. It is therefore intentionally NOT called from the loop;
/// the series renders the supervisor's own process only and agentctl derives step
/// counts from `loop.step` log lines.
pub fn record_loop_step() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.loop_steps.fetch_add(1, Ordering::Relaxed);
}

/// A refusal / guard trip by reason (drives `agent_refusals_total`).
///
/// `reason` is the closed domain (`trifecta`/`rate`/`budget`/`depth`/`mcp`); an
/// unknown value buckets under `other`.
///
/// **Process-local / unwired:** refusals trip inside the re-exec'd child loop
/// (the orchestrator self-tool / scope checks), so a bump here would only reach
/// the child's process-local registry, never the supervisor scrape — there is no
/// cross-process rollup. Intentionally not called; the headline safety signal is
/// the refusal / `scope.trifecta_refused` log line.
pub fn record_refusal(reason: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_refusal(reason);
    #[cfg(not(feature = "metrics"))]
    let _ = reason;
}

/// A hard bound trip (`limit.exceeded`).
///
/// `limit` is the closed domain (`steps`/`tokens`/`deadline`/`depth`/
/// `tree_tokens`/`restart_storm`/`spawn_rate`); an unknown value buckets under
/// `other`.
///
/// **Partially wired:** the `tree_tokens` leg is the supervisor's own tree-ceiling
/// trip (`supervisor::reactor`, this process), so it is live and reaches the
/// scrape. The `steps`/`tokens`/`deadline`/`depth` legs trip inside the re-exec'd
/// child loop and are therefore process-local (not called from the child for the
/// scrape — derive those from `limit.exceeded` log lines).
pub fn record_limit_exceeded(limit: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_limit_exceeded(limit);
    #[cfg(not(feature = "metrics"))]
    let _ = limit;
}

/// A subagent was spawned (`subagent.spawn`).
pub fn record_subagent_spawned() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY
        .subagents_spawned
        .fetch_add(1, Ordering::Relaxed);
}

/// A subagent exited with a terminal `status` (`subagent.exit`). Drives
/// `agent_subagents_exited_total{status}` over the closed status vocabulary.
pub fn record_subagent_exited(status: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_subagent_exited(status);
    #[cfg(not(feature = "metrics"))]
    let _ = status;
}

/// A subagent was restarted by the governor (`subagent.restart`). Drives
/// `agent_subagent_restarts_total{reason}`.
pub fn record_subagent_restart(reason: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_subagent_restart(reason);
    #[cfg(not(feature = "metrics"))]
    let _ = reason;
}

/// A wedged/stuck subagent was killed (`subagent.stuck` — the reliability
/// headline). Drives `agent_subagent_stuck_kills_total{signal}`; `signal` ∈
/// `term`\|`kill` (an unknown value buckets `other`).
pub fn record_subagent_stuck_kill(signal: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_subagent_stuck_kill(signal);
    #[cfg(not(feature = "metrics"))]
    let _ = signal;
}

/// An intelligence call was made (`intel.call`). Drives
/// `agent_intel_calls_total`.
pub fn record_intel_call() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.intel_calls.fetch_add(1, Ordering::Relaxed);
}

/// An intelligence-endpoint error by reason. Drives
/// `agent_intel_errors_total`. `reason` ∈ `unreachable`\|`auth`\|`timeout`\|
/// `5xx` (an unknown value buckets `other`).
pub fn record_intel_error(reason: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_intel_error(reason);
    #[cfg(not(feature = "metrics"))]
    let _ = reason;
}

/// An MCP connect attempt failed for a declared `server` (`mcp.connect.fail`).
/// Drives `agent_mcp_connect_failures_total`. `server` is the declared server
/// name — bounded, because the declared set is small and fixed at config time;
/// an over-capacity name buckets under `other` so the series stays bounded.
pub fn record_mcp_connect_failure(server: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_mcp_connect_failure(server);
    #[cfg(not(feature = "metrics"))]
    let _ = server;
}

/// A drain phase transition. Drives `agent_drains_total`. `phase` ∈
/// `started`\|`completed`\|`forced` (an unknown value buckets `other`).
pub fn record_drain(phase: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_drain(phase);
    #[cfg(not(feature = "metrics"))]
    let _ = phase;
}

/// A supervisor process restart was observed (rebuild + reconcile). Drives
/// `agent_restarts_total` — distinct from the breaker-trip counter
/// [`record_restart_tripped`].
///
/// **Reserved / unwired in metrics_schema 1.0:** there is no in-process
/// rebuild+reconcile restart path to call it from, because a pod restart is a
/// fresh process with a zeroed registry — an orchestrator counts those, not the
/// binary. The series renders (always 0) so the frozen contract stays
/// discoverable; this fn is the hook a reconcile path would call.
pub fn record_supervisor_restart() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY
        .supervisor_restarts
        .fetch_add(1, Ordering::Relaxed);
}

/// A wedged-reactor liveness trip. Drives `agent_reactor_stalls_total`.
///
/// **Reserved / unwired in metrics_schema 1.0:** a wedged reactor is surfaced as a
/// `/healthz` 503 (a per-scrape read of the heartbeat age in `obs::serve`), not as
/// a one-shot in-process event, so there is no clean site to bump this exactly
/// once. The series renders (always 0) for discoverability; the live alerting
/// signal is the 503 itself.
pub fn record_reactor_stall() {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.reactor_stalls.fetch_add(1, Ordering::Relaxed);
}

/// Point-in-time set of the intelligence-endpoint reachability gauge
/// (`agent_intel_up`).
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;
}

/// Point-in-time set of the tree-pause gauge (`agent_paused`) — 1 while the
/// `pause` operator tool has frozen the agentic loops, 0 after `resume`.
/// No-op-safe / metrics-gated, mirroring `set_intel_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;
}

/// Point-in-time set of the intelligence all-endpoints-down gauge
/// (`agent_intel_all_down`) — 1 while every model
/// endpoint is down (the latched, eventually-consistent last-child-experience
/// truth a subagent reports up via `AgentMsg::IntelHealth`; the same flag flips
/// `/readyz` NotReady). 0 once any endpoint is usable again. Distinct from
/// `agent_intel_up` (the active endpoint's reachability): all-down is the
/// fleet-routing signal (no endpoint usable at all). No-op-safe / metrics-gated.
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;
}

/// Point-in-time set of the subagent-tree shape gauges
/// (`agent_active_subagents` / `agent_tree_depth` / `agent_tree_breadth`).
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);
}

/// Point-in-time set of the reactive backlog gauges — the scaling signal set an
/// autoscaler reads (`agent_pending_events` / `agent_inflight_reactions` /
/// `agent_subscriptions_active` / `agent_reaction_lag_ms`).
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);
}

/// A config hot reload reached a terminal disposition. Drives
/// `agent_config_reload_total{result}` with the closed domain
/// `applied`\|`rejected` (an unknown value buckets `other`). A `rejected` reload
/// is a clean no-op (the running config is unchanged); `applied` bumps the
/// generation gauge via [`set_config_generation`].
pub fn record_config_reload(result: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_config_reload(result);
    #[cfg(not(feature = "metrics"))]
    let _ = result;
}

/// A turn worker ran (`agent_turns_total{kind}`, agentd).
pub fn record_turn(kind: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_turn(kind);
    #[cfg(not(feature = "metrics"))]
    let _ = kind;
}

/// A workflow step reached a terminal status (`agent_steps_total{status}`).
pub fn record_step(status: &str) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.record_step(status);
    #[cfg(not(feature = "metrics"))]
    let _ = status;
}

/// A remote-store op completed (`agent_store_ops_total{result}` + latency sum).
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);
}

/// Point-in-time set of the durable inbox backlog (`agent_inbox_pending`).
pub fn set_inbox_pending(n: u64) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.set_inbox_pending(n);
    #[cfg(not(feature = "metrics"))]
    let _ = n;
}

/// Point-in-time set of the largest live context's token estimate
/// (`agent_context_tokens`).
pub fn set_context_tokens(n: u64) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY.set_context_tokens(n);
    #[cfg(not(feature = "metrics"))]
    let _ = n;
}

/// Point-in-time set of the config-generation gauge
/// (`agent_config_generation`): the count of successfully-applied reloads, so a
/// scraper can detect "this instance has picked up generation N" against
/// agentctl's desired generation. Monotonic in practice — the reload loop only
/// ever increments it.
pub fn set_config_generation(generation: u64) {
    #[cfg(feature = "metrics")]
    imp::REGISTRY
        .config_generation
        .store(generation, Ordering::Relaxed);
    #[cfg(not(feature = "metrics"))]
    let _ = generation;
}

/// Point-in-time set of the lifetime-budget balance gauge
/// (`agent_budget_tokens_remaining`): tokens left before the
/// per-instance cumulative cap is reached — the alerting/scaling hook for the
/// threshold event. Only ever set when a budget is installed (absent = the
/// gauge stays at its 0 default, which a scraper reads together with the fact
/// that no budget metric transitions occurred; unbounded instances simply never
/// call this).
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;
}

/// Point-in-time set of the resource-pressure gauges (`agent_pressure_level`,
/// `agent_disk_free_bytes`): the shed/drain state the admission gates act on
/// and the disk headroom that (usually) drives it. `disk_free: None` = no file
/// store on this instance — the byte gauge is then not emitted at all, because
/// exporting the supervisor's local free space when durability lives elsewhere
/// would invite alerts on the wrong disk.
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);
}

/// Point-in-time set of the work-in-progress gauges (`agent_runs_active`,
/// `agent_turns_queued`): non-terminal workflow runs, and conversation turns
/// waiting for a dispatch slot (parallelism, pause, drain, or shed — the gauge
/// does not say which; the event stream does).
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);
}

/// Render the current counters (+ live cgroup memory gauges) as Prometheus text.
#[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();

    // --- closed label domains -----------------------------------------------
    // Each is a fixed array of `(label_value, AtomicU64)`; an out-of-vocabulary
    // value lands in the trailing `other` slot so the series stays bounded. The
    // arrays ARE the cardinality bound — there is no map, no allocation, no
    // unbounded label key path.

    /// `agent_runs_total{status}` / `agent_subagents_exited_total{status}`
    /// label domain: the closed terminal-status vocabulary (verbatim from
    /// `TerminalStatus::as_str`), plus `other`.
    const STATUS_LABELS: &[&str] = &[
        "completed",
        "refused",
        "exhausted_steps",
        "exhausted_tokens",
        "deadline",
        "stalled",
        "loop_detected",
        "cancelled",
        "crashed",
        "other",
    ];

    /// `agent_refusals_total{reason}` label domain.
    const REFUSAL_REASONS: &[&str] = &["trifecta", "rate", "budget", "depth", "mcp", "other"];

    /// `agent_limit_exceeded_total{limit}` label domain; mirrors the
    /// `limit.exceeded` event's `limit` field.
    const LIMIT_LABELS: &[&str] = &[
        "steps",
        "tokens",
        "deadline",
        "depth",
        "tree_tokens",
        "tokens_lifetime",
        "restart_storm",
        "spawn_rate",
        "other",
    ];

    /// `agent_subagent_restarts_total{reason}` label domain.
    const RESTART_REASONS: &[&str] = &["crashed", "stuck", "rate", "other"];

    /// `agent_subagent_stuck_kills_total{signal}` label domain.
    const SIGNAL_LABELS: &[&str] = &["term", "kill", "other"];

    /// `agent_intel_errors_total{reason}` label domain.
    const INTEL_ERROR_REASONS: &[&str] = &["unreachable", "auth", "timeout", "5xx", "other"];

    /// `agent_drains_total{phase}` label domain.
    const DRAIN_PHASES: &[&str] = &["started", "completed", "forced", "other"];

    /// `agent_tokens_total{type}` direction label domain.
    const TOKEN_TYPES: &[&str] = &["in", "out"];

    /// `agent_config_reload_total{result}` label domain. A hot reload either
    /// `applied` (the reloadable diff took effect) or was `rejected` (invalid /
    /// restart-only / inconsistent → a clean no-op); `other` is the catch-all
    /// that keeps the series bounded.
    const RELOAD_RESULTS: &[&str] = &["applied", "rejected", "other"];

    /// `agent_turns_total{kind}` label domain: a turn worker's context kind — a
    /// `root`/conversation turn, a `preflight` think, a `compaction` think —
    /// plus `other`.
    const TURN_KINDS: &[&str] = &["root", "preflight", "compaction", "knowledge", "other"];

    /// `agent_steps_total{status}` label domain: a workflow step's terminal
    /// status — `done` / `failed` / `skipped` — plus `other`.
    const STEP_STATUS: &[&str] = &["done", "failed", "skipped", "other"];

    /// `agent_store_ops_total{result}` label domain: a remote-store op outcome —
    /// `ok` / `conflict` (a CAS mismatch) / `error` — plus `other`.
    const STORE_RESULTS: &[&str] = &["ok", "conflict", "error", "other"];

    /// A fixed-domain labelled counter family: one atomic per known label value.
    /// `N` matches the backing domain slice length; the trailing slot is the
    /// `other` catch-all that keeps the cardinality bounded.
    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],
            }
        }

        /// Increment the slot for `value`; an unknown value lands in the last
        /// (`other`) slot. `domain` MUST have length `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 {
        // --- bare, unlabelled series (kept alongside the labelled families) --
        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,

        // --- frozen: run lifecycle & terminal-status -------------------------
        runs_total: LabelCounter<{ STATUS_LABELS.len() }>,
        pub(super) loop_steps: AtomicU64,

        // --- frozen: refusal / bound counters --------------------------------
        refusals: LabelCounter<{ REFUSAL_REASONS.len() }>,
        limit_exceeded: LabelCounter<{ LIMIT_LABELS.len() }>,

        // --- frozen: subagent-tree gauges + counters -------------------------
        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() }>,

        // --- frozen: intelligence health -------------------------------------
        pub(super) intel_calls: AtomicU64,
        pub(super) intel_up: AtomicU64,
        // 1 while ALL model endpoints are down (the latched,
        // eventually-consistent last-child-experience truth that also flips
        // `/readyz`). Distinct from `intel_up` (the active endpoint's reachability).
        pub(super) intel_all_down: AtomicU64,
        intel_errors: LabelCounter<{ INTEL_ERROR_REASONS.len() }>,

        // --- tree-pause gauge (0/1) ------------------------------------------
        // Set by the `pause`/`resume` operator tools; 1 while the tree is paused.
        pub(super) paused: AtomicU64,

        // --- frozen: MCP server health ---------------------------------------
        mcp_connect_failures: LabelCounter<{ MCP_SERVER_SLOTS }>,

        // --- frozen: lifecycle events ----------------------------------------
        drains: LabelCounter<{ DRAIN_PHASES.len() }>,
        pub(super) supervisor_restarts: AtomicU64,
        pub(super) reactor_stalls: AtomicU64,

        // --- frozen: token accounting (typed) --------------------------------
        tokens_typed: LabelCounter<{ TOKEN_TYPES.len() }>,

        // --- frozen: reactive backlog (the autoscaler's signal set) ----------
        pending_events: AtomicU64,
        inflight_reactions: AtomicU64,
        subscriptions_active: AtomicU64,
        reaction_lag_ms: AtomicU64,

        // --- hot-reload outcome counter + generation gauge -------------------
        config_reloads: LabelCounter<{ RELOAD_RESULTS.len() }>,
        pub(super) config_generation: AtomicU64,

        // --- per-instance lifetime-budget balance gauge ----------------------
        pub(super) budget_tokens_remaining: AtomicU64,
        pub(super) pressure_level: AtomicU64,
        /// `u64::MAX` = unknown/no file store; then the gauge is not emitted.
        pub(super) disk_free_bytes: AtomicU64,
        pub(super) runs_active: AtomicU64,
        pub(super) turns_queued: AtomicU64,

        // --- runtime series: turns, workflow steps, store, inbox, context ----
        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) {
            // The bare, unlabelled counters.
            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);
            // Frozen `agent_runs_total{status}` — a COARSE projection of the three
            // `RunOutcome` variants this hook carries onto the closed status
            // domain. A precise status comes in through `record_run_status`.
            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);
            // Frozen `agent_tokens_total{type}` (the `model` label is not
            // available at the `AgentMsg::Usage` hook, so it stays absent).
            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);

            // --- liveness / readiness gauges ---------------------------------
            // `agent_up` is always 1 while we can render. `agent_ready` reads the
            // same process-wide drain/lame-duck state `/readyz` reports —
            // read-only, no extra call site.
            gauge(&mut s, "agent_up", "1 while the process is alive", 1);
            // `agent_ready` mirrors `/readyz` exactly: NotReady when draining,
            // lame-ducked, OR all intelligence endpoints are down — the same three
            // conditions the readiness probe consults.
            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,
            );
            // `agent_paused`: 1 while the tree is paused at turn boundaries.
            // Pause is NOT readiness — a paused instance can still be ready (the
            // `ready` gauge above ignores pause, only drain/lame-duck).
            gauge(
                &mut s,
                "agent_paused",
                "1 while the agentic tree is paused at turn boundaries",
                g(&self.paused),
            );

            // --- run lifecycle & terminal-status -----------------------------
            labelled_counter(
                &mut s,
                "agent_runs_total",
                "Runs by terminal status.",
                "status",
                STATUS_LABELS,
                &self.runs_total,
            );
            // `agent_loop_steps_total` is driven by `loop.step`, which is emitted
            // INSIDE the re-exec'd child agentic loop — a different process from the
            // supervisor this scrape reflects. `record_loop_step` is intentionally
            // left unwired here: bumping it would only touch the child's own
            // process-local registry, never this supervisor's. The series is
            // rendered (so the frozen contract stays discoverable) but reads the
            // supervisor's own process only; there is no cross-process rollup.
            // agentctl derives per-run step counts from `loop.step` log lines,
            // not from this counter.
            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),
            );

            // --- token / cost accounting -------------------------------------
            // `agent_tokens_total{type}`: the frozen `model` label is DEFERRED in
            // metrics_schema 1.0 — the only call site (`record_tokens`, fed by
            // `AgentMsg::Usage` up the control channel) does not carry the model
            // identifier, and adding it needs an emit site at the intelligence
            // boundary. The label key stays reserved and absent rather than
            // faked. agentctl gets per-model token splits from
            // `intel.result.usage` log lines.
            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,
            );
            // `agent_intel_calls_total`: same `model`-label deferral as tokens
            // (the `record_intel_call` site carries no model id). Additionally
            // process-local — `IntelClient::complete` runs in the re-exec'd child
            // (the supervisor makes no LLM calls), so this reflects only the
            // scraped process. Derive per-model call counts from `intel.call` logs.
            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),
            );

            // --- refusal / bound counters ------------------------------------
            // `agent_refusals_total` is driven by the model/loop refusing or a
            // guard tripping — all INSIDE the re-exec'd child loop (orchestrator
            // self-tool / scope checks), so `record_refusal` is left unwired: it
            // would only bump the child's process-local registry. Rendered for
            // contract discoverability but process-local — the supervisor scrape
            // reflects its own process and nothing rolls up across processes.
            // agentctl derives refusals from the refusal/`scope.trifecta_refused`
            // log lines.
            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,
            );
            // `agent_limit_exceeded_total{limit}` is PARTIALLY wired: the
            // `tree_tokens` leg is the supervisor's own tree-ceiling trip
            // (`supervisor::reactor`, this process → reaches the scrape), so it is
            // live. The `steps`/`tokens`/`deadline`/`depth` legs trip inside the
            // re-exec'd child loop and are therefore process-local (unwired here;
            // derived from `limit.exceeded` log lines). Same cross-process boundary
            // as the rest of this module.
            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,
            );

            // --- subagent-tree gauges + counters -----------------------------
            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,
            );

            // --- intelligence health -----------------------------------------
            gauge(
                &mut s,
                "agent_intel_up",
                "1 when the intelligence endpoint is reachable.",
                g(&self.intel_up),
            );
            // `agent_intel_all_down`: 1 while EVERY model endpoint is down — the
            // fleet-routing signal (the same latch that flips /readyz).
            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 server health -------------------------------------------
            // `agent_mcp_up{server}` is gauge-per-declared-server; there is no
            // declared-server registration hook here, so it is RESERVED and not
            // emitted at all — the honest-absence shape the rest of this module
            // follows. The connect-failure counter below IS wired — the
            // daemon's supervisor-process connect path (initial + hot-reload add,
            // `triggers::mode`) calls `record_mcp_connect_failure(server)`, so a
            // failing declared server shows up here labelled by `server`. (A
            // child-side connect failure is process-local and does not reach this
            // supervisor scrape — the same process boundary as everything else
            // in this module.)
            mcp_servers().render_connect_failures(&mut s, &self.mcp_connect_failures);

            // --- tool-call accounting — RESERVED ------------------------------
            // `agent_tool_calls_total{server,tool,ok}` is keyed off `tool.result`,
            // whose boundary (`McpClient::call_tool`) runs predominantly INSIDE the
            // re-exec'd child loop (the subagent's tool use); the only supervisor-
            // process call sites are the reactor's own management/lease calls
            // (`cluster` claim gate), not the agent's tool use the dashboard wants.
            // A scrape-side counter would therefore be process-local and misleading
            // (it would NOT reflect the children's tool calls), so the series is
            // RESERVED here — rendered as a HELP/TYPE marker, no fabricated 0 — and
            // agentctl reads tool calls from `tool.result` log lines. This mirrors
            // the `agent_mcp_up` honest-absence shape.
            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).",
            );
            // `agent_tool_call_duration_ms` / `agent_intel_call_duration_ms` /
            // `agent_run_duration_ms` are frozen HISTOGRAMS. This crate has no
            // histogram exposition machinery (no bucket/sum/count emission, by
            // design — the surface is hand-written counter/gauge text), so they are
            // RESERVED: rendered as HELP/TYPE markers only, no fabricated buckets.
            // A half-built histogram would be worse than an honest marker. Latency
            // lives in the `dur_ms` field of the matching 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).",
            );

            // --- lifecycle events ---------------------------------------------
            // `agent_drains_total{phase}` is wired: the reactor's per-run teardown
            // (`supervisor::reactor`) and the daemon's graceful wind-down
            // (`triggers::mode`) both run in this (supervisor) process and bump
            // `started`/`completed`/`forced`.
            labelled_counter(
                &mut s,
                "agent_drains_total",
                "Drain phase transitions.",
                "phase",
                DRAIN_PHASES,
                &self.drains,
            );
            // `agent_restarts_total` is RESERVED in metrics_schema 1.0: it counts
            // a supervisor process *restart* (rebuild + reconcile), and there is no
            // such in-process restart path to emit it from — a pod restart is a
            // fresh process with a zeroed registry, so an orchestrator counts
            // those, not the binary. Rendered (always 0) so the frozen series
            // stays discoverable; `record_supervisor_restart` is the hook a
            // reconcile path would call but is deliberately unwired.
            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),
            );
            // `agent_reactor_stalls_total` is RESERVED in metrics_schema 1.0: a
            // wedged reactor is surfaced as a `/healthz` 503 (a derived read of the
            // heartbeat age in `obs::serve`, evaluated per scrape), not as a
            // one-shot in-process event, so there is no clean emit site to bump a
            // counter exactly once. Rendered (always 0) for discoverability;
            // `record_reactor_stall` stays unwired until a dedicated
            // stall-detection edge exists. The liveness signal an operator alerts
            // on is the 503 itself, not this counter.
            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),
            );

            // --- hot reload ---------------------------------------------------
            // `agent_config_reload_total{result}` over the closed applied/rejected
            // domain, plus `agent_config_generation` (applied-reload count) so a
            // scraper detects "generation N is effective" against the desired one.
            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),
            );

            // --- lifetime budget balance --------------------------------------
            // Tokens remaining before the per-instance cumulative cap; the
            // alerting/scaling hook. 0 both when unbounded (never set) and when
            // exhausted — a scraper distinguishes them via the budget event/limit
            // metric, so unbounded instances read as "no budget in play".
            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),
            );

            // --- resource pressure + work in progress -------------------------
            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),
            );

            // --- reactive backlog — the autoscaler's signal set ----------------
            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),
            );

            // --- bare, unlabelled series --------------------------------------
            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),
            );

            // --- runtime series: turns, workflow steps, store, inbox, context -
            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
        }
    }

    // --- `agent_mcp_connect_failures_total{server}` -------------------------
    // The `server` label is bounded — the declared set is small and fixed — but
    // its *values* are config-time strings, not a compile-time enum. We bound it
    // structurally with a fixed slot table that interns server names on first use;
    // once full, further names fold into `other` so the series can never grow
    // unbounded. The table is a process-global behind a Mutex —
    // a slow path touched only on a connect failure, never on the render hot path
    // beyond a read snapshot.

    /// Max distinct `server` label values held before folding into `other`.
    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()),
            }
        }

        /// Index for `server`; interns on first use, or the `other` slot
        /// (`MCP_SERVER_SLOTS - 1`) once the table is full. Poisoning is ignored,
        /// because telemetry must never crash the agent.
        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);
        }

        /// Emit one `agent_mcp_connect_failures_total{server="…"}` line per
        /// interned server with a non-zero count, plus the `other` overflow slot.
        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
    }

    /// One counter family in Prometheus text exposition format.
    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}");
    }

    /// A frozen series whose machinery is not implemented: render the
    /// `# HELP`/`# TYPE` headers — so the contract stays discoverable from the
    /// scrape and a silent drop is catchable — WITHOUT a fabricated always-0
    /// sample line. This is the same honest-absence shape as `agent_mcp_up`: a
    /// marker, not a value. `kind` is the Prometheus type the series will carry
    /// once implemented (`counter`/`histogram`); `help` MUST say it is reserved
    /// and why (cross-process boundary, or no histogram exposition).
    fn reserved(s: &mut String, name: &str, kind: &str, help: &str) {
        let _ = writeln!(s, "# HELP {name} {help}");
        let _ = writeln!(s, "# TYPE {name} {kind}");
    }

    /// One gauge family (point-in-time value) in Prometheus text format.
    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}");
    }

    /// One labelled counter family: a single HELP/TYPE header, then one series
    /// line per closed-domain label value — the domain *is* the cardinality
    /// bound. `domain` and the `LabelCounter` slots are the same length.
    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}");
        }
    }

    /// Live cgroup v2 memory gauges, emitted only for fields the kernel exposes
    /// (kept out of `Registry::render` so the counter set stays deterministic).
    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"));
            // No file store → no byte reading → the gauge is NOT emitted (an
            // exported 0 would read as "disk full" to an alert).
            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();
            // The liveness/readiness gauges are label-free.
            assert!(out.contains("# TYPE agent_up gauge"));
            assert!(out.contains("agent_up 1"));
            assert!(out.contains("# TYPE agent_ready gauge"));
            // ready is 0/1; in a bare test process (no drain) it is 1.
            assert!(out.contains("agent_ready "));
        }

        #[test]
        fn paused_gauge_renders_zero_then_one() {
            // `agent_paused` is a 0/1 gauge, default 0.
            let r = Registry::new();
            let out = r.render();
            assert!(out.contains("# TYPE agent_paused gauge"));
            assert!(out.contains("agent_paused 0"));
            // Set via the same atomic `set_paused` writes; renders 1.
            r.paused.store(1, Ordering::Relaxed);
            assert!(r.render().contains("agent_paused 1"));
        }

        #[test]
        fn intel_all_down_gauge_renders_zero_then_one() {
            // `agent_intel_all_down` is a 0/1 gauge, default 0, set from the
            // latched all-down flag (the same one /readyz reads).
            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"));
            // Set via the same atomic `set_intel_all_down` writes; renders 1.
            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");
            // an out-of-vocabulary status buckets under `other`, never a new label
            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"));
            // every closed-domain value is present (zero-valued series included)
            assert!(out.contains("agent_runs_total{status=\"loop_detected\"} 0"));
            // exactly one HELP/TYPE header for the family
            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"));
            // closed domains: a stray reason never widens the label set
            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() {
            // The reload counter has the closed applied/rejected domain (every
            // value rendered, zero-valued included), and the generation gauge
            // tracks applied reloads.
            let r = Registry::new();
            let out = r.render();
            // Both closed-domain series are present even at zero.
            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"));
            // They increment over the closed domain; an unknown buckets `other`.
            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"));
            // Exactly one HELP/TYPE header for the counter family.
            assert_eq!(
                out.matches("# TYPE agent_config_reload_total counter")
                    .count(),
                1
            );
        }

        #[test]
        fn budget_gauge_and_lifetime_limit_render() {
            // The balance gauge (present at 0 by default) plus the
            // `tokens_lifetime` value of the closed `agent_limit_exceeded_total`
            // domain.
            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() {
            // Cardinality: render must never contain a run_id/agent_path-style
            // label key. We assert the only label keys present are the bounded set.
            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;
            // a limited cgroup → two gauge families
            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);
            // no cgroup → no gauge lines (keeps /metrics clean off-cgroup)
            assert!(memory_gauges(MemorySnapshot::default()).is_empty());
        }

        #[test]
        fn frozen_schema_4_3_series_all_present_emitted_or_reserved() {
            // Honesty gate: every frozen series MUST be discoverable from the
            // render — either as a live counter/gauge or as a reserved HELP/TYPE
            // marker. This catches a silent drop of a frozen series (which is a
            // major-bump-only change) at test time.
            let r = Registry::new();
            let out = r.render();
            // The full metric-name set (the names are the frozen contract).
            for name in [
                // liveness/readiness gauges
                "agent_up",
                "agent_ready",
                // run lifecycle + tokens + intel
                "agent_runs_total",
                "agent_run_duration_ms", // reserved (histogram)
                "agent_loop_steps_total",
                "agent_tokens_total",
                "agent_intel_calls_total",
                "agent_intel_call_duration_ms", // reserved (histogram)
                // refusal / bound
                "agent_refusals_total",
                "agent_limit_exceeded_total",
                // subagent tree
                "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",
                // intelligence health
                "agent_intel_up",
                "agent_intel_errors_total",
                // MCP server health
                "agent_mcp_connect_failures_total",
                // tool-call accounting (reserved)
                "agent_tool_calls_total",
                "agent_tool_call_duration_ms", // reserved (histogram)
                // lifecycle events
                "agent_drains_total",
                "agent_restarts_total",       // reserved (no emit site)
                "agent_reactor_stalls_total", // reserved (no emit site)
                // reactive backlog
                "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}"
                );
            }
            // The three histograms + the deferred tool-call counter are RESERVED:
            // a HELP/TYPE marker, NO fabricated sample line (the honest-absence
            // shape — no `name <value>` and no `name{...} <value>`).
            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}"
                );
                // No sample line for the reserved series (only the two `#` headers).
                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:?}"
                    );
                }
            }
            // The reserved markers say so (honest HELP text).
            assert!(out.contains("reserved in metrics_schema 1.0"));
        }

        #[test]
        fn wired_supervisor_counters_increment() {
            // The supervisor-process counters increment via the same registry
            // methods the emit sites call. (The emit sites live in
            // `supervisor::reactor` / `triggers::mode`; here we exercise the
            // registry contract those call sites depend on.)
            let r = Registry::new();
            // subagent spawn/exit (reactor.rs).
            r.subagents_spawned.fetch_add(1, Ordering::Relaxed);
            r.record_subagent_exited("completed");
            r.record_subagent_exited("cancelled");
            // stuck-kill ladder (reactor.rs drive_drain Term/Kill).
            r.record_subagent_stuck_kill("term");
            r.record_subagent_stuck_kill("kill");
            // drain phases (reactor.rs begin_drain/Done/timeout + mode.rs daemon).
            r.record_drain("started");
            r.record_drain("completed");
            r.record_drain("forced");
            // restart governor respawn (mode.rs Backoff branch).
            r.record_subagent_restart("crashed");
            // mcp connect failure (mode.rs connect + hot-reload add).
            r.record_mcp_connect_failure("github");
            // tree-token bound trip (reactor.rs Usage handler).
            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() {
            // `agent_restarts_total` (supervisor restart) and
            // `agent_reactor_stalls_total` have no in-process emit site; they
            // render reserved-but-present at 0 so the contract stays discoverable
            // without falsely claiming a non-zero value.
            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"));
            // Their HELP marks them reserved (not silently permanent-0). Both
            // reserved-counter HELP lines carry the marker phrase.
            assert!(out.matches("reserved in metrics_schema 1.0").count() >= 2);
        }
    }
}