use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use axum::{
Router,
extract::{Query, State},
http::StatusCode,
response::IntoResponse,
routing::{any, get, post},
};
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(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,
mpm_metrics_cache: MetricsCache,
http_client: Arc<reqwest::Client>,
stream_client: Arc<reqwest::Client>,
analyze_handle: Arc<McpServiceHandle>,
mcp_handles: Arc<HashMap<String, Arc<McpServiceHandle>>>,
pub(crate) search_socket: Option<Arc<PathBuf>>,
}
impl AppState {
pub fn new(connectors: Vec<Box<dyn ServiceConnector>>) -> Self {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.pool_max_idle_per_host(0)
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("reqwest client init");
let stream_client = reqwest::Client::builder()
.read_timeout(Duration::from_secs(60))
.pool_max_idle_per_host(0)
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("reqwest stream 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 mpm_handle = Arc::new(McpServiceHandle::new(
"trusty-mpm",
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));
handles.insert("trusty-mpm".to_string(), Arc::clone(&mpm_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(),
mpm_metrics_cache: MetricsCache::new(),
http_client: Arc::new(client),
stream_client: Arc::new(stream_client),
analyze_handle,
mcp_handles: Arc::new(handles),
search_socket: None,
}
}
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 mpm_metrics_cache(&self) -> &MetricsCache {
&self.mpm_metrics_cache
}
pub fn http_client(&self) -> Arc<reqwest::Client> {
Arc::clone(&self.http_client)
}
pub fn stream_client(&self) -> Arc<reqwest::Client> {
Arc::clone(&self.stream_client)
}
}
pub fn build_router(state: AppState) -> Router {
build_router_with_self_origins(state, crate::routes::origin_guard::SelfOrigins::default())
}
pub fn build_router_with_webhooks(
state: AppState,
self_origins: crate::routes::origin_guard::SelfOrigins,
ingress: crate::webhook::WebhookIngress,
) -> Router {
build_router_inner(state, self_origins, Some(ingress))
}
pub fn build_router_with_self_origins(
state: AppState,
self_origins: crate::routes::origin_guard::SelfOrigins,
) -> Router {
build_router_inner(state, self_origins, None)
}
fn build_router_inner(
state: AppState,
self_origins: crate::routes::origin_guard::SelfOrigins,
webhook: Option<crate::webhook::WebhookIngress>,
) -> Router {
let core = 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/mpm", get(metrics_mpm_handler))
.route(
"/api/console/sessions",
get(crate::routes::sessions::list_handler).post(crate::routes::sessions::new_handler),
)
.route(
"/api/console/sessions/supervisor",
get(crate::routes::sessions::supervisor_handler),
)
.route(
"/api/console/sessions/supervisor/auto-resume",
axum::routing::post(crate::routes::sessions::auto_resume_handler),
)
.route(
"/api/console/sessions/bulk-delete",
axum::routing::post(crate::routes::sessions::bulk_delete_handler),
)
.route(
"/api/console/sessions/{id}",
get(crate::routes::sessions::get_handler)
.delete(crate::routes::sessions::decommission_handler),
)
.route(
"/api/console/sessions/{id}/activity",
get(crate::routes::sessions::activity_handler),
)
.route(
"/api/console/sessions/{id}/stop",
axum::routing::post(crate::routes::sessions::stop_handler),
)
.route(
"/api/console/sessions/{id}/resume",
axum::routing::post(crate::routes::sessions::resume_handler),
)
.route(
"/api/console/config/mpm",
get(crate::routes::config::get_handler).post(crate::routes::config::post_handler),
)
.route(
"/api/console/memory/palaces/{id}",
axum::routing::delete(crate::routes::deletes::delete_palace_handler),
)
.route(
"/api/console/search/indexes/{id}",
axum::routing::delete(crate::routes::deletes::delete_index_handler),
)
.route(
"/api/console/search/prune-indexes",
post(crate::routes::cleanup::prune_indexes_handler),
)
.route(
"/api/console/memory/palaces/{id}/compact",
post(crate::routes::cleanup::compact_palace_handler),
)
.route(
"/api/console/search/deregister-unjudged",
post(crate::routes::unjudged::deregister_unjudged_handler),
)
.route(
"/api/console/metrics/analyze/indexes",
get(analyze_indexes_handler),
)
.route(
"/api/console/metrics/analyze/visualize",
get(analyze_visualize_handler),
)
.route(
"/api/search/{*path}",
any(crate::search_uds::routes::search_api_handler),
)
.route(
"/proxy/search/{*path}",
any(crate::search_uds::routes::deprecated_search_api_handler),
)
.route("/api/{service}/{*path}", any(crate::proxy::proxy_handler))
.route(
"/proxy/{daemon}/{*path}",
any(crate::proxy::deprecated_proxy_handler),
)
.route("/tools/search", get(crate::tools_ui::search_ui_redirect))
.route("/tools/search/", get(crate::tools_ui::search_ui_index))
.route(
"/tools/search/{*path}",
get(crate::tools_ui::search_ui_asset),
)
.route("/", get(crate::console_ui::spa_index_handler))
.route("/ui", get(crate::console_ui::spa_index_handler))
.route("/ui/", get(crate::console_ui::spa_index_handler))
.route("/ui/{*path}", get(crate::console_ui::spa_asset_handler))
.with_state(state);
let router = match webhook {
Some(ingress) => core.merge(
Router::new()
.route(
"/api/webhooks/{source}",
axum::routing::post(crate::webhook::webhook_handler),
)
.route(
"/api/console/metrics/webhooks",
get(crate::webhook::metrics_webhooks_handler),
)
.with_state(ingress)
.layer(axum::extract::DefaultBodyLimit::max(
crate::webhook::MAX_WEBHOOK_BODY_BYTES,
)),
),
None => core,
};
router
.layer(axum::middleware::from_fn_with_state(
self_origins,
crate::routes::origin_guard::guard_write_origin,
))
.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;
crate::detect::order_for_display(&mut services);
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;
crate::detect::order_for_display(&mut infos); 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(),
}
}
async fn metrics_mpm_handler(State(state): State<AppState>) -> axum::response::Response {
match state.mpm_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()
}
#[cfg(test)]
mod tests;