diff --git a/docs/4-implementation/scheduler/observability.md b/docs/4-implementation/scheduler/observability.md index dd193ba5..958a6d58 100644 --- a/docs/4-implementation/scheduler/observability.md +++ b/docs/4-implementation/scheduler/observability.md @@ -411,25 +411,27 @@ of their own — see [ACP-reported cost](#acp-reported-cost). -| Metric | Unit | Kind | Attributes | -| ---------------------------------------- | --------- | --------- | --------------------------------------------------------------------------------------------------------------------- | -| `hyperhive.agent.turn.duration` | `ms` | histogram | `wake_from`, `result_kind`, `model` | -| `hyperhive.agent.turn.count` | — | counter | `wake_from`, `result_kind`, `model` | -| `hyperhive.agent.session.count` | — | counter | `model` (incremented once per fresh, non-`--continue`'d session) | -| `hyperhive.agent.loose_ends.threads` | — | gauge | none | -| `hyperhive.agent.loose_ends.reminders` | — | gauge | none | -| `hyperhive.agent.todos.open` | — | gauge | none — un-acked todos in the harness-local todo store (reminders excluded), recorded every 30 seconds | -| `hyperhive.agent.claude_md.lines` | — | gauge | none — recorded from the `AGENTS.md`-size watch's own ~15-minute tick, **not** per turn like the rows above | -| `hyperhive.agent.subagent_role.lines` | — | gauge | `role` (the file stem) — one series per role file in `subagent_roles/`, on the same ~15-minute tick | -| `hyperhive.agent.claude_usage.percent` | `%` | gauge | `window` (`five_hour`, `seven_day`, … as the usage endpoint names them) — polled every 5 minutes, **not** per turn | -| `hyperhive.agent.claude_usage.resets_at` | `s` | gauge | `window` — unix seconds at which that window resets; same 5-minute poll | -| `hyperhive.agent.context.used` | `{token}` | gauge | none — tokens in the context window at turn end, the numerator of the web UI's ctx% | -| `hyperhive.agent.context.size` | `{token}` | gauge | none — the context window those tokens fill, ctx%'s denominator | -| `hyperhive.agent.cost.usage` | — | counter | `model`, `currency` — ACP agents only, see below | -| `hyperhive.agent.tool.calls` | — | counter | `tool` — the tool's name, or `Skill:` for a skill; counted when a turn ends | -| `hyperhive.agent.bash.commands` | — | counter | `command` — the command name a bash task ran (the first real command, past `cd` and `sudo`), counted when a turn ends | -| `hyperhive.agent.reminders` | — | counter | `event` (`scheduled` or `delivered`) | -| `hyperhive.agent.first_turn.context` | `{token}` | gauge | none — input tokens of the first turn of a fresh session, the cold cost of the system prompt and `AGENTS.md` | +| Metric | Unit | Kind | Attributes | +| --------------------------------------------- | --------- | --------- | ---------------------------------------------------------------------------------------------------------------------------------------- | +| `hyperhive.agent.turn.duration` | `ms` | histogram | `wake_from`, `result_kind`, `model` | +| `hyperhive.agent.turn.count` | — | counter | `wake_from`, `result_kind`, `model` | +| `hyperhive.agent.session.count` | — | counter | `model` (incremented once per fresh, non-`--continue`'d session) | +| `hyperhive.agent.loose_ends.threads` | — | gauge | none | +| `hyperhive.agent.loose_ends.reminders` | — | gauge | none | +| `hyperhive.agent.todos.open` | — | gauge | none — un-acked todos in the harness-local todo store (reminders excluded), recorded every 30 seconds | +| `hyperhive.agent.claude_md.lines` | — | gauge | none — recorded from the `AGENTS.md`-size watch's own ~15-minute tick, **not** per turn like the rows above | +| `hyperhive.agent.subagent_role.lines` | — | gauge | `role` (the file stem) — one series per role file in `subagent_roles/`, on the same ~15-minute tick | +| `hyperhive.agent.claude_usage.percent` | `%` | gauge | `window` (`five_hour`, `seven_day`, … as the usage endpoint names them) — polled every 5 minutes, **not** per turn | +| `hyperhive.agent.claude_usage.resets_at` | `s` | gauge | `window` — unix seconds at which that window resets; same 5-minute poll | +| `hyperhive.agent.context.used` | `{token}` | gauge | none — tokens in the context window at turn end, the numerator of the web UI's ctx% | +| `hyperhive.agent.context.size` | `{token}` | gauge | none — the context window those tokens fill, ctx%'s denominator | +| `hyperhive.agent.cost.usage` | — | counter | `model`, `currency` — ACP agents only, see below | +| `hyperhive.agent.tool.calls` | — | counter | `tool` — the tool's name, or `Skill:` for a skill; counted when a turn ends | +| `hyperhive.agent.bash.commands` | — | counter | `command` — the command name a bash task ran (the first real command, past `cd` and `sudo`), counted when a turn ends | +| `hyperhive.agent.reminders` | — | counter | `event` (`scheduled` or `delivered`) | +| `hyperhive.agent.first_turn.context` | `{token}` | gauge | none — input tokens of the first turn of a fresh session, the cold cost of the system prompt and `AGENTS.md` | +| `hyperhive.agent.swarm_queue.connected` | — | gauge | none — `1` while the swarm-queue client holds a connection, `0` otherwise (including before the first connect), sampled every 30 seconds | +| `hyperhive.agent.swarm_queue.publish_skipped` | — | counter | `site` (`terminal` or `agent_state`) — a publish dropped because the client had no connection | An agent with no `subagent_roles/` directory exports no `subagent_role.lines` series. A role file deleted while the harness runs @@ -450,6 +452,14 @@ API-reported window, else the model's default. A turn that parsed no usage leaves both where the previous turn put them. The dashboard's **Context window used by agent** panel (health tab) divides one by the other. +The `swarm_queue.connected` gauge reads the queue client's connection state: +only `Connected` records `1`. The health tab's **Swarm queue connection by +agent** panel colours `0` red and shows an agent that reports no +`swarm_queue.connected` series, such as one on an older build, as grey with +the text no data, taking the agent list from the container metrics. An agent whose first +connect keeps failing exits after about five minutes, so its gauge reads `0` +only until then. + Resource attributes (`service.name`, `agent`, `swarm`) come from the same container-wide `OTEL_RESOURCE_ATTRIBUTES` as everything else in this section, and the swarm's collector adds `hive` — nothing extra to configure. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 3f404811..c635b74c 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -42,6 +42,7 @@ mod swarm_inference; mod swarm_input; mod swarm_permissions; mod swarm_queue; +mod swarm_queue_gauge; mod swarm_term; mod term_msg; mod todo_server; @@ -489,6 +490,9 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { // not on a publisher's first call, holds that whichever publishers start. swarm_queue::init()?; tokio::spawn(swarm_queue::client()); + // Whether that connection is up, for the agents dashboard: the client + // reconnects in the background, so a detached agent otherwise looks alive. + tokio::spawn(swarm_queue_gauge::run()); // Config fact, stamped once — see `harness_state::write_api_key_mode`'s // doc for why hive-c0re needs this to stop reporting `needs_login` for // an agent whose `~/.claude/` is empty by design. diff --git a/hive-agent/src/otel_turn_metrics.rs b/hive-agent/src/otel_turn_metrics.rs index 4b5651d1..36b4022b 100644 --- a/hive-agent/src/otel_turn_metrics.rs +++ b/hive-agent/src/otel_turn_metrics.rs @@ -18,12 +18,12 @@ //! 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 +//! 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 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`]. +//! 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` @@ -104,6 +104,12 @@ struct Instruments { /// 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, + /// `1` while the swarm-queue client reports `Connected`, else `0`; recorded + /// from [`crate::swarm_queue_gauge`]'s own tick. + swarm_queue_connected: Gauge, + /// Publishes dropped because the queue client was not connected, keyed by + /// `site` (the publisher that dropped it). + swarm_queue_publish_skipped: Counter, } /// Lazily built on the first call to [`record`]. `None` when OTEL isn't @@ -232,6 +238,25 @@ pub fn record_claude_usage(window: &str, percent: f64, resets_at: Option) { } } +/// 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`]. @@ -310,6 +335,12 @@ fn build() -> Option { .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, }) } diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 4c538ced..6b971480 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -315,6 +315,7 @@ async fn publish( // has announced its own, which is smaller than any deployment sets. if let Err(e) = swarm_queue_client::ensure_connected(client) { tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm agent state: publish skipped"); + crate::otel_turn_metrics::record_swarm_queue_publish_skipped("agent_state"); return; } // No degrade path, unlike `swarm_term`'s: every field here is a number, diff --git a/hive-agent/src/swarm_queue_gauge.rs b/hive-agent/src/swarm_queue_gauge.rs new file mode 100644 index 00000000..df239568 --- /dev/null +++ b/hive-agent/src/swarm_queue_gauge.rs @@ -0,0 +1,82 @@ +//! Feeds whether the swarm-queue client is connected into an OTEL gauge on a +//! timer. The client reconnects in the background and keeps running while +//! detached, so the gauge is the only place a detachment shows. + +use std::time::Duration; + +use async_nats::connection::State; + +/// Half the exporter's one-minute interval, so every export carries a +/// value at most one tick old. +const TICK: Duration = Duration::from_secs(30); + +/// Background loop: record the client's connection state every [`TICK`]. +/// Detached task — runs for the harness's lifetime. +/// +/// Records `0` until the first connect has produced a client, so an agent that +/// never reaches the queue reports disconnected instead of nothing. +pub async fn run() { + crate::otel_turn_metrics::record_swarm_queue_connected(false); + let client = crate::swarm_queue::client().await; + watch( + || client.connection_state(), + crate::otel_turn_metrics::record_swarm_queue_connected, + TICK, + ) + .await; +} + +/// Whether `state` counts as connected. `Pending` (never connected) and +/// `Disconnected` both do not. +fn is_connected(state: &State) -> bool { + *state == State::Connected +} + +/// Sample `state` every `tick` and hand `record` whether it is connected. +async fn watch(state: impl Fn() -> State, mut record: impl FnMut(bool), tick: Duration) { + loop { + record(is_connected(&state())); + tokio::time::sleep(tick).await; + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use async_nats::connection::State; + + use super::{is_connected, watch}; + + #[test] + fn only_a_connected_client_counts_as_connected() { + assert!(is_connected(&State::Connected)); + assert!(!is_connected(&State::Disconnected)); + assert!(!is_connected(&State::Pending)); + } + + #[tokio::test(start_paused = true)] + async fn the_gauge_follows_the_connection_state_as_it_changes() { + let state = Arc::new(Mutex::new(State::Connected)); + let seen = Arc::new(Mutex::new(Vec::new())); + let task = tokio::spawn(watch( + { + let state = Arc::clone(&state); + move || state.lock().unwrap().clone() + }, + { + let seen = Arc::clone(&seen); + move |connected| seen.lock().unwrap().push(connected) + }, + Duration::from_secs(10), + )); + tokio::time::sleep(Duration::from_secs(15)).await; + *state.lock().unwrap() = State::Disconnected; + tokio::time::sleep(Duration::from_secs(10)).await; + *state.lock().unwrap() = State::Connected; + tokio::time::sleep(Duration::from_secs(10)).await; + task.abort(); + assert_eq!(*seen.lock().unwrap(), [true, true, false, true]); + } +} diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index b496abc2..413ed5f9 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -83,6 +83,7 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) { // would degrade rows that would have fit. if let Err(e) = swarm_queue_client::ensure_connected(client) { tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm terminal: publish skipped"); + crate::otel_turn_metrics::record_swarm_queue_publish_skipped("terminal"); return; } let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM); diff --git a/nix/host-modules/swarm-grafana/dashboards/agents.json b/nix/host-modules/swarm-grafana/dashboards/agents.json index d468fec1..aee67313 100644 --- a/nix/host-modules/swarm-grafana/dashboards/agents.json +++ b/nix/host-modules/swarm-grafana/dashboards/agents.json @@ -3698,6 +3698,182 @@ } } } + }, + "panel-83": { + "kind": "Panel", + "spec": { + "id": 83, + "title": "Swarm queue connection by agent", + "description": "Whether each agent is connected to the swarm queue. Grey: no data, agent on an older build.", + "links": [], + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "query": { + "kind": "DataQuery", + "group": "prometheus", + "version": "v0", + "datasource": { + "name": "@datasourceUid@" + }, + "spec": { + "expr": "max by (agent) (last_over_time({__name__=\"hyperhive.agent.swarm_queue.connected\", hive=~\"$hive\", agent=~\"$agent\"}[5m])) or (max by (agent) (last_over_time({__name__=\"hyperhive.container.cpu.percent\", hive=~\"$hive\", agent=~\"$agent\"}[5m])) * 0 - 1)", + "legendFormat": "{{agent}}" + } + }, + "refId": "A", + "hidden": false + } + } + ], + "transformations": [], + "queryOptions": {} + } + }, + "vizConfig": { + "kind": "VizConfig", + "group": "state-timeline", + "version": "", + "spec": { + "options": { + "alignValue": "left", + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "mergeValues": true, + "rowHeight": 0.9, + "showValue": "auto", + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "custom": { + "fillOpacity": 80, + "lineWidth": 0 + }, + "mappings": [ + { + "type": "value", + "options": { + "1": { + "text": "connected", + "color": "green", + "index": 0 + }, + "0": { + "text": "disconnected", + "color": "red", + "index": 1 + }, + "-1": { + "text": "no data", + "color": "text", + "index": 2 + } + } + } + ], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "value": null, + "color": "text" + } + ] + } + }, + "overrides": [] + } + } + } + } + }, + "panel-84": { + "kind": "Panel", + "spec": { + "id": 84, + "title": "Skipped queue publishes per hour by agent", + "description": "Publishes an agent dropped because it was not connected to the swarm queue, per hour.", + "links": [], + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "query": { + "kind": "DataQuery", + "group": "prometheus", + "version": "v0", + "datasource": { + "name": "@datasourceUid@" + }, + "spec": { + "expr": "sum by (agent) (rate({__name__=\"hyperhive.agent.swarm_queue.publish_skipped\", hive=~\"$hive\", agent=~\"$agent\"}[$__rate_interval])) * 3600", + "legendFormat": "{{agent}}" + } + }, + "refId": "A", + "hidden": false + } + } + ], + "transformations": [], + "queryOptions": {} + } + }, + "vizConfig": { + "kind": "VizConfig", + "group": "timeseries", + "version": "", + "spec": { + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 1, + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisSoftMin": 0, + "drawStyle": "line", + "fillOpacity": 0, + "lineWidth": 1, + "showPoints": "never", + "spanNulls": false + } + }, + "overrides": [] + } + } + } + } } }, "layout": { @@ -4270,6 +4446,32 @@ "name": "panel-68" } } + }, + { + "kind": "GridLayoutItem", + "spec": { + "x": 0, + "y": 39, + "width": 24, + "height": 7, + "element": { + "kind": "ElementReference", + "name": "panel-83" + } + } + }, + { + "kind": "GridLayoutItem", + "spec": { + "x": 0, + "y": 46, + "width": 24, + "height": 8, + "element": { + "kind": "ElementReference", + "name": "panel-84" + } + } } ] } diff --git a/nix/module-eval/grafana.nix b/nix/module-eval/grafana.nix index 29377bca..37c5acc2 100644 --- a/nix/module-eval/grafana.nix +++ b/nix/module-eval/grafana.nix @@ -505,6 +505,32 @@ let ] && (lib.head xf).spec.options.byField == "agent"; } + { + # The agents board's health tab reads the swarm-queue connection gauge and + # the skipped-publish counter the harness records. A renamed instrument is a + # pair of empty panels, so each must still be spelled in the harness; and the + # connection panel only tells an older build (no series) from a detached + # agent (0) while it falls back to the container metrics for the agent list. + name = "the agents dashboard's health tab reads the swarm-queue metrics the harness records"; + ok = + let + board = builtins.fromJSON (builtins.readFile ../host-modules/swarm-grafana/dashboards/agents.json); + tab = lib.findFirst (t: t.spec.title == "health") null board.spec.layout.spec.tabs; + names = map (i: i.spec.element.name) tab.spec.layout.spec.items; + exprs = lib.concatMap ( + n: map (q: q.spec.query.spec.expr) board.spec.elements.${n}.spec.data.spec.queries + ) names; + recorded = builtins.readFile ../../hive-agent/src/otel_turn_metrics.rs; + connected = lib.findFirst (e: lib.hasInfix "swarm_queue.connected" e) "" exprs; + in + tab != null + && lib.all (m: lib.hasInfix "\"${m}\"" recorded) [ + "hyperhive.agent.swarm_queue.connected" + "hyperhive.agent.swarm_queue.publish_skipped" + ] + && lib.any (e: lib.hasInfix "swarm_queue.publish_skipped" e) exprs + && lib.hasInfix "hyperhive.container.cpu.percent" connected; + } { # The agents board's turns tab reads the turn, tool, bash command, reminder # and session metrics the harness records, grouped by the labels it sets on them. A renamed instrument or