gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-26 23:50:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add heartbeat integration, planning phase, and auto-repair

Browse Source 
- Add HeartbeatConfig for proactive periodic execution with channel notifications
- Add use_planning option to Worker for ActionPlan generation before tool execution
- Implement tool failure tracking in database (V3 migration)
- Add auto-repair via Builder for broken WASM tools in self_repair.rs
- Record tool failures in Worker for self-repair tracking
- Update .env.example with new configuration options

Co-Authored-By: Claude Opus 4.5 <[email protected]>
This commit is contained in:
Illia Polosukhin
2026-02-03 09:32:01 -08:00
co-authored by Claude Opus 4.5
parent 2df4a4f5f0
commit 235f6aae18
11 changed files with 551 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/agent_loop.rs
+63 -2
View File
@@ -8,13 +8,16 @@ use uuid::Uuid;
use crate::agent::compaction::ContextCompactor;
use crate::agent::context_monitor::ContextMonitor;
use crate::agent::heartbeat::spawn_heartbeat;
use crate::agent::self_repair::DefaultSelfRepair;
use crate::agent::session::{Session, ThreadState};
use crate::agent::session_manager::SessionManager;
use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
use crate::agent::{MessageIntent, RepairTask, Router, Scheduler};
use crate::agent::{
HeartbeatConfig as AgentHeartbeatConfig, MessageIntent, RepairTask, Router, Scheduler,
};
use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusUpdate};
use crate::config::AgentConfig;
use crate::config::{AgentConfig, HeartbeatConfig};
use crate::context::ContextManager;
use crate::error::Error;
use crate::history::Store;
@@ -37,6 +40,7 @@ pub struct Agent {
session_manager: Arc<SessionManager>,
context_monitor: ContextMonitor,
workspace: Option<Arc<Workspace>>,
heartbeat_config: Option<HeartbeatConfig>,
}
impl Agent {
@@ -49,6 +53,7 @@ impl Agent {
tools: Arc<ToolRegistry>,
channels: ChannelManager,
workspace: Option<Arc<Workspace>>,
heartbeat_config: Option<HeartbeatConfig>,
) -> Self {
let context_manager = Arc::new(ContextManager::new(config.max_parallel_jobs));
@@ -74,6 +79,7 @@ impl Agent {
session_manager: Arc::new(SessionManager::new()),
context_monitor: ContextMonitor::new(),
workspace,
heartbeat_config,
}
}
@@ -94,6 +100,58 @@ impl Agent {
repair_task.run().await;
});
// Spawn heartbeat if enabled
let heartbeat_handle = if let Some(ref hb_config) = self.heartbeat_config {
if hb_config.enabled {
if let Some(ref workspace) = self.workspace {
let config = AgentHeartbeatConfig::default()
.with_interval(std::time::Duration::from_secs(hb_config.interval_secs));
// Set up notification channel if configured
let (notify_tx, mut notify_rx) =
tokio::sync::mpsc::channel::<OutgoingResponse>(16);
// Spawn notification forwarder
// We can't clone ChannelManager directly, so we just log the notifications
// The heartbeat system will handle notifications via the response_tx
let notify_channel = hb_config.notify_channel.clone();
let notify_user = hb_config.notify_user.clone();
tokio::spawn(async move {
while let Some(response) = notify_rx.recv().await {
if let (Some(ch), Some(user)) = (&notify_channel, &notify_user) {
// Log the heartbeat notification
// In a full implementation, we'd route this through a shared channel reference
tracing::info!(
"Heartbeat notification for {}/{}: {}",
ch,
user,
&response.content
);
}
}
});
tracing::info!(
"Heartbeat enabled with {}s interval",
hb_config.interval_secs
);
Some(spawn_heartbeat(
config,
workspace.clone(),
self.llm.clone(),
Some(notify_tx),
))
} else {
tracing::warn!("Heartbeat enabled but no workspace available");
None
}
} else {
None
}
} else {
None
};
// Main message loop
tracing::info!("Agent {} ready and listening", self.config.name);
@@ -123,6 +181,9 @@ impl Agent {
// Cleanup
tracing::info!("Agent shutting down...");
repair_handle.abort();
if let Some(handle) = heartbeat_handle {
handle.abort();
}
self.scheduler.stop_all().await;
self.channels.shutdown_all().await?;
src/agent/mod.rs
+1 -1
View File
@@ -29,7 +29,7 @@ pub use context_monitor::{CompactionStrategy, ContextBreakdown, ContextMonitor};
pub use heartbeat::{HeartbeatConfig, HeartbeatResult, HeartbeatRunner, spawn_heartbeat};
pub use router::{MessageIntent, Router};
pub use scheduler::Scheduler;
pub use self_repair::{RepairResult, RepairTask, SelfRepair, StuckJob};
pub use self_repair::{BrokenTool, RepairResult, RepairTask, SelfRepair, StuckJob};
pub use session::{Session, Thread, ThreadState, Turn, TurnState};
pub use session_manager::SessionManager;
pub use submission::{Submission, SubmissionParser, SubmissionResult};
src/agent/scheduler.rs
+1
View File
@@ -120,6 +120,7 @@ impl Scheduler {
self.tools.clone(),
self.store.clone(),
self.config.job_timeout,
self.config.use_planning,
);
// Spawn worker task
src/agent/self_repair.rs
+147 -9
View File
@@ -9,6 +9,8 @@ use uuid::Uuid;
use crate::context::{ContextManager, JobState};
use crate::error::RepairError;
use crate::history::Store;
use crate::tools::{BuildRequirement, Language, SoftwareBuilder, SoftwareType, ToolRegistry};
/// A job that has been detected as stuck.
#[derive(Debug, Clone)]
@@ -26,7 +28,10 @@ pub struct BrokenTool {
pub name: String,
pub failure_count: u32,
pub last_error: Option<String>,
pub first_failure: DateTime<Utc>,
pub last_failure: DateTime<Utc>,
pub last_build_result: Option<serde_json::Value>,
pub repair_attempts: u32,
}
/// Result of a repair attempt.
@@ -63,6 +68,9 @@ pub struct DefaultSelfRepair {
context_manager: Arc<ContextManager>,
stuck_threshold: Duration,
max_repair_attempts: u32,
store: Option<Arc<Store>>,
builder: Option<Arc<dyn SoftwareBuilder>>,
tools: Option<Arc<ToolRegistry>>,
}
impl DefaultSelfRepair {
@@ -76,8 +84,28 @@ impl DefaultSelfRepair {
context_manager,
stuck_threshold,
max_repair_attempts,
store: None,
builder: None,
tools: None,
}
}
/// Add a Store for tool failure tracking.
pub fn with_store(mut self, store: Arc<Store>) -> Self {
self.store = Some(store);
self
}
/// Add a Builder and ToolRegistry for automatic tool repair.
pub fn with_builder(
mut self,
builder: Arc<dyn SoftwareBuilder>,
tools: Arc<ToolRegistry>,
) -> Self {
self.builder = Some(builder);
self.tools = Some(tools);
self
}
}
#[async_trait]
@@ -151,19 +179,129 @@ impl SelfRepair for DefaultSelfRepair {
}
async fn detect_broken_tools(&self) -> Vec<BrokenTool> {
// TODO: Implement tool failure tracking
// Would need to track tool failures in the database
vec![]
let Some(ref store) = self.store else {
return vec![];
};
// Threshold: 5 failures before considering a tool broken
match store.get_broken_tools(5).await {
Ok(tools) => {
if !tools.is_empty() {
tracing::info!("Detected {} broken tools needing repair", tools.len());
}
tools
}
Err(e) => {
tracing::warn!("Failed to detect broken tools: {}", e);
vec![]
}
}
}
async fn repair_broken_tool(&self, tool: &BrokenTool) -> Result<RepairResult, RepairError> {
// TODO: Implement tool repair via ToolBuilder
Ok(RepairResult::ManualRequired {
message: format!(
"Tool '{}' repair not implemented - manual intervention required",
tool.name
let Some(ref builder) = self.builder else {
return Ok(RepairResult::ManualRequired {
message: format!("Builder not available for repairing tool '{}'", tool.name),
});
};
let Some(ref store) = self.store else {
return Ok(RepairResult::ManualRequired {
message: "Store not available for tracking repair".to_string(),
});
};
// Check repair attempt limit
if tool.repair_attempts >= self.max_repair_attempts {
return Ok(RepairResult::ManualRequired {
message: format!(
"Tool '{}' exceeded max repair attempts ({})",
tool.name, self.max_repair_attempts
),
});
}
tracing::info!(
"Attempting to repair tool '{}' (attempt {})",
tool.name,
tool.repair_attempts + 1
);
// Increment repair attempts
if let Err(e) = store.increment_repair_attempts(&tool.name).await {
tracing::warn!("Failed to increment repair attempts: {}", e);
}
// Create BuildRequirement for repair
let requirement = BuildRequirement {
name: tool.name.clone(),
description: format!(
"Repair broken WASM tool.\n\n\
Tool name: {}\n\
Previous error: {}\n\
Failure count: {}\n\n\
Analyze the error, fix the implementation, and rebuild.",
tool.name,
tool.last_error.as_deref().unwrap_or("Unknown error"),
tool.failure_count
),
})
software_type: SoftwareType::WasmTool,
language: Language::Rust,
input_spec: None,
output_spec: None,
dependencies: vec![],
capabilities: vec!["http".to_string(), "workspace".to_string()],
};
// Attempt to build/repair
match builder.build(&requirement).await {
Ok(result) if result.success => {
tracing::info!(
"Successfully rebuilt tool '{}' after {} iterations",
tool.name,
result.iterations
);
// Mark as repaired in database
if let Err(e) = store.mark_tool_repaired(&tool.name).await {
tracing::warn!("Failed to mark tool as repaired: {}", e);
}
// Log if the tool was auto-registered
if result.registered {
tracing::info!("Repaired tool '{}' auto-registered", tool.name);
}
Ok(RepairResult::Success {
message: format!(
"Tool '{}' repaired successfully after {} iterations",
tool.name, result.iterations
),
})
}
Ok(result) => {
// Build completed but failed
tracing::warn!(
"Repair build for '{}' completed but failed: {:?}",
tool.name,
result.error
);
Ok(RepairResult::Retry {
message: format!(
"Repair attempt {} for '{}' failed: {}",
tool.repair_attempts + 1,
tool.name,
result.error.unwrap_or_else(|| "Unknown error".to_string())
),
})
}
Err(e) => {
tracing::error!("Repair build for '{}' errored: {}", tool.name, e);
Ok(RepairResult::Retry {
message: format!("Repair build error: {}", e),
})
}
}
}
}
src/agent/worker.rs
+153 -1
View File
@@ -12,7 +12,9 @@ use crate::agent::task::TaskOutput;
use crate::context::{ContextManager, JobState};
use crate::error::Error;
use crate::history::Store;
use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext, ToolSelection};
use crate::llm::{
ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, ToolSelection,
};
use crate::safety::SafetyLayer;
use crate::tools::ToolRegistry;
@@ -25,6 +27,8 @@ pub struct Worker {
tools: Arc<ToolRegistry>,
store: Option<Arc<Store>>,
timeout: Duration,
/// Whether to use planning before tool execution.
use_planning: bool,
}
/// Result of a tool execution with metadata for context building.
@@ -44,6 +48,7 @@ impl Worker {
tools: Arc<ToolRegistry>,
store: Option<Arc<Store>>,
timeout: Duration,
use_planning: bool,
) -> Self {
Self {
job_id,
@@ -53,6 +58,7 @@ impl Worker {
tools,
store,
timeout,
use_planning,
}
}
@@ -144,6 +150,50 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
let max_iterations = 50;
let mut iteration = 0;
// Generate plan if planning is enabled
let plan = if self.use_planning {
match reasoning.plan(reason_ctx).await {
Ok(p) => {
tracing::info!(
"Created plan for job {}: {} actions, {:.0}% confidence",
self.job_id,
p.actions.len(),
p.confidence * 100.0
);
// Add plan to context as assistant message
reason_ctx.messages.push(ChatMessage::assistant(format!(
"I've created a plan to accomplish this goal: {}\n\nSteps:\n{}",
p.goal,
p.actions
.iter()
.enumerate()
.map(|(i, a)| format!("{}. {} - {}", i + 1, a.tool_name, a.reasoning))
.collect::<Vec<_>>()
.join("\n")
)));
Some(p)
}
Err(e) => {
tracing::warn!(
"Planning failed for job {}, falling back to direct selection: {}",
self.job_id,
e
);
None
}
}
} else {
None
};
// If we have a plan, execute it
if let Some(ref plan) = plan {
return self.execute_plan(rx, reasoning, reason_ctx, plan).await;
}
// Otherwise, use direct tool selection loop
loop {
// Check for stop signal
if let Ok(msg) = rx.try_recv() {
@@ -401,6 +451,19 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
e
);
// Record failure for self-repair tracking
if let Some(ref store) = self.store {
let store = store.clone();
let tool_name = selection.tool_name.clone();
let error_msg = e.to_string();
tokio::spawn(async move {
if let Err(db_err) = store.record_tool_failure(&tool_name, &error_msg).await
{
tracing::warn!("Failed to record tool failure: {}", db_err);
}
});
}
reason_ctx.messages.push(ChatMessage::tool_result(
"tool_call_id",
&selection.tool_name,
@@ -412,6 +475,95 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
}
}
/// Execute a pre-generated plan.
async fn execute_plan(
&self,
rx: &mut mpsc::Receiver<WorkerMessage>,
reasoning: &Reasoning,
reason_ctx: &mut ReasoningContext,
plan: &ActionPlan,
) -> Result<(), Error> {
for (i, action) in plan.actions.iter().enumerate() {
// Check for stop signal
if let Ok(msg) = rx.try_recv() {
match msg {
WorkerMessage::Stop => {
tracing::debug!(
"Worker for job {} received stop signal during plan execution",
self.job_id
);
return Ok(());
}
WorkerMessage::Ping => {
tracing::trace!("Worker for job {} received ping", self.job_id);
}
WorkerMessage::Start => {}
}
}
tracing::debug!(
"Job {} executing planned action {}/{}: {} - {}",
self.job_id,
i + 1,
plan.actions.len(),
action.tool_name,
action.reasoning
);
// Execute the planned tool
let result = self
.execute_tool(&action.tool_name, &action.parameters)
.await;
// Create a synthetic ToolSelection for process_tool_result
let selection = ToolSelection {
tool_name: action.tool_name.clone(),
parameters: action.parameters.clone(),
reasoning: action.reasoning.clone(),
alternatives: vec![],
};
// Process the result
let completed = self
.process_tool_result(reason_ctx, &selection, result)
.await?;
if completed {
return Ok(());
}
// Small delay between actions
tokio::time::sleep(Duration::from_millis(100)).await;
}
// Plan completed, check with LLM if job is done
reason_ctx.messages.push(ChatMessage::user(
"All planned actions have been executed. Is the job complete? If not, what else needs to be done?",
));
let response = reasoning.respond(reason_ctx).await?;
reason_ctx.messages.push(ChatMessage::assistant(&response));
let response_lower = response.to_lowercase();
if response_lower.contains("complete")
|| response_lower.contains("finished")
|| response_lower.contains("done")
{
self.mark_completed().await?;
} else {
// Job not complete, could re-plan or fall back to direct selection
tracing::info!(
"Job {} plan completed but work remains, falling back to direct selection",
self.job_id
);
// Continue with standard execution loop by returning (will be picked up by main loop)
self.mark_stuck("Plan completed but job incomplete - needs re-planning")
.await?;
}
Ok(())
}
async fn execute_tool(
&self,
tool_name: &str,