feat(#2007): export per-agent container cpu/mem/disk via otel

hive-c0re already samples each agent container's cgroup load for the
dashboard (stats/container_stats.rs); this rides those gauges out to the
configured OTLP endpoint, reusing the existing services.hyperhive.otel
config (endpoint + auth header) — no new toggle.

- New stats/otel_metrics.rs: exports via the OpenTelemetry Rust SDK (same
  crates as hive-metric) with the semconv container.* metric names +
  container.name attribute so off-the-shelf OTel/Grafana dashboards work,
  plus the hive agent label. container.cpu.time (counter, s, from cumulative
  cpu.stat usage_usec), container.memory.usage, container.memory.usage.limit;
  memory peak / on-disk storage / instantaneous cpu percent stay hyperhive.*
  custom (no semconv equivalent). Observable instruments read a shared
  snapshot an async task refreshes (gather() is async; SDK callbacks sync).
- container_stats: expose cpu_time_usec (cumulative) on ContainerResource.
- The OTLP auth header is loaded onto hive-c0re's own unit via systemd
  LoadCredential and read from $CREDENTIALS_DIRECTORY/otel-headers.
- docs/observability.md documents the host-emitted semconv metrics.

Host-side export, so it covers containers even when their agent is idle.

hive-c0re/src/lib.rs

@@ -44,7 +44,7 @@ pub mod workers;
 // Root re-exports: keep every pre-grouping `crate::<module>` /
 // `hive_c0re::<module>` path compiling without touching consumers.
 pub use agent_config::{capabilities, limits, tool_groups, topology};
-pub use stats::{container_stats, hive_stats, host_stats, sweep_health, warnings};
+pub use stats::{container_stats, hive_stats, host_stats, otel_metrics, sweep_health, warnings};
 pub use stores::{
     approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
 };

hive-c0re/src/main.rs

@@ -428,6 +428,11 @@ async fn cmd_serve(
     // poll stays cheap cgroup-only reads. Feeds `disk_bytes` on the LOAD
     // tab. See container_stats::disk_sampler_loop.
     hive_c0re::container_stats::spawn_disk_sampler();
+    // Per-agent container-resource OTEL export: rides the same
+    // cgroup gauges out to the configured OTLP endpoint, reusing the hive
+    // `services.hyperhive.otel` config (endpoint + LoadCredential auth).
+    // No-op when OTEL isn't configured.
+    hive_c0re::otel_metrics::spawn_exporter();
     // build_logs.sqlite vacuum: c0re-side (single db). Failures kept
     // 30d, successes 24h — see `build_logs::vacuum` for the rule.
     hive_c0re::build_logs::spawn_vacuum(&coord);

hive-c0re/src/stats/container_stats.rs

@@ -43,6 +43,11 @@ pub struct ContainerResource {
     /// Host-normalised CPU usage over the sample interval, as a
     /// percentage of total host CPU (0..100 across all cores).
     pub cpu_pct: f64,
+    /// Cumulative CPU time (`cpu.stat` `usage_usec`, microseconds,
+    /// monotonic). `None` if the read failed. Feeds the OTEL semconv
+    /// `container.cpu.time` counter (which wants absolute cumulative time,
+    /// converted to seconds), distinct from the sampled `cpu_pct`.
+    pub cpu_time_usec: Option<u64>,
     /// Current memory usage (`memory.current`), bytes.
     pub mem_current_bytes: u64,
     /// High-water memory usage since container start (`memory.peak`),
@@ -262,7 +267,11 @@ pub async fn gather() -> Vec<ContainerResource> {
 
     let mut out: Vec<ContainerResource> = Vec::with_capacity(candidates.len());
     for (i, (name, dir)) in candidates.iter().enumerate() {
-        let cpu_pct = match (t0[i], read_usage_usec(dir)) {
+        // The second cumulative read — also exposed raw as `cpu_time_usec`
+        // for the OTEL `container.cpu.time` counter (which wants absolute
+        // cumulative CPU time, not the sampled percent).
+        let usage_now = read_usage_usec(dir);
+        let cpu_pct = match (t0[i], usage_now) {
             (Some(a), Some(b)) => {
                 #[allow(
                     clippy::cast_precision_loss,
@@ -276,6 +285,7 @@ pub async fn gather() -> Vec<ContainerResource> {
         out.push(ContainerResource {
             name: name.clone(),
             cpu_pct,
+            cpu_time_usec: usage_now,
             mem_current_bytes: read_u64(&dir.join("memory.current")).unwrap_or(0),
             mem_peak_bytes: read_u64(&dir.join("memory.peak")),
             mem_max_bytes: read_mem_max(&dir.join("memory.max")),

hive-c0re/src/stats/mod.rs

@@ -6,5 +6,6 @@
 pub mod container_stats;
 pub mod hive_stats;
 pub mod host_stats;
+pub mod otel_metrics;
 pub mod sweep_health;
 pub mod warnings;

hive-c0re/src/stats/otel_metrics.rs

@@ -0,0 +1,318 @@
+//! Per-agent container-resource OTEL export. hive-c0re already samples each
+//! agent container's cgroup load for the dashboard
+//! ([`super::container_stats`]); this rides those same gauges out to the
+//! hive-wide OTLP endpoint, reusing the SAME `services.hyperhive.otel` config
+//! (endpoint + auth header) that Claude Code's in-container SDK export uses —
+//! no new toggle. The auth header is loaded onto hive-c0re's own unit via
+//! systemd `LoadCredential` (see `nix/host-modules/hive-c0re`) and read from
+//! `$CREDENTIALS_DIRECTORY/otel-headers`.
+//!
+//! Emits the OTEL **semconv `container.*`** metrics with the standard
+//! `container.name` attribute (so off-the-shelf OTEL/Grafana container
+//! dashboards work), plus the hive-specific `agent` / `hive` / `swarm` labels
+//! for our own dashboards. Uses the OpenTelemetry Rust SDK (same crates as
+//! `hive-metric`); the blocking OTLP client is deliberate — the metrics SDK's
+//! `PeriodicReader` runs on a background thread with no Tokio reactor.
+//!
+//! Bridging async→sync: [`container_stats::gather`] is async, but the SDK's
+//! observable-instrument callbacks are sync. So an async task refreshes a
+//! shared snapshot on an interval, and the (sync) callbacks read it.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::Duration;
+
+use anyhow::{Context, Result};
+use opentelemetry::KeyValue;
+use opentelemetry::metrics::MeterProvider as _;
+use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig};
+use opentelemetry_sdk::Resource;
+use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
+
+use super::container_stats::{self, ContainerResource};
+
+/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset.
+const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);
+
+/// Latest container-stats snapshot: written by the async refresher, read by
+/// the sync observable-instrument callbacks.
+static SNAPSHOT: OnceLock<Arc<Mutex<Vec<ContainerResource>>>> = OnceLock::new();
+
+/// Keep the provider alive for the process lifetime — the `PeriodicReader`
+/// exports only while the provider lives.
+static PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
+
+/// Spawn the container-resource OTEL exporter if OTEL is configured
+/// (`HYPERHIVE_OTEL_ENDPOINT` non-empty — the same enable signal
+/// [`crate::meta::otel_config`] uses). No-op otherwise. Call once at startup.
+pub fn spawn_exporter() {
+    let Some(endpoint) = endpoint() else {
+        tracing::debug!("otel container-metrics: no endpoint configured, exporter disabled");
+        return;
+    };
+    let snapshot = SNAPSHOT
+        .get_or_init(|| Arc::new(Mutex::new(Vec::new())))
+        .clone();
+    let interval = interval();
+
+    // Async refresher: keep the shared snapshot current for the sync callbacks.
+    let refresh = snapshot.clone();
+    tokio::spawn(async move {
+        loop {
+            let fresh = container_stats::gather().await;
+            if let Ok(mut g) = refresh.lock() {
+                *g = fresh;
+            }
+            tokio::time::sleep(interval).await;
+        }
+    });
+
+    match build_provider(&endpoint, interval, snapshot) {
+        Ok(provider) => {
+            let _ = PROVIDER.set(provider);
+            tracing::info!(%endpoint, ?interval, "otel container-metrics: exporter enabled");
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "otel container-metrics: exporter init failed");
+        }
+    }
+}
+
+fn build_provider(
+    endpoint: &str,
+    interval: Duration,
+    snapshot: Arc<Mutex<Vec<ContainerResource>>>,
+) -> Result<SdkMeterProvider> {
+    // http/json — the only OTLP transport this crate enables (matching
+    // hive-metric). The Claude SDK path honours `HYPERHIVE_OTEL_PROTOCOL` for
+    // its own export; this exporter is always http/json.
+    let mut builder = MetricExporter::builder()
+        .with_http()
+        .with_endpoint(endpoint)
+        .with_protocol(Protocol::HttpJson);
+    let headers = auth_headers();
+    if !headers.is_empty() {
+        builder = builder.with_headers(headers);
+    }
+    let exporter = builder.build().context("build OTLP metric exporter")?;
+    // Drive the reader's export at `interval` so `HYPERHIVE_OTEL_METRIC_INTERVAL_MS`
+    // is the real export cadence (not just the snapshot-refresh cadence). The
+    // refresher runs at the same interval, gather-first, so the snapshot is
+    // populated before the first export.
+    let reader = PeriodicReader::builder(exporter)
+        .with_interval(interval)
+        .build();
+    let provider = SdkMeterProvider::builder()
+        .with_reader(reader)
+        .with_resource(resource())
+        .build();
+    register_instruments(&provider, snapshot);
+    Ok(provider)
+}
+
+/// Register the observable instruments. Each callback reads the shared
+/// snapshot and reports one data point per agent. Instruments are held by the
+/// meter/provider (kept alive in `PROVIDER`).
+fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc<Mutex<Vec<ContainerResource>>>) {
+    let meter = provider.meter("hyperhive.container_stats");
+
+    // semconv `container.cpu.time` — cumulative CPU seconds (monotonic counter).
+    let snap = snapshot.clone();
+    meter
+        .f64_observable_counter("container.cpu.time")
+        .with_unit("s")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    if let Some(usec) = c.cpu_time_usec {
+                        #[allow(clippy::cast_precision_loss)]
+                        obs.observe(usec as f64 / 1_000_000.0, &attrs(c));
+                    }
+                }
+            }
+        })
+        .build();
+
+    // semconv `container.memory.usage` — current usage bytes.
+    let snap = snapshot.clone();
+    meter
+        .u64_observable_gauge("container.memory.usage")
+        .with_unit("By")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    obs.observe(c.mem_current_bytes, &attrs(c));
+                }
+            }
+        })
+        .build();
+
+    // The cgroup memory ceiling (omit when unlimited). Kept `hyperhive.`
+    // custom — semconv defines `container.memory.usage` but not a matching
+    // `.limit` metric, so don't claim a spec name that isn't in the spec.
+    let snap = snapshot.clone();
+    meter
+        .u64_observable_gauge("hyperhive.container.memory.limit")
+        .with_unit("By")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    if let Some(limit) = c.mem_max_bytes {
+                        obs.observe(limit, &attrs(c));
+                    }
+                }
+            }
+        })
+        .build();
+
+    // Custom (`hyperhive.`) metrics semconv doesn't standardise: memory peak,
+    // on-disk footprint, and the instantaneous cpu percent the dashboard shows.
+    let snap = snapshot.clone();
+    meter
+        .u64_observable_gauge("hyperhive.container.memory.peak")
+        .with_unit("By")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    if let Some(peak) = c.mem_peak_bytes {
+                        obs.observe(peak, &attrs(c));
+                    }
+                }
+            }
+        })
+        .build();
+
+    let snap = snapshot.clone();
+    meter
+        .u64_observable_gauge("hyperhive.container.storage.usage")
+        .with_unit("By")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    if let Some(disk) = c.disk_bytes {
+                        obs.observe(disk, &attrs(c));
+                    }
+                }
+            }
+        })
+        .build();
+
+    let snap = snapshot;
+    meter
+        .f64_observable_gauge("hyperhive.container.cpu.percent")
+        .with_unit("%")
+        .with_callback(move |obs| {
+            if let Ok(g) = snap.lock() {
+                for c in g.iter() {
+                    obs.observe(c.cpu_pct, &attrs(c));
+                }
+            }
+        })
+        .build();
+}
+
+/// Per-data-point attributes: the spec `container.name` plus the hive `agent`
+/// label. (`hive` / `swarm` are constant across this c0re, so they live on the
+/// resource.)
+fn attrs(c: &ContainerResource) -> Vec<KeyValue> {
+    vec![
+        KeyValue::new("container.name", format!("h-{}", c.name)),
+        KeyValue::new("agent", c.name.clone()),
+    ]
+}
+
+/// Resource: `service.name = hyperhive-c0re` plus whatever the operator set in
+/// `extraResourceAttributes` (where the hive `hive` / `swarm` labels live, same
+/// channel the agent-side export uses).
+fn resource() -> Resource {
+    let mut builder = Resource::builder().with_service_name("hyperhive-c0re");
+    for (k, v) in resource_attributes() {
+        builder = builder.with_attribute(KeyValue::new(k, v));
+    }
+    builder.build()
+}
+
+/// `HYPERHIVE_OTEL_ENDPOINT`, non-empty. The enable signal.
+fn endpoint() -> Option<String> {
+    std::env::var("HYPERHIVE_OTEL_ENDPOINT")
+        .ok()
+        .map(|s| s.trim().to_owned())
+        .filter(|s| !s.is_empty())
+}
+
+/// Export cadence from `HYPERHIVE_OTEL_METRIC_INTERVAL_MS`, else the default.
+fn interval() -> Duration {
+    std::env::var("HYPERHIVE_OTEL_METRIC_INTERVAL_MS")
+        .ok()
+        .and_then(|s| s.trim().parse::<u64>().ok())
+        .filter(|ms| *ms > 0)
+        .map_or(DEFAULT_INTERVAL, Duration::from_millis)
+}
+
+/// Extra resource attributes from `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES`
+/// (`key=value,key=value`), same env the agent-side export reads.
+fn resource_attributes() -> Vec<(String, String)> {
+    std::env::var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES")
+        .ok()
+        .map(|s| parse_kv(&s))
+        .unwrap_or_default()
+}
+
+/// OTLP auth headers from the systemd credential at
+/// `$CREDENTIALS_DIRECTORY/otel-headers` (an `OTEL_EXPORTER_OTLP_HEADERS`-style
+/// `Key=Value` line). Empty when the credential is absent — the collector is
+/// then assumed unauthenticated.
+fn auth_headers() -> HashMap<String, String> {
+    let Some(dir) = std::env::var("CREDENTIALS_DIRECTORY").ok() else {
+        return HashMap::new();
+    };
+    let path = std::path::Path::new(&dir).join("otel-headers");
+    let Ok(raw) = std::fs::read_to_string(&path) else {
+        tracing::warn!(
+            "otel container-metrics: no auth header at $CREDENTIALS_DIRECTORY/otel-headers; \
+             exporting without Authorization"
+        );
+        return HashMap::new();
+    };
+    parse_kv(&raw).into_iter().collect()
+}
+
+/// Parse `key=value` pairs separated by commas and/or newlines. The value
+/// keeps any `=` after the first (so `Authorization=Bearer x=y` → `Bearer x=y`).
+fn parse_kv(s: &str) -> Vec<(String, String)> {
+    s.split([',', '\n'])
+        .filter_map(|pair| {
+            let pair = pair.trim();
+            if pair.is_empty() {
+                return None;
+            }
+            let (k, v) = pair.split_once('=')?;
+            let k = k.trim();
+            if k.is_empty() {
+                None
+            } else {
+                Some((k.to_owned(), v.trim().to_owned()))
+            }
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_kv_keeps_bearer_value() {
+        assert_eq!(
+            parse_kv("Authorization=Bearer abc=def"),
+            vec![("Authorization".to_owned(), "Bearer abc=def".to_owned())]
+        );
+        assert_eq!(
+            parse_kv("a=1,\n b=2 ,=bad,"),
+            vec![
+                ("a".to_owned(), "1".to_owned()),
+                ("b".to_owned(), "2".to_owned())
+            ]
+        );
+    }
+}