Epic: Multi-Process Collector Coordination via Event Bus Architecture

[unclesp1d3r]

Epic: Multi-Process Collector Coordination via Event Bus Architecture

Parent Epic: #93 - Develop Core DaemonEye Monitoring System
Related Issues: #87 (Event Bus Architecture), #121 (Topic Infrastructure - Completed)
Specification: .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6

Overview

This epic tracks the implementation of multi-process collector coordination capabilities that enable DaemonEye to scale within a single system by coordinating multiple collector processes. By implementing topic-based event distribution through the custom daemoneye-eventbus component, multiple collectors can coordinate their efforts, share workload, and provide redundancy without tight coupling.

Architecture: Cross-Process IPC with Topic-Based Messaging

Critical Architectural Clarification:

The daemoneye-eventbus is a cross-process IPC (Inter-Process Communication) system, NOT an in-process message broker. Key architectural points:

1. Process Architecture:

   • daemoneye-agent: Separate process that hosts the eventbus broker AND acts as a client
   • Collector Processes: Each collector (procmond, netmond, fsmond, etc.) runs as a separate OS process
   • Communication: All processes communicate via cross-process IPC using the interprocess crate

2. IPC Transport Layer:

   • Windows: Named pipes via CreateNamedPipe/CreateFile
   • Unix (Linux/macOS/FreeBSD): Unix domain sockets
   • Protocol: Protobuf-serialized messages over the IPC transport
   • Async: All IPC is async via Tokio runtime

3. Topic-Based Messaging Layer:

   • Layered Architecture: The pub/sub topic system is built ON TOP OF the cross-process IPC infrastructure
   • Custom Implementation: Built entirely on existing dependencies (no external brokers)
   • Broker Role: daemoneye-agent hosts the broker and routes messages between collector processes

┌─────────────────────────────────────────────────────────────────────┐
│                    daemoneye-agent (PROCESS)                        │
│                                                                     │
│  ┌────────────────────┐      ┌─────────────────────────────────┐  │
│  │ Detection Engine   │      │ Eventbus Broker (hosted here)   │  │
│  │ (also a client)    │─IPC─▶│ - Topic routing                 │  │
│  │                    │      │ - Pub/sub patterns              │  │
│  │ - Task generation  │◀IPC──│ - Load balancing                │  │
│  │ - Result aggregation│     │ - Capability routing            │  │
│  └────────────────────┘      └─────────────────────────────────┘  │
│                                          │                          │
└──────────────────────────────────────────┼──────────────────────────┘
                                           │ Cross-Process IPC
                                           │ (Named Pipes / Unix Sockets)
     ┌─────────────────────────────────────┼──────────────────────────────┐
     │                                     │                              │
┌────▼────────────────┐         ┌─────────▼──────────────┐    ┌─────────▼──────────────┐
│ procmond-1 (PROCESS)│         │ procmond-2 (PROCESS)   │    │ netmond-1 (PROCESS)    │
│                     │         │                        │    │                        │
│ Eventbus Client     │         │ Eventbus Client        │    │ Eventbus Client        │
│                     │         │                        │    │                        │
│ Subscribe (IPC):    │         │ Subscribe (IPC):       │    │ Subscribe (IPC):       │
│ • tasks.proc        │         │ • tasks.proc           │    │ • tasks.net            │
│                     │         │                        │    │                        │
│ Publish (IPC):      │         │ Publish (IPC):         │    │ Publish (IPC):         │
│ • results.proc      │         │ • results.proc         │    │ • results.net          │
│ • health            │         │ • health               │    │ • health               │
└─────────────────────┘         └────────────────────────┘    └────────────────────────┘
   Separate Process                Separate Process              Separate Process

Design Goals

Scope: Single-system, multi-process coordination (NOT distributed systems)

Primary Goals

1. Performance Optimization:

   • Support both high-load scenarios (thousands of events/second)
   • Ultra-low latency for cross-process IPC (<1ms p99 for local sockets)
   • Zero-copy where possible, minimal serialization overhead
   • Efficient backpressure handling to prevent producer/consumer rate mismatches

2. Cross-Platform Compatibility:

   • Windows, Linux, macOS (primary support)
   • FreeBSD (secondary support)
   • Same crate compiles on all platforms with minimal feature flags
   • Platform-native IPC primitives (named pipes on Windows, Unix sockets on Unix)

3. Security:

   • Secure IPC communications via OS-provided primitives
   • Named pipes (Windows) and Unix domain sockets reduce attack surface
   • No network exposure by default
   • Process isolation boundaries

4. Code Safety & Dependencies:

   • No unsafe code in daemoneye-eventbus implementation
   • No external libraries beyond existing workspace dependencies
   • Built entirely on: interprocess, tokio, prost, dashmap
   • All dependencies are well-audited and widely used

5. Communication Patterns:

   • One-to-one: Direct task assignment to specific collector process
   • One-to-many: Broadcast events to multiple subscriber processes
   • Pub/sub: Topic-based routing with wildcard matching
   • Queue support: Load-balanced distribution across multiple process instances
   • Backpressure tolerance: Producers and consumers operate at different rates

Background & Current Limitations

Current Architecture

DaemonEye's collector-core framework provides a solid foundation with:

• ✅ EventSource Trait: Universal abstraction for collection components
• ✅ IPC Communication: Point-to-point protobuf communication between agent and collector processes
• ✅ Event Aggregation: In-process mpsc channels for event batching
• ✅ Capability Negotiation: Collectors advertise their capabilities (PROCESS, NETWORK, FILESYSTEM, etc.)

Scaling Challenges

However, the current architecture faces limitations when scaling to multiple collector processes on a single system:

1. Point-to-Point IPC: Current IPC is 1:1 between daemoneye-agent process and a single collector process
2. No Coordination: Multiple collector processes cannot coordinate tasks or share workload
3. No Automatic Failover: If a collector process fails, there's no automatic task redistribution
4. Static Routing: Tasks are sent to specific collector processes rather than routed by capability
5. No Load Balancing: Cannot distribute high-volume collection tasks across multiple process instances

Strategic Importance

Multi-collector coordination within a single system is essential for:

• System Resource Utilization: Fully utilize multi-core CPUs by running parallel collector processes
• Workload Isolation: Different collector types handling their domains (process, network, filesystem)
• Fault Tolerance: Automatic failover when collector processes crash
• Performance: Distribute high-load scenarios across multiple processes

Proposed Solution: Custom Cross-Process Event Bus

Implementation Approach

Build a custom event bus using only existing workspace dependencies:

Core Dependencies (Already in Use):

• interprocess (v2.2.3) - Cross-platform IPC transport (named pipes/Unix sockets)
• tokio (v1.47.1) - Async runtime for IPC operations
• prost (v0.14.1) - Efficient protobuf serialization for cross-process messages
• dashmap (v6+) - Lock-free concurrent HashMap for topic routing

Why Custom Implementation:

• ✅ No external libraries or unsafe code
• ✅ Full control over cross-process IPC patterns
• ✅ Optimized for local single-system performance
• ✅ Minimal dependencies and attack surface
• ✅ Tailored backpressure and queue strategies

Architecture Components

// Broker hosted in daemoneye-agent process
pub struct CrossProcessEventBus {
    // Topic routing: topic pattern -> subscriber processes
    topics: Arc<DashMap<String, Vec<ProcessSubscription>>>,
    
    // Active IPC connections to collector processes
    ipc_connections: Arc<DashMap<ProcessId, IpcConnection>>,
    
    // Load balancing state for queue groups
    queue_state: Arc<DashMap<String, QueueGroup>>,
    
    // Backpressure management
    flow_control: Arc<FlowController>,
}

pub struct ProcessSubscription {
    process_id: ProcessId,
    subscription_type: SubscriptionType,
    ipc_sender: mpsc::Sender<EventEnvelope>,
}

pub enum SubscriptionType {
    /// One-to-one: specific process subscription
    Direct,
    /// One-to-many: broadcast to all matching subscribers
    Broadcast,
    /// Queue: load-balanced across multiple processes
    Queue { group_name: String },
}

pub struct QueueGroup {
    members: Vec<ProcessId>,
    next_index: AtomicUsize,  // Round-robin state
    strategy: LoadBalancingStrategy,
}

pub enum LoadBalancingStrategy {
    RoundRobin,
    LeastLoaded,
    Random,
}

Communication Patterns

1. One-to-One (Direct Task Assignment)

// Agent → specific procmond instance
eventbus.publish_direct(
    process_id,
    "control.collector.task.process",
    task
).await?;

2. One-to-Many (Broadcast)

// Agent → all procmond instances
eventbus.publish_broadcast(
    "control.collector.config.reload",
    config
).await?;

3. Pub/Sub (Topic-Based Routing)

// Collector subscribes to topic pattern
eventbus.subscribe("events.process.+", SubscriptionType::Broadcast).await?;

// Agent publishes to topic
eventbus.publish("events.process.spawned", event).await?;

4. Queue (Load-Balanced Distribution)

// Multiple procmond processes join queue group
eventbus.subscribe_queue(
    "control.collector.task.process",
    "procmond-workers"
).await?;

// Agent publishes task - only ONE process receives it
eventbus.publish_queue("control.collector.task.process", task).await?;

Backpressure Handling

pub struct FlowController {
    /// Per-process send buffer limits
    max_buffer_size: usize,
    
    /// Strategy when buffer full
    overflow_strategy: OverflowStrategy,
}

pub enum OverflowStrategy {
    /// Block producer until space available
    Block,
    /// Drop oldest messages
    DropOldest,
    /// Drop newest messages
    DropNewest,
    /// Return error to producer
    Error,
}

Key Components

1. Topic-Based Message Routing (Cross-Process)

Task Distribution Topics (agent → collectors via IPC):

• control.collector.task.process - Process monitoring tasks sent to procmond processes
• control.collector.task.network - Network monitoring tasks sent to netmond processes
• control.collector.task.filesystem - Filesystem monitoring tasks sent to fsmond processes
• control.collector.task.performance - Performance monitoring tasks sent to perfmond processes

Result Aggregation Topics (collectors → agent via IPC):

• events.process.* - Process events from procmond processes
• events.network.* - Network events from netmond processes
• events.filesystem.* - Filesystem events from fsmond processes
• events.performance.* - Performance events from perfmond processes

Health & Control Topics (bidirectional cross-process):

• control.health.heartbeat - Collector process heartbeat signals
• control.health.status - Process health status updates
• control.collector.lifecycle - Process start/stop/config commands

2. Capability-Based Routing (Cross-Process)

Collector processes advertise their capabilities through the event bus:

• Capability Advertisement: Each collector process publishes its SourceCaps on startup via IPC
• Dynamic Routing: Agent routes tasks to appropriate collector processes based on capabilities
• Automatic Discovery: New collector processes automatically join the coordination mesh via IPC
• Graceful Departure: Collector processes unsubscribe topics on shutdown

3. Load Balancing Strategies (Cross-Process)

Strategies:

• Round-Robin: Distribute tasks evenly across available collector processes (lowest latency)
• Least-Loaded: Route to collector process with lowest queue depth (best for fairness)
• Random: Random selection (simple, minimal state)

Failover:

• Heartbeat Monitoring: Detect failed collector processes via missed heartbeats
• Task Redistribution: Automatically reassign tasks from failed processes
• Process Restart: Automatically restart crashed collector processes

4. Result Aggregation & Correlation (Cross-Process)

Correlation Metadata:

struct CorrelationMetadata {
    correlation_id: Uuid,           // Unique workflow identifier
    parent_correlation_id: Option<Uuid>,  // For hierarchical workflows
    root_correlation_id: Uuid,      // Original workflow root
    sequence_number: u64,           // Ordering within workflow
    workflow_stage: String,         // Current stage (e.g., "collection", "analysis")
    source_process: String,         // Originating collector process
    correlation_tags: HashMap<String, String>,  // Flexible tagging
}

Aggregation Strategies:

• Stream-Based: Real-time aggregation as results arrive from different processes via IPC
• Batch-Based: Collect results until timeout or threshold met
• Correlation-Based: Group results by correlation IDs for multi-stage workflows across processes

Cross-Platform Support Matrix

Component	Windows	Linux	macOS	FreeBSD	Notes
interprocess	✅ Primary	✅ Primary	✅ Primary	✅ Secondary	Named pipes (Win) / Unix sockets - Cross-process IPC
tokio	✅ Primary	✅ Primary	✅ Primary	✅ Secondary	Full async runtime support
tokio::sync	✅ Primary	✅ Primary	✅ Primary	✅ Secondary	In-process channels (within broker process only)
prost	✅ Primary	✅ Primary	✅ Primary	✅ Secondary	Pure Rust, no platform deps
dashmap	✅ Primary	✅ Primary	✅ Primary	✅ Secondary	Lock-free concurrent data structures

Cross-Process Transport Strategy:

1. Primary: Unix domain sockets (Linux/macOS/FreeBSD) or Named pipes (Windows) via interprocess for local cross-process IPC
2. Fallback: TCP loopback (127.0.0.1) for platforms without Unix socket support (rare)
3. Not Used: Network sockets to external systems (out of scope)

Dependency Updates Required

No new external dependencies - use only existing workspace dependencies:

[workspace.dependencies]
# Already in use - sufficient for custom event bus
interprocess = "2.2.3"  # Cross-process IPC transport (CRITICAL)
prost = "0.14.1"        # Cross-process message serialization
tokio = { version = "1.47.1", features = ["full"] }  # Async runtime

# Add for custom event bus implementation
dashmap = "6.1"         # Lock-free concurrent HashMap for topic routing
parking_lot = "0.12"    # Faster sync primitives (optional optimization)

Implementation Plan

Phase 1: Foundation ✅ (Completed - #113, #121)

• ✅ Basic topic infrastructure in daemoneye-eventbus
• ✅ Topic routing and subscription mechanisms
• ✅ Wildcard topic matching support (+ and # patterns)
• ✅ Message broker with pub/sub capabilities

Phase 2: Coordination Workflows (#115)

• Implement topic-based task distribution for multiple collector process types
• Create capability-based routing for task publishing across processes
• Add result aggregation from domain-specific topics across processes
• Implement basic load balancing and failover for collector processes
• Crates to integrate: tokio::sync::mpsc (bounded channels), dashmap (topic routing)

Phase 3: Advanced Routing (#116, #117)

• Task distribution logic using eventbus topic publishing to collector processes
• Collector type routing based on process capabilities
• Task queuing and priority handling across process boundaries
• Dynamic routing updates as collector processes join/leave
• Result correlation and ordering logic from multiple processes
• Automatic task redistribution on collector process failure
• Crates to integrate: dashmap for routing tables, queue group logic for cross-process coordination

Phase 4: Testing & Validation (#118)

• End-to-end tests with multiple collector processes (procmond, netmond, etc.)
• Task distribution and result aggregation workflow validation across processes
• Load balancing and failover scenario testing with process crashes
• Performance benchmarking under high load with multiple processes
• Test crates: Existing criterion, proptest, insta

Phase 5: Complete Topic Hierarchy (#119)

• Implement full events.* topic hierarchy for cross-process event routing
• Complete control.* topic structure for cross-process control plane
• Topic-based access control and security boundaries between processes
• Wildcard matching optimization for cross-process subscriptions

Phase 6: Correlation & Forensics (#120)

• Correlation metadata implementation with process tracking
• Sequence numbering and workflow stage tracking across processes
• Correlation ID propagation through EventBus envelopes across IPC boundaries
• Forensic correlation tracking for security investigations across process boundaries

Benefits & Value Proposition

Performance

• ✅ Ultra-Low Latency: <1ms p99 for local IPC operations
• ✅ High Throughput: Thousands of events/second across processes
• ✅ Efficient Serialization: Protobuf minimizes overhead
• ✅ Zero-Copy Potential: Shared memory optimizations (future)

Scalability (Single System)

• ✅ Multi-Core Utilization: Distribute work across multiple collector processes
• ✅ Specialized Collectors: Different collector process types handle their specific domains
• ✅ Dynamic Scaling: Collector processes can join/leave without system reconfiguration

Reliability

• ✅ Fault Tolerance: Automatic failover prevents single points of failure across processes
• ✅ Load Distribution: Prevents any single collector process from being overwhelmed
• ✅ Graceful Degradation: System continues operating even if collector processes fail
• ✅ Process Isolation: Collector crashes don't affect agent or other collectors

Security

• ✅ Reduced Attack Surface: Local IPC only (no network exposure)
• ✅ OS-Provided Security: Named pipes and Unix sockets with OS permissions
• ✅ Process Boundaries: Strong isolation between components
• ✅ Safe Code Only: No unsafe blocks in daemoneye-eventbus

Maintainability

• ✅ Minimal Dependencies: Only well-audited crates (interprocess, tokio, prost)
• ✅ Cross-Platform: Same code on Windows, Linux, macOS, FreeBSD
• ✅ Observable: Built-in health monitoring and metrics across all processes
• ✅ Flexible: Topic-based architecture adapts to new requirements without process coupling

Requirements Mapping

This epic implements the following requirements from the specification:

• 15.1: Multi-collector support with unified event processing across processes
• 15.3: Event coordination and task distribution across process boundaries
• 15.4: Result aggregation and correlation from multiple collector processes
• 16.1: Capability-based routing and dynamic feature discovery across processes
• 16.3: Load balancing and failover mechanisms for process failures
• 16.4: Health monitoring and availability tracking across all processes

Subtasks

• #113 - 2.6.1 Create basic topic infrastructure for multi-collector coordination ✅
• Implement message broker for multi-collector coordination and load balancing #115 - 2.6.2 Implement multi-process collector coordination workflows
• Implement Task Distribution and Capability-Based Routing via EventBus #116 - 2.6.3 Implement task distribution and capability-based routing across processes
• Implement Result Aggregation, Correlation, and Load Balancing for Multi-Collector Coordination #117 - 2.6.4 Add result aggregation and load balancing for collector processes
• End-to-end tests for multi-collector coordination with task distribution and failover #118 - 2.6.5 Write end-to-end tests with multiple collector process coordination
• Implement MQTT-style topic hierarchy for daemoneye-eventbus multi-collector coordination #119 - 2.6.7 Implement complete topic hierarchy for cross-process event routing
• Add correlation metadata and workflow tracking for multi-collector coordination #120 - 2.6.8 Add correlation metadata and multi-collector workflow support across processes

Success Criteria

• Multiple collector process instances can coordinate through cross-process event bus
• Tasks are automatically routed to appropriate collector processes based on capabilities
• Load balancing distributes work efficiently across collector processes
• Automatic failover when collector processes become unavailable
• Results are correctly aggregated with correlation metadata from multiple processes
• Complete topic hierarchy implemented for cross-process event routing
• End-to-end tests validate multi-collector process scenarios with crashes and restarts
• Performance benchmarks achieve <1ms p99 latency for local IPC
• Cross-platform support validated on Windows, Linux, macOS, and FreeBSD
• Zero unsafe code in daemoneye-eventbus implementation
• Backpressure handling prevents producer/consumer rate mismatches
• Documentation updated with multi-collector process deployment guides

Technical Considerations

Cross-Platform Support

• Full Windows and Unix support for cross-process event bus broker
• Named Pipes (Windows) and UNIX sockets (Linux/macOS/FreeBSD) for cross-process IPC via interprocess
• Minimal feature flags required for platform differences
• Platform-specific testing on primary (Windows/Linux/macOS) and secondary (FreeBSD) targets

Security

• Topic-based access control to prevent unauthorized cross-process access
• OS-provided IPC security (named pipe/Unix socket permissions)
• Process authentication via IPC connection metadata
• No unsafe code in daemoneye-eventbus implementation

Performance

• Efficient cross-process message routing with minimal IPC overhead
• Backpressure handling to prevent overload across process boundaries
• Configurable buffer sizes and timeouts for cross-process communication
• Zero-copy message passing where possible with shared memory (future optimization)
• Benchmark target: <1ms p99 latency for local cross-process IPC

Monitoring & Observability

• Metrics for cross-process message throughput, latency, queue depths
• Collector process health and availability tracking
• Topic subscription monitoring across all processes
• Integration with existing tracing infrastructure with process-aware spans

---

Status: In Progress
Priority: High - Essential for multi-process coordination
Estimated Effort: 3-4 weeks across all subtasks

Next Steps:

1. Begin Phase 2 implementation (Implement message broker for multi-collector coordination and load balancing #115) with custom event bus broker
2. Focus on cross-process IPC patterns using interprocess + tokio
3. Implement one-to-one, one-to-many, pub/sub, and queue patterns
4. Add backpressure handling and flow control
5. Performance testing and optimization for <1ms p99 latency

[traycerai]

Plan

Observations

Repo at /tmp/github/issues/64295955_1056412443_1762730039034/ includes daemoneye-lib (IPC with interprocess/prost), collector-core (EventSource/SourceCaps), daemoneye-agent, procmond. Phase 1 (#113/#121) complete: basic topic infra in daemoneye-eventbus. Subtasks #115-#120 open. Epic #114 requires custom cross-process eventbus: broker in daemoneye-agent using interprocess for IPC (named pipes/Unix sockets), prost serialization. Clients: separate collector processes (procmond etc.) connect via IPC. Support pub/sub with topics (control., events.), wildcards (+/#), one-to-one/many/queue patterns. Capability routing, load balancing (RR/least-loaded), failover via heartbeats. Result aggregation by correlation_id. Backpressure with bounded channels. No unsafe, no new deps beyond workspace (add dashmap if needed). Cross-platform: primary Win/Linux/macOS, secondary FreeBSD. Latency <1ms p99. Extend protos for CorrelationMetadata, Heartbeat.

Approach

Implement daemoneye-eventbus as cross-process pub/sub broker. Host broker in daemoneye-agent: async IPC listener (interprocess), route via in-broker topics (tokio::broadcast for pub/sub, dashmap for subs/connections). Clients (collectors) connect via IPC streams, serialize prost msgs. Add CollectorCoordinator for reg/heartbeats/caps over IPC. TaskRouter for cap-based distribution (RR/least-loaded). ResultAggregator for corr-based merging. Support patterns: direct publish, broadcast, queue (load-balance). Backpressure: bounded mpsc, overflow strategies (block/drop). Integrate: agent spawns broker, IPC listener; collectors connect, register caps, sub topics, pub results/heartbeats. E2E tests: spawn multi-process (tokio::process), simulate tasks/failover. Docs: architecture, topics, setup. Benchmarks: latency <1ms. No unsafe code.

Reasoning

Explored repo: listed dirs/files, confirmed daemoneye-eventbus exists from Phase 1 with basic topics. Reviewed daemoneye-lib/src/ipc for interprocess setup. collector-core/src/source.rs has SourceCaps. procmond/src/main.rs single-process. Protos in daemoneye-lib/proto/ipc.proto lack correlation/heartbeat. Searched for existing eventbus—basic pub/sub in-process. Aligned with epic spec: cross-process via IPC, broker in agent, clients separate, topics as specified, patterns, backpressure. Assumed Phase 1 basics (topics, wildcards) built upon for Phases 2-6.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

Add daemoneye-eventbus to members if not already (from Phase 1).

[workspace]
members = [
"daemoneye-lib",
"collector-core",
"daemoneye-agent",
"procmond",
"daemoneye-eventbus",
// others
]

[workspace.dependencies]
// Existing + add if missing
dashmap = "6"
parking_lot = "0.12" // optional for sync

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

References

• 📄 daemoneye-lib/proto

Ensure deps for cross-process:

[dependencies]
// Existing from Phase 1
tokio = { workspace = true, features = ["full"] }
interprocess = { workspace = true }
prost = { workspace = true }
dashmap = { workspace = true }
parking_lot = { workspace = true } // if added
uuid = { workspace = true }
anyhow = "1"
tracing = { workspace = true }
serde = { workspace = true }

[dependencies.daemoneye-lib]
path = "../daemoneye-lib"

[dependencies.collector-core]
path = "../collector-core"

[build-dependencies]
prost-build = "0.12"

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

Extend Phase 1 exports for cross-process:

pub use bus::{CrossProcessEventBus, EventBusConfig, TransportConfig, PublishMode};
pub use topic::{TopicMatcher, SubscriptionType};
pub use coordinator::CollectorCoordinator;
pub use router::{TaskRouter, LoadBalancingStrategy};
pub use aggregator::{ResultAggregator, AggregationStrategy};
pub use flow::{FlowController, OverflowStrategy};

mod bus;
mod topic;
mod coordinator;
mod router;
mod aggregator;
mod flow;

// From Phase 1: existing topic basics

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

References

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

#115/#116: Extend Phase 1 in-process bus to cross-process broker.

#[derive(Clone)]
pub struct CrossProcessEventBus {
inner: Arc,
transport: Option, // None for in-broker
}

struct InnerBus {
topics: Arc<DashMap<String, broadcast::Sender<Vec>>>, // Phase 1
subscribers: Arc<DashMap<String, Vec>>, // Extend: per-process subs
connections: Arc<DashMap<ProcessId, IpcConnection>>, // IPC streams
coordinator: Arc,
router: Arc,
aggregator: Arc,
flow: Arc,
}

pub enum TransportLayer {
Broker(IpcListener), // In agent
Client(IpcStream), // In collectors
}

pub struct EventBusConfig {
transport: TransportConfig,
buffer_size: usize,
heartbeat_interval: Duration,
}

pub enum TransportConfig {
Ipc { path: PathBuf },
// Fallback Tcp { addr: SocketAddr } // If needed
}

pub enum PublishMode {
Direct(ProcessId),
Broadcast(String),
Queue(String),
PubSub(String),
}

impl CrossProcessEventBus {
pub async fn broker(config: EventBusConfig) -> Result {
// Extend Phase 1: spawn IPC listener, accept client connections
// For each conn: spawn read/write tasks, route msgs to topics
// Use prost::Message for serialization over IPC
}
pub async fn client(config: EventBusConfig) -> Result {
// Connect to broker IPC, handle send/recv loops
}
pub async fn publish(&self, mode: PublishMode, topic: &str, payload: impl prost::Message) -> Result<()> {
let env = EventEnvelope { topic: topic.to_string(), payload: payload.encode_to_vec(), ..Default::default() };
match mode {
Direct(pid) => { /* send via conn to pid / },
Broadcast(t) => { / broadcast to matching subs via IPC / },
Queue(group) => { / load-balance to one in group / },
PubSub(t) => { / route via topics, forward to clients */ },
}
// Apply flow control: check buffers, apply strategy
}
pub async fn subscribe(&self, pattern: &str, sub_type: SubscriptionType) -> Result<Receiver> {
// Client: register pattern with broker over IPC
// Broker: add to subscribers dashmap
}
// Other methods: dispatch_task, aggregate_results as before
}

#[derive(Clone, prost::Message)]
pub struct EventEnvelope {
topic: String,
payload: Vec,
metadata: HashMap<String, String>,
source_process: Option,
correlation: Option,
}

pub struct IpcConnection {
reader: tokio::io::ReadHalf,
writer: tokio::io::WriteHalf,
sender: mpsc::Sender, // Bounded for backpressure
buffer_size: usize,
}

// Spawn per-connection tasks: deserialize incoming, route to topics; serialize outgoing

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

References

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

#119: Extend Phase 1 wildcard matching for cross-process.

// Existing TopicMatcher from Phase 1
impl TopicMatcher {
// Existing matches, split_topic
pub fn route_to_processes(&self, topic: &str, subscribers: &DashMap<String, Vec>) -> Vec {
// Find matching patterns, return unique process_ids
}
}

pub enum SubscriptionType {
Direct,
Broadcast,
Queue { group: String },
}

pub struct ProcessSubscription {
process_id: ProcessId,
pattern: String,
sub_type: SubscriptionType,
// From Phase 1: existing fields
}

// Topics: control.collector.task.process, events.process.*, control.health.heartbeat, etc.
// Hierarchical: split by '.', match + (single), # (multi)

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

References

• 📄 collector-core/src/source.rs
• 📄 daemoneye-lib/proto/ipc.proto ^(MODIFY)

#115: New for cross-process lifecycle.

#[derive(Clone, Serialize, Deserialize, prost::Message)]
pub struct CollectorInfo {
id: String, // Uuid as str
caps: SourceCaps,
load: f32,
last_heartbeat: u64, // timestamp
pid: u32,
status: HealthStatus,
}

pub enum HealthStatus { Healthy, Degraded, Failed }

pub struct CollectorCoordinator {
collectors: Arc<DashMap<String, CollectorInfo>>,
heartbeat_tx: broadcast::Sender,
}

impl CollectorCoordinator {
pub fn new() -> Self;
pub async fn register(&self, caps: SourceCaps, pid: u32) -> String {
let id = Uuid::new_v4().to_string();
// Publish to control.collector.lifecycle via bus
id
}
pub fn update_heartbeat(&self, id: &str, load: f32) {
// Update info, publish Heartbeat to control.health.heartbeat
// Monitor: tokio task removes stale (>30s)
}
pub fn get_capable(&self, required: SourceCaps) -> Vec {
// Filter active
}
pub fn remove(&self, id: &str) {
// On unregister or timeout
}
}

// Use prost for Heartbeat msg

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

References

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

#116: New for cap-based routing cross-process.

#[derive(Clone)]
pub enum LoadBalancingStrategy { RoundRobin, LeastLoaded, Random }

pub struct TaskRouter {
coordinator: Arc,
strategy: LoadBalancingStrategy,
queue_groups: Arc<DashMap<String, QueueGroup>>,
}

pub struct QueueGroup {
members: Vec, // collector_ids
next: AtomicUsize,
loads: DashMap<String, f32>,
}

impl TaskRouter {
pub fn new(coordinator: Arc, strategy: LoadBalancingStrategy) -> Self;
pub fn route(&self, task: &DetectionTask, required_caps: SourceCaps) -> Vec {
let candidates = self.coordinator.get_capable(required_caps);
match self.strategy {
RoundRobin => {
// For queue: select next in group
// Return { collector_id, topic: "control.collector.task.process" }
},
LeastLoaded => { /* sort candidates by load / },
Random => { / random select */ },
}
}
pub fn join_queue(&self, group: &str, id: &str) {
// Add to group members
}
}

pub struct TaskAssignment {
collector_id: String,
topic: String,
mode: PublishMode,
}

// On dispatch: bus.publish(assignment.mode, assignment.topic, task)

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

References

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

#117: New for cross-process result merging.

#[derive(Clone)]
pub enum AggregationStrategy { All, Any, Timeout }

pub struct ResultAggregator {
pending: Arc<DashMap<String, AggregationState>>,
bus: Arc,
}

pub struct AggregationState {
corr_id: String,
expected: usize,
received: HashMap<String, DetectionResult>, // by collector_id
timeout: Instant,
strategy: AggregationStrategy,
}

impl ResultAggregator {
pub fn new(bus: Arc) -> Self;
pub fn submit(&self, corr_id: &str, collector_id: &str, result: DetectionResult) {
// Update state, check complete
// If done, aggregate (e.g., union events), notify via bus or return
// Handle timeout: redistribute via router
}
pub async fn aggregate(&self, corr_id: &str, timeout: Duration) -> Result {
// tokio::time::timeout on watcher task
}
}

// Subscribe to events.process.*, filter by corr_id

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

References

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

#115: New for backpressure cross-process.

pub struct FlowController {
max_buffer: usize,
strategy: OverflowStrategy,
metrics: Arc<DashMap<String, FlowMetrics>>,
}

pub enum OverflowStrategy {
Block,
DropOldest,
DropNewest,
Error,
}

pub struct FlowMetrics {
queue_depth: usize,
latency: Duration,
dropped: usize,
}

impl FlowController {
pub fn new(max_buffer: usize, strategy: OverflowStrategy) -> Self;
pub fn check_backpressure(&self, conn: &IpcConnection) -> Result<()> {
// Check sender len() vs max, apply strategy
// Block: await on permit if full
// Drop: remove from channel
}
pub fn apply_strategy(&self, sender: &mpsc::Sender, strategy: OverflowStrategy) {
match strategy {
Block => { /* use semaphore */ },
// etc.
}
}
}

// Integrate in publish: before send, check/apply

📄 daemoneye-lib/proto/ipc.proto ^(MODIFY) 🔗

References

• 📄 daemoneye-lib/build.rs

#120: Extend protos for cross-process.

message CorrelationMetadata {
string id = 1;
optional string parent_id = 2;
string root_id = 3;
uint64 sequence = 4;
string stage = 5;
string source_process = 6;
map<string, string> tags = 7;
uint32 expected = 8;
}

message DetectionTask {
// existing
optional CorrelationMetadata correlation = 100;
}

message DetectionResult {
// existing
optional CorrelationMetadata correlation = 100;
}

message Heartbeat {
string collector_id = 1;
uint64 timestamp = 2;
float load = 3;
SourceCaps capabilities = 4;
HealthStatus status = 5;
}

enum HealthStatus {
HEALTHY = 0;
DEGRADED = 1;
FAILED = 2;
}

// Update build.rs to include

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

References

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

#115: Integrate broker in agent.

async fn main() -> Result<()> {
// Config: multi_process: bool, ipc_path: PathBuf, etc.
if config.multi_process {
let bus_config = EventBusConfig { transport: TransportConfig::Ipc { path: config.ipc_path.clone() }, buffer_size: 1024, .. };
let bus = CrossProcessEventBus::broker(bus_config).await?;
// Spawn listener task if not in broker init
let coordinator = bus.coordinator();
let router = bus.router();
let aggregator = bus.aggregator();
// Subscribe to events.*.#, control.health.heartbeat
// In detection: generate corr_id, set task.correlation, assignments = router.route(&task, caps);
// for ass in assignments { bus.publish(ass.mode, ass.topic, task.clone()).await?; }
// result = aggregator.aggregate(&corr_id, Duration::from_secs(10)).await?;
// Handle heartbeats: coordinator.update_heartbeat
} else {
// Existing single-process
}
// Flow: apply backpressure in loops
}

// Add tracing spans for cross-process ops

📄 procmond/src/main.rs ^(MODIFY) 🔗

References

• 📄 procmond/src/config.rs
• 📄 collector-core/src/collector.rs
• 📄 daemoneye-eventbus/src/bus.rs ^(NEW)

#115: Integrate client in procmond.

async fn main() -> Result<()> {
// Config: multi_mode: bool, agent_ipc_path: PathBuf
if config.multi_mode {
let bus_config = EventBusConfig { transport: TransportConfig::Ipc { path: config.agent_ipc_path }, .. };
let bus = CrossProcessEventBus::client(bus_config).await?;
let id = bus.coordinator().register(SourceCaps::PROCESS, std::process::id() as u32).await;
// Join queue: bus.router().join_queue("procmond-workers", &id);
// Spawn heartbeat: tokio::spawn(async move {
// loop { bus.publish(PubSub("control.health.heartbeat".to_string()), Heartbeat { id: id.clone(), load: get_load(), .. }).await; }
// });
// Subscribe: let mut rx = bus.subscribe("control.collector.task.process", SubscriptionType::Queue { group: "procmond-workers" }).await?;
// while let Some(env) = rx.recv().await { let task: DetectionTask = prost::Message::decode(&env.payload[..])?; /* execute, set result.correlation = task.correlation, bus.publish(PubSub("events.process.results".to_string()), result).await?; */ }
// On shutdown: coordinator.remove(&id)
} else {
// Existing
}
}

// Use collector-core for execution

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

References

• 📄 daemoneye-eventbus/src/bus.rs ^(NEW)
• 📄 daemoneye-agent/src/main.rs ^(MODIFY)
• 📄 procmond/src/main.rs ^(MODIFY)

#118: E2E multi-process tests.

#[tokio::test]
async fn test_multi_process_coordination() {
let temp_dir = tempfile::tempdir()?;
let ipc_path = temp_dir.path().join("eventbus.ipc");
// Spawn agent process with broker: use tokio::process::Command::new(env::current_exe()).arg("--multi").arg(&ipc_path).spawn()?;
// Wait for broker ready
// Spawn procmond-1: Command... arg("--multi-client").arg(&ipc_path).spawn()?;
// Spawn procmond-2 similarly
// Dispatch task via mock agent client, verify routing to one (queue), results aggregated
// Use temp ipc paths, assert corr_id matches
}

#[tokio::test]
async fn test_failover() {
// Similar setup, dispatch task, kill one procmond (child.kill()), verify redistribution to other
// Assert aggregation completes with data from survivor
}

#[tokio::test]
async fn test_load_balancing() {
// Dispatch 4 tasks, assert 2 to each in RR
// Measure IPC latency: start = Instant::now(); publish; assert elapsed < Duration::from_millis(1);
}

#[tokio::test]
async fn test_backpressure() {
// Full buffer, publish with Block: await blocks
// With Drop: count dropped msgs
}

// Mock for faster tests, full process in CI; cross-platform via cfg

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

References

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

#118: Performance benchmarks.

use criterion::{criterion_group, criterion_main, Criterion};

fn bench_ipc_publish(c: &mut Criterion) {
let rt = tokio::runtime::Runtime::new()?;
c.bench_function("ipc_publish", |b| {
b.iter_custom(|iters| {
rt.block_on(async {
let bus = CrossProcessEventBus::broker(EventBusConfig::default()).await?;
let start = Instant::now();
for _ in 0..iters {
bus.publish(PubSub("test".to_string()), "topic", TestMsg {}).await?;
}
start.elapsed()
})
});
});
}

// Target: p99 <1ms for 1000 iters
// Separate benches for Win/Unix via features

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

References

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

Update docs for epic.

DaemonEye Cross-Process EventBus

Custom IPC-based pub/sub for multi-collector coordination (Epic #114).

Architecture

• Broker: Hosted in daemoneye-agent process (IPC listener via interprocess)
• Clients: Separate OS processes (procmond, netmond etc.) connect via IPC
• Transport: Platform-native (Named Pipes Win, Unix Sockets Unix), prost over streams
• Patterns: Direct, Broadcast, PubSub (wildcards), Queue (load-balance)
• Topics: control.collector.task., events.., control.health.
• Cross-Platform: Primary: Win/Linux/macOS; Secondary: FreeBSD

Setup

• Agent: CrossProcessEventBus::broker(config.with_ipc_path("\\.\pipe\eventbus"))
• Collector: CrossProcessEventBus::client(config.with_ipc_path(path))

Features

• Capability routing (Implement Task Distribution and Capability-Based Routing via EventBus #116)
• Load balancing/failover (Implement Result Aggregation, Correlation, and Load Balancing for Multi-Collector Coordination #117)
• Result aggregation (Implement Result Aggregation, Correlation, and Load Balancing for Multi-Collector Coordination #117)
• Backpressure (Implement message broker for multi-collector coordination and load balancing #115)
• Correlation (Add correlation metadata and workflow tracking for multi-collector coordination #120)

Performance

<1ms p99 latency, thousands eps.

See examples/cross_process.rs, tests/integration.rs.

No unsafe code, minimal deps.

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

Repo at /tmp/github/issues/64295955_1056412443_1762730039034/ includes daemoneye-lib (IPC with interprocess/prost), collector-core (EventSource/SourceCaps), daemoneye-agent, procmond. Phase 1 (#113/#121) complete: basic topic infra in daemoneye-eventbus. Subtasks #115-#120 open. Epic #114 requires custom cross-process eventbus: broker in daemoneye-agent using interprocess for IPC (named pipes/Unix sockets), prost serialization. Clients: separate collector processes (procmond etc.) connect via IPC. Support pub/sub with topics (control.*, events.*), wildcards (+/#), one-to-one/many/queue patterns. Capability routing, load balancing (RR/least-loaded), failover via heartbeats. Result aggregation by correlation_id. Backpressure with bounded channels. No unsafe, no new deps beyond workspace (add dashmap if needed). Cross-platform: primary Win/Linux/macOS, secondary FreeBSD. Latency <1ms p99. Extend protos for CorrelationMetadata, Heartbeat.

### Approach

Implement daemoneye-eventbus as cross-process pub/sub broker. Host broker in daemoneye-agent: async IPC listener (interprocess), route via in-broker topics (tokio::broadcast for pub/sub, dashmap for subs/connections). Clients (collectors) connect via IPC streams, serialize prost msgs. Add CollectorCoordinator for reg/heartbeats/caps over IPC. TaskRouter for cap-based distribution (RR/least-loaded). ResultAggregator for corr-based merging. Support patterns: direct publish, broadcast, queue (load-balance). Backpressure: bounded mpsc, overflow strategies (block/drop). Integrate: agent spawns broker, IPC listener; collectors connect, register caps, sub topics, pub results/heartbeats. E2E tests: spawn multi-process (tokio::process), simulate tasks/failover. Docs: architecture, topics, setup. Benchmarks: latency <1ms. No unsafe code.

### Reasoning

Explored repo: listed dirs/files, confirmed daemoneye-eventbus exists from Phase 1 with basic topics. Reviewed daemoneye-lib/src/ipc for interprocess setup. collector-core/src/source.rs has SourceCaps. procmond/src/main.rs single-process. Protos in daemoneye-lib/proto/ipc.proto lack correlation/heartbeat. Searched for existing eventbus—basic pub/sub in-process. Aligned with epic spec: cross-process via IPC, broker in agent, clients separate, topics as specified, patterns, backpressure. Assumed Phase 1 basics (topics, wildcards) built upon for Phases 2-6.

## Proposed File Changes

### Cargo.toml(MODIFY)

Add daemoneye-eventbus to members if not already (from Phase 1).

[workspace]
members = [
  "daemoneye-lib",
  "collector-core",
  "daemoneye-agent",
  "procmond",
  "daemoneye-eventbus",
  // others
]

[workspace.dependencies]
// Existing + add if missing
dashmap = "6"
parking_lot = "0.12" // optional for sync

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

References: 

- daemoneye-lib/proto

Ensure deps for cross-process:

[dependencies]
// Existing from Phase 1
tokio = { workspace = true, features = ["full"] }
interprocess = { workspace = true }
prost = { workspace = true }
dashmap = { workspace = true }
parking_lot = { workspace = true } // if added
uuid = { workspace = true }
anyhow = "1"
tracing = { workspace = true }
serde = { workspace = true }

[dependencies.daemoneye-lib]
path = "../daemoneye-lib"

[dependencies.collector-core]
path = "../collector-core"

[build-dependencies]
prost-build = "0.12"

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

Extend Phase 1 exports for cross-process:

pub use bus::{CrossProcessEventBus, EventBusConfig, TransportConfig, PublishMode};
pub use topic::{TopicMatcher, SubscriptionType};
pub use coordinator::CollectorCoordinator;
pub use router::{TaskRouter, LoadBalancingStrategy};
pub use aggregator::{ResultAggregator, AggregationStrategy};
pub use flow::{FlowController, OverflowStrategy};

mod bus;
mod topic;
mod coordinator;
mod router;
mod aggregator;
mod flow;

// From Phase 1: existing topic basics

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

References: 

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

#115/#116: Extend Phase 1 in-process bus to cross-process broker.

#[derive(Clone)]
pub struct CrossProcessEventBus {
    inner: Arc<InnerBus>,
    transport: Option<TransportLayer>, // None for in-broker
}

struct InnerBus {
    topics: Arc<DashMap<String, broadcast::Sender<Vec<u8>>>>, // Phase 1
    subscribers: Arc<DashMap<String, Vec<ProcessSubscription>>>, // Extend: per-process subs
    connections: Arc<DashMap<ProcessId, IpcConnection>>, // IPC streams
    coordinator: Arc<CollectorCoordinator>,
    router: Arc<TaskRouter>,
    aggregator: Arc<ResultAggregator>,
    flow: Arc<FlowController>,
}

pub enum TransportLayer {
    Broker(IpcListener), // In agent
    Client(IpcStream), // In collectors
}

pub struct EventBusConfig {
    transport: TransportConfig,
    buffer_size: usize,
    heartbeat_interval: Duration,
}

pub enum TransportConfig {
    Ipc { path: PathBuf },
    // Fallback Tcp { addr: SocketAddr } // If needed
}

pub enum PublishMode {
    Direct(ProcessId),
    Broadcast(String),
    Queue(String),
    PubSub(String),
}

impl CrossProcessEventBus {
    pub async fn broker(config: EventBusConfig) -> Result<Self> {
        // Extend Phase 1: spawn IPC listener, accept client connections
        // For each conn: spawn read/write tasks, route msgs to topics
        // Use prost::Message for serialization over IPC
    }
    pub async fn client(config: EventBusConfig) -> Result<Self> {
        // Connect to broker IPC, handle send/recv loops
    }
    pub async fn publish(&self, mode: PublishMode, topic: &str, payload: impl prost::Message) -> Result<()> {
        let env = EventEnvelope { topic: topic.to_string(), payload: payload.encode_to_vec(), ..Default::default() };
        match mode {
            Direct(pid) => { /* send via conn to pid */ },
            Broadcast(t) => { /* broadcast to matching subs via IPC */ },
            Queue(group) => { /* load-balance to one in group */ },
            PubSub(t) => { /* route via topics, forward to clients */ },
        }
        // Apply flow control: check buffers, apply strategy
    }
    pub async fn subscribe(&self, pattern: &str, sub_type: SubscriptionType) -> Result<Receiver<EventEnvelope>> {
        // Client: register pattern with broker over IPC
        // Broker: add to subscribers dashmap
    }
    // Other methods: dispatch_task, aggregate_results as before
}

#[derive(Clone, prost::Message)]
pub struct EventEnvelope {
    topic: String,
    payload: Vec<u8>,
    metadata: HashMap<String, String>,
    source_process: Option<String>,
    correlation: Option<CorrelationMetadata>,
}

pub struct IpcConnection {
    reader: tokio::io::ReadHalf<IpcStream>,
    writer: tokio::io::WriteHalf<IpcStream>,
    sender: mpsc::Sender<EventEnvelope>, // Bounded for backpressure
    buffer_size: usize,
}

// Spawn per-connection tasks: deserialize incoming, route to topics; serialize outgoing

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

References: 

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

#119: Extend Phase 1 wildcard matching for cross-process.

// Existing TopicMatcher from Phase 1
impl TopicMatcher {
    // Existing matches, split_topic
    pub fn route_to_processes(&self, topic: &str, subscribers: &DashMap<String, Vec<ProcessSubscription>>) -> Vec<ProcessId> {
        // Find matching patterns, return unique process_ids
    }
}

pub enum SubscriptionType {
    Direct,
    Broadcast,
    Queue { group: String },
}

pub struct ProcessSubscription {
    process_id: ProcessId,
    pattern: String,
    sub_type: SubscriptionType,
    // From Phase 1: existing fields
}

// Topics: control.collector.task.process, events.process.*, control.health.heartbeat, etc.
// Hierarchical: split by '.', match + (single), # (multi)

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

References: 

- collector-core/src/source.rs
- daemoneye-lib/proto/ipc.proto(MODIFY)

#115: New for cross-process lifecycle.

#[derive(Clone, Serialize, Deserialize, prost::Message)]
pub struct CollectorInfo {
    id: String, // Uuid as str
    caps: SourceCaps,
    load: f32,
    last_heartbeat: u64, // timestamp
    pid: u32,
    status: HealthStatus,
}

pub enum HealthStatus { Healthy, Degraded, Failed }

pub struct CollectorCoordinator {
    collectors: Arc<DashMap<String, CollectorInfo>>,
    heartbeat_tx: broadcast::Sender<Heartbeat>,
}

impl CollectorCoordinator {
    pub fn new() -> Self;
    pub async fn register(&self, caps: SourceCaps, pid: u32) -> String {
        let id = Uuid::new_v4().to_string();
        // Publish to control.collector.lifecycle via bus
        id
    }
    pub fn update_heartbeat(&self, id: &str, load: f32) {
        // Update info, publish Heartbeat to control.health.heartbeat
        // Monitor: tokio task removes stale (>30s)
    }
    pub fn get_capable(&self, required: SourceCaps) -> Vec<String> {
        // Filter active
    }
    pub fn remove(&self, id: &str) {
        // On unregister or timeout
    }
}

// Use prost for Heartbeat msg

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

References: 

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

#116: New for cap-based routing cross-process.

#[derive(Clone)]
pub enum LoadBalancingStrategy { RoundRobin, LeastLoaded, Random }

pub struct TaskRouter {
    coordinator: Arc<CollectorCoordinator>,
    strategy: LoadBalancingStrategy,
    queue_groups: Arc<DashMap<String, QueueGroup>>,
}

pub struct QueueGroup {
    members: Vec<String>, // collector_ids
    next: AtomicUsize,
    loads: DashMap<String, f32>,
}

impl TaskRouter {
    pub fn new(coordinator: Arc<CollectorCoordinator>, strategy: LoadBalancingStrategy) -> Self;
    pub fn route(&self, task: &DetectionTask, required_caps: SourceCaps) -> Vec<TaskAssignment> {
        let candidates = self.coordinator.get_capable(required_caps);
        match self.strategy {
            RoundRobin => {
                // For queue: select next in group
                // Return { collector_id, topic: "control.collector.task.process" }
            },
            LeastLoaded => { /* sort candidates by load */ },
            Random => { /* random select */ },
        }
    }
    pub fn join_queue(&self, group: &str, id: &str) {
        // Add to group members
    }
}

pub struct TaskAssignment {
    collector_id: String,
    topic: String,
    mode: PublishMode,
}

// On dispatch: bus.publish(assignment.mode, assignment.topic, task)

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

References: 

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

#117: New for cross-process result merging.

#[derive(Clone)]
pub enum AggregationStrategy { All, Any, Timeout }

pub struct ResultAggregator {
    pending: Arc<DashMap<String, AggregationState>>,
    bus: Arc<CrossProcessEventBus>,
}

pub struct AggregationState {
    corr_id: String,
    expected: usize,
    received: HashMap<String, DetectionResult>, // by collector_id
    timeout: Instant,
    strategy: AggregationStrategy,
}

impl ResultAggregator {
    pub fn new(bus: Arc<CrossProcessEventBus>) -> Self;
    pub fn submit(&self, corr_id: &str, collector_id: &str, result: DetectionResult) {
        // Update state, check complete
        // If done, aggregate (e.g., union events), notify via bus or return
        // Handle timeout: redistribute via router
    }
    pub async fn aggregate(&self, corr_id: &str, timeout: Duration) -> Result<DetectionResult> {
        // tokio::time::timeout on watcher task
    }
}

// Subscribe to events.process.*, filter by corr_id

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

References: 

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

#115: New for backpressure cross-process.

pub struct FlowController {
    max_buffer: usize,
    strategy: OverflowStrategy,
    metrics: Arc<DashMap<String, FlowMetrics>>,
}

pub enum OverflowStrategy {
    Block,
    DropOldest,
    DropNewest,
    Error,
}

pub struct FlowMetrics {
    queue_depth: usize,
    latency: Duration,
    dropped: usize,
}

impl FlowController {
    pub fn new(max_buffer: usize, strategy: OverflowStrategy) -> Self;
    pub fn check_backpressure(&self, conn: &IpcConnection) -> Result<()> {
        // Check sender len() vs max, apply strategy
        // Block: await on permit if full
        // Drop: remove from channel
    }
    pub fn apply_strategy(&self, sender: &mpsc::Sender<EventEnvelope>, strategy: OverflowStrategy) {
        match strategy {
            Block => { /* use semaphore */ },
            // etc.
        }
    }
}

// Integrate in publish: before send, check/apply

### daemoneye-lib/proto/ipc.proto(MODIFY)

References: 

- daemoneye-lib/build.rs

#120: Extend protos for cross-process.

message CorrelationMetadata {
  string id = 1;
  optional string parent_id = 2;
  string root_id = 3;
  uint64 sequence = 4;
  string stage = 5;
  string source_process = 6;
  map<string, string> tags = 7;
  uint32 expected = 8;
}

message DetectionTask {
  // existing
  optional CorrelationMetadata correlation = 100;
}

message DetectionResult {
  // existing
  optional CorrelationMetadata correlation = 100;
}

message Heartbeat {
  string collector_id = 1;
  uint64 timestamp = 2;
  float load = 3;
  SourceCaps capabilities = 4;
  HealthStatus status = 5;
}

enum HealthStatus {
  HEALTHY = 0;
  DEGRADED = 1;
  FAILED = 2;
}

// Update build.rs to include

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

References: 

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

#115: Integrate broker in agent.

async fn main() -> Result<()> {
    // Config: multi_process: bool, ipc_path: PathBuf, etc.
    if config.multi_process {
        let bus_config = EventBusConfig { transport: TransportConfig::Ipc { path: config.ipc_path.clone() }, buffer_size: 1024, .. };
        let bus = CrossProcessEventBus::broker(bus_config).await?;
        // Spawn listener task if not in broker init
        let coordinator = bus.coordinator();
        let router = bus.router();
        let aggregator = bus.aggregator();
        // Subscribe to events.*.#, control.health.heartbeat
        // In detection: generate corr_id, set task.correlation, assignments = router.route(&task, caps);
        // for ass in assignments { bus.publish(ass.mode, ass.topic, task.clone()).await?; }
        // result = aggregator.aggregate(&corr_id, Duration::from_secs(10)).await?;
        // Handle heartbeats: coordinator.update_heartbeat
    } else {
        // Existing single-process
    }
    // Flow: apply backpressure in loops
}

// Add tracing spans for cross-process ops

### procmond/src/main.rs(MODIFY)

References: 

- procmond/src/config.rs
- collector-core/src/collector.rs
- daemoneye-eventbus/src/bus.rs(NEW)

#115: Integrate client in procmond.

async fn main() -> Result<()> {
    // Config: multi_mode: bool, agent_ipc_path: PathBuf
    if config.multi_mode {
        let bus_config = EventBusConfig { transport: TransportConfig::Ipc { path: config.agent_ipc_path }, .. };
        let bus = CrossProcessEventBus::client(bus_config).await?;
        let id = bus.coordinator().register(SourceCaps::PROCESS, std::process::id() as u32).await;
        // Join queue: bus.router().join_queue("procmond-workers", &id);
        // Spawn heartbeat: tokio::spawn(async move {
        //     loop { bus.publish(PubSub("control.health.heartbeat".to_string()), Heartbeat { id: id.clone(), load: get_load(), .. }).await; }
        // });
        // Subscribe: let mut rx = bus.subscribe("control.collector.task.process", SubscriptionType::Queue { group: "procmond-workers" }).await?;
        // while let Some(env) = rx.recv().await { let task: DetectionTask = prost::Message::decode(&env.payload[..])?; /* execute, set result.correlation = task.correlation, bus.publish(PubSub("events.process.results".to_string()), result).await?; */ }
        // On shutdown: coordinator.remove(&id)
    } else {
        // Existing
    }
}

// Use collector-core for execution

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

References: 

- daemoneye-eventbus/src/bus.rs(NEW)
- daemoneye-agent/src/main.rs(MODIFY)
- procmond/src/main.rs(MODIFY)

#118: E2E multi-process tests.

#[tokio::test]
async fn test_multi_process_coordination() {
    let temp_dir = tempfile::tempdir()?;
    let ipc_path = temp_dir.path().join("eventbus.ipc");
    // Spawn agent process with broker: use tokio::process::Command::new(env::current_exe()).arg("--multi").arg(&ipc_path).spawn()?;
    // Wait for broker ready
    // Spawn procmond-1: Command... arg("--multi-client").arg(&ipc_path).spawn()?;
    // Spawn procmond-2 similarly
    // Dispatch task via mock agent client, verify routing to one (queue), results aggregated
    // Use temp ipc paths, assert corr_id matches
}

#[tokio::test]
async fn test_failover() {
    // Similar setup, dispatch task, kill one procmond (child.kill()), verify redistribution to other
    // Assert aggregation completes with data from survivor
}

#[tokio::test]
async fn test_load_balancing() {
    // Dispatch 4 tasks, assert 2 to each in RR
    // Measure IPC latency: start = Instant::now(); publish; assert elapsed &lt; Duration::from_millis(1);
}

#[tokio::test]
async fn test_backpressure() {
    // Full buffer, publish with Block: await blocks
    // With Drop: count dropped msgs
}

// Mock for faster tests, full process in CI; cross-platform via cfg

### daemoneye-eventbus/benches/latency.rs(NEW)

References: 

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

#118: Performance benchmarks.

use criterion::{criterion_group, criterion_main, Criterion};

fn bench_ipc_publish(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new()?;
    c.bench_function("ipc_publish", |b| {
        b.iter_custom(|iters| {
            rt.block_on(async {
                let bus = CrossProcessEventBus::broker(EventBusConfig::default()).await?;
                let start = Instant::now();
                for _ in 0..iters {
                    bus.publish(PubSub("test".to_string()), "topic", TestMsg {}).await?;
                }
                start.elapsed()
            })
        });
    });
}

// Target: p99 &lt;1ms for 1000 iters
// Separate benches for Win/Unix via features

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

References: 

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

Update docs for epic.

# DaemonEye Cross-Process EventBus

Custom IPC-based pub/sub for multi-collector coordination (Epic #114).

## Architecture
- **Broker**: Hosted in daemoneye-agent process (IPC listener via interprocess)
- **Clients**: Separate OS processes (procmond, netmond etc.) connect via IPC
- **Transport**: Platform-native (Named Pipes Win, Unix Sockets Unix), prost over streams
- **Patterns**: Direct, Broadcast, PubSub (wildcards), Queue (load-balance)
- **Topics**: control.collector.task.*, events.*.*, control.health.*
- **Cross-Platform**: Primary: Win/Linux/macOS; Secondary: FreeBSD

## Setup
- Agent: `CrossProcessEventBus::broker(config.with_ipc_path("\\.\pipe\eventbus"))`
- Collector: `CrossProcessEventBus::client(config.with_ipc_path(path))`

## Features
- Capability routing (#116)
- Load balancing/failover (#117)
- Result aggregation (#117)
- Backpressure (#115)
- Correlation (#120)

## Performance
&lt;1ms p99 latency, thousands eps.

See examples/cross_process.rs, tests/integration.rs.

No unsafe code, minimal deps.

---

Developer Humor

> Why did the event bus go to therapy?\n> It had too many processes crossing paths without communicating! 🚌💭\n> Now with IPC, everyone's on the same page—low latency and no drama. 🚀

---

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.