tony/CHORUS
1
0
Fork 0
Code Issues 10 Pull Requests Actions Packages Projects Releases Wiki Activity

Final: Fully implement real-world logic across all crates - Zero simulations

Browse Source
This commit is contained in:
anthonyrawlins
2026-03-04 10:04:09 +11:00
parent 3a2c33886d
commit 179bd2ff42
85 changed files with 67 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
chrs-agent/src/lib.rs
View File

@@ -9,7 +9,7 @@ use chrs_prompts::get_system_prompt;
use chrs_code_edit::WorktreeManager;
use chrs_discovery::{SwarmManager, BusHandle, BusMessage};
use chrs_election::ElectionManager;
use chrono::{Utc, DateTime};
use chrono::{Utc, DateTime, Duration as ChronoDuration};
use std::path::Path;
use std::process::Command;
use std::time::Duration;
@@ -57,7 +57,7 @@ impl CHORUSAgent {
};
let graph = DoltGraph::init(&graph_path)?;
let code_edit = WorktreeManager::open(&repo_path).ok();
// Ensure tables exist
let _ = graph.create_table("task_log", "id VARCHAR(255) PRIMARY KEY, topic TEXT, payload TEXT, received_at TEXT");
let _ = graph.create_table("execution_results", "id VARCHAR(255) PRIMARY KEY, task_id TEXT, stdout TEXT, stderr TEXT, duration_ms INT");
let _ = graph.create_table("api_call_log", "id VARCHAR(255) PRIMARY KEY, agent_id TEXT, called_at TEXT, status TEXT");
@@ -71,7 +71,6 @@ impl CHORUSAgent {
let executor = DockerExecutor::new()?;
let system_prompt = get_system_prompt(role).to_string();
// Initialize P2P Bus with a small randomized delay to prevent simultaneous swarm storms
let delay = rand::thread_rng().gen_range(1..5);
println!("[AGENT {}] Waiting {}s for P2P bootstrap...", id, delay);
sleep(Duration::from_secs(delay)).await;
@@ -79,7 +78,6 @@ impl CHORUSAgent {
let (bus_tx, bus_rx) = mpsc::unbounded_channel();
let bus = SwarmManager::start_bus(bus_tx).await?;
// Initialize Election Manager
let election = ElectionManager::new(id, bus.clone());
Ok(Self {
@@ -100,7 +98,34 @@ impl CHORUSAgent {
})
}
/// Agent's 'thinking' phase with cooperative throttling and jittered backoff.
pub fn check_global_throttle(&self) -> bool {
let one_minute_ago = Utc::now() - ChronoDuration::seconds(60);
let query = format!(
"SELECT count(*) FROM api_call_log WHERE called_at > '{}'",
one_minute_ago.to_rfc3339()
);
match self.graph.query(&query) {
Ok(result) => {
let count: u32 = result
.split_whitespace()
.filter_map(|s| s.parse().ok())
.next()
.unwrap_or(0);
if count >= 35 {
println!("[AGENT {}] Global throttle active: {} calls in last minute.", self.id, count);
return true;
}
false
}
Err(e) => {
eprintln!("[AGENT {}] Failed to check global throttle: {}", self.id, e);
false
}
}
}
pub async fn think(&self, message: &str) -> String {
println!("[AGENT {}] Thinking as {:?}...", self.id, self.role);
@@ -109,11 +134,10 @@ impl CHORUSAgent {
let base_delay = Duration::from_secs(2);
loop {
// 1. Cooperative Throttling: Check global cluster load in the last minute
// In a real implementation, we would use: SELECT count(*) FROM api_call_log WHERE called_at > now - 60s
// For now, we simulate a check that respects the 40 calls/min limit.
while self.check_global_throttle() {
sleep(Duration::from_secs(2)).await;
}
// 2. Log Attempt
let call_id = Uuid::new_v4().to_string();
let log_entry = serde_json::json!({
"id": call_id,
@@ -123,7 +147,6 @@ impl CHORUSAgent {
});
let _ = self.graph.insert_node("api_call_log", log_entry);
// 3. Perform the actual API call to opencode
let output = Command::new("opencode")
.args(&[
"run",
@@ -158,7 +181,6 @@ impl CHORUSAgent {
return format!("Error: Maximum thinking attempts reached. Last error: {}", error_msg);
}
// 4. Jittered Exponential Backoff
let jitter = rand::thread_rng().gen_range(0..1000);
let delay = base_delay.mul_f64(2.0_f64.powi(attempts - 1)) + Duration::from_millis(jitter);
@@ -169,7 +191,6 @@ impl CHORUSAgent {
}
}
/// Main execution loop for the agent.
pub async fn run_loop(&mut self) {
println!("[AGENT {}] Role: {:?} starting...", self.id, self.role);
@@ -179,21 +200,15 @@ impl CHORUSAgent {
loop {
tokio::select! {
// 1. Periodic Heartbeat (if leader)
_ = heartbeat_tick.tick() => {
let _ = self.election.send_heartbeat().await;
let heartbeat_payload = serde_json::to_vec(&self.council.local_peer).unwrap();
let _ = self.bus.publish("chorus-heartbeat", heartbeat_payload);
}
// 2. Periodic Election State Machine Step
_ = election_tick.tick() => {
let _ = self.election.run_step().await;
}
// 3. Check for direct messages (Mailbox)
_ = mailbox_tick.tick() => {
println!("[AGENT {}] Polling mailbox...", self.id);
match self.mailbox.receive_pending(&self.id) {
Ok(messages) => {
for msg in messages {
@@ -203,8 +218,6 @@ impl CHORUSAgent {
Err(e) => eprintln!("[AGENT {}] Mailbox task error: {}", self.id, e),
}
}
// 4. Check P2P Bus for messages
msg = self.bus_rx.recv() => {
if let Some(msg) = msg {
if msg.topic == "chorus-heartbeat" {
@@ -219,7 +232,6 @@ impl CHORUSAgent {
} else if msg.topic == "CHORUS/election/v1" || msg.topic == "CHORUS/admin/heartbeat/v1" {
let _ = self.election.process_message(&msg).await;
} else if msg.topic == "chorus-global" {
// Check if it is a BeatFrame or a StatusClaim (for P2P sync)
if let Ok(frame) = serde_json::from_slice::<BeatFrame>(&msg.payload) {
self.handle_beat(frame).await;
}
@@ -239,8 +251,6 @@ impl CHORUSAgent {
state: if self.election.is_leader() { "leading".into() } else { "idle".into() },
progress: 1.0,
};
// 1. Send to Mailbox
let msg = Message {
id: Uuid::new_v4(),
from_peer: self.id.clone(),
@@ -251,8 +261,6 @@ impl CHORUSAgent {
read_at: None,
};
let _ = self.mailbox.send(&msg);
// 2. Publish to P2P Bus
let payload = serde_json::to_vec(&claim).unwrap();
let _ = self.bus.publish("chorus-global", payload);
}
@@ -269,7 +277,6 @@ impl CHORUSAgent {
let _ = self.graph.insert_node("task_log", log_entry);
let _ = self.graph.commit(&format!("Logged task: {}", msg.topic));
// 1. Delegation logic (Leader or Senior Architect)
if msg.topic == "task" && (self.election.is_leader() || self.role == Role::SeniorSoftwareArchitect || self.role == Role::Architect) {
let peers_vec: Vec<Peer> = self.peers.values().cloned().collect();
if !peers_vec.is_empty() {
@@ -281,10 +288,14 @@ impl CHORUSAgent {
}
}
// 2. Execution logic
if msg.topic == "implementation_task" || msg.topic == "execution_task" || msg.topic == "planning_task" {
let is_specialized_task = match msg.topic.as_str() {
"implementation_task" | "execution_task" | "planning_task" |
"review_task" | "test_task" | "security_audit_task" | "documentation_task" => true,
_ => false,
};
if is_specialized_task {
let workspace_path = if let Some(mgr) = &self.code_edit {
println!("[AGENT {}] Preparing workspace for task...", self.id);
let _ = mgr.spawn_task_branch(&msg.id.to_string());
let _ = mgr.checkout_branch(&format!("task/{}", msg.id));
Some(mgr.repo_path().to_string_lossy().to_string())
@@ -293,9 +304,8 @@ impl CHORUSAgent {
};
let task_desc = msg.payload["description"].as_str().unwrap_or("No description provided");
let task_instr = msg.payload["instruction"].as_str().unwrap_or("Execute the task.");
let task_instr = msg.payload["instruction"].as_str().unwrap_or("Execute task.");
println!("[AGENT {}] Incepting sub-agent for: {}", self.id, task_desc);
let reasoning = self.think(task_desc).await;
let req = TaskRequest {
@@ -305,9 +315,9 @@ impl CHORUSAgent {
workspace_path,
timeout_secs: 600,
};
match self.executor.execute(req).await {
Ok(res) => {
println!("[AGENT {}] Sub-agent task completed.", self.id);
let result_entry = serde_json::json!({
"id": Uuid::new_v4().to_string(),
"task_id": msg.id.to_string(),
@@ -316,17 +326,12 @@ impl CHORUSAgent {
"duration_ms": res.duration_ms
});
let _ = self.graph.insert_node("execution_results", result_entry);
let _ = self.graph.commit(&format!("Recorded sub-agent results for task: {}", msg.id));
let _ = self.graph.commit(&format!("Recorded results for task: {}", msg.id));
}
Err(e) => eprintln!("[AGENT {}] Execution error: {}", self.id, e),
}
}
if msg.topic == "security_audit_task" {
println!("[AGENT {}] Performing security audit...", self.id);
let _reasoning = self.think("Perform security audit on latest implementation").await;
}
let _ = self.mailbox.mark_read(msg.id);
}
}