use std::collections::HashSet;
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use serde_json::{Value, json};
use tokio::sync::Mutex;
use tracing::{debug, warn};
use trusty_common::console_metrics::{CONSOLE_METRICS_METHOD, ConsoleMetricsReport, parse_report};
use trusty_common::stdio_mcp_client::StdioMcpClient;
#[cfg(test)]
mod tests;
pub(super) const BACKOFF_BASE_MS: u64 = 1_000;
pub(super) const BACKOFF_CAP_MS: u64 = 60_000;
const DEGRADED_HINT: &str = "reachable but `console_metrics` tool not registered — \
check `serve --stdio` wiring / restart the daemon";
#[derive(Debug)]
pub enum McpHandleError {
Absent,
Backoff {
failure_count: u32,
next_attempt_in: Duration,
},
Degraded {
hint: String,
},
ToolUnavailable {
tool: String,
hint: String,
},
Other(anyhow::Error),
}
impl std::fmt::Display for McpHandleError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Absent => write!(f, "McpServiceHandle: binary not installed on this machine"),
Self::Backoff {
failure_count,
next_attempt_in,
} => write!(
f,
"McpServiceHandle: in backoff after {failure_count} failure(s); \
next attempt in {next_attempt_in:.2?}"
),
Self::Degraded { hint } => {
write!(f, "McpServiceHandle: degraded — {hint}")
}
Self::ToolUnavailable { tool, hint } => {
write!(f, "McpServiceHandle: tool `{tool}` unavailable — {hint}")
}
Self::Other(e) => write!(f, "{e:#}"),
}
}
}
impl std::error::Error for McpHandleError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Other(e) => e.source(),
_ => None,
}
}
}
pub(super) enum HandleState {
Absent,
Connected {
client: Arc<Mutex<Box<StdioMcpClient>>>,
tool_names: HashSet<String>,
daemon_version: String,
},
Degraded,
}
pub(super) struct SpawnBackoff {
pub(super) failure_count: u32,
pub(super) next_attempt: Instant,
}
impl SpawnBackoff {
pub(super) fn new() -> Self {
Self {
failure_count: 0,
next_attempt: Instant::now(),
}
}
pub(super) fn record_failure(&mut self) {
self.failure_count = self.failure_count.saturating_add(1);
let delay_ms = compute_backoff_delay(self.failure_count, BACKOFF_BASE_MS, BACKOFF_CAP_MS);
self.next_attempt = Instant::now() + Duration::from_millis(delay_ms);
}
pub(super) fn reset(&mut self) {
self.failure_count = 0;
self.next_attempt = Instant::now();
}
pub(super) fn should_attempt(&self) -> bool {
Instant::now() >= self.next_attempt
}
}
pub struct McpServiceHandle {
pub(super) binary: String,
pub(super) args: Vec<String>,
pub(super) state: Arc<Mutex<(Option<HandleState>, SpawnBackoff)>>,
}
impl McpServiceHandle {
pub fn new(binary: impl Into<String>, args: Vec<String>) -> Self {
Self {
binary: binary.into(),
args,
state: Arc::new(Mutex::new((None, SpawnBackoff::new()))),
}
}
#[cfg(test)]
pub async fn prime_degraded_for_test(&self) {
self.prime_degraded_with_backoff_for_test(Duration::from_secs(60))
.await;
}
#[cfg(test)]
pub async fn prime_degraded_with_backoff_for_test(&self, future_backoff: Duration) {
let mut guard = self.state.lock().await;
let (state_opt, backoff) = &mut *guard;
*state_opt = Some(HandleState::Degraded);
backoff.failure_count = 1;
backoff.next_attempt = Instant::now() + future_backoff;
}
#[cfg(test)]
pub async fn prime_connected_missing_tool_for_test(&self, missing_tool: &str) {
let client =
trusty_common::stdio_mcp_client::StdioMcpClient::spawn("cat", &[], "test-client")
.await
.expect("cat must be present for test setup");
let client_arc = Arc::new(Mutex::new(Box::new(client)));
let mut tool_names = HashSet::new();
tool_names.insert(CONSOLE_METRICS_METHOD.to_string());
for t in &["extract_graph", "list_entities", "cluster_concepts"] {
if *t != missing_tool {
tool_names.insert(t.to_string());
}
}
tool_names.remove(missing_tool);
let mut guard = self.state.lock().await;
let (state_opt, backoff) = &mut *guard;
backoff.reset();
*state_opt = Some(HandleState::Connected {
client: Arc::clone(&client_arc),
tool_names,
daemon_version: "0.0.0-test".to_string(),
});
}
pub async fn degraded_hint(&self) -> Option<String> {
let guard = self.state.lock().await;
let (state_opt, _) = &*guard;
if matches!(state_opt, Some(HandleState::Degraded)) {
Some(DEGRADED_HINT.to_string())
} else {
None
}
}
pub async fn poll_metrics(&self) -> Result<ConsoleMetricsReport, McpHandleError> {
let (client_arc, _tool_names) = self.ensure_connected().await?;
let mut client_guard = client_arc.lock().await;
let raw = client_guard
.call_tool(CONSOLE_METRICS_METHOD, json!({}))
.await
.with_context(|| {
format!(
"McpServiceHandle: {} tool call failed for {}",
CONSOLE_METRICS_METHOD, self.binary
)
});
drop(client_guard);
match raw {
Ok(value) => {
self.on_call_success().await;
parse_report(&value)
.with_context(|| {
format!("McpServiceHandle: parse_report failed for {}", self.binary)
})
.map_err(McpHandleError::Other)
}
Err(e) => {
self.on_call_failure().await;
Err(McpHandleError::Other(e))
}
}
}
pub async fn call_tool_raw(&self, tool: &str, args: Value) -> Result<Value, McpHandleError> {
let (client_arc, _tool_names) = self.ensure_connected().await?;
let mut client_guard = client_arc.lock().await;
let result = client_guard.call_tool(tool, args).await.with_context(|| {
format!(
"McpServiceHandle: {} tool call failed for {}",
tool, self.binary
)
});
drop(client_guard);
match result {
Ok(raw) => {
self.on_call_success().await;
Ok(unwrap_mcp_content(raw))
}
Err(e) => {
self.on_call_failure().await;
Err(McpHandleError::Other(e))
}
}
}
pub async fn call_tool_checked(
&self,
tool: &str,
args: Value,
) -> Result<Value, McpHandleError> {
let (client_arc, tool_names) = self.ensure_connected().await?;
if !tool_names.contains(tool) {
let hint = format!(
"{} does not expose `{tool}` — rebuild/upgrade the daemon \
(run `cargo install {}`)",
self.binary, self.binary
);
warn!(
binary = %self.binary,
tool = %tool,
"McpServiceHandle: capability-gate rejected call — tool not in cached tool set"
);
return Err(McpHandleError::ToolUnavailable {
tool: tool.to_string(),
hint,
});
}
let mut client_guard = client_arc.lock().await;
let result = client_guard.call_tool(tool, args).await.with_context(|| {
format!(
"McpServiceHandle: {} tool call failed for {}",
tool, self.binary
)
});
drop(client_guard);
match result {
Ok(raw) => {
self.on_call_success().await;
Ok(unwrap_mcp_content(raw))
}
Err(e) => {
self.on_call_failure().await;
Err(McpHandleError::Other(e))
}
}
}
pub async fn daemon_version(&self) -> Option<String> {
let guard = self.state.lock().await;
let (state_opt, _) = &*guard;
if let Some(HandleState::Connected { daemon_version, .. }) = state_opt {
if daemon_version.is_empty() {
None
} else {
Some(daemon_version.clone())
}
} else {
None
}
}
async fn ensure_connected(
&self,
) -> Result<(Arc<Mutex<Box<StdioMcpClient>>>, HashSet<String>), McpHandleError> {
let mut guard = self.state.lock().await;
let maybe_probe: Option<(Arc<Mutex<Box<StdioMcpClient>>>, String)>;
{
let (state_opt, backoff) = &mut *guard;
if matches!(state_opt, Some(HandleState::Degraded)) && backoff.should_attempt() {
warn!(
binary = %self.binary,
"McpServiceHandle: Degraded handle backoff window elapsed — \
re-probing to attempt self-heal"
);
*state_opt = None;
}
if state_opt.is_none() {
let resolved = which::which(&self.binary).ok();
if resolved.is_none() {
warn!(
binary = %self.binary,
"McpServiceHandle: binary not found on PATH — marking as Absent"
);
*state_opt = Some(HandleState::Absent);
maybe_probe = None;
} else {
if !backoff.should_attempt() {
let next_in = backoff
.next_attempt
.saturating_duration_since(Instant::now());
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
next_attempt_secs = ?next_in,
"McpServiceHandle: spawn is in backoff — skipping this cycle"
);
return Err(McpHandleError::Backoff {
failure_count: backoff.failure_count,
next_attempt_in: next_in,
});
}
debug!(binary = %self.binary, "McpServiceHandle: spawning MCP child");
let args_ref: Vec<&str> = self.args.iter().map(String::as_str).collect();
match StdioMcpClient::spawn(&self.binary, &args_ref, "trusty-console").await {
Ok(mut client) => {
let server_info = match client.initialize().await.with_context(|| {
format!(
"McpServiceHandle: MCP initialize failed for {}",
self.binary
)
}) {
Ok(info) => info,
Err(e) => {
backoff.record_failure();
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
error = %e,
"McpServiceHandle: initialize failed — will retry after backoff"
);
return Err(McpHandleError::Other(e));
}
};
let client_arc = Arc::new(Mutex::new(Box::new(client)));
maybe_probe = Some((Arc::clone(&client_arc), server_info.version));
}
Err(e) => {
backoff.record_failure();
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
next_attempt_secs = ?backoff.next_attempt.saturating_duration_since(Instant::now()),
error = %e,
"McpServiceHandle: spawn failed — will retry after backoff"
);
return Err(McpHandleError::Other(e.context(format!(
"McpServiceHandle: failed to spawn {} (failure #{})",
self.binary, backoff.failure_count
))));
}
}
}
} else {
maybe_probe = None;
}
}
if let Some((client_arc, daemon_version)) = maybe_probe {
drop(guard);
let probe_result = client_arc.lock().await.list_tools().await;
guard = self.state.lock().await;
let (state_opt, backoff) = &mut *guard;
match probe_result {
Ok(tools) => {
let tool_names: HashSet<String> =
tools.iter().map(|t| t.name.clone()).collect();
let has_metrics = tool_names.contains(CONSOLE_METRICS_METHOD);
if !has_metrics {
backoff.record_failure();
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
next_attempt_secs = ?backoff.next_attempt.saturating_duration_since(Instant::now()),
"McpServiceHandle: tools/list OK but \
`console_metrics` not listed — marking Degraded; \
will self-heal after backoff window"
);
*state_opt = Some(HandleState::Degraded);
} else {
backoff.reset();
*state_opt = Some(HandleState::Connected {
client: Arc::clone(&client_arc),
tool_names,
daemon_version,
});
}
}
Err(e) => {
backoff.record_failure();
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
error = %e,
"McpServiceHandle: tools/list failed after \
initialize — will retry after backoff"
);
return Err(McpHandleError::Other(e.context(format!(
"McpServiceHandle: tools/list failed for {}",
self.binary
))));
}
}
}
let (final_state, _) = &*guard;
match final_state.as_ref() {
Some(HandleState::Absent) => Err(McpHandleError::Absent),
Some(HandleState::Degraded) => Err(McpHandleError::Degraded {
hint: DEGRADED_HINT.to_string(),
}),
Some(HandleState::Connected {
client, tool_names, ..
}) => Ok((Arc::clone(client), tool_names.clone())),
None => unreachable!("guard must be Some after init block"),
}
}
async fn on_call_success(&self) {
let mut guard = self.state.lock().await;
let (_state_opt, backoff) = &mut *guard;
backoff.reset();
}
async fn on_call_failure(&self) {
let mut guard = self.state.lock().await;
let (state_opt, backoff) = &mut *guard;
backoff.record_failure();
*state_opt = None;
warn!(
binary = %self.binary,
failure_count = backoff.failure_count,
next_attempt_secs = ?backoff.next_attempt.saturating_duration_since(Instant::now()),
"McpServiceHandle: tool call/respawn failed — resetting to None, \
will retry after backoff"
);
}
}
pub(super) fn unwrap_mcp_content(raw: Value) -> Value {
if let Some(text) = raw
.get("content")
.and_then(|c| c.as_array())
.and_then(|arr| arr.first())
.and_then(|item| item.get("text"))
.and_then(|t| t.as_str())
{
match serde_json::from_str::<Value>(text) {
Ok(inner) => return inner,
Err(_) => return Value::String(text.to_string()),
}
}
raw
}
pub fn compute_backoff_delay(attempt: u32, base_ms: u64, cap_ms: u64) -> u64 {
let shift = attempt.saturating_sub(1).min(62);
let raw = base_ms.saturating_mul(1u64 << shift);
raw.min(cap_ms)
}