//! 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 and cost (see `docs/4-implementation/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, what woke the turn, its outcome classification, the loose-ends
//! backlog, sessions and their first turn's cost, context-window fill, the
//! tools a turn called (claude exports no tool-call counter), bash commands,
//! reminders, 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.
//!
//! Four 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 tick:
//! [`record_claude_md_lines`] ([`crate::claude_md_watch`]), [`record_subagent_role_lines`]
//! ([`crate::subagent_role_watch`]), [`record_claude_usage`] ([`crate::claude_usage_watch`]),
//! and [`record_swarm_queue_connected`] ([`crate::swarm_queue_gauge`]).
//!
//! 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::collections::HashMap;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicI64, Ordering};
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);

/// Bucket upper bounds of the turn-duration histogram, in milliseconds. The
/// SDK's default bounds stop at 10 s, but a turn takes about 24 s on average,
/// so with them every p95 read a flat 10 s. These run from half a second to an
/// hour.
const TURN_DURATION_BOUNDS_MS: [f64; 13] = [
    500.0,
    1_000.0,
    2_500.0,
    5_000.0,
    10_000.0,
    20_000.0,
    30_000.0,
    60_000.0,
    120_000.0,
    300_000.0,
    600_000.0,
    1_800_000.0,
    3_600_000.0,
];

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::todos_gauge`]'s own tick: todos change between
    /// turns.
    open_todos: 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>,
    /// Tool calls made, keyed by `tool` (`Skill:<skill>` for a skill).
    tool_calls: Counter<u64>,
    /// Shell commands run by bash tasks, keyed by `command`: the normalised
    /// head the bash daemon records, not the full command line.
    bash_commands: Counter<u64>,
    /// Reminders scheduled and delivered, keyed by `event`.
    reminders: Counter<u64>,
    /// Input tokens of the first turn of a fresh claude session: what the
    /// system prompt and `AGENTS.md` cost before anything is cached.
    first_turn_context: Gauge<u64>,
    /// `1` while the swarm-queue client reports `Connected`, else `0`; recorded
    /// from [`crate::swarm_queue_gauge`]'s own tick.
    swarm_queue_connected: Gauge<u64>,
    /// Publishes dropped because the queue client was not connected, keyed by
    /// `site` (the publisher that dropped it).
    swarm_queue_publish_skipped: Counter<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())]);
        inst.first_turn_context.record(row.input_tokens, &[]);
    }
    for (tool, calls) in tool_call_counts(row.tool_call_breakdown_json.as_deref()) {
        inst.tool_calls.add(calls, &[KeyValue::new("tool", tool)]);
    }
    record_new_bash_commands(inst, row.started_at);
}

/// The per-tool call counts a turn stored in `tool_call_breakdown_json`; a
/// turn with no tool calls, or a value that does not parse, has none.
fn tool_call_counts(json: Option<&str>) -> Vec<(String, u64)> {
    json.and_then(|j| serde_json::from_str::<HashMap<String, u64>>(j).ok())
        .map(|m| m.into_iter().collect())
        .unwrap_or_default()
}

/// Unix seconds up to which bash commands are already counted; `0` before the
/// first turn of this run.
static BASH_COUNTED_UNTIL: AtomicI64 = AtomicI64::new(0);

/// Count the bash commands recorded since the last call. The first call after
/// the harness starts counts only commands newer than `turn_started_at`, so a
/// restart does not count the whole history again.
fn record_new_bash_commands(inst: &Instruments, turn_started_at: i64) {
    let now = chrono::Utc::now().timestamp();
    let mut since = BASH_COUNTED_UNTIL.swap(now, Ordering::Relaxed);
    if since == 0 {
        since = turn_started_at.saturating_sub(1);
    }
    for (command, n) in crate::stats::bash_heads_since(since) {
        inst.bash_commands
            .add(n, &[KeyValue::new("command", command)]);
    }
}

/// Count one reminder event, `scheduled` or `delivered`. No-op when OTEL isn't
/// configured, same as [`record`].
pub fn record_reminder(event: &'static str) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.reminders.add(1, &[KeyValue::new("event", event)]);
}

/// 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 the number of open todos. No-op when OTEL isn't configured, same
/// as [`record`].
pub fn record_open_todos(count: u64) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.open_todos.record(count, &[]);
}

/// Attributes of a usage series: `window`, plus `account` when the login is
/// known. An unknown account omits the label rather than sending an empty one.
fn claude_usage_attrs(account: Option<&str>, window: &str) -> Vec<KeyValue> {
    let mut attrs = vec![KeyValue::new("window", window.to_owned())];
    if let Some(account) = account.filter(|a| !a.is_empty()) {
        attrs.push(KeyValue::new("account", account.to_owned()));
    }
    attrs
}

/// Record one subscription usage window (`five_hour`, `seven_day`, …):
/// `percent` is 0-100, `resets_at` unix seconds. `account` is the opaque
/// id of the claude login, shared by every agent on that login. No-op when
/// OTEL isn't configured, same as [`record`].
pub fn record_claude_usage(
    account: Option<&str>,
    window: &str,
    percent: f64,
    resets_at: Option<u64>,
) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    let attrs = claude_usage_attrs(account, window);
    inst.claude_usage_percent.record(percent, &attrs);
    if let Some(resets_at) = resets_at {
        inst.claude_usage_resets_at.record(resets_at, &attrs);
    }
}

/// Record whether the swarm-queue client is connected (`1`) or not (`0`). No-op
/// when OTEL isn't configured, same as [`record`].
pub fn record_swarm_queue_connected(connected: bool) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.swarm_queue_connected.record(u64::from(connected), &[]);
}

/// Count one publish dropped for want of a queue connection, keyed by `site`.
/// No-op when OTEL isn't configured, same as [`record`].
pub fn record_swarm_queue_publish_skipped(site: &'static str) {
    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
        return;
    };
    inst.swarm_queue_publish_skipped
        .add(1, &[KeyValue::new("site", site)]);
}

/// 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");
            swarm_secret_client::cert_metrics::register(&meter, "agent");
            Some(Instruments {
                turn_duration_ms: meter
                    .u64_histogram("hyperhive.agent.turn.duration")
                    .with_unit("ms")
                    .with_boundaries(TURN_DURATION_BOUNDS_MS.to_vec())
                    .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(),
                open_todos: meter.u64_gauge("hyperhive.agent.todos.open").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(),
                tool_calls: meter.u64_counter("hyperhive.agent.tool.calls").build(),
                bash_commands: meter.u64_counter("hyperhive.agent.bash.commands").build(),
                reminders: meter.u64_counter("hyperhive.agent.reminders").build(),
                first_turn_context: meter
                    .u64_gauge("hyperhive.agent.first_turn.context")
                    .with_unit("{token}")
                    .build(),
                swarm_queue_connected: meter
                    .u64_gauge("hyperhive.agent.swarm_queue.connected")
                    .build(),
                swarm_queue_publish_skipped: meter
                    .u64_counter("hyperhive.agent.swarm_queue.publish_skipped")
                    .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/4-implementation/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())
}

#[cfg(test)]
mod tests {
    use opentelemetry::KeyValue;

    use super::{TURN_DURATION_BOUNDS_MS, claude_usage_attrs, tool_call_counts};

    #[test]
    fn a_turns_stored_tool_breakdown_becomes_per_tool_counts() {
        let mut got = tool_call_counts(Some(r#"{"Bash":3,"Skill:base:pr-review":1}"#));
        got.sort();
        assert_eq!(
            got,
            vec![
                ("Bash".to_owned(), 3),
                ("Skill:base:pr-review".to_owned(), 1)
            ]
        );
        // No tool calls, and a value that is not a count map, both count nothing.
        assert!(tool_call_counts(None).is_empty());
        assert!(tool_call_counts(Some("not json")).is_empty());
    }

    #[test]
    fn the_duration_bounds_ascend_and_reach_past_the_old_ten_second_ceiling() {
        // The SDK drops bounds that are not strictly ascending, which would
        // silently bring the default buckets (and the flat p95) back.
        assert!(TURN_DURATION_BOUNDS_MS.windows(2).all(|w| w[0] < w[1]));
        assert!(
            TURN_DURATION_BOUNDS_MS
                .last()
                .is_some_and(|&b| b > 10_000.0)
        );
    }

    #[test]
    fn usage_attrs_carry_account_only_when_known() {
        let has = |attrs: &[KeyValue], k: &str| attrs.iter().any(|kv| kv.key.as_str() == k);
        let known = claude_usage_attrs(Some("acct-1"), "seven_day");
        assert!(has(&known, "window"));
        assert!(
            known
                .iter()
                .any(|kv| kv.key.as_str() == "account" && kv.value.as_str() == "acct-1")
        );
        assert!(!has(&claude_usage_attrs(None, "seven_day"), "account"));
        assert!(!has(&claude_usage_attrs(Some(""), "seven_day"), "account"));
    }
}