use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use axum::{
Router,
body::Body,
extract::{Path, Query, State},
http::{Response, StatusCode, header},
response::IntoResponse,
routing::{any, get},
};
use rust_embed::RustEmbed;
use serde::Deserialize;
use serde_json::json;
use tower_http::cors::CorsLayer;
use tower_http::trace::TraceLayer;
use crate::connector::{ServiceConnector, ServiceInfo, ServiceStatus};
use crate::mcp_handle::{McpHandleError, McpServiceHandle};
use crate::metrics_poller::MetricsCache;
use crate::poller::PollerCache;
#[derive(RustEmbed)]
#[folder = "ui/dist/"]
struct UiAssets;
#[derive(Clone)]
pub struct AppState {
connectors: Arc<Vec<Box<dyn ServiceConnector>>>,
poller_cache: PollerCache,
metrics_cache: MetricsCache,
memory_metrics_cache: MetricsCache,
search_metrics_cache: MetricsCache,
review_metrics_cache: MetricsCache,
http_client: Arc<reqwest::Client>,
analyze_handle: Arc<McpServiceHandle>,
mcp_handles: Arc<HashMap<String, Arc<McpServiceHandle>>>,
}
impl AppState {
pub fn new(connectors: Vec<Box<dyn ServiceConnector>>) -> Self {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.build()
.expect("reqwest client init");
let analyze_handle = Arc::new(McpServiceHandle::new(
"trusty-analyze",
vec!["mcp".to_string()],
));
let memory_handle = Arc::new(McpServiceHandle::new(
"trusty-memory",
vec!["serve".to_string(), "--stdio".to_string()],
));
let search_handle = Arc::new(McpServiceHandle::new(
"trusty-search",
vec!["serve".to_string()],
));
let review_handle = Arc::new(McpServiceHandle::new(
"trusty-review",
vec!["serve".to_string(), "--stdio".to_string()],
));
let mut handles: HashMap<String, Arc<McpServiceHandle>> = HashMap::new();
handles.insert("trusty-analyze".to_string(), Arc::clone(&analyze_handle));
handles.insert("trusty-memory".to_string(), Arc::clone(&memory_handle));
handles.insert("trusty-search".to_string(), Arc::clone(&search_handle));
handles.insert("trusty-review".to_string(), Arc::clone(&review_handle));
Self {
connectors: Arc::new(connectors),
poller_cache: PollerCache::new(),
metrics_cache: MetricsCache::new(),
memory_metrics_cache: MetricsCache::new(),
search_metrics_cache: MetricsCache::new(),
review_metrics_cache: MetricsCache::new(),
http_client: Arc::new(client),
analyze_handle,
mcp_handles: Arc::new(handles),
}
}
pub fn mcp_handles(&self) -> Arc<HashMap<String, Arc<McpServiceHandle>>> {
Arc::clone(&self.mcp_handles)
}
pub fn analyze_handle(&self) -> Arc<McpServiceHandle> {
Arc::clone(&self.analyze_handle)
}
pub fn connectors(&self) -> Arc<Vec<Box<dyn ServiceConnector>>> {
Arc::clone(&self.connectors)
}
pub fn poller_cache(&self) -> &PollerCache {
&self.poller_cache
}
pub fn metrics_cache(&self) -> &MetricsCache {
&self.metrics_cache
}
pub fn memory_metrics_cache(&self) -> &MetricsCache {
&self.memory_metrics_cache
}
pub fn search_metrics_cache(&self) -> &MetricsCache {
&self.search_metrics_cache
}
pub fn review_metrics_cache(&self) -> &MetricsCache {
&self.review_metrics_cache
}
pub fn http_client(&self) -> Arc<reqwest::Client> {
Arc::clone(&self.http_client)
}
}
pub fn build_router(state: AppState) -> Router {
Router::new()
.route("/health", get(health_handler))
.route("/api/console/services", get(services_handler))
.route("/api/console/metrics/analyze", get(metrics_analyze_handler))
.route("/api/console/metrics/memory", get(metrics_memory_handler))
.route("/api/console/metrics/search", get(metrics_search_handler))
.route("/api/console/metrics/review", get(metrics_review_handler))
.route(
"/api/console/metrics/analyze/indexes",
get(analyze_indexes_handler),
)
.route(
"/api/console/metrics/analyze/visualize",
get(analyze_visualize_handler),
)
.route("/proxy/{daemon}/{*path}", any(crate::proxy::proxy_handler))
.route("/", get(spa_index_handler))
.route("/ui", get(spa_index_handler))
.route("/ui/", get(spa_index_handler))
.route("/ui/{*path}", get(spa_asset_handler))
.with_state(state)
.layer(CorsLayer::permissive())
.layer(TraceLayer::new_for_http())
}
async fn health_handler() -> impl IntoResponse {
axum::Json(json!({
"status": "ok",
"version": env!("CARGO_PKG_VERSION"),
}))
}
async fn apply_handle_overrides(
infos: &mut [ServiceInfo],
handles: &HashMap<String, Arc<McpServiceHandle>>,
) {
for info in infos.iter_mut() {
if info.status == ServiceStatus::Absent {
continue;
}
if let Some(handle) = handles.get(&info.id) {
if let Some(hint) = handle.degraded_hint().await {
info.status = ServiceStatus::Degraded;
info.hint = Some(hint);
}
if info.version.is_none()
&& let Some(ver) = handle.daemon_version().await
{
info.version = Some(ver);
}
}
}
}
async fn services_handler(State(state): State<AppState>) -> axum::response::Response {
let handles = state.mcp_handles();
if let Some(snap) = state.poller_cache().snapshot().await {
let mut services = snap.services;
apply_handle_overrides(&mut services, &handles).await;
return axum::Json(services).into_response();
}
let connectors = state.connectors();
match tokio::task::spawn_blocking(move || {
connectors.iter().map(|c| c.detect()).collect::<Vec<_>>()
})
.await
{
Ok(mut infos) => {
apply_handle_overrides(&mut infos, &handles).await;
axum::Json(infos).into_response()
}
Err(e) => {
tracing::error!("service detection task panicked: {e}");
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
}
async fn metrics_analyze_handler(State(state): State<AppState>) -> axum::response::Response {
match state.metrics_cache().get().await {
Some(report) => axum::Json(report).into_response(),
None => StatusCode::SERVICE_UNAVAILABLE.into_response(),
}
}
async fn metrics_memory_handler(State(state): State<AppState>) -> axum::response::Response {
match state.memory_metrics_cache().get().await {
Some(report) => axum::Json(report).into_response(),
None => StatusCode::SERVICE_UNAVAILABLE.into_response(),
}
}
async fn metrics_search_handler(State(state): State<AppState>) -> axum::response::Response {
match state.search_metrics_cache().get().await {
Some(report) => axum::Json(report).into_response(),
None => StatusCode::SERVICE_UNAVAILABLE.into_response(),
}
}
async fn metrics_review_handler(State(state): State<AppState>) -> axum::response::Response {
match state.review_metrics_cache().get().await {
Some(report) => axum::Json(report).into_response(),
None => StatusCode::SERVICE_UNAVAILABLE.into_response(),
}
}
#[derive(Deserialize)]
struct VisualizeQuery {
index: Option<String>,
}
async fn analyze_indexes_handler(State(state): State<AppState>) -> axum::response::Response {
match state
.analyze_handle()
.call_tool_checked("list_analyze_indexes", serde_json::json!({}))
.await
{
Ok(val) => axum::Json(val).into_response(),
Err(McpHandleError::ToolUnavailable { tool, hint }) => {
tracing::warn!(
tool = %tool,
hint = %hint,
"analyze_indexes_handler: tool not available — capability-gate triggered"
);
(
StatusCode::SERVICE_UNAVAILABLE,
axum::Json(serde_json::json!({
"status": "degraded",
"hint": hint,
})),
)
.into_response()
}
Err(
McpHandleError::Absent
| McpHandleError::Backoff { .. }
| McpHandleError::Degraded { .. },
) => StatusCode::SERVICE_UNAVAILABLE.into_response(),
Err(e) => {
tracing::warn!("analyze_indexes_handler error: {e:#}");
StatusCode::BAD_GATEWAY.into_response()
}
}
}
async fn analyze_visualize_handler(
State(state): State<AppState>,
Query(params): Query<VisualizeQuery>,
) -> axum::response::Response {
let index_id = match params.index {
Some(id) if !id.is_empty() => id,
_ => {
return (
StatusCode::BAD_REQUEST,
axum::Json(json!({"error": "missing required query param: index"})),
)
.into_response();
}
};
let handle = state.analyze_handle();
let args = serde_json::json!({ "index_id": index_id });
let (graph_res, entities_res, clusters_res) = tokio::join!(
handle.call_tool_checked("extract_graph", args.clone()),
handle.call_tool_checked("list_entities", args.clone()),
handle.call_tool_checked("cluster_concepts", {
let mut a = args.clone();
if let Some(m) = a.as_object_mut() {
m.insert("k".to_string(), serde_json::json!(8));
}
a
}),
);
match &graph_res {
Err(McpHandleError::ToolUnavailable { tool, hint }) => {
tracing::warn!(
tool = %tool,
hint = %hint,
"analyze_visualize_handler: tool not available — capability-gate triggered"
);
return (
StatusCode::SERVICE_UNAVAILABLE,
axum::Json(serde_json::json!({
"status": "degraded",
"hint": hint,
})),
)
.into_response();
}
Err(
McpHandleError::Absent
| McpHandleError::Backoff { .. }
| McpHandleError::Degraded { .. },
) => {
return StatusCode::SERVICE_UNAVAILABLE.into_response();
}
Err(e) => {
tracing::warn!("analyze_visualize_handler graph error: {e:#}");
return StatusCode::BAD_GATEWAY.into_response();
}
Ok(_) => {}
}
if let Err(McpHandleError::ToolUnavailable { tool, .. }) = &entities_res {
tracing::warn!(
tool = %tool,
"analyze_visualize_handler: list_entities tool unavailable — returning partial payload"
);
}
if let Err(McpHandleError::ToolUnavailable { tool, .. }) = &clusters_res {
tracing::warn!(
tool = %tool,
"analyze_visualize_handler: cluster_concepts tool unavailable — returning partial payload"
);
}
let combined = json!({
"graph": graph_res.unwrap_or(serde_json::Value::Null),
"entities": entities_res.unwrap_or(serde_json::Value::Null),
"clusters": clusters_res.unwrap_or(serde_json::Value::Null),
});
axum::Json(combined).into_response()
}
async fn spa_index_handler() -> impl IntoResponse {
serve_asset("index.html")
}
async fn spa_asset_handler(Path(path): Path<String>) -> impl IntoResponse {
let path = path.trim_start_matches('/');
serve_asset(path)
}
fn serve_asset(path: &str) -> Response<Body> {
match UiAssets::get(path) {
Some(content) => {
let mime = mime_guess::from_path(path).first_or_octet_stream();
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, mime.as_ref())
.body(Body::from(content.data.to_vec()))
.unwrap_or_else(|_| {
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::empty())
.expect("static response")
})
}
None => {
match UiAssets::get("index.html") {
Some(content) => Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "text/html")
.body(Body::from(content.data.to_vec()))
.unwrap_or_else(|_| {
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::empty())
.expect("static response")
}),
None => Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("not found"))
.expect("static 404"),
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::header::CONTENT_TYPE;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt;
use crate::connector::{ServiceInfo, ServiceStatus};
struct StubConnector {
id: &'static str,
display_name: &'static str,
status: ServiceStatus,
}
impl ServiceConnector for StubConnector {
fn id(&self) -> &'static str {
self.id
}
fn display_name(&self) -> &'static str {
self.display_name
}
fn detect(&self) -> ServiceInfo {
ServiceInfo {
id: self.id.to_string(),
display_name: self.display_name.to_string(),
status: self.status.clone(),
version: None,
url: None,
hint: None,
}
}
}
fn make_test_state() -> AppState {
AppState::new(vec![
Box::new(StubConnector {
id: "trusty-search",
display_name: "Trusty Search",
status: ServiceStatus::Running,
}),
Box::new(StubConnector {
id: "trusty-memory",
display_name: "Trusty Memory",
status: ServiceStatus::Available,
}),
Box::new(StubConnector {
id: "trusty-analyze",
display_name: "Trusty Analyze",
status: ServiceStatus::Absent,
}),
])
}
async fn get_bytes(resp: axum::http::Response<Body>) -> Vec<u8> {
resp.into_body()
.collect()
.await
.expect("collect body")
.to_bytes()
.to_vec()
}
#[tokio::test]
async fn test_services_route_returns_json() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/services")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let bytes = get_bytes(resp).await;
let body: Vec<serde_json::Value> = serde_json::from_slice(&bytes).expect("parse json");
assert_eq!(body.len(), 3);
assert_eq!(body[0]["id"], "trusty-search");
assert_eq!(body[0]["status"], "running");
assert_eq!(body[0]["display_name"], "Trusty Search");
assert_eq!(body[1]["id"], "trusty-memory");
assert_eq!(body[1]["status"], "available");
assert_eq!(body[2]["id"], "trusty-analyze");
assert_eq!(body[2]["status"], "absent");
}
#[tokio::test]
async fn test_health_route() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/health")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let bytes = get_bytes(resp).await;
let body: serde_json::Value = serde_json::from_slice(&bytes).expect("parse json");
assert_eq!(body["status"], "ok");
assert!(body["version"].is_string());
}
#[tokio::test]
async fn test_services_route_returns_degraded_with_hint() {
use crate::connector::ServiceInfo;
struct DegradedConnector;
impl ServiceConnector for DegradedConnector {
fn id(&self) -> &'static str {
"trusty-analyze"
}
fn display_name(&self) -> &'static str {
"Trusty Analyze"
}
fn detect(&self) -> ServiceInfo {
ServiceInfo {
id: "trusty-analyze".to_string(),
display_name: "Trusty Analyze".to_string(),
status: ServiceStatus::Degraded,
version: None,
url: None,
hint: Some("reachable but `console_metrics` tool not registered".to_string()),
}
}
}
let state = AppState::new(vec![Box::new(DegradedConnector)]);
let router = build_router(state);
let req = Request::builder()
.uri("/api/console/services")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let bytes = get_bytes(resp).await;
let body: Vec<serde_json::Value> = serde_json::from_slice(&bytes).expect("parse json");
assert_eq!(body.len(), 1);
assert_eq!(body[0]["status"], "degraded");
assert!(
body[0].get("hint").is_some(),
"degraded service must include hint field"
);
assert!(
body[0]["hint"]
.as_str()
.unwrap_or("")
.contains("console_metrics"),
"hint must mention console_metrics"
);
}
#[tokio::test]
async fn test_spa_root_returns_html() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let ct = resp
.headers()
.get(CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
assert!(ct.contains("text/html"), "expected text/html, got: {ct}");
}
struct PanicConnector;
impl ServiceConnector for PanicConnector {
fn id(&self) -> &'static str {
"panic-svc"
}
fn display_name(&self) -> &'static str {
"Panic Service"
}
fn detect(&self) -> ServiceInfo {
panic!("intentional test panic from PanicConnector");
}
}
#[tokio::test]
async fn test_services_handler_returns_500_on_panic() {
let state = AppState::new(vec![Box::new(PanicConnector)]);
let router = build_router(state);
let req = Request::builder()
.uri("/api/console/services")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
#[tokio::test]
async fn test_metrics_analyze_route_cold_cache_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/analyze")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn test_proxy_unknown_daemon_returns_400() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/proxy/unknown/health")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn test_proxy_known_daemon_cold_cache_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/proxy/search/health")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn test_metrics_memory_route_cold_cache_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/memory")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn test_metrics_search_route_cold_cache_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/search")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn test_metrics_review_route_cold_cache_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/review")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn test_analyze_indexes_absent_binary_returns_503() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/analyze/indexes")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_ne!(
resp.status(),
StatusCode::INTERNAL_SERVER_ERROR,
"indexes route must not 500 when binary absent"
);
}
#[tokio::test]
async fn test_analyze_visualize_handler_no_index_returns_json_error() {
let router = build_router(make_test_state());
let req = Request::builder()
.uri("/api/console/metrics/analyze/visualize")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(
resp.status(),
StatusCode::BAD_REQUEST,
"missing index param must return 400"
);
let bytes = get_bytes(resp).await;
let body: serde_json::Value = serde_json::from_slice(&bytes).expect("parse json");
assert!(
body.get("error").is_some(),
"expected error field, got: {body}"
);
}
#[tokio::test]
async fn test_services_route_handle_degraded_overlay() {
let state = AppState::new(vec![
Box::new(StubConnector {
id: "trusty-search",
display_name: "Trusty Search",
status: ServiceStatus::Running,
}),
Box::new(StubConnector {
id: "trusty-analyze",
display_name: "Trusty Analyze",
status: ServiceStatus::Absent,
}),
]);
{
let handles = state.mcp_handles();
let search_handle = handles
.get("trusty-search")
.expect("search handle must exist");
search_handle.prime_degraded_for_test().await;
}
let router = build_router(state);
let req = Request::builder()
.uri("/api/console/services")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let bytes = get_bytes(resp).await;
let body: Vec<serde_json::Value> = serde_json::from_slice(&bytes).expect("parse json");
assert_eq!(body.len(), 2);
let search = body
.iter()
.find(|s| s["id"] == "trusty-search")
.expect("search entry");
assert_eq!(
search["status"], "degraded",
"Running service whose handle is Degraded must report degraded, got: {search}"
);
let hint = search["hint"].as_str().unwrap_or("");
assert!(
!hint.is_empty(),
"degraded service must include a non-empty hint"
);
assert!(
hint.contains("console_metrics"),
"hint must mention console_metrics, got: {hint}"
);
let analyze = body
.iter()
.find(|s| s["id"] == "trusty-analyze")
.expect("analyze entry");
assert_eq!(
analyze["status"], "absent",
"Absent service must not be overridden to degraded"
);
}
#[tokio::test]
#[cfg(unix)]
async fn test_analyze_indexes_tool_unavailable_returns_degraded_hint() {
let state = make_test_state();
{
let analyze_handle = state.analyze_handle();
analyze_handle
.prime_connected_missing_tool_for_test("list_analyze_indexes")
.await;
}
let router = build_router(state);
let req = Request::builder()
.uri("/api/console/metrics/analyze/indexes")
.body(Body::empty())
.expect("request");
let resp = router.oneshot(req).await.expect("response");
assert_eq!(
resp.status(),
StatusCode::SERVICE_UNAVAILABLE,
"missing tool must return 503 SERVICE_UNAVAILABLE, not 502 BAD_GATEWAY"
);
let bytes = get_bytes(resp).await;
let body: serde_json::Value = serde_json::from_slice(&bytes).expect("parse json body");
assert_eq!(
body["status"], "degraded",
"response body must have status=degraded, got: {body}"
);
let hint = body["hint"].as_str().unwrap_or("");
assert!(
!hint.is_empty(),
"response body must include a non-empty hint, got: {body}"
);
assert!(
hint.contains("list_analyze_indexes"),
"hint must mention the missing tool name, got: {hint}"
);
}
}