soma_cli/
watch.rs

1//! `example watch` — health monitor for the MCP HTTP server.
2//!
3//! Polls `{url}/health` on a fixed interval and emits a single stdout line
4//! whenever the server state changes. Stdout is the event stream; the plugin
5//! monitor runtime delivers each line to Claude as a notification.
6//!
7//! States:
8//!   UP       — /health returned 2xx
9//!   DEGRADED — /health returned a non-2xx HTTP response
10//!   DOWN     — connection refused / timeout / DNS failure
11//!
12//! Only state *changes* produce output, so Claude isn't spammed while the
13//! server is stable. The initial state always emits one line.
14
15use std::time::{Duration, Instant};
16
17use anyhow::Result;
18
19/// Server reachability state.
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21enum ServerState {
22    Up,
23    Degraded(u16),
24    Down,
25}
26
27impl std::fmt::Display for ServerState {
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        match self {
30            ServerState::Up => write!(f, "UP"),
31            ServerState::Degraded(code) => write!(f, "DEGRADED(HTTP {code})"),
32            ServerState::Down => write!(f, "DOWN"),
33        }
34    }
35}
36
37/// Run the health watch loop. Exits only on CTRL+C or fatal error.
38///
39/// Each state change emits exactly one line to stdout. The first poll always
40/// emits regardless of state. All other output goes to stderr so it doesn't
41/// pollute the event stream.
42pub async fn run_watch(base_url: &str, interval_secs: u64) -> Result<()> {
43    if interval_secs == 0 {
44        return Err(anyhow::anyhow!("interval must be at least 1 second"));
45    }
46    let health_url = format!("{}/health", base_url.trim_end_matches('/'));
47    let interval = Duration::from_secs(interval_secs);
48
49    let client = reqwest::Client::builder()
50        .timeout(Duration::from_secs(5))
51        .build()?;
52
53    eprintln!(
54        "[example watch] polling {} every {}s — emitting stdout on state change",
55        health_url, interval_secs
56    );
57
58    let mut last_state: Option<ServerState> = None;
59    let mut state_since = Instant::now();
60
61    loop {
62        let current = probe(&client, &health_url).await;
63        let changed = last_state.map(|s| s != current).unwrap_or(true);
64
65        if changed {
66            let line = format_event(
67                base_url,
68                &current,
69                last_state,
70                state_since.elapsed(),
71                interval_secs,
72            );
73            println!("{line}");
74            state_since = Instant::now();
75            last_state = Some(current);
76        }
77
78        tokio::time::sleep(interval).await;
79    }
80}
81
82/// Probe the health endpoint and classify the result.
83async fn probe(client: &reqwest::Client, url: &str) -> ServerState {
84    match client.get(url).send().await {
85        Ok(r) if r.status().is_success() => ServerState::Up,
86        Ok(r) => ServerState::Degraded(r.status().as_u16()),
87        Err(e) => {
88            // Timeouts and connection refusals are expected transient DOWN states.
89            // Anything else (TLS error, DNS failure, etc.) likely indicates a
90            // configuration problem — log it to stderr so the operator can diagnose.
91            if !e.is_timeout() && !e.is_connect() {
92                eprintln!("[example watch] probe error (may indicate config issue): {e}");
93            }
94            ServerState::Down
95        }
96    }
97}
98
99#[cfg(test)]
100#[path = "watch_tests.rs"]
101mod tests;
102
103/// Format a state-change event line for the monitor stream.
104fn format_event(
105    base_url: &str,
106    current: &ServerState,
107    prev: Option<ServerState>,
108    prev_duration: Duration,
109    interval_secs: u64,
110) -> String {
111    match (current, prev) {
112        // Recovery — bind prev_state so we avoid unwrap and display it cleanly.
113        (ServerState::Up, Some(prev_state @ (ServerState::Down | ServerState::Degraded(_)))) => {
114            format!(
115                "[soma] UP — {} recovered after {}s (was {})",
116                base_url,
117                prev_duration.as_secs(),
118                prev_state,
119            )
120        }
121        // Initial healthy state
122        (ServerState::Up, _) => format!("[soma] UP — {} is healthy", base_url),
123        // Went down
124        (ServerState::Down, _) => format!(
125            "[soma] DOWN — {} is unreachable (retrying every {}s)",
126            base_url, interval_secs
127        ),
128        // Degraded (non-2xx)
129        (ServerState::Degraded(code), _) => {
130            format!("[soma] DEGRADED — {} returned HTTP {code}", base_url)
131        }
132    }
133}
soma_cli/watch.rs

soma_cli/
watch.rs
