Implement Task Distribution and Capability-Based Routing via EventBus

[unclesp1d3r]

Implement task distribution and capability-based routing across processes

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

Design Goals (from Parent #114)

Scope: Single-system, local multi-process coordination

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

Critical Architectural Context

Task distribution across local processes:

• daemoneye-agent (separate process): hosts broker, routes tasks
• Collector processes (procmond-1, procmond-2, netmond-1, etc.): separate OS processes on same machine
• Communication via local cross-process IPC using interprocess crate
• Task distribution and routing happens across process boundaries on local system

Overview

This task implements the task distribution and capability-based routing system within daemoneye-eventbus, enabling the agent to intelligently route tasks to appropriate collector processes on the local system based on their advertised capabilities.

Architecture

Routing Flow (Local Cross-Process)

1. Collector Process Startup (Local Process)
   ├─▶ Connect to broker via local IPC
   ├─▶ Advertise capabilities (PROCESS, NETWORK, etc.)
   └─▶ Subscribe to relevant task topics

2. Task Creation (In Agent Process)
   ├─▶ Analyze required capabilities
   ├─▶ Match against available local collectors
   └─▶ Route to appropriate topic

3. Broker Routing (Within Agent Process)
   ├─▶ Match topic to subscribed local processes
   ├─▶ Apply load balancing strategy
   └─▶ Send via local IPC to selected process

4. Task Execution (In Collector Process)
   ├─▶ Receive task via local IPC
   ├─▶ Execute collection
   └─▶ Publish results via local IPC

Capability Registry (In Broker Process)

/// Registry of local collector process capabilities
pub struct CapabilityRegistry {
    // Map of process IDs to their capabilities (local processes only)
    collectors: Arc<DashMap<ProcessId, CollectorInfo>>,
    
    // Capability index for fast lookups (local processes)
    capability_index: Arc<DashMap<SourceCaps, Vec<ProcessId>>>,
}

pub struct CollectorInfo {
    process_id: ProcessId,
    capabilities: SourceCaps,
    health: CollectorHealth,
    last_heartbeat: Instant,
    topics: Vec<String>,
}

impl CapabilityRegistry {
    /// Register local collector process
    pub async fn register(&self, info: CollectorInfo) -> Result<()> {
        // Add to local process registry
    }
    
    /// Find local collectors matching capability
    pub fn find_by_capability(&self, cap: SourceCaps) -> Vec<ProcessId> {
        // Return matching local process IDs
    }
}

Task Router (In Broker Process)

/// Routes tasks to appropriate local collector processes
pub struct TaskRouter {
    capability_registry: Arc<CapabilityRegistry>,
    load_balancer: Arc<LoadBalancer>,
    eventbus: Arc<EventBusBroker>,
}

impl TaskRouter {
    /// Route task to appropriate local collector process(es)
    pub async fn route_task(&self, task: DetectionTask) -> Result<()> {
        // 1. Determine required capabilities
        let required_caps = task.required_capabilities();
        
        // 2. Find matching local collectors
        let candidates = self.capability_registry
            .find_by_capability(required_caps)?;
        
        // 3. Apply load balancing (local processes)
        let target = self.load_balancer
            .select(candidates, LoadBalancingStrategy::RoundRobin)?;
        
        // 4. Publish to topic (routed to local process via IPC)
        let topic = self.task_topic_for_capability(required_caps);
        self.eventbus.publish_local(&topic, task).await?;
        
        Ok(())
    }
}

Load Balancing (Across Local Processes)

pub struct LoadBalancer {
    // Track load per local collector process
    process_load: Arc<DashMap<ProcessId, ProcessLoad>>,
}

pub struct ProcessLoad {
    active_tasks: AtomicUsize,
    queue_depth: AtomicUsize,
    last_task_time: Instant,
}

pub enum LoadBalancingStrategy {
    /// Distribute evenly across local processes
    RoundRobin,
    /// Route to least-loaded local process
    LeastLoaded,
    /// Random selection among local processes
    Random,
}

Topic Mapping (Cross-Process)

// Map capabilities to task topics for local process routing
fn task_topic_for_capability(cap: SourceCaps) -> &'static str {
    match cap {
        SourceCaps::PROCESS => "control.collector.task.process",
        SourceCaps::NETWORK => "control.collector.task.network",
        SourceCaps::FILESYSTEM => "control.collector.task.filesystem",
        SourceCaps::PERFORMANCE => "control.collector.task.performance",
    }
}

Implementation Steps

1. Capability Registry

Note: "In-process" refers to code running WITHIN the broker process (daemoneye-agent), NOT within collector processes. All communication TO collectors uses cross-process IPC.

// In-process implementation within broker (uses tokio::sync for broker-internal coordination)
// IPC used for all communication WITH collector processes

impl CapabilityRegistry {
    pub fn new() -> Self {
        Self {
            collectors: Arc::new(DashMap::new()),
            capability_index: Arc::new(DashMap::new()),
        }
    }
    
    /// Register local collector process (called when process connects via IPC)
    pub async fn register(&self, info: CollectorInfo) -> Result<()> {
        let process_id = info.process_id;
        let caps = info.capabilities;
        
        // Store local collector info
        self.collectors.insert(process_id, info);
        
        // Update capability index
        self.capability_index
            .entry(caps)
            .or_insert_with(Vec::new)
            .push(process_id);
        
        Ok(())
    }
    
    /// Unregister local collector process (when process disconnects IPC)
    pub async fn unregister(&self, process_id: ProcessId) -> Result<()> {
        if let Some((_, info)) = self.collectors.remove(&process_id) {
            // Remove from capability index
            if let Some(mut pids) = self.capability_index.get_mut(&info.capabilities) {
                pids.retain(|&id| id != process_id);
            }
        }
        Ok(())
    }
}

2. Task Router

impl TaskRouter {
    pub fn new(
        capability_registry: Arc<CapabilityRegistry>,
        load_balancer: Arc<LoadBalancer>,
        eventbus: Arc<EventBusBroker>,
    ) -> Self {
        Self {
            capability_registry,
            load_balancer,
            eventbus,
        }
    }
    
    /// Route single task to local collector process
    pub async fn route_task(&self, task: DetectionTask) -> Result<()> {
        let required_caps = task.required_capabilities();
        let candidates = self.capability_registry.find_by_capability(required_caps)?;
        
        if candidates.is_empty() {
            return Err(Error::NoCapableCollector(required_caps));
        }
        
        let target = self.load_balancer.select(candidates, LoadBalancingStrategy::RoundRobin)?;
        let topic = self.task_topic_for_capability(required_caps);
        
        // Publish to local process via IPC
        self.eventbus.publish_local(&topic, task).await?;
        
        Ok(())
    }
    
    /// Broadcast task to all local collectors with capability
    pub async fn broadcast_task(&self, task: DetectionTask) -> Result<()> {
        let required_caps = task.required_capabilities();
        let topic = self.task_topic_for_capability(required_caps);
        
        // Broadcast to all local subscribers via IPC
        self.eventbus.publish_broadcast_local(&topic, task).await?;
        
        Ok(())
    }
}

3. Load Balancer

impl LoadBalancer {
    /// Select best local process for task
    pub fn select(
        &self,
        candidates: Vec<ProcessId>,
        strategy: LoadBalancingStrategy,
    ) -> Result<ProcessId> {
        match strategy {
            LoadBalancingStrategy::RoundRobin => {
                self.round_robin_select(candidates)
            }
            LoadBalancingStrategy::LeastLoaded => {
                self.least_loaded_select(candidates)
            }
            LoadBalancingStrategy::Random => {
                self.random_select(candidates)
            }
        }
    }
    
    fn least_loaded_select(&self, candidates: Vec<ProcessId>) -> Result<ProcessId> {
        candidates
            .into_iter()
            .min_by_key(|&pid| {
                self.process_load
                    .get(&pid)
                    .map(|load| load.active_tasks.load(Ordering::Relaxed))
                    .unwrap_or(usize::MAX)
            })
            .ok_or(Error::NoCollectorAvailable)
    }
}

Acceptance Criteria

• Capability registry implemented for local collector processes
• Task router with capability matching (local processes)
• Load balancing strategies (RoundRobin, LeastLoaded, Random) for local processes
• Topic-based task distribution via local IPC
• Dynamic collector registration/unregistration for local processes
• Health-based routing decisions (exclude unhealthy local processes)
• Performance: <1ms p99 latency for routing decisions
• Zero unsafe code in routing implementation
• Unit tests for routing logic
• Integration tests with multiple local collector types
• Cross-platform validation (Windows, Linux, macOS, FreeBSD)
• Documentation of routing strategies

Testing Strategy

Unit Tests

• Capability matching logic
• Load balancing algorithms
• Topic selection
• Health-based filtering

Integration Tests

• Route tasks to correct local collector types
• Load distribution across multiple local processes
• Failover when local collector becomes unhealthy
• Dynamic registration/unregistration of local processes

Performance Tests

• Routing decision latency (<1ms p99 target)
• Throughput with many concurrent tasks
• Memory overhead of routing tables

Performance Requirements

• Routing Decision Time: <1ms p99 for selecting target local process
• Registration Time: <100μs for adding/removing local collector
• Lookup Time: <100μs for capability matching

Code Safety

• Zero unsafe code in routing and balancing logic
• Built entirely on: interprocess (IPC), tokio (async), prost (serialization), dashmap (routing tables)

---

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

[traycerai]

Plan

Observations

Codebase Analysis:\n- daemoneye-lib/proto/ipc.proto defines DetectionTask, DetectionResult, CollectionCapabilities with MonitoringDomain enum (PROCESS=0, etc.), but lacks SourceCaps—map to bitflags or vec of domains.\n- daemoneye-agent/src/main.rs uses IpcClientManager for direct enumerate_processes() calls; replace with TaskRouter for capability-based distribution.\n- collector-core/src/collector.rs manages event sources and IPC server; add registration to broker via IPC on startup, subscribe to topics.\n- No daemoneye-eventbus crate exists; create for broker, registry, router.\n- Ticket emphasizes local cross-process IPC via interprocess, zero unsafe, deps: tokio, prost, dashmap.\n- Align SourceCaps to CollectionCapabilities; use ProcessId from models.

Approach

Implementation Strategy:\n1. Create daemoneye-eventbus crate with EventBusBroker using interprocess for Unix sockets/named pipes cross-process pub/sub.\n2. Implement CapabilityRegistry with DashMap<ProcessId, CollectorInfo> and index for fast capability lookup; handle register/unregister on IPC connect/disconnect.\n3. Build LoadBalancer with strategies (RoundRobin via atomic counter, LeastLoaded by active_tasks, Random via rand).\n4. Develop TaskRouter to analyze DetectionTask for required domains, find candidates, select via balancer, publish to topic like "control.collector.task.process".\n5. Integrate into agent: init broker/registry/router, route tasks instead of direct IPC.\n6. Update collector: on startup, register capabilities via IPC to broker, subscribe to relevant topics, execute on receive, publish results.\n7. Add metrics, tests, docs; ensure <1ms routing latency.

Reasoning

I examined the workspace structure to identify relevant crates like daemoneye-lib, daemoneye-agent, and collector-core. I inspected ipc.proto for message definitions including DetectionTask and CollectionCapabilities. I reviewed main.rs in daemoneye-agent to understand current direct IPC usage for process enumeration. I analyzed collector.rs to see runtime and IPC server setup. This informed adaptations for cross-process event bus, capability registry, and routing per ticket specs, mapping proto domains to SourceCaps equivalent.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

References

• 📄 daemoneye-lib/Cargo.toml

Add daemoneye-eventbus to the [workspace.members] array to include the new crate, enabling workspace builds and dependency sharing with daemoneye-lib.

📄 daemoneye-eventbus/Cargo.toml ^(NEW) 🔗

References

• 📄 daemoneye-lib/Cargo.toml

Create Cargo.toml for daemoneye-eventbus: package.name = "daemoneye-eventbus", version = "0.1.0", edition = "2021". Dependencies: tokio = { version = "1", features = ["full"] }, prost = "0.12", interprocess = "1.2", dashmap = "5.5", thiserror = "1.0", async-trait = "0.1", rand = "0.8", prometheus = "0.13", daemoneye-lib = { path = "../daemoneye-lib" }. Set lib.crate-type = ["cdylib", "rlib"] for flexibility.

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

References

• 📄 daemoneye-lib/src/models/process.rs

Declare modules: mod registry; mod router; mod balancer; mod eventbus; mod types; mod error;. Re-export: pub use self::eventbus::EventBusBroker; pub use self::registry::CapabilityRegistry; pub use self::router::TaskRouter; pub use self::balancer::LoadBalancer; pub use self::types::{ProcessId, CollectorInfo, SourceCaps, ProcessLoad, LoadBalancingStrategy}; pub use self::error::Error;. Define ProcessId as u32 (align with models::ProcessId::raw()). Document cross-process local IPC focus.

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

References

• 📄 daemoneye-lib/proto/ipc.proto

Define SourceCaps as bitflags from MonitoringDomain: #[derive(Clone, Debug)] pub struct SourceCaps(u32); with impls for from_vec/domains, intersection check. CollectorInfo { process_id: ProcessId, capabilities: SourceCaps, health: bool, last_heartbeat: Instant, topics: Vec } deriving Clone, Debug. ProcessLoad { active_tasks: AtomicUsize, queue_depth: AtomicUsize, last_task_time: Instant }. LoadBalancingStrategy enum { RoundRobin, LeastLoaded, Random } deriving Clone, Debug, PartialEq.

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

References

• 📄 daemoneye-lib/src/proto.rs

Implement EventBusBroker struct with interprocess::local_socket::LocalSocketListener for cross-process: broker: LocalSocketListener, subscribers: Arc<DashMap<String, Vec<LocalSocketStream>>>. new(path: &str) binds listener. async fn publish_local(&self, topic: &str, msg: impl prost::Message) -> Result<(), Error> serializes msg to bytes via prost, writes to subscribers on topic. async fn subscribe_local(&self, topic: &str) -> Result<LocalSocketStream, Error> accepts connection, filters by topic. Handle connect/disconnect for registry updates. Add publish_broadcast_local for all subscribers.

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

References

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

Implement CapabilityRegistry as per ticket: collectors: Arc<DashMap<ProcessId, CollectorInfo>>, capability_index: Arc<DashMap<SourceCaps, Vec<ProcessId>>>. new() initializes DashMaps. async fn register(&self, info: CollectorInfo) inserts to collectors, pushes to index entry. async fn unregister(&self, process_id: ProcessId) removes from collectors and index. fn find_by_capability(&self, cap: SourceCaps) -> Vec<ProcessId> returns index.get(&cap).cloned().unwrap_or_default(), filters healthy (last_heartbeat within 30s). Add health check in find.

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

References

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

Implement LoadBalancer { process_load: Arc<DashMap<ProcessId, ProcessLoad>> }. new() initializes DashMap. fn select(&self, candidates: Vec<ProcessId>, strategy: LoadBalancingStrategy) -> Result<ProcessId, Error>: for RoundRobin use atomic counter % len, LeastLoaded min_by active_tasks.load(), Random via rand::thread_rng().sample_iter(). For LeastLoaded: candidates.into_iter().min_by_key(|&pid| self.process_load.get(&pid).map(|l| l.active_tasks.load(Ordering::Relaxed)).unwrap_or(usize::MAX)).ok_or(Error::NoCollectorAvailable)?. Update load on task assign: active_tasks.fetch_add(1).

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

References

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

Implement TaskRouter { capability_registry: Arc<CapabilityRegistry>, load_balancer: Arc<LoadBalancer>, eventbus: Arc<EventBusBroker> }. new(...) as per ticket. async fn route_task(&self, task: DetectionTask) -> Result<(), Error>: let required_caps = task.required_capabilities(); (impl fn mapping task_type to SourceCaps, e.g., EnumerateProcesses -> PROCESS). let candidates = self.capability_registry.find_by_capability(required_caps)?; if empty Err(NoCapableCollector). let target = self.load_balancer.select(candidates, LoadBalancingStrategy::RoundRobin)?; let topic = self.task_topic_for_capability(required_caps);(match cap to "control.collector.task.process" etc.).self.eventbus.publish_local(&topic, &task).await?; Update load for target.Similar forbroadcast_task`.

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

#[derive(thiserror::Error, Debug)] pub enum Error { #[error("No capable collector for {0}")] NoCapableCollector(SourceCaps), #[error("No collector available")] NoCollectorAvailable, #[error("IPC error: {0}")] Ipc(#[from] interprocess::OsError), #[error("Protobuf error: {0}")] Protobuf(#[from] prost::EncodeError), // add more } Implement Display, From for common errors.

📄 daemoneye-agent/Cargo.toml ^(MODIFY) 🔗

References

• 📄 daemoneye-eventbus/Cargo.toml ^(NEW)

Add dependency daemoneye-eventbus = { path = "../daemoneye-eventbus" } to integrate routing and event bus.

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

References

• 📄 daemoneye-agent/src/ipc_client.rs
• 📄 daemoneye-eventbus/src/router.rs ^(NEW)

After IPC init, create let socket_path = "/tmp/daemoneye-broker.sock"; let eventbus = Arc::new(EventBusBroker::new(socket_path)?); let registry = Arc::new(CapabilityRegistry::new()); let balancer = Arc::new(LoadBalancer::new()); let router = Arc::new(TaskRouter::new(registry.clone(), balancer.clone(), eventbus.clone()));. Spawn task to listen for registrations: on collector connect, registry.register(CollectorInfo { process_id: pid, capabilities: from_proto(caps), .. }). In loop, replace ipc_manager.enumerate_processes() with let task = DetectionTask::new_enumerate_processes("enum-".to_string() + &iteration.to_string(), None); router.route_task(task).await?; Then subscribe to result topic, await result via stream. Add shutdown for broker.

📄 collector-core/Cargo.toml ^(MODIFY) 🔗

References

• 📄 daemoneye-eventbus/Cargo.toml ^(NEW)

Add optional daemoneye-eventbus = { path = "../../daemoneye-eventbus", optional = true } with feature eventbus for conditional event bus integration.

📄 collector-core/src/collector.rs ^(MODIFY) 🔗

References

• 📄 collector-core/src/ipc.rs
• 📄 daemoneye-eventbus/src/types.rs ^(NEW)

In Collector::new, add optional eventbus_path: Option<String>. In run, if path provided, connect to broker: let stream = EventBusBroker::connect(&path)?; let caps = self.capabilities(); let info = CollectorInfo { process_id: std::process::id() as u32, capabilities: SourceCaps::from_proto(&caps), health: true, last_heartbeat: Instant::now(), topics: vec!["control.collector.task.process".to_string()] }; Send registration via IPC or direct publish. Subscribe to topics based on caps, e.g., for PROCESS: let mut rx = stream.subscribe("control.collector.task.process")?; while let Some(msg) = rx.recv().await { let task: DetectionTask = prost::Message::decode(&msg)?; // execute, publish result to "results.process" }. Maintain IPC fallback if no eventbus. Update health with heartbeats every 30s.

📄 daemoneye-eventbus/tests/registry_test.rs ^(NEW) 🔗

References

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

Unit tests: Create registry, register multiple collectors with caps, verify find_by_capability returns correct PIDs, test unregister removes, health filter excludes old heartbeats. Use tokio::test, assert_eq! on vecs.

📄 daemoneye-eventbus/tests/router_integration.rs ^(NEW) 🔗

References

• 📄 daemoneye-lib/tests/integration.rs

Integration tests: Mock broker with in-memory IPC, spawn mock collectors, register, route task requiring PROCESS, verify publish to correct topic, simulate execution and result receipt. Test load balancing with multiple candidates, failover on unregister. Measure latency <1ms.

📄 daemoneye-eventbus/README.md ^(NEW) 🔗

References

• 📄 docs/architecture.md

Document: Crate purpose for local cross-process task routing. Setup: Add to Cargo.toml, init broker in agent. API: EventBusBroker, CapabilityRegistry usage examples. Topics: control.collector.task.{domain}. Integration with daemoneye-agent/collector-core. Performance: <1ms routing. Reference ticket #116.

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

**Codebase Analysis:**\n- `daemoneye-lib/proto/ipc.proto` defines `DetectionTask`, `DetectionResult`, `CollectionCapabilities` with `MonitoringDomain` enum (PROCESS=0, etc.), but lacks `SourceCaps`—map to bitflags or vec of domains.\n- `daemoneye-agent/src/main.rs` uses `IpcClientManager` for direct `enumerate_processes()` calls; replace with `TaskRouter` for capability-based distribution.\n- `collector-core/src/collector.rs` manages event sources and IPC server; add registration to broker via IPC on startup, subscribe to topics.\n- No `daemoneye-eventbus` crate exists; create for broker, registry, router.\n- Ticket emphasizes local cross-process IPC via `interprocess`, zero unsafe, deps: tokio, prost, dashmap.\n- Align `SourceCaps` to `CollectionCapabilities`; use `ProcessId` from models.

### Approach

**Implementation Strategy:**\n1. Create `daemoneye-eventbus` crate with `EventBusBroker` using `interprocess` for Unix sockets/named pipes cross-process pub/sub.\n2. Implement `CapabilityRegistry` with `DashMap<ProcessId, CollectorInfo>` and index for fast capability lookup; handle register/unregister on IPC connect/disconnect.\n3. Build `LoadBalancer` with strategies (RoundRobin via atomic counter, LeastLoaded by active_tasks, Random via rand).\n4. Develop `TaskRouter` to analyze `DetectionTask` for required domains, find candidates, select via balancer, publish to topic like \"control.collector.task.process\".\n5. Integrate into agent: init broker/registry/router, route tasks instead of direct IPC.\n6. Update collector: on startup, register capabilities via IPC to broker, subscribe to relevant topics, execute on receive, publish results.\n7. Add metrics, tests, docs; ensure &lt;1ms routing latency.

### Reasoning

I examined the workspace structure to identify relevant crates like daemoneye-lib, daemoneye-agent, and collector-core. I inspected ipc.proto for message definitions including DetectionTask and CollectionCapabilities. I reviewed main.rs in daemoneye-agent to understand current direct IPC usage for process enumeration. I analyzed collector.rs to see runtime and IPC server setup. This informed adaptations for cross-process event bus, capability registry, and routing per ticket specs, mapping proto domains to SourceCaps equivalent.

## Proposed File Changes

### Cargo.toml(MODIFY)

References: 

- daemoneye-lib/Cargo.toml

Add `daemoneye-eventbus` to the `[workspace.members]` array to include the new crate, enabling workspace builds and dependency sharing with daemoneye-lib.

### daemoneye-eventbus/Cargo.toml(NEW)

References: 

- daemoneye-lib/Cargo.toml

Create Cargo.toml for `daemoneye-eventbus`: `package.name = "daemoneye-eventbus"`, `version = "0.1.0"`, `edition = "2021"`. Dependencies: `tokio = { version = "1", features = ["full"] }`, `prost = "0.12"`, `interprocess = "1.2"`, `dashmap = "5.5"`, `thiserror = "1.0"`, `async-trait = "0.1"`, `rand = "0.8"`, `prometheus = "0.13"`, `daemoneye-lib = { path = "../daemoneye-lib" }`. Set `lib.crate-type = ["cdylib", "rlib"]` for flexibility.

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

References: 

- daemoneye-lib/src/models/process.rs

Declare modules: `mod registry; mod router; mod balancer; mod eventbus; mod types; mod error;`. Re-export: `pub use self::eventbus::EventBusBroker; pub use self::registry::CapabilityRegistry; pub use self::router::TaskRouter; pub use self::balancer::LoadBalancer; pub use self::types::{ProcessId, CollectorInfo, SourceCaps, ProcessLoad, LoadBalancingStrategy}; pub use self::error::Error;`. Define `ProcessId` as `u32` (align with models::ProcessId::raw()). Document cross-process local IPC focus.

### daemoneye-eventbus/src/types.rs(NEW)

References: 

- daemoneye-lib/proto/ipc.proto

Define `SourceCaps` as bitflags from `MonitoringDomain`: `#[derive(Clone, Debug)] pub struct SourceCaps(u32);` with impls for from_vec/domains, intersection check. `CollectorInfo` { process_id: ProcessId, capabilities: SourceCaps, health: bool, last_heartbeat: Instant, topics: Vec<String> } deriving Clone, Debug. `ProcessLoad` { active_tasks: AtomicUsize, queue_depth: AtomicUsize, last_task_time: Instant }. `LoadBalancingStrategy` enum { RoundRobin, LeastLoaded, Random } deriving Clone, Debug, PartialEq.

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

References: 

- daemoneye-lib/src/proto.rs

Implement `EventBusBroker` struct with `interprocess::local_socket::LocalSocketListener` for cross-process: `broker: LocalSocketListener, subscribers: Arc<DashMap<String, Vec<LocalSocketStream>>>`. `new(path: &str)` binds listener. `async fn publish_local(&self, topic: &str, msg: impl prost::Message) -> Result<(), Error>` serializes msg to bytes via prost, writes to subscribers on topic. `async fn subscribe_local(&self, topic: &str) -> Result<LocalSocketStream, Error>` accepts connection, filters by topic. Handle connect/disconnect for registry updates. Add `publish_broadcast_local` for all subscribers.

### daemoneye-eventbus/src/registry.rs(NEW)

References: 

- daemoneye-eventbus/src/types.rs(NEW)

Implement `CapabilityRegistry` as per ticket: `collectors: Arc<DashMap<ProcessId, CollectorInfo>>`, `capability_index: Arc<DashMap<SourceCaps, Vec<ProcessId>>>`. `new()` initializes DashMaps. `async fn register(&self, info: CollectorInfo)` inserts to collectors, pushes to index entry. `async fn unregister(&self, process_id: ProcessId)` removes from collectors and index. `fn find_by_capability(&self, cap: SourceCaps) -> Vec<ProcessId>` returns index.get(&cap).cloned().unwrap_or_default(), filters healthy (last_heartbeat within 30s). Add health check in find.

### daemoneye-eventbus/src/balancer.rs(NEW)

References: 

- daemoneye-eventbus/src/types.rs(NEW)

Implement `LoadBalancer` { `process_load: Arc<DashMap<ProcessId, ProcessLoad>>` }. `new()` initializes DashMap. `fn select(&self, candidates: Vec<ProcessId>, strategy: LoadBalancingStrategy) -> Result<ProcessId, Error>`: for RoundRobin use atomic counter % len, LeastLoaded min_by active_tasks.load(), Random via rand::thread_rng().sample_iter(). For LeastLoaded: `candidates.into_iter().min_by_key(|&pid| self.process_load.get(&pid).map(|l| l.active_tasks.load(Ordering::Relaxed)).unwrap_or(usize::MAX)).ok_or(Error::NoCollectorAvailable)?`. Update load on task assign: `active_tasks.fetch_add(1)`.

### daemoneye-eventbus/src/router.rs(NEW)

References: 

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

Implement `TaskRouter` { `capability_registry: Arc<CapabilityRegistry>`, `load_balancer: Arc<LoadBalancer>`, `eventbus: Arc<EventBusBroker>` }. `new(...)` as per ticket. `async fn route_task(&self, task: DetectionTask) -> Result<(), Error>`: `let required_caps = task.required_capabilities();` (impl fn mapping task_type to SourceCaps, e.g., EnumerateProcesses -> PROCESS). `let candidates = self.capability_registry.find_by_capability(required_caps)?; if empty Err(NoCapableCollector). `let target = self.load_balancer.select(candidates, LoadBalancingStrategy::RoundRobin)?; let topic = self.task_topic_for_capability(required_caps);` (match cap to "control.collector.task.process" etc.). `self.eventbus.publish_local(&topic, &task).await?; Update load for target.` Similar for `broadcast_task`.

### daemoneye-eventbus/src/error.rs(NEW)

#[derive(thiserror::Error, Debug)] pub enum Error { #[error("No capable collector for {0}")] NoCapableCollector(SourceCaps), #[error("No collector available")] NoCollectorAvailable, #[error("IPC error: {0}")] Ipc(#[from] interprocess::OsError), #[error("Protobuf error: {0}")] Protobuf(#[from] prost::EncodeError), // add more } Implement Display, From for common errors.

### daemoneye-agent/Cargo.toml(MODIFY)

References: 

- daemoneye-eventbus/Cargo.toml(NEW)

Add dependency `daemoneye-eventbus = { path = "../daemoneye-eventbus" }` to integrate routing and event bus.

### daemoneye-agent/src/main.rs(MODIFY)

References: 

- daemoneye-agent/src/ipc_client.rs
- daemoneye-eventbus/src/router.rs(NEW)

After IPC init, create `let socket_path = "/tmp/daemoneye-broker.sock"; let eventbus = Arc::new(EventBusBroker::new(socket_path)?); let registry = Arc::new(CapabilityRegistry::new()); let balancer = Arc::new(LoadBalancer::new()); let router = Arc::new(TaskRouter::new(registry.clone(), balancer.clone(), eventbus.clone()));`. Spawn task to listen for registrations: on collector connect, `registry.register(CollectorInfo { process_id: pid, capabilities: from_proto(caps), .. })`. In loop, replace `ipc_manager.enumerate_processes()` with `let task = DetectionTask::new_enumerate_processes("enum-".to_string() + &iteration.to_string(), None); router.route_task(task).await?;` Then subscribe to result topic, await result via stream. Add shutdown for broker.

### collector-core/Cargo.toml(MODIFY)

References: 

- daemoneye-eventbus/Cargo.toml(NEW)

Add optional `daemoneye-eventbus = { path = "../../daemoneye-eventbus", optional = true }` with feature `eventbus` for conditional event bus integration.

### collector-core/src/collector.rs(MODIFY)

References: 

- collector-core/src/ipc.rs
- daemoneye-eventbus/src/types.rs(NEW)

In `Collector::new`, add optional `eventbus_path: Option<String>`. In `run`, if path provided, connect to broker: `let stream = EventBusBroker::connect(&path)?; let caps = self.capabilities(); let info = CollectorInfo { process_id: std::process::id() as u32, capabilities: SourceCaps::from_proto(&caps), health: true, last_heartbeat: Instant::now(), topics: vec!["control.collector.task.process".to_string()] };` Send registration via IPC or direct publish. Subscribe to topics based on caps, e.g., for PROCESS: `let mut rx = stream.subscribe("control.collector.task.process")?; while let Some(msg) = rx.recv().await { let task: DetectionTask = prost::Message::decode(&msg)?; // execute, publish result to "results.process" }`. Maintain IPC fallback if no eventbus. Update health with heartbeats every 30s.

### daemoneye-eventbus/tests/registry_test.rs(NEW)

References: 

- daemoneye-eventbus/src/registry.rs(NEW)

Unit tests: Create registry, register multiple collectors with caps, verify find_by_capability returns correct PIDs, test unregister removes, health filter excludes old heartbeats. Use tokio::test, assert_eq! on vecs.

### daemoneye-eventbus/tests/router_integration.rs(NEW)

References: 

- daemoneye-lib/tests/integration.rs

Integration tests: Mock broker with in-memory IPC, spawn mock collectors, register, route task requiring PROCESS, verify publish to correct topic, simulate execution and result receipt. Test load balancing with multiple candidates, failover on unregister. Measure latency &lt;1ms.

### daemoneye-eventbus/README.md(NEW)

References: 

- docs/architecture.md

Document: Crate purpose for local cross-process task routing. Setup: Add to Cargo.toml, init broker in agent. API: EventBusBroker, CapabilityRegistry usage examples. Topics: control.collector.task.{domain}. Integration with daemoneye-agent/collector-core. Performance: &lt;1ms routing. Reference ticket #116.

---

Developer Humor

\n> Why did the TaskRouter go to school?\n> To improve its balancing act!\n> With capabilities sharp as a tack,\n> It routes tasks without looking back.\n> Local IPC, fast and secure,\n> No network drama to endure! 🚀🛤️

---

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.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash

# Get current issue details
echo "=== Current Issue #116 Details ==="
gh issue view 116 --json title,body,labels,assignees,milestone,state

echo -e "\n=== Reading specification file ==="
cat .kiro/specs/daemoneye-core-monitoring/tasks.md

echo -e "\n=== Finding related files ==="
# Look for eventbus and collector related files
fd -e go -e rb "eventbus|collector" --type f --max-depth 3

echo -e "\n=== Check project structure ==="
ls -la

echo -e "\n=== Find README for context ==="
cat README.md 2>/dev/null || echo "No README.md found"

Length of output: 50533

---

🏁 Script executed:

#!/bin/bash

echo "=== Searching for eventbus references ==="
rg -n "eventbus|EventBus" --type rust -C2

echo -e "\n=== Searching for collector-core capability code ==="
rg -n "capability|Capability|SourceCaps" --type rust -C2 | head -100

echo -e "\n=== Check collector-core structure ==="
find collector-core -type f -name "*.rs" 2>/dev/null | head -20

echo -e "\n=== Search for task distribution and routing ==="
rg -n "task.*distribut|routing|TaskRouter|task.*queue" --type rust -C2 | head -100

echo -e "\n=== Find requirements documentation ==="
fd -e md requirements

echo -e "\n=== Check parent issue #114 ==="
gh issue view 114 --json title,body,labels

echo -e "\n=== Find related issues ==="
gh issue list --label "daemoneye" --label "architecture" --limit 10 --json number,title,state

Length of output: 17366

---

🏁 Script executed:

#!/bin/bash

# Update issue with enhanced content
gh issue edit 116 \
  --title "Implement Task Distribution and Capability-Based Routing via EventBus" \
  --body "## Overview

This issue implements a robust task distribution system using an event-driven architecture with capability-based routing for multi-collector coordination. This is part of the broader multi-process collector coordination initiative tracked in #114.

## Background & Context

**Current State:**
- ✅ Collector-core framework with \`EventSource\` trait and \`SourceCaps\` capabilities
- ✅ IPC client/server infrastructure with protobuf messaging
- ✅ Basic capability negotiation between daemoneye-agent and collectors
- ✅ SQL-to-IPC translation pipeline with pushdown optimization
- ✅ Simple task routing via \`send_task_with_routing()\`

**What's Missing:**
- ❌ Event-driven publish/subscribe architecture for decoupled communication
- ❌ Task queuing and priority handling for workload management
- ❌ Dynamic routing updates when collectors join/leave the topology
- ❌ Fallback routing for unavailable or overloaded collectors
- ❌ Load-aware task distribution across multiple collector instances

## Motivation

The current point-to-point IPC approach works for single collector scenarios but doesn't scale to multi-collector deployments. An event bus enables:

1. **Decoupling**: Collectors subscribe to task topics without tight coupling to daemoneye-agent
2. **Scalability**: Horizontal scaling with multiple collectors handling different capabilities
3. **Resilience**: Automatic failover when collectors become unavailable
4. **Flexibility**: Dynamic addition/removal of collectors without service interruption

## Proposed Solution

### 1. Create daemoneye-eventbus Library

Create a new workspace crate \`daemoneye-eventbus\` with topic-based publish/subscribe:

\`\`\`rust
// Core event bus traits and types
pub trait EventBus: Send + Sync {
    async fn publish(&self, topic: &str, event: Event) -> Result<()>;
    async fn subscribe(&self, topic: &str) -> Result<EventStream>;
}

pub struct Event {
    pub id: EventId,
    pub topic: String,
    pub payload: Bytes,
    pub metadata: HashMap<String, String>,
    pub timestamp: SystemTime,
}

// In-process implementation using tokio::sync::broadcast
pub struct InProcessEventBus {
    topics: Arc<RwLock<HashMap<String, broadcast::Sender<Event>>>>,
}
\`\`\`

**Topic Hierarchy:**
\`\`\`
tasks/
  ├── process/       # Process monitoring tasks
  ├── network/       # Network monitoring tasks  
  ├── filesystem/    # Filesystem monitoring tasks
  └── custom/        # Custom detection tasks

collectors/
  ├── announce/      # Collector registration/heartbeat
  ├── capabilities/  # Capability advertisements
  └── status/        # Health status updates

results/
  ├── success/       # Successful detection results
  └── failure/       # Failed task executions
\`\`\`

### 2. Implement Task Distribution Logic

Extend \`IpcClientManager\` or create new \`TaskDistributor\`:

\`\`\`rust
pub struct TaskDistributor {
    event_bus: Arc<dyn EventBus>,
    router: Arc<CapabilityRouter>,
    queue: TaskQueue,
}

impl TaskDistributor {
    pub async fn distribute_task(&self, task: DetectionTask) -> Result<TaskHandle> {
        // 1. Determine required capabilities from task
        let required_caps = self.analyze_task_requirements(&task)?;
        
        // 2. Find capable collectors via router
        let collectors = self.router.find_collectors(required_caps).await?;
        
        // 3. Apply load balancing and priority
        let target = self.select_target(&collectors, task.priority)?;
        
        // 4. Publish to appropriate topic with fallback
        let topic = self.task_topic_for(&task);
        self.event_bus.publish(&topic, task.into()).await?;
        
        // 5. Return handle for result tracking
        Ok(TaskHandle::new(task.task_id))
    }
}
\`\`\`

### 3. Capability-Based Router

\`\`\`rust
pub struct CapabilityRouter {
    // Map of collector_id -> (capabilities, last_seen, load_factor)
    collectors: Arc<RwLock<HashMap<String, CollectorInfo>>>,
}

impl CapabilityRouter {
    pub async fn register_collector(&self, info: CollectorInfo) {
        // Handle collectors/announce events
    }
    
    pub async fn find_collectors(&self, caps: SourceCaps) -> Vec<CollectorId> {
        // Filter by capability match and health status
        self.collectors.read().await
            .iter()
            .filter(|(_, info)| info.capabilities.contains(caps) && info.is_healthy())
            .map(|(id, _)| id.clone())
            .collect()
    }
    
    pub async fn select_best(&self, candidates: &[CollectorId], strategy: LoadStrategy) -> Option<CollectorId> {
        // Round-robin, least-loaded, or random selection
    }
}
\`\`\`

### 4. Task Queue with Priority Handling

\`\`\`rust
pub struct TaskQueue {
    high_priority: Arc<Mutex<VecDeque<DetectionTask>>>,
    normal_priority: Arc<Mutex<VecDeque<DetectionTask>>>,
    low_priority: Arc<Mutex<VecDeque<DetectionTask>>>,
    semaphore: Arc<Semaphore>,
}

impl TaskQueue {
    pub async fn enqueue(&self, task: DetectionTask) {
        match task.priority {
            Priority::High => self.high_priority.lock().await.push_back(task),
            Priority::Normal => self.normal_priority.lock().await.push_back(task),
            Priority::Low => self.low_priority.lock().await.push_back(task),
        }
    }
    
    pub async fn dequeue(&self) -> Option<DetectionTask> {
        // Drain high priority first, then normal, then low
        self.high_priority.lock().await.pop_front()
            .or_else(|| self.normal_priority.lock().await.pop_front())
            .or_else(|| self.low_priority.lock().await.pop_front())
    }
}
\`\`\`

### 5. Dynamic Routing Updates

Collectors publish heartbeat/capability messages:

\`\`\`rust
// In collector-core runtime
pub async fn start_heartbeat(&self) {
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    loop {
        interval.tick().await;
        
        let announcement = CollectorAnnouncement {
            collector_id: self.id.clone(),
            capabilities: self.capabilities(),
            load_factor: self.current_load(),
            timestamp: SystemTime::now(),
        };
        
        self.event_bus.publish(\"collectors/announce\", announcement.into()).await?;
    }
}
\`\`\`

Router watches for announcements and updates routing table:

\`\`\`rust
pub async fn watch_collector_updates(&self) {
    let mut stream = self.event_bus.subscribe(\"collectors/announce\").await?;
    
    while let Some(event) = stream.next().await {
        let announcement: CollectorAnnouncement = event.try_into()?;
        self.collectors.write().await.insert(
            announcement.collector_id.clone(),
            CollectorInfo {
                capabilities: announcement.capabilities,
                last_seen: announcement.timestamp,
                load_factor: announcement.load_factor,
            }
        );
    }
}
\`\`\`

### 6. Fallback Routing

\`\`\`rust
impl TaskDistributor {
    async fn distribute_with_fallback(&self, task: DetectionTask) -> Result<()> {
        let mut collectors = self.router.find_collectors(task.required_caps).await?;
        
        // Try primary collectors first
        for collector_id in collectors.iter().take(3) {
            match self.try_distribute(&task, collector_id).await {
                Ok(_) => return Ok(()),
                Err(e) => {
                    warn!(\"Failed to distribute to {}: {}\", collector_id, e);
                    continue;
                }
            }
        }
        
        // Fall back to any collector with partial capabilities
        let fallback = self.router.find_fallback_collectors(task.required_caps).await?;
        for collector_id in fallback {
            if self.try_distribute(&task, collector_id).await.is_ok() {
                return Ok(());
            }
        }
        
        Err(Error::NoAvailableCollector)
    }
}
\`\`\`

## Implementation Checklist

- [ ] Create \`daemoneye-eventbus\` workspace crate with \`EventBus\` trait
- [ ] Implement \`InProcessEventBus\` using tokio broadcast channels
- [ ] Define topic hierarchy and event schemas (tasks, collectors, results)
- [ ] Create \`TaskDistributor\` with capability-aware distribution logic
- [ ] Implement \`CapabilityRouter\` with dynamic collector registry
- [ ] Add \`TaskQueue\` with priority handling (high/normal/low)
- [ ] Integrate eventbus into \`daemoneye-agent\` main loop
- [ ] Update \`collector-core\` to publish heartbeats and capabilities
- [ ] Implement fallback routing for unavailable collectors
- [ ] Add load balancing strategies (round-robin, least-loaded, random)
- [ ] Create \`TaskHandle\` for result tracking and timeout handling
- [ ] Add metrics and observability (tasks distributed, routing decisions, failures)
- [ ] Write unit tests for eventbus, router, and queue components
- [ ] Add integration tests with multiple collectors
- [ ] Write end-to-end tests with failover scenarios
- [ ] Document eventbus API and usage patterns
- [ ] Update architecture diagrams with eventbus components

## Testing Strategy

### Unit Tests
- EventBus publish/subscribe with multiple subscribers
- CapabilityRouter collector registration and lookup
- TaskQueue priority ordering (high > normal > low)
- Load balancing strategy selection

### Integration Tests
- Multi-collector task distribution workflows
- Collector join/leave with routing table updates
- Fallback routing when primary collectors fail
- Priority task preemption under load

### End-to-End Tests  
- Full workflow: SQL rule → task distribution → collector execution → result aggregation
- Chaos testing: random collector failures during task execution
- Performance testing: 1000+ tasks distributed across 10+ collectors
- Load testing: sustained high-priority task stream

## Requirements

This implementation satisfies:
- **15.1**: daemoneye-cli query capabilities (via improved task routing)
- **15.3**: Extended CLI with advanced query capabilities
- **16.1**: Management commands for rules, alerts, health (via better orchestration)

## Related Issues

- Parent: #114 (Multi-process collector coordination via daemoneye-eventbus)
- Depends on: #113 (Basic topic infrastructure - COMPLETED)
- Related: #115 (Multi-process collector coordination workflows)
- Related: #117 (Result aggregation and load balancing)

## References

- Specification: \`.kiro/specs/daemoneye-core-monitoring/tasks.md\` (Section 2.6)
- Existing code: \`daemoneye-lib/src/ipc/client.rs\` (\`send_task_with_routing\`)
- Existing code: \`collector-core/src/source.rs\` (\`SourceCaps\`, \`EventSource\`)" \
  --add-label "enhancement" \
  --add-assignee "unclesp1d3r"

echo "Issue #116 has been updated successfully!"

Length of output: 94

---

@unclesp1d3r I've successfully enhanced issue #116 with comprehensive context and a detailed solution proposal. Here's what was updated:

🎯 Key Enhancements Made

1. Improved Title

Changed from: 2.6.3 Implement task distribution and capability-based routing
To: Implement Task Distribution and Capability-Based Routing via EventBus

2. Added Comprehensive Context

• Current State: Documented existing infrastructure (collector-core, IPC, capability negotiation)
• What's Missing: Clear gap analysis identifying the need for event-driven architecture
• Motivation: Explained why eventbus approach enables decoupling, scalability, resilience, and flexibility

3. Detailed Technical Solution

Provided complete implementation guidance with code examples for:

• EventBus Library: New daemoneye-eventbus crate with publish/subscribe pattern
• Topic Hierarchy: Structured topics for tasks, collectors, and results
• TaskDistributor: Capability-aware task distribution with load balancing
• CapabilityRouter: Dynamic collector registry with health tracking
• TaskQueue: Priority-based queuing (high/normal/low)
• Dynamic Routing: Heartbeat-based collector discovery and updates
• Fallback Routing: Resilient task distribution with multiple fallback strategies

4. Implementation Checklist

Added 17 concrete tasks covering:

• Core eventbus implementation
• Task distribution and routing logic
• Integration with existing components
• Metrics and observability
• Comprehensive testing
• Documentation

5. Testing Strategy

Defined multi-level testing approach:

• Unit Tests: Component-level validation
• Integration Tests: Multi-collector workflows
• End-to-End Tests: Full system validation with chaos testing

6. Metadata Updates

• ✅ Added enhancement label
• ✅ Assigned to you (@unclesp1d3r)
• ✅ Linked to parent issue Epic: Multi-Process Collector Coordination via Event Bus Architecture #114 and related issues
• ✅ Referenced specification and existing code files

The issue is now ready for implementation with clear technical direction and comprehensive context! 🚀

View the enhanced issue: #116