riz/mcp-memory
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: consolidate hypotheses, agent_signals, and process_logs into action-based smart tools

Browse Source
This commit is contained in:
Riz Ashraf committed 2026-10-07 12:00:20 +01:00
1 parent 79209da711
commit 3b146f91c2
17 files changed
+438 -359
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/handlers/meta.rs
+161 -176
View File
@@ -1384,84 +1384,73 @@ impl McpTool for GetNextActionableTasksHandler {
}
}
pub struct LogHypothesisHandler;
pub struct HypothesesHandler;
#[async_trait]
impl McpTool for LogHypothesisHandler {
impl McpTool for HypothesesHandler {
fn name(&self) -> &'static str {
"log_hypothesis"
"hypotheses"
}
fn schema(&self) -> Value {
crate::mcp::tool_def::<LogHypothesisTool>(
"log_hypothesis",
"Log a diagnostic hypothesis and associated evidence for a task",
crate::mcp::tool_def::<HypothesesTool>(
"hypotheses",
"Manage diagnostic hypotheses, tested evidence, and status during problem solving: log new hypotheses or query existing ones.",
)
}
async fn execute(&self, args: Value, state: Arc<MemoryState>) -> crate::error::Result<String> {
let req: LogHypothesisTool = serde_json::from_value(args).map_err(|e| e.to_string())?;
let hyp_id = format!(
"HYP-{}",
uuid::Uuid::new_v4().to_string()[..8].to_uppercase()
);
let timestamp = now_secs();
let req: HypothesesTool = serde_json::from_value(args).map_err(|e| e.to_string())?;
let record = crate::models::Hypothesis {
id: hyp_id.clone(),
task_id: req.task_id,
hypothesis: req.hypothesis,
status: req.status.unwrap_or_else(|| "unverified".to_string()),
evidence: req.evidence,
timestamp,
..Default::default()
};
match req.action {
HypothesisAction::Log => {
let hyp_text = req.hypothesis.ok_or_else(|| {
crate::error::AppError::Internal("Missing required 'hypothesis' for action 'log'".to_string())
})?;
let hyp_id = format!(
"HYP-{}",
uuid::Uuid::new_v4().to_string()[..8].to_uppercase()
);
let timestamp = now_secs();
state.code.hypotheses.modify(|h| h.push(record));
let record = crate::models::Hypothesis {
id: hyp_id.clone(),
task_id: req.task_id,
hypothesis: hyp_text,
status: req.status.unwrap_or_else(|| "unverified".to_string()),
evidence: req.evidence,
timestamp,
..Default::default()
};
Ok(format!("Hypothesis '{}' logged successfully.", hyp_id))
}
}
state.code.hypotheses.modify(|h| h.push(record));
pub struct QueryHypothesesHandler;
Ok(format!("Hypothesis '{}' logged successfully.", hyp_id))
}
HypothesisAction::Query => {
let hypotheses = state.code.hypotheses.read_with(|h| h.clone());
let filtered: Vec<_> = hypotheses
.into_iter()
.filter(|h| {
if let Some(tid) = &req.task_id {
if h.task_id.as_ref() != Some(tid) {
return false;
}
}
if let Some(q) = &req.query {
let lq = q.to_lowercase();
return h.hypothesis.to_lowercase().contains(&lq)
|| h.evidence
.as_ref()
.map_or(false, |e| e.to_lowercase().contains(&lq));
}
true
})
.collect();
#[async_trait]
impl McpTool for QueryHypothesesHandler {
fn name(&self) -> &'static str {
"query_hypotheses"
}
fn schema(&self) -> Value {
crate::mcp::tool_def::<QueryHypothesesTool>(
"query_hypotheses",
"Query active diagnostic hypotheses and evidence by task ID or keyword",
)
}
async fn execute(&self, args: Value, state: Arc<MemoryState>) -> crate::error::Result<String> {
let req: QueryHypothesesTool = serde_json::from_value(args).map_err(|e| e.to_string())?;
let hypotheses = state.code.hypotheses.read_with(|h| h.clone());
let filtered: Vec<_> = hypotheses
.into_iter()
.filter(|h| {
if let Some(tid) = &req.task_id {
if h.task_id.as_ref() != Some(tid) {
return false;
}
}
if let Some(q) = &req.query {
let lq = q.to_lowercase();
return h.hypothesis.to_lowercase().contains(&lq)
|| h.evidence
.as_ref()
.map_or(false, |e| e.to_lowercase().contains(&lq));
}
true
})
.collect();
Ok(serde_json::to_string_pretty(&filtered)?)
Ok(serde_json::to_string_pretty(&filtered)?)
}
}
}
}
@@ -1542,117 +1531,111 @@ impl McpTool for GetPreflightContextHandler {
}
}
pub struct BroadcastAgentSignalHandler;
pub struct AgentSignalsHandler;
#[async_trait]
impl McpTool for BroadcastAgentSignalHandler {
impl McpTool for AgentSignalsHandler {
fn name(&self) -> &'static str {
"broadcast_agent_signal"
"agent_signals"
}
fn schema(&self) -> Value {
crate::mcp::tool_def::<BroadcastAgentSignalTool>(
"broadcast_agent_signal",
"Broadcast real-time inter-agent signal to peer subagents.",
crate::mcp::tool_def::<AgentSignalsTool>(
"agent_signals",
"Real-time inter-agent communication bus: broadcast signals or query active signals from peer subagents.",
)
}
async fn execute(&self, args: Value, state: Arc<MemoryState>) -> crate::error::Result<String> {
let req: BroadcastAgentSignalTool =
serde_json::from_value(args).map_err(|e| e.to_string())?;
let req: AgentSignalsTool = serde_json::from_value(args).map_err(|e| e.to_string())?;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
match req.action {
AgentSignalAction::Broadcast => {
let sender = req.sender.ok_or_else(|| {
crate::error::AppError::Internal("Missing required 'sender' for action 'broadcast'".to_string())
})?;
let signal_type = req.signal_type.ok_or_else(|| {
crate::error::AppError::Internal("Missing required 'signal_type' for action 'broadcast'".to_string())
})?;
let payload = req.payload.ok_or_else(|| {
crate::error::AppError::Internal("Missing required 'payload' for action 'broadcast'".to_string())
})?;
let sig_id = format!("sig_{}", timestamp);
let signal = crate::models::AgentSignal {
id: sig_id.clone(),
sender: req.sender.clone(),
signal_type: req.signal_type.clone(),
payload: req.payload,
timestamp,
ttl_seconds: req.ttl_seconds,
..Default::default()
};
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
state.telemetry.agent_signals.modify(|s| {
s.retain(|sig| {
if let Some(ttl) = sig.ttl_seconds {
timestamp <= sig.timestamp + ttl
} else {
true
}
});
s.push(signal);
if s.len() > 500 {
s.remove(0);
let sig_id = format!("sig_{}", timestamp);
let signal = crate::models::AgentSignal {
id: sig_id.clone(),
sender: sender.clone(),
signal_type: signal_type.clone(),
payload,
timestamp,
ttl_seconds: req.ttl_seconds,
..Default::default()
};
state.telemetry.agent_signals.modify(|s| {
s.retain(|sig| {
if let Some(ttl) = sig.ttl_seconds {
timestamp <= sig.timestamp + ttl
} else {
true
}
});
s.push(signal);
if s.len() > 500 {
s.remove(0);
}
});
state.record_activity(
"agent_signal",
&format!("{}: {}", sender, signal_type),
None,
);
Ok(format!(
"Broadcasted signal '{}' from agent '{}'.",
sig_id, sender
))
}
});
state.record_activity(
"agent_signal",
&format!("{}: {}", req.sender, req.signal_type),
None,
);
AgentSignalAction::Query => {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
Ok(format!(
"Broadcasted signal '{}' from agent '{}'.",
sig_id, req.sender
))
}
}
let filtered = state.telemetry.agent_signals.read_with(|signals| {
signals
.iter()
.filter(|s| {
if let Some(ttl) = s.ttl_seconds {
if now > s.timestamp + ttl {
return false;
}
}
if let Some(sender) = &req.sender {
if s.sender.to_lowercase() != sender.to_lowercase() {
return false;
}
}
if let Some(st) = &req.signal_type {
if s.signal_type.to_lowercase() != st.to_lowercase() {
return false;
}
}
true
})
.cloned()
.take(req.limit.unwrap_or(20))
.collect::<Vec<_>>()
});
pub struct QueryAgentSignalsHandler;
#[async_trait]
impl McpTool for QueryAgentSignalsHandler {
fn name(&self) -> &'static str {
"query_agent_signals"
}
fn schema(&self) -> Value {
crate::mcp::tool_def::<QueryAgentSignalsTool>(
"query_agent_signals",
"Query active inter-agent signals from subagent signal bus.",
)
}
async fn execute(&self, args: Value, state: Arc<MemoryState>) -> crate::error::Result<String> {
let req: QueryAgentSignalsTool = serde_json::from_value(args).map_err(|e| e.to_string())?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let filtered = state.telemetry.agent_signals.read_with(|signals| {
signals
.iter()
.filter(|s| {
if let Some(ttl) = s.ttl_seconds {
if now > s.timestamp + ttl {
return false;
}
}
if let Some(sender) = &req.sender {
if s.sender.to_lowercase() != sender.to_lowercase() {
return false;
}
}
if let Some(st) = &req.signal_type {
if s.signal_type.to_lowercase() != st.to_lowercase() {
return false;
}
}
true
})
.cloned()
.take(req.limit.unwrap_or(20))
.collect::<Vec<_>>()
});
Ok(serde_json::to_string_pretty(&filtered)?)
Ok(serde_json::to_string_pretty(&filtered)?)
}
}
}
}
@@ -2048,11 +2031,12 @@ mod tests {
.await;
assert!(res_td_res.is_ok());
// LogHypothesis & QueryHypotheses
let log_hyp = LogHypothesisHandler;
let hyp_res = log_hyp
// Hypotheses
let hyp_handler = HypothesesHandler;
let hyp_res = hyp_handler
.execute(
serde_json::json!({
"action": "log",
"hypothesis": "Caching improves response speed",
"status": "testing"
}),
@@ -2062,9 +2046,8 @@ mod tests {
.unwrap();
assert!(hyp_res.contains("Hypothesis"));
let q_hyp = QueryHypothesesHandler;
let q_hyp_res = q_hyp
.execute(serde_json::json!({}), state.clone())
let q_hyp_res = hyp_handler
.execute(serde_json::json!({"action": "query"}), state.clone())
.await
.unwrap();
assert!(q_hyp_res.contains("Caching improves response speed"));
@@ -2185,10 +2168,11 @@ mod tests {
assert!(q_lin_res.contains("timeline"));
// Agent Signals
let bcast = BroadcastAgentSignalHandler;
let bcast_res = bcast
let sig_handler = AgentSignalsHandler;
let bcast_res = sig_handler
.execute(
serde_json::json!({
"action": "broadcast",
"sender": "Agent1",
"signal_type": "info",
"payload": "Agent starting task"
@@ -2199,9 +2183,8 @@ mod tests {
.unwrap();
assert!(bcast_res.contains("Broadcasted signal"));
let q_sig = QueryAgentSignalsHandler;
let q_sig_res = q_sig
.execute(serde_json::json!({}), state.clone())
let q_sig_res = sig_handler
.execute(serde_json::json!({"action": "query"}), state.clone())
.await
.unwrap();
assert!(q_sig_res.contains("Agent1"));
@@ -2464,11 +2447,12 @@ mod tests {
.unwrap();
assert!(preflight_res2.contains("active_tasks"));
// QueryHypotheses with task_id and query
let q_hyp = QueryHypothesesHandler;
let q_hyp_res = q_hyp
// Hypotheses query with task_id and query
let hyp_handler = HypothesesHandler;
let q_hyp_res = hyp_handler
.execute(
serde_json::json!({
"action": "query",
"task_id": "t-1",
"query": "Caching"
}),
@@ -2563,10 +2547,11 @@ mod tests {
assert!(q_lin_res.contains("lineage_count"));
// Agent signals filtering
let bcast_sig = BroadcastAgentSignalHandler;
bcast_sig
let sig_handler = AgentSignalsHandler;
sig_handler
.execute(
serde_json::json!({
"action": "broadcast",
"sender": "AgentA",
"signal_type": "handshake",
"payload": "status: ready",
@@ -2577,10 +2562,10 @@ mod tests {
.await
.unwrap();
let q_sig = QueryAgentSignalsHandler;
let q_sig_res = q_sig
let q_sig_res = sig_handler
.execute(
serde_json::json!({
"action": "query",
"sender": "AgentA",
"signal_type": "handshake"
}),