use std::sync::OnceLock;
use std::time::Instant;
use serde::Serialize;
use crate::core::a2a::dlq::DeadLetterQueue;
use super::capsule::global_capsule_store;
use super::registry::OclaRegistry;
use super::response_cache::global_response_cache;
use super::tracing::initialized_collector;
use super::types::{OCLA_API_VERSION, OclaCapability, OclaCapabilityKind, OclaCapabilityStatus};
use super::unified_ledger::{FileUnifiedLedger, UnifiedLedger};
static STARTED_AT: OnceLock<Instant> = OnceLock::new();
static DLQ: OnceLock<DeadLetterQueue> = OnceLock::new();
pub(crate) fn dead_letter_queue() -> &'static DeadLetterQueue {
DLQ.get_or_init(DeadLetterQueue::new)
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
pub struct DlqHealthDetails {
pub total: usize,
pub oldest_age_secs: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub total_entries: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub total_bytes: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_depth: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub hits: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub misses: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub evictions: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub span_count: Option<usize>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct ComponentHealth {
pub name: String,
pub status: HealthStatus,
pub latency_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub details: Option<DlqHealthDetails>,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum HealthStatus {
Healthy,
Degraded(String),
Unhealthy(String),
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct SystemHealth {
pub overall: HealthStatus,
pub components: Vec<ComponentHealth>,
pub uptime_seconds: u64,
pub version: String,
}
pub fn check_system_health() -> SystemHealth {
let started_at = STARTED_AT.get_or_init(Instant::now);
let registry = OclaRegistry::global();
let mut components = Vec::with_capacity(OclaCapabilityKind::ALL.len() + 7);
components.push(poll_capability("observation_hook", || {
registry.observation_hook.capability()
}));
components.push(poll_capability("usage_sink", || {
registry.usage_sink.capability()
}));
components.push(poll_capability("metrics_exporter", || {
registry.metrics_exporter.capability()
}));
components.push(poll_capability("savings_ledger", || {
registry.savings_ledger.capability()
}));
components.push(poll_capability("intent_classifier", || {
registry.intent_classifier.capability()
}));
components.push(poll_capability("outcome_tracker", || {
registry.outcome_tracker.capability()
}));
components.push(poll_capability("compression_provider", || {
registry.compression_provider.capability()
}));
components.push(poll_capability("response_optimizer", || {
registry.response_optimizer.capability()
}));
components.push(poll_capability("model_router", || {
registry.model_router.capability()
}));
components.push(poll_capability("efficiency_analyzer", || {
registry.efficiency_analyzer.capability()
}));
components.push(poll_capability("config_tuner", || {
registry.config_tuner.capability()
}));
components.push(poll_capability("experiment_runner", || {
registry.experiment_runner.capability()
}));
components.push(poll_capability("connector_scheduler", || {
registry.connector_scheduler.capability()
}));
components.push(poll_capability("agent_gateway", || {
registry.agent_gateway.capability()
}));
components.push(check_a2a_bus());
components.push(check_ledger());
components.push(check_budget());
components.push(check_dlq(dead_letter_queue()));
components.push(check_capsule_store());
components.push(check_response_cache());
components.push(check_tracing());
let overall = aggregate_statuses(&components);
SystemHealth {
overall,
components,
uptime_seconds: started_at.elapsed().as_secs(),
version: OCLA_API_VERSION.to_string(),
}
}
fn poll_capability<F>(name: &str, poll: F) -> ComponentHealth
where
F: FnOnce() -> OclaCapability,
{
let started_at = Instant::now();
let capability = poll();
let status = match capability.status {
OclaCapabilityStatus::Available => HealthStatus::Healthy,
OclaCapabilityStatus::Degraded => {
HealthStatus::Degraded("capability reports degraded".into())
}
OclaCapabilityStatus::Unavailable => {
HealthStatus::Unhealthy("capability unavailable".into())
}
};
ComponentHealth {
name: name.to_string(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: None,
}
}
fn check_a2a_bus() -> ComponentHealth {
let started_at = Instant::now();
let status = if crate::core::agents::AgentRegistry::load().is_some() {
HealthStatus::Healthy
} else {
HealthStatus::Degraded("A2A agent registry is unavailable".into())
};
ComponentHealth {
name: "a2a_bus".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: None,
}
}
fn check_ledger() -> ComponentHealth {
let started_at = Instant::now();
let status = match FileUnifiedLedger::from_data_dir().and_then(|ledger| ledger.verify_chain()) {
Ok(true) => HealthStatus::Healthy,
Ok(false) => HealthStatus::Unhealthy("ledger chain integrity check failed".into()),
Err(error) => HealthStatus::Unhealthy(format!("ledger is inaccessible: {error}")),
};
ComponentHealth {
name: "ledger".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: None,
}
}
fn check_budget() -> ComponentHealth {
let started_at = Instant::now();
let snapshot = crate::core::budget_tracker::BudgetTracker::global().check();
let status = match snapshot.worst_level() {
crate::core::budget_tracker::BudgetLevel::Ok => HealthStatus::Healthy,
crate::core::budget_tracker::BudgetLevel::Warning => {
HealthStatus::Degraded("runtime budget warning".into())
}
crate::core::budget_tracker::BudgetLevel::Exhausted => {
HealthStatus::Unhealthy("runtime budget exhausted".into())
}
};
ComponentHealth {
name: "budget".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: None,
}
}
fn check_dlq(queue: &DeadLetterQueue) -> ComponentHealth {
let started_at = Instant::now();
let stats = queue.stats();
let status = if stats.total > 500 {
HealthStatus::Unhealthy(format!("DLQ contains {} entries", stats.total))
} else if stats.total > 100 {
HealthStatus::Degraded(format!("DLQ contains {} entries", stats.total))
} else {
HealthStatus::Healthy
};
ComponentHealth {
name: "dlq".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: Some(DlqHealthDetails {
total: stats.total,
oldest_age_secs: stats.oldest_age_seconds,
..Default::default()
}),
}
}
fn check_capsule_store() -> ComponentHealth {
let started_at = Instant::now();
let stats = global_capsule_store().stats();
let status = if stats.total_entries > 10_000 {
HealthStatus::Degraded(format!(
"capsule store contains {} entries",
stats.total_entries
))
} else {
HealthStatus::Healthy
};
ComponentHealth {
name: "capsule_store".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: Some(DlqHealthDetails {
total_entries: Some(stats.total_entries),
total_bytes: Some(stats.total_bytes),
max_depth: Some(stats.max_depth),
..Default::default()
}),
}
}
fn check_response_cache() -> ComponentHealth {
let started_at = Instant::now();
let stats = global_response_cache().stats();
let status = if stats.evictions > stats.hits {
HealthStatus::Degraded("response cache is thrashing".into())
} else if stats.hit_rate > 0.0 || stats.entries == 0 {
HealthStatus::Healthy
} else {
HealthStatus::Degraded("response cache has no hits".into())
};
ComponentHealth {
name: "response_cache".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: Some(DlqHealthDetails {
total_entries: Some(stats.entries),
hits: Some(stats.hits),
misses: Some(stats.misses),
evictions: Some(stats.evictions),
..Default::default()
}),
}
}
fn check_tracing() -> ComponentHealth {
let started_at = Instant::now();
let (status, span_count) = match initialized_collector() {
Some(collector) => (HealthStatus::Healthy, collector.span_count()),
None => (HealthStatus::Healthy, 0),
};
ComponentHealth {
name: "tracing".into(),
status,
latency_ms: Some(started_at.elapsed().as_millis() as u64),
details: Some(DlqHealthDetails {
span_count: Some(span_count),
..Default::default()
}),
}
}
fn aggregate_statuses(components: &[ComponentHealth]) -> HealthStatus {
if let Some(reason) = components
.iter()
.find_map(|component| match &component.status {
HealthStatus::Unhealthy(reason) => Some(reason.clone()),
_ => None,
})
{
return HealthStatus::Unhealthy(reason);
}
if let Some(reason) = components
.iter()
.find_map(|component| match &component.status {
HealthStatus::Degraded(reason) => Some(reason.clone()),
_ => None,
})
{
return HealthStatus::Degraded(reason);
}
HealthStatus::Healthy
}
#[cfg(test)]
mod tests {
use super::*;
fn component(status: HealthStatus) -> ComponentHealth {
ComponentHealth {
name: "test".into(),
status,
latency_ms: None,
details: None,
}
}
#[test]
fn health_includes_capsule_store() {
let report = check_system_health();
assert!(
report
.components
.iter()
.any(|component| { component.name == "capsule_store" })
);
}
#[test]
fn health_includes_response_cache() {
let report = check_system_health();
assert!(
report
.components
.iter()
.any(|component| { component.name == "response_cache" })
);
}
#[test]
fn health_includes_tracing() {
let report = check_system_health();
assert!(
report
.components
.iter()
.any(|component| { component.name == "tracing" })
);
}
#[test]
fn capsule_store_healthy() {
let health = check_capsule_store();
assert_eq!(health.status, HealthStatus::Healthy);
assert!(
health
.details
.as_ref()
.expect("details")
.total_entries
.is_some(),
"capsule store health must report total_entries"
);
}
#[test]
fn all_healthy_aggregates_to_healthy() {
let components = vec![
component(HealthStatus::Healthy),
component(HealthStatus::Healthy),
];
assert_eq!(aggregate_statuses(&components), HealthStatus::Healthy);
}
#[test]
fn mixed_health_aggregates_to_degraded() {
let components = vec![
component(HealthStatus::Healthy),
component(HealthStatus::Degraded("slow".into())),
];
assert_eq!(
aggregate_statuses(&components),
HealthStatus::Degraded("slow".into())
);
}
#[test]
fn all_unhealthy_aggregates_to_unhealthy() {
let components = vec![
component(HealthStatus::Unhealthy("first failed".into())),
component(HealthStatus::Unhealthy("second failed".into())),
];
assert_eq!(
aggregate_statuses(&components),
HealthStatus::Unhealthy("first failed".into())
);
}
#[test]
fn unhealthy_takes_precedence_over_degraded() {
let components = vec![
component(HealthStatus::Degraded("slow".into())),
component(HealthStatus::Unhealthy("failed".into())),
];
assert_eq!(
aggregate_statuses(&components),
HealthStatus::Unhealthy("failed".into())
);
}
#[test]
fn system_health_reports_all_components() {
let report = check_system_health();
assert_eq!(report.components.len(), 21);
assert_eq!(report.version, OCLA_API_VERSION);
}
#[test]
fn dlq_health_thresholds_and_details_are_reported() {
let queue = DeadLetterQueue::new();
let healthy = check_dlq(&queue);
assert_eq!(healthy.status, HealthStatus::Healthy);
assert_eq!(healthy.details.as_ref().expect("details").total, 0);
for index in 0..101 {
queue.enqueue(crate::core::a2a::dlq::DeadLetter {
id: index.to_string(),
original_message: "message".into(),
target_agent: "agent".into(),
error: "error".into(),
attempts: 1,
first_failed_at: "2026-01-01T00:00:00Z".into(),
last_failed_at: "2026-01-01T00:00:00Z".into(),
});
}
assert!(matches!(
check_dlq(&queue).status,
HealthStatus::Degraded(_)
));
for index in 101..501 {
queue.enqueue(crate::core::a2a::dlq::DeadLetter {
id: index.to_string(),
original_message: "message".into(),
target_agent: "agent".into(),
error: "error".into(),
attempts: 1,
first_failed_at: "2026-01-01T00:00:00Z".into(),
last_failed_at: "2026-01-01T00:00:00Z".into(),
});
}
assert!(matches!(
check_dlq(&queue).status,
HealthStatus::Unhealthy(_)
));
}
}