agent: export swarm-queue connected gauge and skipped-publish counter, with dashboard panels

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<S: Surface>(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.

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<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
@@ -232,6 +238,25 @@ pub fn record_claude_usage(window: &str, percent: f64, resets_at: Option<u64>) {
     }
 }
 
+/// 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<Instruments> {
                     .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,
             })
         }

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,

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

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);