Implement Result Aggregation, Correlation, and Load Balancing for Multi-Collector Coordination

[unclesp1d3r]

Add result aggregation and load balancing for collector processes

Parent Epic: #114 - Multi-Process Collector Coordination via Event Bus Architecture
Related Issues: #115 (Coordination Workflows), #116 (Task Distribution)
Specification: .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6.4

Design Goals (from Parent #114)

Scope: Single-system, local multi-process coordination

1. Performance: <1ms p99 latency for result aggregation from local processes
2. Cross-Platform: Windows, Linux, macOS, FreeBSD
3. Security: Local IPC only (no network exposure)
4. Code Safety: Zero unsafe code in aggregation implementation
5. Dependencies: Only use interprocess, tokio, prost, dashmap

Critical Architectural Context

Cross-process result aggregation:

• daemoneye-agent (separate process): hosts broker, aggregates results from local collectors
• Collector processes (procmond-1, procmond-2, netmond-1, etc.): separate OS processes on same machine
• All communication via local cross-process IPC using interprocess crate
• Result aggregation and load balancing happen across process boundaries

Overview

This task implements result aggregation and advanced load balancing capabilities within daemoneye-eventbus, enabling the agent to efficiently collect and correlate results from multiple collector processes on the local system.

Architecture

Result Aggregation Flow (Cross-Process)

1. Collector Processes Execute Tasks
   ├─▶ procmond-1: Process enumeration
   ├─▶ procmond-2: Process integrity checks
   └─▶ netmond-1: Network connections

2. Results Published via Local IPC
   ├─▶ procmond-1 → events.process.lifecycle
   ├─▶ procmond-2 → events.process.integrity
   └─▶ netmond-1 → events.network.connections

3. Broker Receives Results (In Agent Process)
   ├─▶ Route by topic pattern
   ├─▶ Group by correlation_id
   └─▶ Buffer until complete

4. Aggregated Results (In Agent Process)
   ├─▶ Merge related events from multiple local processes
   ├─▶ Deduplicate if needed
   └─▶ Forward to detection engine

Result Aggregator (In Broker Process)

/// Aggregates results from multiple local collector processes
pub struct ResultAggregator {
    // Pending result sets indexed by correlation ID
    pending_results: Arc<DashMap<Uuid, ResultSet>>,
    
    // Completion callback
    on_complete: Arc<dyn Fn(AggregatedResult) + Send + Sync>,
    
    // Timeout for incomplete result sets
    timeout_ms: u64,
}

pub struct ResultSet {
    correlation_id: Uuid,
    expected_collectors: HashSet<ProcessId>,
    received_results: Vec<(ProcessId, DetectionResult)>,
    created_at: Instant,
}

impl ResultAggregator {
    /// Add result from local collector process
    pub async fn add_result(
        &self,
        process_id: ProcessId,
        correlation_id: Uuid,
        result: DetectionResult,
    ) -> Result<()> {
        let mut result_set = self.pending_results
            .entry(correlation_id)
            .or_insert_with(|| ResultSet::new(correlation_id));
        
        result_set.received_results.push((process_id, result));
        
        // Check if complete
        if result_set.is_complete() {
            let aggregated = self.aggregate(result_set)?;
            (self.on_complete)(aggregated);
            self.pending_results.remove(&correlation_id);
        }
        
        Ok(())
    }
    
    /// Aggregate results from multiple local processes
    fn aggregate(&self, result_set: ResultSet) -> Result<AggregatedResult> {
        // Merge results, deduplicate, correlate
        let merged = self.merge_results(result_set.received_results)?;
        
        Ok(AggregatedResult {
            correlation_id: result_set.correlation_id,
            results: merged,
            collector_count: result_set.expected_collectors.len(),
            complete: true,
        })
    }
}

Advanced Load Balancer (Across Local Processes)

pub struct AdvancedLoadBalancer {
    // Real-time load metrics per local process
    metrics: Arc<DashMap<ProcessId, LoadMetrics>>,
    
    // Historical performance data
    history: Arc<DashMap<ProcessId, PerformanceHistory>>,
}

pub struct LoadMetrics {
    active_tasks: AtomicUsize,
    queue_depth: AtomicUsize,
    cpu_usage: AtomicU64,  // 0-100%
    memory_bytes: AtomicU64,
    last_heartbeat: AtomicU64,  // Unix timestamp
}

pub struct PerformanceHistory {
    avg_task_duration_ms: f64,
    success_rate: f64,
    total_tasks_completed: u64,
}

impl AdvancedLoadBalancer {
    /// Select best local process using weighted scoring
    pub fn select_weighted(
        &self,
        candidates: Vec<ProcessId>,
    ) -> Result<ProcessId> {
        candidates
            .into_iter()
            .max_by_key(|&pid| {
                self.calculate_score(pid)
            })
            .ok_or(Error::NoCollectorAvailable)
    }
    
    /// Calculate fitness score for local process
    fn calculate_score(&self, process_id: ProcessId) -> u64 {
        let metrics = self.metrics.get(&process_id)?;
        let history = self.history.get(&process_id)?;
        
        // Lower is better for load indicators
        let load_score = 1000 - metrics.active_tasks.load(Ordering::Relaxed) * 10;
        let queue_score = 1000 - metrics.queue_depth.load(Ordering::Relaxed) * 5;
        
        // Higher is better for performance
        let perf_score = (history.success_rate * 100.0) as u64;
        
        load_score + queue_score + perf_score
    }
    
    /// Update metrics from local collector heartbeat
    pub fn update_metrics(&self, process_id: ProcessId, heartbeat: CollectorHeartbeat) {
        let mut metrics = self.metrics.entry(process_id).or_insert_with(LoadMetrics::default);
        
        metrics.active_tasks.store(heartbeat.active_tasks as usize, Ordering::Relaxed);
        metrics.queue_depth.store(heartbeat.queue_depth as usize, Ordering::Relaxed);
        metrics.cpu_usage.store(heartbeat.cpu_usage.to_bits(), Ordering::Relaxed);
        metrics.memory_bytes.store(heartbeat.memory_bytes, Ordering::Relaxed);
    }
}

Failover Manager (For Local Processes)

pub struct FailoverManager {
    health_tracker: Arc<HealthTracker>,
    task_router: Arc<TaskRouter>,
}

impl FailoverManager {
    /// Handle local collector process failure
    pub async fn handle_failure(&self, failed_process_id: ProcessId) -> Result<()> {
        // 1. Mark process as unhealthy
        self.health_tracker.mark_unhealthy(failed_process_id).await?;
        
        // 2. Get pending tasks from failed local process
        let pending_tasks = self.get_pending_tasks(failed_process_id).await?;
        
        // 3. Redistribute to healthy local processes
        for task in pending_tasks {
            self.task_router.route_task(task).await?;
        }
        
        // 4. Clean up subscriptions
        self.cleanup_subscriptions(failed_process_id).await?;
        
        Ok(())
    }
    
    /// Detect failed local processes via heartbeat monitoring
    pub async fn monitor_heartbeats(&self) {
        loop {
            tokio::time::sleep(Duration::from_secs(10)).await;
            
            let now = Instant::now();
            let stale_processes = self.health_tracker
                .find_stale_processes(Duration::from_secs(30))
                .await?;
            
            for process_id in stale_processes {
                warn!("Local collector process {} missed heartbeat, initiating failover", process_id);
                self.handle_failure(process_id).await?;
            }
        }
    }
}

Implementation Steps

1. Result Aggregator

impl ResultAggregator {
    pub fn new(timeout_ms: u64, on_complete: impl Fn(AggregatedResult) + Send + Sync + 'static) -> Self {
        Self {
            pending_results: Arc::new(DashMap::new()),
            on_complete: Arc::new(on_complete),
            timeout_ms,
        }
    }
    
    /// Start timeout monitor for incomplete result sets
    pub fn start_timeout_monitor(self: Arc<Self>) {
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_millis(self.timeout_ms / 2)).await;
                
                let now = Instant::now();
                let mut timed_out = Vec::new();
                
                for entry in self.pending_results.iter() {
                    let result_set = entry.value();
                    if now.duration_since(result_set.created_at).as_millis() as u64 > self.timeout_ms {
                        timed_out.push(*entry.key());
                    }
                }
                
                for correlation_id in timed_out {
                    if let Some((_, result_set)) = self.pending_results.remove(&correlation_id) {
                        warn!("Result set {} timed out, partial results available", correlation_id);
                        let partial = self.aggregate(result_set)?;
                        (self.on_complete)(partial);
                    }
                }
            }
        });
    }
}

2. Advanced Load Balancer

impl AdvancedLoadBalancer {
    /// Update performance history from completed task
    pub fn record_task_completion(
        &self,
        process_id: ProcessId,
        duration_ms: u64,
        success: bool,
    ) {
        let mut history = self.history.entry(process_id).or_insert_with(PerformanceHistory::default);
        
        // Update moving average
        let alpha = 0.3; // Smoothing factor
        history.avg_task_duration_ms = 
            alpha * duration_ms as f64 + (1.0 - alpha) * history.avg_task_duration_ms;
        
        // Update success rate
        let total = history.total_tasks_completed as f64;
        history.success_rate = 
            (history.success_rate * total + if success { 1.0 } else { 0.0 }) / (total + 1.0);
        
        history.total_tasks_completed += 1;
    }
}

3. Failover Manager

impl FailoverManager {
    /// Attempt to restart failed local collector process
    pub async fn attempt_restart(&self, failed_process_id: ProcessId) -> Result<()> {
        // Get collector metadata
        let collector_info = self.get_collector_info(failed_process_id)?;
        
        // Spawn new collector process
        let new_pid = self.spawn_collector_process(&collector_info).await?;
        
        // Wait for registration
        tokio::time::timeout(
            Duration::from_secs(30),
            self.wait_for_registration(new_pid)
        ).await??;
        
        info!("Successfully restarted local collector process {}", new_pid);
        Ok(())
    }
}

Acceptance Criteria

• Result aggregator implemented for local collector processes
• Advanced load balancing with weighted scoring (local processes)
• Failover manager with automatic task redistribution (local processes)
• Heartbeat monitoring and stale process detection
• Performance metrics tracking per local collector process
• Result correlation by correlation_id across local processes
• Timeout handling for incomplete result sets
• Automatic process restart on failure (local processes)
• Performance: <1ms p99 latency for result aggregation
• Zero unsafe code in aggregation and balancing logic
• Unit tests for aggregation logic
• Integration tests with multiple local collector failures
• Cross-platform validation (Windows, Linux, macOS, FreeBSD)
• Documentation of aggregation strategies

Testing Strategy

Unit Tests

• Result merging and deduplication
• Weighted scoring algorithm
• Timeout handling
• Heartbeat staleness detection

Integration Tests

• Aggregate results from multiple local processes
• Load balancing across healthy local processes
• Failover when local process crashes
• Task redistribution on failure

Performance Tests

• Aggregation latency with varying local process counts
• Load balancing decision time
• Failover response time (<3 heartbeat intervals = 30s)
• Target: <1ms p99 latency for aggregation

Performance Requirements

• Result Aggregation Latency: <1ms p99 for merging results from local processes
• Load Balancing Decision Time: <1ms for weighted scoring
• Failover Detection: Within 3 heartbeat intervals (default 30s)
• Task Redistribution: <10s for all pending tasks from failed local process

Code Safety

• Zero unsafe code in aggregation and balancing logic
• Built entirely on: interprocess, tokio, prost, dashmap
• All dependencies are well-audited

---

Status: Not Started
Priority: High
Estimated Effort: 1 week
Depends On: #115 (Coordination Workflows), #116 (Task Distribution)

[traycerai]

Plan

Observations

The DaemonEye project requires implementing result aggregation, correlation, and load balancing for multi-collector coordination as per GitHub issue #117 and spec Section 2.6.4. Current codebase has foundational IPC and event bus from related issues #115 and #116, using interprocess, tokio, prost, and dashmap. Gaps include ResultAggregator for grouping by correlation_id, AdvancedLoadBalancer with weighted scoring based on metrics like active_tasks and success_rate, and FailoverManager for heartbeat monitoring and task redistribution. Architecture emphasizes cross-process local IPC, zero unsafe code, and <1ms p99 latency. Plan introduces these components in daemoneye-lib/src/eventbus, integrates with daemoneye-agent, adds configs, tests, and metrics while ensuring thread-safety with Arc and DashMap.

Approach

1. Result Aggregation: Implement ResultAggregator using DashMap for pending_results, add_result to group by correlation_id, check completion based on expected_collectors, aggregate via merge_results, invoke on_complete callback. Add timeout_monitor task for incomplete sets.\n2. Load Balancing: Create AdvancedLoadBalancer with DashMap for metrics and history, select_weighted using calculate_score (load_score + queue_score + perf_score), update_metrics from heartbeats, record_task_completion for moving averages.\n3. Failover Management: Build FailoverManager with health_tracker and task_router, handle_failure to mark unhealthy, redistribute pending_tasks, cleanup subscriptions. Implement monitor_heartbeats loop for stale detection (30s timeout), optional attempt_restart spawning new process.\n4. Integration: Extend EventBus in daemoneye-agent to use these components, subscribe to result topics, publish tasks via balancer, handle failures. Add configs for timeouts, strategies.\n5. Testing & Metrics: Unit tests for aggregation logic, integration tests for multi-process simulation, performance benchmarks for latency. Track metrics like aggregation_latency, selection_time, failover_count.\nEnsure all async with tokio, prost for proto serialization, no unsafe, cross-platform compatibility.

Reasoning

Examined GitHub issue #117 with detailed pseudocode for ResultAggregator, AdvancedLoadBalancer, and FailoverManager. Reviewed related issues #114 (epic), #115 (workflows), #116 (distribution) for context on event bus and IPC. Analyzed spec .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6.4 for requirements on aggregation, balancing, failover. Inspected existing proto files (ipc.proto, eventbus.proto) for DetectionResult, CollectorHeartbeat, ProcessId. Explored daemoneye-lib/src/ipc and src/collectors for cross-process patterns. Identified integration points with daemoneye-agent's broker. Noted dependencies: interprocess, tokio, prost, dashmap; performance goals <1ms latency, zero unsafe.

Proposed File Changes

📄 daemoneye-lib/src/eventbus/result_aggregator.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/proto/ipc.proto
• 📄 daemoneye-lib/proto/eventbus.proto

Implement ResultAggregator struct with pending_results: Arc<DashMap<Uuid, ResultSet>>, on_complete: Arc<dyn Fn(AggregatedResult) + Send + Sync>, timeout_ms: u64.

Define ResultSet with correlation_id: Uuid, expected_collectors: HashSet, received_results: Vec<(ProcessId, DetectionResult)>, created_at: Instant.

Add new(timeout_ms, on_complete) constructor.

Implement async add_result(process_id, correlation_id, result): insert into pending_results, push to received_results, if is_complete() then aggregate and call on_complete, remove entry.

Add fn aggregate(result_set) -> Result merging results, deduplicating, creating AggregatedResult {correlation_id, results: Vec, collector_count, complete: bool}.

Implement ResultSet::is_complete() checking received unique process_ids == expected_collectors.len().

Add pub fn start_timeout_monitor(self: Arc) spawning tokio task to periodically check and handle timeouts > timeout_ms, calling on_complete with partial if timed out.

Define AggregatedResult and use uuid::Uuid, std::collections::HashSet, std::time::Instant. Integrate with proto DetectionResult via prost.

Include derive(Debug) where appropriate, error handling with thiserror.

📄 daemoneye-lib/src/eventbus/advanced_load_balancer.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/proto/eventbus.proto

Define AdvancedLoadBalancer struct with metrics: Arc<DashMap<ProcessId, LoadMetrics>>, history: Arc<DashMap<ProcessId, PerformanceHistory>>.

LoadMetrics: active_tasks: AtomicUsize, queue_depth: AtomicUsize, cpu_usage: AtomicU64, memory_bytes: AtomicU64, last_heartbeat: AtomicU64.

PerformanceHistory: avg_task_duration_ms: f64, success_rate: f64, total_tasks_completed: u64.

Implement new() with empty DashMaps.

Add pub fn select_weighted(&self, candidates: Vec) -> Result iterating candidates, max_by_key calculate_score, or Err if none.

fn calculate_score(&self, process_id) -> u64: get metrics and history, load_score = 1000 - active_tasks10 - queue_depth5, perf_score = (success_rate * 100.0) as u64, sum scores.

pub fn update_metrics(&self, process_id, heartbeat: CollectorHeartbeat): update atomics from heartbeat fields.

pub fn record_task_completion(&self, process_id, duration_ms: u64, success: bool): update history with moving average (alpha=0.3) for duration, success_rate incrementally, increment total.

Use std::sync::atomic::{AtomicUsize, AtomicU64, Ordering}, std::collections::DashMap. Define ProcessId as u32 or String based on proto. Error with thiserror::Error.

📄 daemoneye-lib/src/eventbus/failover_manager.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/src/eventbus/mod.rs ^(NEW)
• 📄 daemoneye-lib/proto/ipc.proto

Implement FailoverManager struct with health_tracker: Arc, task_router: Arc.

Assume HealthTracker and TaskRouter exist or define minimally: HealthTracker with async mark_unhealthy(pid), find_stale_processes(timeout); TaskRouter with async route_task(task).

pub async fn handle_failure(&self, failed_process_id: ProcessId) -> Result<()>: mark_unhealthy, get_pending_tasks (implement as vec![] or from store), for each route_task, cleanup_subscriptions (log or nop).

pub async fn monitor_heartbeats(&self): loop { sleep(Duration::from_secs(10)).await; let stale = find_stale_processes(Duration::from_secs(30)).await?; for pid in stale { warn!("..."), self.handle_failure(pid).await?; } }

Optional: pub async fn attempt_restart(&self, failed_process_id) -> Result<()>: get_collector_info, spawn_collector_process (use std::process::Command for cross-platform), wait_for_registration with timeout 30s.

Use tokio::time::{sleep, Duration, timeout}, log crate for warn!. Define get_pending_tasks, cleanup_subscriptions as placeholders integrating with eventbus.

📄 daemoneye-lib/src/eventbus/mod.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/src/lib.rs ^(MODIFY)

Create module file declaring pub mod result_aggregator, advanced_load_balancer, failover_manager;

Re-export: pub use result_aggregator::{ResultAggregator, ResultSet, AggregatedResult}; pub use advanced_load_balancer::{AdvancedLoadBalancer, LoadMetrics, PerformanceHistory}; pub use failover_manager::FailoverManager;

If existing, add these mods and re-exports. Add doc comments: "Multi-collector coordination components for result aggregation, load balancing, and failover."

📄 daemoneye-lib/src/lib.rs ^(MODIFY) 🔗

References

• 📄 daemoneye-lib/src/eventbus/mod.rs ^(NEW)

Add pub mod eventbus; if not present.

Re-export new types: pub use eventbus::{ResultAggregator, AdvancedLoadBalancer, FailoverManager, AggregatedResult, LoadMetrics};

Update crate documentation to mention new features for multi-process coordination, reference issue #117.

📄 daemoneye-agent/src/eventbus_integration.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/src/eventbus/mod.rs ^(NEW)
• 📄 daemoneye-agent/src/main.rs
• 📄 daemoneye-lib/src/config.rs ^(MODIFY)

Integrate components in agent: Create EventBusManager struct with result_aggregator: Arc, load_balancer: Arc, failover_manager: Arc, eventbus_client.

In new(): init aggregator with on_complete forwarding to detection engine, balancer, manager with trackers/routers.

Async distribute_task(task): select via load_balancer.select_weighted(candidates from config), publish to topic via client, track if needed.

Async start(): aggregator.start_timeout_monitor(), failover_manager.monitor_heartbeats() in spawn, subscribe to result topics, on receive call aggregator.add_result.

Async handle_heartbeat(heartbeat): load_balancer.update_metrics(pid, heartbeat).

Shutdown: stop tasks.

Use configs for timeout_ms etc. Ensure cross-process pub/sub via interprocess IPC.

📄 daemoneye-lib/src/config.rs ^(MODIFY) 🔗

References

• 📄 daemoneye-lib/src/eventbus/advanced_load_balancer.rs ^(NEW)

Extend existing configs: Add to EventBusConfig { timeout_ms: u64 default 30000, heartbeat_interval_secs: u64 default 10 }.

Add LoadBalancerConfig { strategy: String default "weighted" }.

In CollectorConfig add expected_collectors: Vec for aggregation.

Update serde deserialization, validation for positive values.

📄 daemoneye-lib/tests/result_aggregation_tests.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/src/eventbus/mod.rs ^(NEW)

Unit tests with #[tokio::test]:

• Test ResultAggregator::new and add_result: create, add multiple results, assert is_complete triggers on_complete with merged AggregatedResult.

• Test timeout_monitor: spawn, add incomplete set, advance time, assert partial on_complete called.

• Test AdvancedLoadBalancer::select_weighted: register metrics/history, assert max score selected.

• Test calculate_score: vary active_tasks, success_rate, assert score computation.

• Test record_task_completion: assert moving average updates correctly.

• Test FailoverManager::handle_failure: mock trackers, assert route_task called for pendings.

• Test monitor_heartbeats: simulate stale, assert handle_failure invoked.

Use mockall for mocks, assert_eq! for results, coverage >80%.

📄 daemoneye-lib/tests/integration_multi_collector.rs ^(NEW) 🔗

References

• 📄 daemoneye-agent/src/eventbus_integration.rs ^(NEW)
• 📄 daemoneye-lib/src/ipc/mod.rs

Integration tests simulating multi-process:

• Spawn mock collector processes using tokio::process::Command or threads mimicking IPC.

• Test aggregation: publish results from multiple 'collectors', assert correlated output.

• Test load balancing: distribute tasks, assert selection based on metrics.

• Test failover: 'kill' a collector (stop publishing), assert tasks redistributed, heartbeats missed trigger handle_failure.

• Performance: measure aggregation latency with criterion or manual timing, assert <1ms p99 for 100 iterations.

Use tempdir for logs, env vars for config, cross-platform with if cfg!.

End-to-end: init EventBusManager, distribute tasks, collect aggregated, assert no loss on failure.

Diagram

sequenceDiagram
    participant Agent as daemoneye-agent
    participant EB as EventBus (IPC Broker)
    participant RA as ResultAggregator
    participant ALB as AdvancedLoadBalancer
    participant FM as FailoverManager
    participant Coll1 as procmond-1
    participant Coll2 as netmond-1

    Note over Agent,Coll2: Initialization
    Agent->>ALB: Register collectors (Coll1, Coll2)
    Agent->>RA: Init with on_complete
    RA->>RA: start_timeout_monitor()
    FM->>FM: monitor_heartbeats() spawned
    Coll1->>EB: Heartbeat
    EB->>Agent: Receive heartbeat
    Agent->>ALB: update_metrics(Coll1, heartbeat)
    Coll2->>EB: Heartbeat & register
    Agent->>ALB: update_metrics(Coll2, heartbeat)

    Note over Agent,Coll2: Task Distribution
    Agent->>ALB: select_weighted(candidates)
    ALB-->>Agent: Coll1 (highest score)
    Agent->>EB: Publish task to Coll1 topic
    EB->>Coll1: Deliver task
    Coll1->>EB: Publish result (corr_id)
    EB->>Agent: Route to result topic
    Agent->>RA: add_result(Coll1, corr_id, result)
    RA->>RA: Check complete? (No, wait for others)

    Note over Agent,Coll2: Aggregation & Completion
    Agent->>ALB: select_weighted() -> Coll2
    Agent->>EB: Publish related task
    Coll2->>EB: Publish result (same corr_id)
    EB->>Agent: Deliver
    Agent->>RA: add_result(Coll2, corr_id, result)
    alt Complete
        RA->>RA: aggregate() -> AggregatedResult
        RA->>Detection: on_complete(aggregated)
    end

    Note over Agent,Coll2: Failover
    Coll2--xEB: Miss heartbeats
    FM->>FM: Detect stale (30s)
    FM->>RA/ALB: handle_failure(Coll2)
    FM->>Health: mark_unhealthy(Coll2)
    FM->>TaskRouter: Redistribute pending tasks
    alt Restart
        FM->>FM: attempt_restart(Coll2) -> new_pid
    end

Loading

Import In IDE

🤖 Prompt for AI Agents

I have created the following plan after thorough exploration and analysis of the codebase. Follow the below plan verbatim. Trust the files and references. Do not re-verify what's written in the plan. Explore only when absolutely necessary. First implement all the proposed file changes and then I'll review all the changes together at the end.

### Observations

The DaemonEye project requires implementing result aggregation, correlation, and load balancing for multi-collector coordination as per GitHub issue #117 and spec Section 2.6.4. Current codebase has foundational IPC and event bus from related issues #115 and #116, using interprocess, tokio, prost, and dashmap. Gaps include ResultAggregator for grouping by correlation_id, AdvancedLoadBalancer with weighted scoring based on metrics like active_tasks and success_rate, and FailoverManager for heartbeat monitoring and task redistribution. Architecture emphasizes cross-process local IPC, zero unsafe code, and <1ms p99 latency. Plan introduces these components in daemoneye-lib/src/eventbus, integrates with daemoneye-agent, adds configs, tests, and metrics while ensuring thread-safety with Arc and DashMap.

### Approach

1. **Result Aggregation**: Implement ResultAggregator using DashMap for pending_results, add_result to group by correlation_id, check completion based on expected_collectors, aggregate via merge_results, invoke on_complete callback. Add timeout_monitor task for incomplete sets.\n2. **Load Balancing**: Create AdvancedLoadBalancer with DashMap for metrics and history, select_weighted using calculate_score (load_score + queue_score + perf_score), update_metrics from heartbeats, record_task_completion for moving averages.\n3. **Failover Management**: Build FailoverManager with health_tracker and task_router, handle_failure to mark unhealthy, redistribute pending_tasks, cleanup subscriptions. Implement monitor_heartbeats loop for stale detection (30s timeout), optional attempt_restart spawning new process.\n4. **Integration**: Extend EventBus in daemoneye-agent to use these components, subscribe to result topics, publish tasks via balancer, handle failures. Add configs for timeouts, strategies.\n5. **Testing & Metrics**: Unit tests for aggregation logic, integration tests for multi-process simulation, performance benchmarks for latency. Track metrics like aggregation_latency, selection_time, failover_count.\nEnsure all async with tokio, prost for proto serialization, no unsafe, cross-platform compatibility.

### Reasoning

Examined GitHub issue #117 with detailed pseudocode for ResultAggregator, AdvancedLoadBalancer, and FailoverManager. Reviewed related issues #114 (epic), #115 (workflows), #116 (distribution) for context on event bus and IPC. Analyzed spec .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6.4 for requirements on aggregation, balancing, failover. Inspected existing proto files (ipc.proto, eventbus.proto) for DetectionResult, CollectorHeartbeat, ProcessId. Explored daemoneye-lib/src/ipc and src/collectors for cross-process patterns. Identified integration points with daemoneye-agent's broker. Noted dependencies: interprocess, tokio, prost, dashmap; performance goals <1ms latency, zero unsafe.

## Mermaid Diagram

sequenceDiagram
    participant Agent as daemoneye-agent
    participant EB as EventBus (IPC Broker)
    participant RA as ResultAggregator
    participant ALB as AdvancedLoadBalancer
    participant FM as FailoverManager
    participant Coll1 as procmond-1
    participant Coll2 as netmond-1

    Note over Agent,Coll2: Initialization
    Agent->>ALB: Register collectors (Coll1, Coll2)
    Agent->>RA: Init with on_complete
    RA->>RA: start_timeout_monitor()
    FM->>FM: monitor_heartbeats() spawned
    Coll1->>EB: Heartbeat
    EB->>Agent: Receive heartbeat
    Agent->>ALB: update_metrics(Coll1, heartbeat)
    Coll2->>EB: Heartbeat & register
    Agent->>ALB: update_metrics(Coll2, heartbeat)

    Note over Agent,Coll2: Task Distribution
    Agent->>ALB: select_weighted(candidates)
    ALB-->>Agent: Coll1 (highest score)
    Agent->>EB: Publish task to Coll1 topic
    EB->>Coll1: Deliver task
    Coll1->>EB: Publish result (corr_id)
    EB->>Agent: Route to result topic
    Agent->>RA: add_result(Coll1, corr_id, result)
    RA->>RA: Check complete? (No, wait for others)

    Note over Agent,Coll2: Aggregation & Completion
    Agent->>ALB: select_weighted() -> Coll2
    Agent->>EB: Publish related task
    Coll2->>EB: Publish result (same corr_id)
    EB->>Agent: Deliver
    Agent->>RA: add_result(Coll2, corr_id, result)
    alt Complete
        RA->>RA: aggregate() -> AggregatedResult
        RA->>Detection: on_complete(aggregated)
    end

    Note over Agent,Coll2: Failover
    Coll2--xEB: Miss heartbeats
    FM->>FM: Detect stale (30s)
    FM->>RA/ALB: handle_failure(Coll2)
    FM->>Health: mark_unhealthy(Coll2)
    FM->>TaskRouter: Redistribute pending tasks
    alt Restart
        FM->>FM: attempt_restart(Coll2) -> new_pid
    end

## Proposed File Changes

### daemoneye-lib/src/eventbus/result_aggregator.rs(NEW)

References: 

- daemoneye-lib/proto/ipc.proto
- daemoneye-lib/proto/eventbus.proto

Implement ResultAggregator struct with pending_results: Arc<DashMap<Uuid, ResultSet>>, on_complete: Arc<dyn Fn(AggregatedResult) + Send + Sync>, timeout_ms: u64.

Define ResultSet with correlation_id: Uuid, expected_collectors: HashSet<ProcessId>, received_results: Vec<(ProcessId, DetectionResult)>, created_at: Instant.

Add new(timeout_ms, on_complete) constructor.

Implement async add_result(process_id, correlation_id, result): insert into pending_results, push to received_results, if is_complete() then aggregate and call on_complete, remove entry.

Add fn aggregate(result_set) -> Result<AggregatedResult> merging results, deduplicating, creating AggregatedResult {correlation_id, results: Vec<DetectionResult>, collector_count, complete: bool}.

Implement ResultSet::is_complete() checking received unique process_ids == expected_collectors.len().

Add pub fn start_timeout_monitor(self: Arc<Self>) spawning tokio task to periodically check and handle timeouts > timeout_ms, calling on_complete with partial if timed out.

Define AggregatedResult and use uuid::Uuid, std::collections::HashSet, std::time::Instant. Integrate with proto DetectionResult via prost.

Include derive(Debug) where appropriate, error handling with thiserror.

### daemoneye-lib/src/eventbus/advanced_load_balancer.rs(NEW)

References: 

- daemoneye-lib/proto/eventbus.proto

Define AdvancedLoadBalancer struct with metrics: Arc<DashMap<ProcessId, LoadMetrics>>, history: Arc<DashMap<ProcessId, PerformanceHistory>>.

LoadMetrics: active_tasks: AtomicUsize, queue_depth: AtomicUsize, cpu_usage: AtomicU64, memory_bytes: AtomicU64, last_heartbeat: AtomicU64.

PerformanceHistory: avg_task_duration_ms: f64, success_rate: f64, total_tasks_completed: u64.

Implement new() with empty DashMaps.

Add pub fn select_weighted(&self, candidates: Vec<ProcessId>) -> Result<ProcessId> iterating candidates, max_by_key calculate_score, or Err if none.

fn calculate_score(&self, process_id) -> u64: get metrics and history, load_score = 1000 - active_tasks*10 - queue_depth*5, perf_score = (success_rate * 100.0) as u64, sum scores.

pub fn update_metrics(&self, process_id, heartbeat: CollectorHeartbeat): update atomics from heartbeat fields.

pub fn record_task_completion(&self, process_id, duration_ms: u64, success: bool): update history with moving average (alpha=0.3) for duration, success_rate incrementally, increment total.

Use std::sync::atomic::{AtomicUsize, AtomicU64, Ordering}, std::collections::DashMap. Define ProcessId as u32 or String based on proto. Error with thiserror::Error.

### daemoneye-lib/src/eventbus/failover_manager.rs(NEW)

References: 

- daemoneye-lib/src/eventbus/mod.rs(NEW)
- daemoneye-lib/proto/ipc.proto

Implement FailoverManager struct with health_tracker: Arc<HealthTracker>, task_router: Arc<TaskRouter>.

Assume HealthTracker and TaskRouter exist or define minimally: HealthTracker with async mark_unhealthy(pid), find_stale_processes(timeout); TaskRouter with async route_task(task).

pub async fn handle_failure(&self, failed_process_id: ProcessId) -> Result<()>: mark_unhealthy, get_pending_tasks (implement as vec![] or from store), for each route_task, cleanup_subscriptions (log or nop).

pub async fn monitor_heartbeats(&self): loop { sleep(Duration::from_secs(10)).await; let stale = find_stale_processes(Duration::from_secs(30)).await?; for pid in stale { warn!("..."), self.handle_failure(pid).await?; } }

Optional: pub async fn attempt_restart(&self, failed_process_id) -> Result<()>: get_collector_info, spawn_collector_process (use std::process::Command for cross-platform), wait_for_registration with timeout 30s.

Use tokio::time::{sleep, Duration, timeout}, log crate for warn!. Define get_pending_tasks, cleanup_subscriptions as placeholders integrating with eventbus.

### daemoneye-lib/src/eventbus/mod.rs(NEW)

References: 

- daemoneye-lib/src/lib.rs(MODIFY)

Create module file declaring pub mod result_aggregator, advanced_load_balancer, failover_manager;

Re-export: pub use result_aggregator::{ResultAggregator, ResultSet, AggregatedResult}; pub use advanced_load_balancer::{AdvancedLoadBalancer, LoadMetrics, PerformanceHistory}; pub use failover_manager::FailoverManager;

If existing, add these mods and re-exports. Add doc comments: "Multi-collector coordination components for result aggregation, load balancing, and failover."

### daemoneye-lib/src/lib.rs(MODIFY)

References: 

- daemoneye-lib/src/eventbus/mod.rs(NEW)

Add pub mod eventbus; if not present.

Re-export new types: pub use eventbus::{ResultAggregator, AdvancedLoadBalancer, FailoverManager, AggregatedResult, LoadMetrics};

Update crate documentation to mention new features for multi-process coordination, reference issue #117.

### daemoneye-agent/src/eventbus_integration.rs(NEW)

References: 

- daemoneye-lib/src/eventbus/mod.rs(NEW)
- daemoneye-agent/src/main.rs
- daemoneye-lib/src/config.rs(MODIFY)

Integrate components in agent: Create EventBusManager struct with result_aggregator: Arc<ResultAggregator>, load_balancer: Arc<AdvancedLoadBalancer>, failover_manager: Arc<FailoverManager>, eventbus_client.

In new(): init aggregator with on_complete forwarding to detection engine, balancer, manager with trackers/routers.

Async distribute_task(task): select via load_balancer.select_weighted(candidates from config), publish to topic via client, track if needed.

Async start(): aggregator.start_timeout_monitor(), failover_manager.monitor_heartbeats() in spawn, subscribe to result topics, on receive call aggregator.add_result.

Async handle_heartbeat(heartbeat): load_balancer.update_metrics(pid, heartbeat).

Shutdown: stop tasks.

Use configs for timeout_ms etc. Ensure cross-process pub/sub via interprocess IPC.

### daemoneye-lib/src/config.rs(MODIFY)

References: 

- daemoneye-lib/src/eventbus/advanced_load_balancer.rs(NEW)

Extend existing configs: Add to EventBusConfig { timeout_ms: u64 default 30000, heartbeat_interval_secs: u64 default 10 }.

Add LoadBalancerConfig { strategy: String default "weighted" }.

In CollectorConfig add expected_collectors: Vec<ProcessId> for aggregation.

Update serde deserialization, validation for positive values.

### daemoneye-lib/tests/result_aggregation_tests.rs(NEW)

References: 

- daemoneye-lib/src/eventbus/mod.rs(NEW)

Unit tests with #[tokio::test]:

- Test ResultAggregator::new and add_result: create, add multiple results, assert is_complete triggers on_complete with merged AggregatedResult.

- Test timeout_monitor: spawn, add incomplete set, advance time, assert partial on_complete called.

- Test AdvancedLoadBalancer::select_weighted: register metrics/history, assert max score selected.

- Test calculate_score: vary active_tasks, success_rate, assert score computation.

- Test record_task_completion: assert moving average updates correctly.

- Test FailoverManager::handle_failure: mock trackers, assert route_task called for pendings.

- Test monitor_heartbeats: simulate stale, assert handle_failure invoked.

Use mockall for mocks, assert_eq! for results, coverage >80%.

### daemoneye-lib/tests/integration_multi_collector.rs(NEW)

References: 

- daemoneye-agent/src/eventbus_integration.rs(NEW)
- daemoneye-lib/src/ipc/mod.rs

Integration tests simulating multi-process:

- Spawn mock collector processes using tokio::process::Command or threads mimicking IPC.

- Test aggregation: publish results from multiple 'collectors', assert correlated output.

- Test load balancing: distribute tasks, assert selection based on metrics.

- Test failover: 'kill' a collector (stop publishing), assert tasks redistributed, heartbeats missed trigger handle_failure.

- Performance: measure aggregation latency with criterion or manual timing, assert <1ms p99 for 100 iterations.

Use tempdir for logs, env vars for config, cross-platform with if cfg!.

End-to-end: init EventBusManager, distribute tasks, collect aggregated, assert no loss on failure.

---

Developer Humor

🔄 DaemonEye's Load Balancing Wisdom\n>\n> Why did the collector go to the load balancer?\n> It was feeling overwhelmed with tasks!\n> The balancer replied, \Don't worry, I'll weigh your options.\ 💪\n>\n> \In multi-collector life, failover is just a restart away!\ 🚀

---

Execution Information

Branch: main
Commit: a169b68

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[unclesp1d3r]

@coderabbitai Enhance this issue body with additional context and a proposed solution. Improve the issue title and assign the proper labels and other metadata.