hive-agent: emit harness-only per-turn otel metrics

Cargo.lock

@@ -1593,6 +1593,9 @@ dependencies = [
  "hyper",
  "hyper-util",
  "libc",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry_sdk",
  "reqwest",
  "rmcp",
  "rusqlite",

hive-agent/Cargo.toml

@@ -23,6 +23,9 @@ hive-core-agent-sock.workspace = true
 hive-sh4re.workspace = true
 hive-sock-client.workspace = true
 libc.workspace = true
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+opentelemetry-otlp.workspace = true
 rmcp.workspace = true
 rusqlite.workspace = true
 schemars.workspace = true

hive-agent/src/main.rs

@@ -17,6 +17,7 @@ mod identity;
 mod login;
 mod login_session;
 mod mcp_config;
+mod otel_turn_metrics;
 mod paths;
 mod plugins;
 mod prompt;
@@ -1006,30 +1007,42 @@ async fn handle_turn<S: Surface>(
     let tool_calls = bus.take_tool_calls();
     let todo_wake_checked =
         (from == "todo").then(|| tool_calls.contains_key("mcp__hyperhive__get_loose_ends"));
+    // Read-and-clear once per turn unconditionally (same rationale as
+    // `take_tool_calls` above — the harness gets exactly one chance to
+    // observe this flag) rather than only when the sqlite sink happens to be
+    // configured: the OTEL turn-metrics exporter's session-boundary counter
+    // needs it too, independent of `stats`.
+    let fresh_session = bus.take_fresh_session();
     if let Some(stats) = stats {
         // Fresh session this turn → mint a `sessions` row and set its id on
         // the bus so this turn (and subsequent ones until the next fresh
-        // start) stamp `turn_stats.session_id`. Takes the one-shot flag
+        // start) stamp `turn_stats.session_id`.
-        // `run_claude` set when it suppressed `--continue`.
+        if fresh_session {
-        if bus.take_fresh_session() {
             let sid = stats.start_session(started_at, &model_at_start);
             bus.set_session_id(sid);
         }
-        let ended_at = chrono::Utc::now().timestamp();
+    }
-        let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
+    let ended_at = chrono::Utc::now().timestamp();
-        let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
+    let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
-        let row = serve_common::build_row(serve_common::TurnRowArgs {
+    let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
-            started_at,
+    let row = serve_common::build_row(serve_common::TurnRowArgs {
-            ended_at,
+        started_at,
-            duration_ms,
+        ended_at,
-            model: model_at_start,
+        duration_ms,
-            wake_from: from.clone(),
+        model: model_at_start,
-            outcome: &outcome,
+        wake_from: from.clone(),
-            bus,
+        outcome: &outcome,
-            tool_calls,
+        bus,
-            open_threads_count: open_threads,
+        tool_calls,
-            open_reminders_count: open_reminders,
+        open_threads_count: open_threads,
-        });
+        open_reminders_count: open_reminders,
+    });
+    // Harness-only OTEL metrics (duration/wake_from/result_kind/loose-ends/
+    // session boundaries) — independent of the sqlite sink below, and a
+    // cheap no-op when OTEL isn't configured. See `otel_turn_metrics`'s
+    // module doc for why token/cost/tool-count are deliberately not here.
+    otel_turn_metrics::record(&row, fresh_session);
+    if let Some(stats) = stats {
         stats.record(&row);
     }
     let pending = S::inbox_unread(socket).await;

hive-agent/src/otel_turn_metrics.rs

@@ -0,0 +1,159 @@
+//! Per-turn OTEL metric export for the fields Claude Code's own built-in
+//! OTEL integration cannot know about. Claude's own export already covers
+//! token usage, cost, and tool-call counts (see `docs/observability.md`) —
+//! duplicating those here under different metric names would just give a
+//! collector two disagreeing series for the same number. This module emits
+//! only the harness-only concepts: wall-clock turn duration as *this harness*
+//! measures it (not Claude's own per-request latency), what woke the turn,
+//! the harness's own outcome classification, the loose-ends backlog at turn
+//! end, and session boundaries (fresh vs. `--continue`'d). Scoped this way
+//! per the tracker discussion on the "emit agent stats as OTEL metrics" issue.
+//!
+//! Synchronous instruments, not observable ones: unlike hive-c0re's
+//! container-resource gauges (`hive-c0re/src/stats/otel_metrics.rs`), which
+//! poll a periodically-refreshed snapshot because the underlying cgroup value
+//! is continuously live, every value here only exists at one instant — the
+//! moment a turn ends — so it's recorded directly (`counter.add`,
+//! `histogram.record`, `gauge.record`) from [`record`], called once per turn
+//! from `handle_turn`. The `PeriodicReader` still batches + exports on its
+//! own interval; only the recording is event-driven, not the export.
+//!
+//! Resource attributes (`service.name`, `agent`, `hive`, `swarm`) are picked
+//! up automatically by the SDK from the container-wide `OTEL_RESOURCE_ATTRIBUTES`
+//! env var (same mechanism `hive-metric` relies on) — nothing to set here.
+
+use std::sync::OnceLock;
+use std::time::Duration;
+
+use opentelemetry::KeyValue;
+use opentelemetry::metrics::{Counter, Gauge, Histogram, MeterProvider as _};
+use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig};
+use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
+
+use crate::turn_stats::TurnStatRow;
+
+/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset.
+/// Matches `hive-c0re`'s container-metrics default so the two exporters
+/// batch on the same rhythm by default.
+const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);
+
+struct Instruments {
+    /// Kept alive for the process lifetime — the `PeriodicReader` only
+    /// exports while its provider lives.
+    _provider: SdkMeterProvider,
+    turn_duration_ms: Histogram<u64>,
+    turn_count: Counter<u64>,
+    session_count: Counter<u64>,
+    open_threads: Gauge<u64>,
+    open_reminders: Gauge<u64>,
+}
+
+/// Lazily built on the first call to [`record`]. `None` when OTEL isn't
+/// configured (`OTEL_EXPORTER_OTLP_ENDPOINT` unset/empty) or the exporter
+/// failed to build — either way `record` becomes a cheap no-op.
+static INSTRUMENTS: OnceLock<Option<Instruments>> = OnceLock::new();
+
+/// Record one turn's harness-only metrics. No-op (and cheap — one
+/// `OnceLock` read after the first call) when OTEL isn't configured.
+///
+/// `fresh_session` is the harness's own "did this turn start a brand-new
+/// claude session" flag (`Bus::take_fresh_session`) — pass it through rather
+/// than re-deriving it here, since the harness only gets to observe it once
+/// per turn (the same flag gates minting a `turn_stats` `sessions` row).
+pub fn record(row: &TurnStatRow, fresh_session: bool) {
+    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
+        return;
+    };
+    let attrs = [
+        KeyValue::new("wake_from", row.wake_from.clone()),
+        KeyValue::new("result_kind", row.result_kind),
+        KeyValue::new("model", row.model.clone()),
+    ];
+    // `try_from` rather than `as` — `duration_ms` is never negative in
+    // practice (it's `Instant::elapsed().as_millis()`), but a fallback of 0
+    // beats a lossy cast or a panic if that ever stops holding.
+    inst.turn_duration_ms
+        .record(u64::try_from(row.duration_ms).unwrap_or(0), &attrs);
+    inst.turn_count.add(1, &attrs);
+    if let Some(threads) = row.open_threads_count {
+        inst.open_threads.record(threads, &[]);
+    }
+    if let Some(reminders) = row.open_reminders_count {
+        inst.open_reminders.record(reminders, &[]);
+    }
+    if fresh_session {
+        inst.session_count
+            .add(1, &[KeyValue::new("model", row.model.clone())]);
+    }
+}
+
+fn build() -> Option<Instruments> {
+    if !enabled() {
+        tracing::debug!("otel turn-metrics: no endpoint configured, exporter disabled");
+        return None;
+    }
+    let interval = interval();
+    match build_provider(interval) {
+        Ok(provider) => {
+            tracing::info!(?interval, "otel turn-metrics: exporter enabled");
+            let meter = provider.meter("hyperhive.turn_stats");
+            Some(Instruments {
+                turn_duration_ms: meter
+                    .u64_histogram("hyperhive.agent.turn.duration")
+                    .with_unit("ms")
+                    .build(),
+                turn_count: meter.u64_counter("hyperhive.agent.turn.count").build(),
+                session_count: meter.u64_counter("hyperhive.agent.session.count").build(),
+                open_threads: meter
+                    .u64_gauge("hyperhive.agent.loose_ends.threads")
+                    .build(),
+                open_reminders: meter
+                    .u64_gauge("hyperhive.agent.loose_ends.reminders")
+                    .build(),
+                _provider: provider,
+            })
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "otel turn-metrics: exporter init failed");
+            None
+        }
+    }
+}
+
+fn build_provider(interval: Duration) -> anyhow::Result<SdkMeterProvider> {
+    // http/json, no explicit endpoint/resource: same rationale as
+    // `hive-c0re/src/stats/otel_metrics.rs` for the transport (the blocking
+    // OTLP client is required — `PeriodicReader` drives export from a
+    // background thread with no Tokio reactor), and same rationale as
+    // `hive-metric` for the resource (the SDK reads `OTEL_RESOURCE_ATTRIBUTES`
+    // — set container-wide, see `docs/observability.md::Built-in resource
+    // labels` — on its own; passing an explicit `Resource` here would need to
+    // duplicate `agent`/`hive`/`swarm` this crate has no other reason to know).
+    let exporter = MetricExporter::builder()
+        .with_http()
+        .with_protocol(Protocol::HttpJson)
+        .build()?;
+    let reader = PeriodicReader::builder(exporter)
+        .with_interval(interval)
+        .build();
+    Ok(SdkMeterProvider::builder().with_reader(reader).build())
+}
+
+/// `OTEL_EXPORTER_OTLP_ENDPOINT` non-empty — the same enable signal every
+/// other exporter in this workspace uses (see `hive-c0re/src/stats/
+/// otel_metrics.rs::endpoint` for why this exact variable and not a
+/// hyperhive-specific one: it's also the variable the SDK itself reads to
+/// build the exporter's URL, so "configured" and "where it goes" cannot
+/// disagree).
+fn enabled() -> bool {
+    std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().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)
+}