hyperhive/hive-agent/src/otel_turn_metrics.rs

//! 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/scheduler/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, session boundaries (fresh vs. `--continue`'d), context-window fill,
//! and the cost an ACP agent reports — claude's runtime reports none.
//!
//! 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.
//!
//! Three exceptions share this module's exporter but not [`record`]'s
//! per-turn call site: their values are continuously live (like hive-c0re's
//! cgroup values), so each is recorded from its own watch's periodic tick —
//! [`record_claude_md_lines`] from [`crate::claude_md_watch`],
//! [`record_subagent_role_lines`] from [`crate::subagent_role_watch`], and
//! [`record_claude_usage`] from [`crate::claude_usage_watch`].
//!
//! 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;

/// Export cadence. Matches `hive-c0re`'s container-metrics exporter so the
/// two batch on the same rhythm.
const EXPORT_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>,
    /// Recorded from [`crate::claude_md_watch`]'s own tick, not from
    /// [`record`] — `AGENTS.md` size is a continuously-live value (like
    /// hive-c0re's container gauges), not a once-per-turn one, so it has
    /// its own entry point rather than riding `record`'s per-turn call.
    claude_md_lines: Gauge<u64>,
    /// Recorded from [`crate::subagent_role_watch`]'s own tick, keyed by
    /// `role`.
    subagent_role_lines: Gauge<u64>,
    /// Recorded from [`crate::claude_usage_watch`]'s own tick, keyed by
    /// `window`.
    claude_usage_percent: Gauge<f64>,
    claude_usage_resets_at: Gauge<u64>,
    context_used: Gauge<u64>,
    context_size: Gauge<u64>,
    /// Keyed by `model` and `currency`: an ACP agent reports cost in a
    /// currency of its choosing.
    reported_cost: Counter<f64>,
}

/// 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())]);
    }
}

/// Record the current `AGENTS.md` line count. Called from
/// [`crate::claude_md_watch`]'s periodic tick (every ~15 minutes, not
/// per turn) — see the field doc on [`Instruments::claude_md_lines`] for
/// why this doesn't ride [`record`]. No-op when OTEL isn't configured,
/// same as `record`.
pub fn record_claude_md_lines(lines: u64) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.claude_md_lines.record(lines, &[]);
}

/// Record one subagent role file's line count, keyed by `role` (the file
/// stem). Called from [`crate::subagent_role_watch`]'s periodic tick.
/// No-op when OTEL isn't configured, same as [`record`].
pub fn record_subagent_role_lines(role: String, lines: u64) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.subagent_role_lines
        .record(lines, &[KeyValue::new("role", role)]);
}

/// Record one subscription usage window (`five_hour`, `seven_day`, …):
/// `percent` is 0-100, `resets_at` unix seconds. No-op when OTEL isn't
/// configured, same as [`record`].
pub fn record_claude_usage(window: &str, percent: f64, resets_at: Option<u64>) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    let attrs = [KeyValue::new("window", window.to_owned())];
    inst.claude_usage_percent.record(percent, &attrs);
    if let Some(resets_at) = resets_at {
        inst.claude_usage_resets_at.record(resets_at, &attrs);
    }
}

/// Record the context window's fill at turn end: `used` tokens of `size`,
/// the two numbers the web UI's ctx% divides. No-op when OTEL isn't
/// configured, same as [`record`].
pub fn record_context(used: u64, size: u64) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.context_used.record(used, &[]);
    inst.context_size.record(size, &[]);
}

/// Add what an ACP agent reported spending since the last call. No-op when
/// OTEL isn't configured, same as [`record`].
pub fn record_reported_cost(model: &str, cost: &hive_runtime::ReportedCost) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    let attrs = [
        KeyValue::new("model", model.to_owned()),
        KeyValue::new("currency", cost.currency.clone()),
    ];
    inst.reported_cost.add(cost.amount, &attrs);
}

fn build() -> Option<Instruments> {
    if !enabled() {
        tracing::debug!("otel turn-metrics: no endpoint configured, exporter disabled");
        return None;
    }
    let interval = EXPORT_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(),
                claude_md_lines: meter.u64_gauge("hyperhive.agent.claude_md.lines").build(),
                subagent_role_lines: meter
                    .u64_gauge("hyperhive.agent.subagent_role.lines")
                    .build(),
                claude_usage_percent: meter
                    .f64_gauge("hyperhive.agent.claude_usage.percent")
                    .with_unit("%")
                    .build(),
                claude_usage_resets_at: meter
                    .u64_gauge("hyperhive.agent.claude_usage.resets_at")
                    .with_unit("s")
                    .build(),
                context_used: meter
                    .u64_gauge("hyperhive.agent.context.used")
                    .with_unit("{token}")
                    .build(),
                context_size: meter
                    .u64_gauge("hyperhive.agent.context.size")
                    .with_unit("{token}")
                    .build(),
                reported_cost: meter.f64_counter("hyperhive.agent.cost.usage").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/scheduler/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).
pub fn enabled() -> bool {
    std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().is_empty())
}