diff --git a/src/gateway/http.rs b/src/gateway/http.rs index 16f6295..54f32db 100644 --- a/src/gateway/http.rs +++ b/src/gateway/http.rs @@ -689,6 +689,107 @@ pub struct LimitQuery { limit: Option, } +fn scheduler_snapshot(jobs: &[crate::storage::ScheduledJob]) -> Value { + let enabled = jobs.iter().filter(|job| job.enabled).count(); + let failed_jobs = jobs + .iter() + .filter(|job| { + matches!( + job.last_status.as_deref(), + Some("error" | "timeout" | "delivery_error") + ) + }) + .count(); + let next_run_at = jobs + .iter() + .filter(|job| job.enabled) + .map(|job| job.next_run_at) + .min(); + + json!({ + "jobs": jobs.len(), + "enabled": enabled, + "failed_jobs": failed_jobs, + "next_run_at": next_run_at, + }) +} + +pub async fn get_status(State(state): State>) -> Result, ApiError> { + let reload = state.reload.status(); + let metrics = crate::observability::metrics::global_metrics().snapshot(); + let depths = state.bus().queue_depths(); + let active_lanes = state + .outbound_lanes + .load(std::sync::atomic::Ordering::Relaxed); + let ws_connections = state + .ws_connections + .load(std::sync::atomic::Ordering::Relaxed); + let background_tasks = state.task_supervisor.running_count(); + let sessions_total = state.session_manager.session_count().await; + let active_turns = state.session_manager.active_turn_count().await; + + let mut channels = Vec::new(); + for name in state.channel_manager.list_channel_names().await { + let running = state + .channel_manager + .get_channel(&name) + .await + .is_some_and(|channel| channel.is_running()); + channels.push(json!({ + "name": name, + "status": if running { "connected" } else { "stopped" }, + })); + } + + let jobs = state + .storage + .list_scheduled_jobs() + .await + .map_err(ApiError::internal)?; + let scheduler = scheduler_snapshot(&jobs); + let mcp = crate::mcp::get_mcp_status() + .into_iter() + .map(|status| { + json!({ + "name": status.name, + "connected": status.connected, + "tools": status.tools.len(), + }) + }) + .collect::>(); + + Ok(Json(json!({ + "generation": reload.generation, + "version": env!("CARGO_PKG_VERSION"), + "uptime_secs": crate::gateway::process_uptime_secs(), + "phase": reload.phase, + "ws_connections": ws_connections, + "background_tasks": background_tasks, + "sessions": { + "total": sessions_total, + "active_turns": active_turns, + }, + "metrics": { + "tokens_in": metrics.tokens_in, + "tokens_out": metrics.tokens_out, + "cost": metrics.cost, + "tool_calls": metrics.tool_calls, + "turns": metrics.turns, + "turn_latency_p95_ms": metrics.turn_latency_p95_ms, + }, + "bus": { + "inbound": { "depth": depths.inbound_depth, "cap": depths.inbound_cap }, + "outbound": { "depth": depths.outbound_depth, "cap": depths.outbound_cap }, + "control": { "depth": depths.control_depth, "cap": depths.control_cap }, + "active_lanes": active_lanes, + }, + "providers": metrics.providers, + "channels": channels, + "scheduler": scheduler, + "mcp": mcp, + }))) +} + pub async fn get_tasks( State(state): State>, Query(query): Query, @@ -779,6 +880,69 @@ pub async fn get_memories( mod tests { use super::*; + fn scheduled_job( + id: &str, + enabled: bool, + next_run_at: i64, + last_status: Option<&str>, + ) -> crate::storage::ScheduledJob { + crate::storage::ScheduledJob { + id: id.to_string(), + name: id.to_string(), + schedule: crate::scheduler::Schedule::Every { every_ms: 60_000 }, + prompt: String::new(), + channel: "cli_chat".to_string(), + chat_id: "test".to_string(), + model: None, + job_kind: crate::storage::JobKind::Task, + delivery_policy: crate::storage::DeliveryPolicy::Never, + enabled, + delete_after_run: false, + next_run_at, + last_run_at: None, + last_status: last_status.map(str::to_string), + last_error: None, + created_at: 0, + updated_at: 0, + } + } + + #[test] + fn scheduler_snapshot_classifies_failures_and_next_enabled_run() { + let jobs = vec![ + scheduled_job("healthy", true, 300, Some("ok")), + scheduled_job("error", true, 200, Some("error")), + scheduled_job("timeout", false, 100, Some("timeout")), + scheduled_job("delivery", true, 400, Some("delivery_error")), + scheduled_job("other", false, 50, Some("cancelled")), + ]; + + assert_eq!( + scheduler_snapshot(&jobs), + json!({ + "jobs": 5, + "enabled": 3, + "failed_jobs": 3, + "next_run_at": 200, + }) + ); + } + + #[test] + fn scheduler_snapshot_has_null_next_run_without_enabled_jobs() { + let jobs = vec![scheduled_job("disabled", false, 100, None)]; + + assert_eq!( + scheduler_snapshot(&jobs), + json!({ + "jobs": 1, + "enabled": 0, + "failed_jobs": 0, + "next_run_at": null, + }) + ); + } + #[test] fn secrets_are_redacted_and_restored() { let current = json!({"api_key":"real", "nested":{"access_token":"token"}, "safe":"yes"}); diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 3a85eb2..152543b 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -593,6 +593,7 @@ fn build_router(state: Arc) -> Router { ) .route("/api/logs", routing::get(http::get_logs)) .route("/api/tasks", routing::get(http::get_tasks)) + .route("/api/status", routing::get(http::get_status)) .route("/api/jobs", routing::get(http::get_jobs)) .route("/api/jobs/{id}/runs", routing::get(http::get_job_runs)) .route("/api/memories", routing::get(http::get_memories))