use mofa_kernel::agent::context::AgentContext;
use serde::{Serialize, de::DeserializeOwned};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
#[derive(Debug, Clone)]
pub struct ComponentOutput {
pub component: String,
pub output: serde_json::Value,
pub timestamp_ms: u64,
}
#[derive(Debug, Clone, Default)]
pub struct ExecutionMetrics {
pub start_time_ms: u64,
pub end_time_ms: Option<u64>,
pub component_calls: HashMap<String, u64>,
pub total_tokens: u64,
pub tool_calls: u64,
}
impl ExecutionMetrics {
pub fn new() -> Self {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
Self {
start_time_ms: now,
..Default::default()
}
}
pub fn duration_ms(&self) -> u64 {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
self.end_time_ms.unwrap_or(now) - self.start_time_ms
}
}
#[derive(Clone)]
pub struct RichAgentContext {
inner: Arc<AgentContext>,
outputs: Arc<RwLock<Vec<ComponentOutput>>>,
metrics: Arc<RwLock<ExecutionMetrics>>,
}
impl RichAgentContext {
pub fn new(inner: AgentContext) -> Self {
Self {
inner: Arc::new(inner),
outputs: Arc::new(RwLock::new(Vec::new())),
metrics: Arc::new(RwLock::new(ExecutionMetrics::new())),
}
}
pub async fn record_output(&self, component: impl Into<String>, output: serde_json::Value) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
let mut outputs = self.outputs.write().await;
outputs.push(ComponentOutput {
component: component.into(),
output,
timestamp_ms: now,
});
}
pub async fn get_outputs(&self) -> Vec<ComponentOutput> {
let outputs = self.outputs.read().await;
outputs.clone()
}
pub async fn increment_component_calls(&self, component: &str) {
let mut metrics = self.metrics.write().await;
*metrics
.component_calls
.entry(component.to_string())
.or_insert(0) += 1;
}
pub async fn add_tokens(&self, tokens: u64) {
let mut metrics = self.metrics.write().await;
metrics.total_tokens += tokens;
}
pub async fn increment_tool_calls(&self) {
let mut metrics = self.metrics.write().await;
metrics.tool_calls += 1;
}
pub async fn get_metrics(&self) -> ExecutionMetrics {
let metrics = self.metrics.read().await;
metrics.clone()
}
pub async fn finish(&self) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
let mut metrics = self.metrics.write().await;
metrics.end_time_ms = Some(now);
}
pub async fn duration_ms(&self) -> u64 {
let metrics = self.metrics.read().await;
metrics.duration_ms()
}
pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
self.inner.get(key).await
}
pub async fn set<T: Serialize>(&self, key: &str, value: T) {
self.inner.set(key, value).await
}
pub async fn remove(&self, key: &str) -> Option<serde_json::Value> {
self.inner.remove(key).await
}
pub async fn contains(&self, key: &str) -> bool {
self.inner.contains(key).await
}
pub async fn keys(&self) -> Vec<String> {
self.inner.keys().await
}
pub async fn find<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
self.inner.find(key).await
}
pub fn execution_id(&self) -> &str {
&self.inner.execution_id
}
pub fn session_id(&self) -> Option<&str> {
self.inner.session_id.as_deref()
}
pub fn parent(&self) -> Option<&Arc<AgentContext>> {
self.inner.parent()
}
pub fn is_interrupted(&self) -> bool {
self.inner.is_interrupted()
}
pub fn trigger_interrupt(&self) {
self.inner.trigger_interrupt()
}
pub fn clear_interrupt(&self) {
self.inner.clear_interrupt()
}
pub fn config(&self) -> &mofa_kernel::agent::context::ContextConfig {
self.inner.config()
}
pub async fn emit_event(&self, event: mofa_kernel::agent::context::AgentEvent) {
self.inner.emit_event(event).await
}
pub async fn subscribe(
&self,
event_type: &str,
) -> tokio::sync::mpsc::Receiver<mofa_kernel::agent::context::AgentEvent> {
self.inner.subscribe(event_type).await
}
pub fn inner(&self) -> &AgentContext {
&self.inner
}
}
impl From<AgentContext> for RichAgentContext {
fn from(inner: AgentContext) -> Self {
Self::new(inner)
}
}
impl From<RichAgentContext> for AgentContext {
fn from(rich: RichAgentContext) -> Self {
(*rich.inner).clone()
}
}
impl AsRef<AgentContext> for RichAgentContext {
fn as_ref(&self) -> &AgentContext {
&self.inner
}
}