Implement MQTT-style topic hierarchy for daemoneye-eventbus multi-collector coordination

[unclesp1d3r]

Implement MQTT-style topic hierarchy for daemoneye-eventbus multi-collector coordination

Parent Epic: #114 - Multi-Process Collector Coordination via Event Bus Architecture
Related Issues: #113 (Topic Infrastructure - Completed), #121 (Topic-Based Messaging - Completed)
Specification: .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6.7

Overview

This issue implements the complete topic hierarchy for daemoneye-eventbus, a cross-process IPC pub/sub event bus that enables multi-process collector coordination in DaemonEye. The eventbus provides MQTT-style hierarchical topics with wildcard matching for efficient event routing between multiple separate collector processes (procmond, netmond, fsmond, perfmond) and the daemoneye-agent orchestrator process.

Critical Architectural Context

daemoneye-eventbus is a CROSS-PROCESS IPC system:

• daemoneye-agent: Separate OS process hosting the broker and acting as a client
• Collector Processes: Each collector (procmond, netmond, etc.) runs as a separate OS process
• IPC Transport: Communication via interprocess crate (named pipes on Windows, Unix domain sockets on Unix)
• Topic-Based Messaging: Pub/sub topic routing layered on top of cross-process IPC

All communication in this system happens across process boundaries via local IPC, NOT within a single process.

Problem Statement

DaemonEye's collector-core framework enables multiple collection components to operate independently, but there's currently no mechanism for cross-process coordination:

1. Event Distribution: Collector processes cannot publish events to interested subscriber processes without direct IPC connections
2. Control Plane: No centralized way to manage collector process lifecycle, configuration, and task distribution
3. Health Monitoring: No unified heartbeat and status reporting across distributed collector processes
4. Event Filtering: Cannot selectively subscribe to specific event types without receiving all events across process boundaries
5. Multi-Collector Coordination: No way to coordinate between multiple collector process instances for load balancing and failover

Proposed Solution

Implement a lightweight, cross-process IPC pub/sub event bus with MQTT-inspired topic hierarchy that provides:

Core Topic Namespaces

1. Process Event Topics (events.process.*)

Published by procmond processes → Consumed by daemoneye-agent process via IPC:

• events.process.lifecycle - Process start/stop/exec events
• events.process.metadata - Process metadata updates (CPU, memory, etc.)
• events.process.tree - Process parent-child relationship events
• events.process.integrity - Executable hash verification events
• events.process.anomaly - Behavioral anomaly detection events
• events.process.batch - Bulk enumeration results

2. Collector Control Topics (control.collector.*)

Published by daemoneye-agent process → Consumed by collector processes via IPC:

• control.collector.lifecycle - Start/stop/restart commands sent to collector processes
• control.collector.config - Configuration updates and hot-reload for collector processes
• control.collector.task.process - Detection tasks for procmond processes
• control.collector.task.network - Detection tasks for netmond processes
• control.collector.task.filesystem - Detection tasks for fsmond processes
• control.collector.task.performance - Detection tasks for perfmond processes

3. Health Monitoring Topics (control.health.*)

Bidirectional between all processes via IPC:

• control.health.heartbeat - Periodic keepalive signals from collector processes
• control.health.status - Component status updates from any process
• control.health.diagnostics - Diagnostic information requests to specific processes

4. Future Network/FS/Performance Topics (Reserved)

For future separate collector processes:

• events.network.* - Network monitoring events from future netmond processes
• events.filesystem.* - Filesystem monitoring events from future fsmond processes
• events.performance.* - Performance monitoring events from future perfmond processes

Wildcard Topic Matching (Cross-Process)

Support MQTT-style wildcards for flexible cross-process subscriptions:

• Single-level (+): Matches exactly one level

  • events.process.+ matches events.process.lifecycle but NOT events.process.tree.update
  • Agent process can subscribe to events.+.lifecycle to receive lifecycle events from all collector types

• Multi-level (#): Matches zero or more levels (must be last segment)

  • events.# matches all event topics from all collector processes
  • events.process.# matches all events from procmond processes
  • control.collector.# matches all control messages to collector processes

Topic-Based Access Control (Cross-Process Security)

Implement security boundaries between processes:

• Read-only topics: events.* (collector processes publish, agent process subscribes)
• Control topics: control.* (agent process publishes, collector processes subscribe)
• Admin topics: Restricted to daemoneye-agent process only
• No cross-namespace permissions by default to prevent unauthorized cross-process access

Technical Implementation

Data Structures

use std::collections::HashMap;
use tokio::sync::mpsc;

/// Cross-process event bus broker (hosted in daemoneye-agent process)
pub struct EventBusBroker {
    /// Map of topics to subscriber processes (cross-process connections)
    subscriptions: HashMap<TopicPattern, Vec<ProcessSubscription>>,
    /// Access control for cross-process security
    topic_acl: HashMap<TopicPattern, AccessControl>,
    /// Active IPC connections to collector processes
    process_connections: HashMap<ProcessId, IpcConnection>,
}

/// Represents a subscription from a collector process
pub struct ProcessSubscription {
    /// ID of the subscriber process
    process_id: ProcessId,
    /// IPC channel to send events to this process
    ipc_sender: mpsc::Sender<EventEnvelope>,
}

pub struct TopicPattern {
    segments: Vec<TopicSegment>,
}

pub enum TopicSegment {
    Literal(String),
    SingleWildcard,  // +
    MultiWildcard,   // #
}

/// Event envelope sent across process boundaries via IPC
pub struct EventEnvelope {
    topic: String,
    payload: Vec<u8>,  // Serialized with prost
    timestamp: SystemTime,
    publisher_process_id: ProcessId,
    correlation_metadata: Option<CorrelationMetadata>,
}

pub enum AccessControl {
    ReadOnly,   // Can subscribe (receive via IPC)
    WriteOnly,  // Can publish (send via IPC)
    ReadWrite,  // Both
    Restricted, // daemoneye-agent only
}

Cross-Process Message Flow

1. Collector Process (procmond) → IPC → Broker (agent):
   PublishEvent { topic: "events.process.spawned", payload: ProcessEvent {...} }

2. Broker (agent) → Internal Routing:
   - Match topic against subscriptions
   - Find subscriber processes

3. Broker (agent) → IPC → Subscriber Processes:
   EventEnvelope { topic, payload, ... } sent to each matching process

4. Subscriber Process → Internal Handling:
   - Deserialize payload
   - Process event locally

Implementation Steps

1. Topic Matching Engine (within broker process)

   impl TopicPattern {
       /// Match cross-process subscription patterns against published topics
       pub fn matches(&self, topic: &str) -> bool {
           // Implement MQTT-style wildcard matching
       }
   }

2. Cross-Process Subscription Management

   impl EventBusBroker {
       /// Register subscription from a collector process via IPC
       pub async fn subscribe_from_process(
           &mut self,
           process_id: ProcessId,
           pattern: TopicPattern,
           ipc_channel: mpsc::Sender<EventEnvelope>
       ) -> Result<()> {
           // Validate ACL for this process
           // Register subscription
           // Return acknowledgment via IPC
       }
       
       /// Publish event from a collector process via IPC
       pub async fn publish_from_process(
           &self,
           process_id: ProcessId,
           topic: &str,
           payload: Vec<u8>
       ) -> Result<()> {
           // Validate ACL for this process
           // Route to matching subscriber processes via IPC
       }
   }

3. IPC Integration

   /// Client library for collector processes
   pub struct EventBusClient {
       ipc_connection: IpcConnection,  // interprocess crate connection
       process_id: ProcessId,
   }
   
   impl EventBusClient {
       /// Connect to broker in daemoneye-agent process
       pub async fn connect(broker_address: &str) -> Result<Self> {
           // Establish IPC connection via interprocess crate
       }
       
       /// Subscribe to topics from this collector process
       pub async fn subscribe(&self, pattern: &str) -> Result<EventStream> {
           // Send subscription request via IPC
           // Return stream of events from broker
       }
       
       /// Publish event from this collector process
       pub async fn publish(&self, topic: &str, event: impl Serialize) -> Result<()> {
           // Serialize event
           // Send via IPC to broker
       }
   }

Acceptance Criteria

• Complete topic hierarchy implemented in daemoneye-eventbus crate
• MQTT-style wildcard matching (+, #) working across process boundaries
• Topic-based ACL enforced for cross-process security
• IPC integration with interprocess crate for cross-process communication
• Event serialization/deserialization with prost for cross-process messages
• Cross-process subscription management (register, unregister, enumerate)
• Cross-process event publishing with routing to subscriber processes
• Unit tests for topic matching engine
• Integration tests with multiple collector processes
• Performance benchmarks for cross-process IPC throughput
• Documentation of topic hierarchy and cross-process usage patterns
• Cross-platform validation (Windows named pipes, Unix domain sockets)

Testing Strategy

Unit Tests

• Topic pattern parsing and validation
• Wildcard matching correctness
• ACL enforcement logic

Integration Tests

• Multi-process coordination tests
• Spawn daemoneye-agent broker process
• Spawn multiple collector processes (mock procmond, netmond)
• Verify cross-process pub/sub functionality
• Test process failure and reconnection scenarios

Performance Tests

• Cross-process IPC latency measurement
• Throughput with multiple publisher/subscriber processes
• Memory usage under sustained cross-process load
• Wildcard matching performance at scale

Dependencies

• interprocess (v2.2.3) - Critical for cross-process IPC transport
• prost (v0.14.1) - Cross-process message serialization
• tokio (v1.47.1) - Async runtime for IPC
• dashmap (v6+) - Concurrent topic routing within broker

Technical Considerations

Cross-Process IPC Performance

• Target: <1ms p99 latency for local IPC between processes
• Zero-copy where possible (shared memory future optimization)
• Efficient serialization with prost

Process Isolation & Security

• ACL enforcement prevents unauthorized cross-process access
• Each collector process has limited topic permissions
• Broker validates all cross-process operations

Failure Handling

• Detect when collector processes disconnect (broken pipes, socket errors)
• Clean up subscriptions for terminated processes
• Retry logic for transient IPC errors

Cross-Platform Support

• Windows: Named pipes via CreateNamedPipe/CreateFile
• Unix (Linux/macOS/FreeBSD): Unix domain sockets via socket()
• Validate on all primary platforms (Windows, Linux, macOS) and secondary (FreeBSD)

---

Status: Not Started
Priority: High - Required for multi-collector coordination
Estimated Effort: 1 week
Depends On: #113 (Topic Infrastructure - Completed), #121 (Topic-Based Messaging - Completed)

Next Steps:

1. Design protobuf messages for cross-process EventEnvelope
2. Implement cross-process subscription management in broker
3. Create EventBusClient library for collector processes
4. Add IPC integration tests with multiple processes

Performance Requirements (from Parent #114)

• Topic Routing Latency: <1ms p99 for matching and routing via local IPC
• Wildcard Matching: <100μs for typical patterns (e.g., events.process.+)
• Subscription Management: <100μs for register/unregister local process
• ACL Enforcement: <10μs overhead per publish/subscribe operation
• Zero Unsafe Code: All topic matching and routing logic uses safe Rust only

Code Safety

• Zero unsafe code in topic hierarchy implementation
• Built entirely on: interprocess (IPC), tokio (async), prost (serialization), dashmap (routing)
• All dependencies are well-audited and widely used

[traycerai]

Plan

Observations

The codebase features robust IPC infrastructure in daemoneye-lib/src/ipc/ using interprocess for cross-process communication, protobuf codec, and connection management. Collector-core handles events via event.rs and IPC server in ipc.rs. Daemoneye-agent uses ipc_client.rs for polling. No daemoneye-eventbus crate exists, and related issues #113/#121 show no implementation despite "completed" status. Protobuf schema in proto/ lacks event bus messages. Architecture emphasizes separate OS processes for agent and collectors with Unix sockets/named pipes.

Approach

Develop daemoneye-eventbus as a new workspace crate providing cross-process pub/sub with MQTT-style topics. Host EventBusBroker in daemoneye-agent on dedicated IPC endpoint. Collectors use EventBusClient for publish/subscribe. Extend protobuf with EventEnvelope and control messages. Implement wildcard matching engine and topic ACL for security. Ensure <1ms routing latency via efficient DashMap routing and prost serialization. Integrate optionally in collector-core and agent for gradual adoption.

Reasoning

I examined the workspace Cargo.toml for members and dependencies, explored daemoneye-lib/src/ipc for interprocess transport and codec patterns, reviewed proto files for existing messages, analyzed collector-core/src for event and IPC handling, inspected daemoneye-agent/src/main.rs for startup flow, and confirmed no eventbus code exists across the repository.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

Add daemoneye-eventbus to the workspace members array after daemoneye-lib. Add dashmap =

📁 daemoneye-eventbus ^(NEW) 🔗

Create new directory for the daemoneye-eventbus workspace crate containing broker, client, topic engine, and ACL implementations.

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

Initialize Cargo.toml for daemoneye-eventbus crate. Set package name, version from workspace, edition 2021. Add dependencies: tokio (workspace), prost (workspace), interprocess (workspace), dashmap (workspace), anyhow (workspace), thiserror (workspace), tracing (workspace), uuid (workspace), async-trait (workspace), chrono (workspace for timestamps). Enable lib crate-type. Inherit workspace lints.

📁 daemoneye-eventbus/src ^(NEW) 🔗

Create src directory for daemoneye-eventbus source code.

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

Create lib.rs with public mod declarations: pub mod broker; pub mod client; pub mod topic; pub mod acl; pub mod error; pub mod types; pub mod proto; Re-export key types: pub use broker::EventBusBroker; pub use client::EventBusClient; pub use topic::{TopicPattern, TopicSegment}; pub use acl::{AccessControl, TopicAcl}; pub use error::EventBusError; pub use types::*; Add crate-level docs explaining cross-process pub/sub architecture, MQTT topics, and integration with existing IPC.

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

References

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

Implement TopicPattern and TopicSegment for MQTT-style hierarchy. TopicPattern holds Vec. Parse topics like "events.process.lifecycle" into segments, handling / as separator, escaping. TopicSegment: Literal(String), SingleLevelWildcard, MultiLevelWildcard. Implement parse_from_str with validation (no consecutive /, # only at end). Implement matches method: recursive segment matching, + matches one level, # matches remainder. Add tests for exact, +, # patterns, invalid inputs. Ensure <100μs matching time.

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

Define AccessControl enum: ReadOnly, WriteOnly, ReadWrite, Restricted. TopicAcl struct with HashMap<TopicPattern, AccessControl> and default_fallback. Implement new with defaults: events.# -> ReadOnly (collectors publish, agent reads), control.# -> WriteOnly (agent publishes, collectors read), admin.# -> Restricted (agent only). Methods: can_publish(process_type: ProcessType, topic: &str) -> bool, can_subscribe(process_type, pattern: &str) -> bool. ProcessType enum: Agent, Collector. Use topic.matches for permission checks. Tests for namespace permissions.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 daemoneye-lib/src/ipc/codec.rs

Implement EventBusBroker struct: subscriptions: DashMap<TopicPattern, Vec>, acl: TopicAcl, connections: DashMap<ProcessId, Arc<Mutex>>, runtime: tokio::runtime::Handle. ProcessSubscription { process_id: ProcessId, sender: mpsc::UnboundedSender }. new() initializes acl. async start(endpoint: &str) spawns IPC server using interprocess::local_socket::LocalSocketListener, accepts connections, authenticates process_type. subscribe(process_id, pattern, sender) validates acl, inserts to dashmap. publish(process_id, envelope) validates acl, iterates subscriptions, matches topic, sends to matching senders. handle_connection loop deserializes requests (SubscribeRequest, PublishRequest), dispatches. shutdown() closes listener, drains connections. Integrate tracing for metrics.

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

References

• 📄 daemoneye-lib/src/ipc/client.rs
• 📄 daemoneye-agent/src/ipc_client.rs

Implement EventBusClient struct: connection: IpcConnection, process_id: ProcessId, process_type: ProcessType, codec: IpcCodec. async connect(endpoint: &str, process_type) establishes LocalSocketStream, sends handshake ProcessIdentity, receives ack. async subscribe(pattern: &str) -> mpsc::Receiver sends SubscribeRequest, creates local channel, spawns receive task deserializing EventEnvelope from stream, forwards to receiver. async publish(topic: &str, payload: Vec) creates EventEnvelope, serializes PublishRequest, sends via codec. async unsubscribe(sub_id). Handle disconnects with retry (exponential backoff). Use patterns from daemoneye-lib/src/ipc/client.rs.

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

References

• 📄 daemoneye-lib/src/ipc/codec.rs

#[derive(thiserror::Error, Debug)] pub enum EventBusError { #[error("IPC connection failed: {0}")] Connection(anyhow::Error), #[error("ACL violation for topic '{topic}' by process {process}")] AclViolation { topic: String, process: ProcessId }, #[error("Invalid topic pattern: {0}")] InvalidTopic(String), #[error("Subscription not found: {0}")] NoSubscription(ProcessId), #[error("Serialization error: {0}")] Serde(prost::EncodeError), #[error("Deserialization error: {0}")] De( prost::DecodeError), #[error("Wildcard mismatch")] WildcardError, #[error("Process authentication failed")] AuthFailed, } Implement From for anyhow::Error, prost errors. Context-aware messages for cross-process debugging.

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

References

• 📄 daemoneye-lib/proto/common.proto

pub type ProcessId = uuid::Uuid; pub type SubscriptionId = u64; pub type EventStream = mpsc::Receiver; #[derive(Debug, Clone, PartialEq)] pub enum ProcessType { Agent, Collector, } #[derive(Debug, Clone)] pub struct ProcessIdentity { pub process_id: ProcessId, pub process_type: ProcessType, pub name: Option, } Add derive for Serialize/Deserialize if needed for future extensions.

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

References

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

syntax = "proto3"; package daemoneye.eventbus; import "common.proto"; message EventEnvelope { string topic = 1; bytes payload = 2; int64 timestamp_ms = 3; string publisher_id = 4; CorrelationMetadata correlation = 5; } message CorrelationMetadata { string correlation_id = 1; optional string parent_id = 2; string root_id = 3; uint64 sequence = 4; optional string stage = 5; string source = 6; map<string, string> tags = 7; } message SubscribeRequest { string id = 1; string pattern = 2; ProcessIdentity identity = 3; } message SubscribeResponse { string id = 1; bool success = 2; optional string error = 3; } message PublishRequest { EventEnvelope envelope = 1; } message PublishResponse { bool success = 1; optional string error = 2; } message UnsubscribeRequest { string id = 1; } message HandshakeRequest { ProcessIdentity identity = 1; } message HandshakeResponse { bool accepted = 1; optional string error = 2; } message ProcessIdentity { string id = 1; ProcessType type = 2; optional string name = 3; } enum ProcessType { AGENT = 0; COLLECTOR = 1; } Follow existing proto style from ipc.proto.

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

In prost_build::Config, add .file("proto/eventbus.proto") to compile_protos. Ensure generated code includes EventEnvelope, etc. Output to OUT_DIR as before. Add include!("$OUT_DIR/eventbus.rs"); in proto.rs if needed.

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

Add pub use statements for eventbus types: pub use prost::eventbus::{EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, HandshakeRequest, HandshakeResponse, ProcessIdentity, ProcessType}; Add impl helpers, e.g., impl EventEnvelope { pub fn new(topic: &str, payload: Vec) -> Self { ... } } Similar to existing ipc types.

📁 daemoneye-eventbus/tests ^(NEW) 🔗

Create tests directory for unit and integration tests.

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

#[cfg(test)] mod topic_tests { use super::*; #[test] fn exact_match() { assert!(TopicPattern::parse("events.process.lifecycle").unwrap().matches("events.process.lifecycle")); } #[test] fn single_wildcard() { let pat = TopicPattern::parse("events.process.+").unwrap(); assert!(pat.matches("events.process.lifecycle")); assert!(!pat.matches("events.process.tree.update")); } #[test] fn multi_wildcard() { let pat = TopicPattern::parse("events.#").unwrap(); assert!(pat.matches("events.process.anomaly")); } #[test] fn invalid_pattern() { assert!(TopicPattern::parse("events.#.invalid").is_err()); } // Add 20+ tests for edge cases, performance. }

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

#[cfg(test)] mod acl_tests { use super::*; let acl = TopicAcl::default(); #[test] fn collector_publish_events() { assert!(acl.can_publish(ProcessType::Collector, "events.process.lifecycle")); assert!(!acl.can_publish(ProcessType::Collector, "control.collector.task")); } #[test] fn agent_subscribe_events() { assert!(acl.can_subscribe(ProcessType::Agent, "events.#")); } // Tests for restricted, fallbacks. }

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

References

• 📄 collector-core/tests/integration_test.rs

#[tokio::test] async fn multi_process_pubsub() { use tempfile::tempdir; let dir = tempdir().unwrap(); let endpoint = format!("{}eventbus.sock", dir.path().display()); // Spawn broker task let (broker_tx, mut broker_rx) = mpsc::channel(1); tokio::spawn(async move { let mut broker = EventBusBroker::new(); broker.start(&endpoint).await.unwrap(); }); // Spawn mock collector tokio::spawn(async move { let client = EventBusClient::connect(&endpoint, ProcessType::Collector).await.unwrap(); let mut stream = client.subscribe("control.#").await.unwrap(); client.publish("events.process.spawn", vec![]).await.unwrap(); // Wait and assert receipt }); // Assert events routed. Use tokio::time::sleep. } // Add tests for disconnect, reconnect, multiple clients.

📁 daemoneye-eventbus/benches ^(NEW) 🔗

Create benches directory for criterion benchmarks.

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

References

• 📄 daemoneye-lib/benches

use criterion::{criterion_group, criterion_main, Criterion}; fn bench_matching(c: &mut Criterion) { let patterns = vec!["events.process.+".to_string(), "events.#".to_string()]; c.bench_function("wildcard_match", |b| { b.iter(|| { for pat in &patterns { let _ = TopicPattern::parse(pat).unwrap().matches("events.process.lifecycle.anomaly"); } }) }); } criterion_group!(benches, bench_matching); criterion_main!(benches); // Benchmarks for routing, serialization. Target <1ms p99.

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

References

• 📄 daemoneye-lib/src/telemetry.rs

pub struct AgentEventBus { broker: EventBusBroker, handle: JoinHandle<()>, } impl AgentEventBus { pub async fn start(endpoint: &str) -> Result { let mut broker = EventBusBroker::new(); let handle = tokio::spawn(async move { broker.start(endpoint).await }); // Subscribe agent to events.# Ok(Self { broker, handle }) } pub async fn publish(&self, envelope: EventEnvelope) -> Result<()> { self.broker.publish(ProcessId::nil(), &envelope).await } pub async fn subscribe(&mut self, pattern: &str) -> EventStream { // Internal subscription since broker is local } pub async fn shutdown(self) { self.broker.shutdown().await; self.handle.abort(); } } Add tracing integration with daemoneye-lib/src/telemetry.rs.

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

References

• 📄 daemoneye-agent/src/ipc_client.rs

Add mod eventbus; In async fn run(), after let mut ipc_client = IpcClientManager::new(...).await?, add: let endpoint = env::var("DAEMON_EYE_EVENTBUS_ENDPOINT").unwrap_or_else(|_| "/tmp/daemoneye-eventbus.sock".to_string()); let mut eventbus = AgentEventBus::start(&endpoint).await?; // Subscribe to events broker.subscribe("events.#").await; In shutdown: eventbus.shutdown().await; Ensure graceful handling.

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

Add daemoneye-eventbus = { path = "../daemoneye-eventbus" } to dependencies.

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

References

• 📄 collector-core/src/event.rs
• 📄 collector-core/src/ipc.rs

#[cfg(feature = "eventbus")] pub struct CollectorEventBus { client: EventBusClient, } impl CollectorEventBus { pub async fn connect(endpoint: &str) -> Result { let client = EventBusClient::connect(endpoint, ProcessType::Collector).await?; // Subscribe to control.collector.# let mut stream = client.subscribe("control.collector.#").await?; // Spawn handler task to process control messages Ok(Self { client }) } pub async fn publish_event(&self, event: &CollectionEvent) -> Result<()> { let topic = match event { CollectionEvent::ProcessSpawn(_) => "events.process.lifecycle", // etc. }; let payload = bincode::serialize(event)?; // Or prost if needed self.client.publish(topic, payload).await } }

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

#[cfg(feature = "eventbus")] pub mod eventbus; #[cfg(feature = "eventbus")] pub use eventbus::CollectorEventBus; Add to existing mod declarations.

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

Add [features] default = [] eventbus = ["daemoneye-eventbus"] Add daemoneye-eventbus = { path = "../daemoneye-eventbus", optional = true } to dependencies.

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

References

• 📄 collector-core/README.md

daemoneye-eventbus

Cross-process pub/sub event bus for DaemonEye multi-collector coordination.

Architecture

• Broker hosted in daemoneye-agent
• Clients in separate collector processes
• IPC via interprocess (sockets/pipes)
• MQTT-style topics with + and # wildcards

Topics

• events.process.*: Process events from procmond
• control.collector.*: Tasks/config from agent
• control.health.*: Heartbeats/status

Usage

let broker = EventBusBroker::new();
broker.start("/tmp/eventbus.sock").await;

let client = EventBusClient::connect("/tmp/eventbus.sock", ProcessType::Collector).await;
let mut stream = client.subscribe("events.#").await;

See docs for ACL, performance.

📄 docs/src/technical/eventbus.md ^(NEW) 🔗

References

• 📄 docs/src/technical/ipc-implementation.md

Event Bus Architecture

Topic Hierarchy

Events

• events.process.lifecycle: Process start/stop
• events.process.metadata: Resource usage
• etc.

Control

• control.collector.task.process: Detection tasks

Wildcards

• events.process.+: Specific subtopic
• events.#: All events

Security

ACL enforces: collectors publish events., subscribe control.; agent full access.

Cross-Process Flow

See sequence diagram in plan.

Performance

<1ms latency target via DashMap routing.

📄 docs/src/SUMMARY.md ^(MODIFY) 🔗

Under Technical Documentation, add - Event Bus after IPC Implementation.

Diagram

sequenceDiagram
    participant A as daemoneye-agent (Broker)
    participant C1 as procmond (Client)
    participant C2 as netmond (Client)
    
    A->>A: EventBusBroker::start(\"/tmp/eventbus.sock\")
    
    C1->>A: connect() via IPC
    A->>C1: accept, handshake
    C1->>A: subscribe(\"control.collector.task.process\")
    A->>A: validate ACL, register subscription
    
    C2->>A: connect() via IPC
    C2->>A: subscribe(\"events.process.#\")
    A->>A: register subscription
    
    A->>C1: publish(\"control.collector.task.process\", task)
    A->>A: match subscriptions, route to C1
    A-->>C1: EventEnvelope via IPC
    
    C1->>A: publish(\"events.process.lifecycle\", event)
    A->>A: match, route to C2
    A-->>C2: EventEnvelope via IPC
    
    Note over A,C2: Wildcard matching: events.process.# receives lifecycle

Loading

Import In IDE

🤖 Prompt for AI Agents

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

### Observations

The codebase features robust IPC infrastructure in `daemoneye-lib/src/ipc/` using `interprocess` for cross-process communication, protobuf codec, and connection management. Collector-core handles events via `event.rs` and IPC server in `ipc.rs`. Daemoneye-agent uses `ipc_client.rs` for polling. No `daemoneye-eventbus` crate exists, and related issues #113/#121 show no implementation despite "completed" status. Protobuf schema in `proto/` lacks event bus messages. Architecture emphasizes separate OS processes for agent and collectors with Unix sockets/named pipes.

### Approach

Develop **daemoneye-eventbus** as a new workspace crate providing cross-process pub/sub with MQTT-style topics. Host **EventBusBroker** in **daemoneye-agent** on dedicated IPC endpoint. Collectors use **EventBusClient** for publish/subscribe. Extend protobuf with **EventEnvelope** and control messages. Implement wildcard matching engine and topic ACL for security. Ensure <1ms routing latency via efficient DashMap routing and prost serialization. Integrate optionally in collector-core and agent for gradual adoption.

### Reasoning

I examined the workspace Cargo.toml for members and dependencies, explored daemoneye-lib/src/ipc for interprocess transport and codec patterns, reviewed proto files for existing messages, analyzed collector-core/src for event and IPC handling, inspected daemoneye-agent/src/main.rs for startup flow, and confirmed no eventbus code exists across the repository.

## Mermaid Diagram

sequenceDiagram
    participant A as daemoneye-agent (Broker)
    participant C1 as procmond (Client)
    participant C2 as netmond (Client)
    
    A->>A: EventBusBroker::start(\"/tmp/eventbus.sock\")
    
    C1->>A: connect() via IPC
    A->>C1: accept, handshake
    C1->>A: subscribe(\"control.collector.task.process\")
    A->>A: validate ACL, register subscription
    
    C2->>A: connect() via IPC
    C2->>A: subscribe(\"events.process.#\")
    A->>A: register subscription
    
    A->>C1: publish(\"control.collector.task.process\", task)
    A->>A: match subscriptions, route to C1
    A-->>C1: EventEnvelope via IPC
    
    C1->>A: publish(\"events.process.lifecycle\", event)
    A->>A: match, route to C2
    A-->>C2: EventEnvelope via IPC
    
    Note over A,C2: Wildcard matching: events.process.# receives lifecycle

## Proposed File Changes

### Cargo.toml(MODIFY)

Add **daemoneye-eventbus** to the workspace members array after **daemoneye-lib**. Add **dashmap** =

### daemoneye-eventbus(NEW)

Create new directory for the daemoneye-eventbus workspace crate containing broker, client, topic engine, and ACL implementations.

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

Initialize Cargo.toml for daemoneye-eventbus crate. Set package name, version from workspace, edition 2021. Add dependencies: tokio (workspace), prost (workspace), interprocess (workspace), dashmap (workspace), anyhow (workspace), thiserror (workspace), tracing (workspace), uuid (workspace), async-trait (workspace), chrono (workspace for timestamps). Enable lib crate-type. Inherit workspace lints.

### daemoneye-eventbus/src(NEW)

Create src directory for daemoneye-eventbus source code.

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

Create lib.rs with public mod declarations: pub mod broker; pub mod client; pub mod topic; pub mod acl; pub mod error; pub mod types; pub mod proto; Re-export key types: pub use broker::EventBusBroker; pub use client::EventBusClient; pub use topic::{TopicPattern, TopicSegment}; pub use acl::{AccessControl, TopicAcl}; pub use error::EventBusError; pub use types::*; Add crate-level docs explaining cross-process pub/sub architecture, MQTT topics, and integration with existing IPC.

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

References: 

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

Implement TopicPattern and TopicSegment for MQTT-style hierarchy. TopicPattern holds Vec<TopicSegment>. Parse topics like "events.process.lifecycle" into segments, handling / as separator, escaping. TopicSegment: Literal(String), SingleLevelWildcard, MultiLevelWildcard. Implement parse_from_str with validation (no consecutive /, # only at end). Implement matches method: recursive segment matching, + matches one level, # matches remainder. Add tests for exact, +, # patterns, invalid inputs. Ensure <100μs matching time.

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

Define AccessControl enum: ReadOnly, WriteOnly, ReadWrite, Restricted. TopicAcl struct with HashMap<TopicPattern, AccessControl> and default_fallback. Implement new with defaults: events.# -> ReadOnly (collectors publish, agent reads), control.# -> WriteOnly (agent publishes, collectors read), admin.# -> Restricted (agent only). Methods: can_publish(process_type: ProcessType, topic: &str) -> bool, can_subscribe(process_type, pattern: &str) -> bool. ProcessType enum: Agent, Collector. Use topic.matches for permission checks. Tests for namespace permissions.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- daemoneye-lib/src/ipc/codec.rs

Implement EventBusBroker struct: subscriptions: DashMap<TopicPattern, Vec<ProcessSubscription>>, acl: TopicAcl, connections: DashMap<ProcessId, Arc<Mutex<IpcConnection>>>, runtime: tokio::runtime::Handle. ProcessSubscription { process_id: ProcessId, sender: mpsc::UnboundedSender<EventEnvelope> }. new() initializes acl. async start(endpoint: &str) spawns IPC server using interprocess::local_socket::LocalSocketListener, accepts connections, authenticates process_type. subscribe(process_id, pattern, sender) validates acl, inserts to dashmap. publish(process_id, envelope) validates acl, iterates subscriptions, matches topic, sends to matching senders. handle_connection loop deserializes requests (SubscribeRequest, PublishRequest), dispatches. shutdown() closes listener, drains connections. Integrate tracing for metrics.

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

References: 

- daemoneye-lib/src/ipc/client.rs
- daemoneye-agent/src/ipc_client.rs

Implement EventBusClient struct: connection: IpcConnection, process_id: ProcessId, process_type: ProcessType, codec: IpcCodec. async connect(endpoint: &str, process_type) establishes LocalSocketStream, sends handshake ProcessIdentity, receives ack. async subscribe(pattern: &str) -> mpsc::Receiver<EventEnvelope> sends SubscribeRequest, creates local channel, spawns receive task deserializing EventEnvelope from stream, forwards to receiver. async publish(topic: &str, payload: Vec<u8>) creates EventEnvelope, serializes PublishRequest, sends via codec. async unsubscribe(sub_id). Handle disconnects with retry (exponential backoff). Use patterns from daemoneye-lib/src/ipc/client.rs.

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

References: 

- daemoneye-lib/src/ipc/codec.rs

#[derive(thiserror::Error, Debug)] pub enum EventBusError { #[error("IPC connection failed: {0}")] Connection(anyhow::Error), #[error("ACL violation for topic '{topic}' by process {process}")] AclViolation { topic: String, process: ProcessId }, #[error("Invalid topic pattern: {0}")] InvalidTopic(String), #[error("Subscription not found: {0}")] NoSubscription(ProcessId), #[error("Serialization error: {0}")] Serde(prost::EncodeError), #[error("Deserialization error: {0}")] De( prost::DecodeError), #[error("Wildcard mismatch")] WildcardError, #[error("Process authentication failed")] AuthFailed, } Implement From for anyhow::Error, prost errors. Context-aware messages for cross-process debugging.

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

References: 

- daemoneye-lib/proto/common.proto

pub type ProcessId = uuid::Uuid; pub type SubscriptionId = u64; pub type EventStream = mpsc::Receiver<EventEnvelope>; #[derive(Debug, Clone, PartialEq)] pub enum ProcessType { Agent, Collector, } #[derive(Debug, Clone)] pub struct ProcessIdentity { pub process_id: ProcessId, pub process_type: ProcessType, pub name: Option<String>, } Add derive for Serialize/Deserialize if needed for future extensions.

### daemoneye-lib/proto/eventbus.proto(NEW)

References: 

- daemoneye-lib/proto/ipc.proto
- daemoneye-lib/proto/common.proto

syntax = "proto3"; package daemoneye.eventbus; import "common.proto"; message EventEnvelope { string topic = 1; bytes payload = 2; int64 timestamp_ms = 3; string publisher_id = 4; CorrelationMetadata correlation = 5; } message CorrelationMetadata { string correlation_id = 1; optional string parent_id = 2; string root_id = 3; uint64 sequence = 4; optional string stage = 5; string source = 6; map<string, string> tags = 7; } message SubscribeRequest { string id = 1; string pattern = 2; ProcessIdentity identity = 3; } message SubscribeResponse { string id = 1; bool success = 2; optional string error = 3; } message PublishRequest { EventEnvelope envelope = 1; } message PublishResponse { bool success = 1; optional string error = 2; } message UnsubscribeRequest { string id = 1; } message HandshakeRequest { ProcessIdentity identity = 1; } message HandshakeResponse { bool accepted = 1; optional string error = 2; } message ProcessIdentity { string id = 1; ProcessType type = 2; optional string name = 3; } enum ProcessType { AGENT = 0; COLLECTOR = 1; } Follow existing proto style from ipc.proto.

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

In prost_build::Config, add .file("proto/eventbus.proto") to compile_protos. Ensure generated code includes EventEnvelope, etc. Output to OUT_DIR as before. Add include!("$OUT_DIR/eventbus.rs"); in proto.rs if needed.

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

Add pub use statements for eventbus types: pub use prost::eventbus::{EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, HandshakeRequest, HandshakeResponse, ProcessIdentity, ProcessType}; Add impl helpers, e.g., impl EventEnvelope { pub fn new(topic: &str, payload: Vec<u8>) -> Self { ... } } Similar to existing ipc types.

### daemoneye-eventbus/tests(NEW)

Create tests directory for unit and integration tests.

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

#[cfg(test)] mod topic_tests { use super::*; #[test] fn exact_match() { assert!(TopicPattern::parse("events.process.lifecycle").unwrap().matches("events.process.lifecycle")); } #[test] fn single_wildcard() { let pat = TopicPattern::parse("events.process.+").unwrap(); assert!(pat.matches("events.process.lifecycle")); assert!(!pat.matches("events.process.tree.update")); } #[test] fn multi_wildcard() { let pat = TopicPattern::parse("events.#").unwrap(); assert!(pat.matches("events.process.anomaly")); } #[test] fn invalid_pattern() { assert!(TopicPattern::parse("events.#.invalid").is_err()); } // Add 20+ tests for edge cases, performance. }

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

#[cfg(test)] mod acl_tests { use super::*; let acl = TopicAcl::default(); #[test] fn collector_publish_events() { assert!(acl.can_publish(ProcessType::Collector, "events.process.lifecycle")); assert!(!acl.can_publish(ProcessType::Collector, "control.collector.task")); } #[test] fn agent_subscribe_events() { assert!(acl.can_subscribe(ProcessType::Agent, "events.#")); } // Tests for restricted, fallbacks. }

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

References: 

- collector-core/tests/integration_test.rs

#[tokio::test] async fn multi_process_pubsub() { use tempfile::tempdir; let dir = tempdir().unwrap(); let endpoint = format!("{}eventbus.sock", dir.path().display()); // Spawn broker task let (broker_tx, mut broker_rx) = mpsc::channel(1); tokio::spawn(async move { let mut broker = EventBusBroker::new(); broker.start(&endpoint).await.unwrap(); }); // Spawn mock collector tokio::spawn(async move { let client = EventBusClient::connect(&endpoint, ProcessType::Collector).await.unwrap(); let mut stream = client.subscribe("control.#").await.unwrap(); client.publish("events.process.spawn", vec![]).await.unwrap(); // Wait and assert receipt }); // Assert events routed. Use tokio::time::sleep. } // Add tests for disconnect, reconnect, multiple clients.

### daemoneye-eventbus/benches(NEW)

Create benches directory for criterion benchmarks.

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

References: 

- daemoneye-lib/benches

use criterion::{criterion_group, criterion_main, Criterion}; fn bench_matching(c: &mut Criterion) { let patterns = vec!["events.process.+".to_string(), "events.#".to_string()]; c.bench_function("wildcard_match", |b| { b.iter(|| { for pat in &patterns { let _ = TopicPattern::parse(pat).unwrap().matches("events.process.lifecycle.anomaly"); } }) }); } criterion_group!(benches, bench_matching); criterion_main!(benches); // Benchmarks for routing, serialization. Target <1ms p99.

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

References: 

- daemoneye-lib/src/telemetry.rs

pub struct AgentEventBus { broker: EventBusBroker, handle: JoinHandle<()>, } impl AgentEventBus { pub async fn start(endpoint: &str) -> Result<Self> { let mut broker = EventBusBroker::new(); let handle = tokio::spawn(async move { broker.start(endpoint).await }); // Subscribe agent to events.# Ok(Self { broker, handle }) } pub async fn publish(&self, envelope: EventEnvelope) -> Result<()> { self.broker.publish(ProcessId::nil(), &envelope).await } pub async fn subscribe(&mut self, pattern: &str) -> EventStream { // Internal subscription since broker is local } pub async fn shutdown(self) { self.broker.shutdown().await; self.handle.abort(); } } Add tracing integration with daemoneye-lib/src/telemetry.rs.

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

References: 

- daemoneye-agent/src/ipc_client.rs

Add mod eventbus; In async fn run(), after let mut ipc_client = IpcClientManager::new(...).await?, add: let endpoint = env::var("DAEMON_EYE_EVENTBUS_ENDPOINT").unwrap_or_else(|_| "/tmp/daemoneye-eventbus.sock".to_string()); let mut eventbus = AgentEventBus::start(&endpoint).await?; // Subscribe to events broker.subscribe("events.#").await; In shutdown: eventbus.shutdown().await; Ensure graceful handling.

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

Add daemoneye-eventbus = { path = "../daemoneye-eventbus" } to dependencies.

### collector-core/src/eventbus.rs(NEW)

References: 

- collector-core/src/event.rs
- collector-core/src/ipc.rs

#[cfg(feature = "eventbus")] pub struct CollectorEventBus { client: EventBusClient, } impl CollectorEventBus { pub async fn connect(endpoint: &str) -> Result<Self> { let client = EventBusClient::connect(endpoint, ProcessType::Collector).await?; // Subscribe to control.collector.# let mut stream = client.subscribe("control.collector.#").await?; // Spawn handler task to process control messages Ok(Self { client }) } pub async fn publish_event(&self, event: &CollectionEvent) -> Result<()> { let topic = match event { CollectionEvent::ProcessSpawn(_) => "events.process.lifecycle", // etc. }; let payload = bincode::serialize(event)?; // Or prost if needed self.client.publish(topic, payload).await } }

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

#[cfg(feature = "eventbus")] pub mod eventbus; #[cfg(feature = "eventbus")] pub use eventbus::CollectorEventBus; Add to existing mod declarations.

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

Add [features] default = [] eventbus = ["daemoneye-eventbus"] Add daemoneye-eventbus = { path = "../daemoneye-eventbus", optional = true } to dependencies.

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

References: 

- collector-core/README.md

# daemoneye-eventbus

Cross-process pub/sub event bus for DaemonEye multi-collector coordination.

## Architecture

- Broker hosted in daemoneye-agent
- Clients in separate collector processes
- IPC via interprocess (sockets/pipes)
- MQTT-style topics with + and # wildcards

## Topics

- `events.process.*`: Process events from procmond
- `control.collector.*`: Tasks/config from agent
- `control.health.*`: Heartbeats/status

## Usage

```rust
let broker = EventBusBroker::new();
broker.start("/tmp/eventbus.sock").await;

let client = EventBusClient::connect("/tmp/eventbus.sock", ProcessType::Collector).await;
let mut stream = client.subscribe("events.#").await;
```

See docs for ACL, performance.

### docs/src/technical/eventbus.md(NEW)

References: 

- docs/src/technical/ipc-implementation.md

# Event Bus Architecture

## Topic Hierarchy

### Events
- `events.process.lifecycle`: Process start/stop
- `events.process.metadata`: Resource usage
- etc.

### Control
- `control.collector.task.process`: Detection tasks

## Wildcards

- `events.process.+`: Specific subtopic
- `events.#`: All events

## Security

ACL enforces: collectors publish events.*, subscribe control.*; agent full access.

## Cross-Process Flow

See sequence diagram in plan.

## Performance

<1ms latency target via DashMap routing.

### docs/src/SUMMARY.md(MODIFY)

Under Technical Documentation, add - [Event Bus](technical/eventbus.md) after IPC Implementation.

---

Developer Humor

🐛 Why did the collector subscribe to \events.process.+?\n> Because it wanted to catch every spawn without the drama of full # wildcards!\n> The broker chuckled: \That's +hilarious, but remember ACL— no control topics for you!\n> In the IPC bar, topics flowed freely, but only with proper permissions. 🍻

---

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 #119 details
gh issue view 119 --json title,body,labels,assignees,milestone,state

# Read the task specification file
cat .kiro/specs/daemoneye-core-monitoring/tasks.md

# Get parent issue #114 details
gh issue view 114 --json title,body,labels

# Search for architecture documents related to topics
fd -e md -e txt . .kiro/specs | head -20

# Search for topic hierarchy design in architecture docs
rg -n "topic.*hierarchy" --type md .kiro/

Length of output: 40682

---

🏁 Script executed:

#!/bin/bash
# Read design and requirements documents for context
cat .kiro/specs/daemoneye-core-monitoring/design.md

echo "=== REQUIREMENTS FILE ==="
cat .kiro/specs/daemoneye-core-monitoring/requirements.md

# Search for any existing topic-related code
rg -n "topic|Topic" --type rust -C2 | head -100

# Look for eventbus related code
fd -e rs eventbus

# Check for any MQTT or pub/sub patterns
rg -n "mqtt|pubsub|publish|subscribe" --type rust -C2 | head -50

Length of output: 68463

---

🏁 Script executed:

#!/bin/bash
# Read section 2.6 from tasks.md to get full context
sed -n '/^## 2\.6 /,/^## [0-9]/p' .kiro/specs/daemoneye-core-monitoring/tasks.md | head -200

# Search for eventbus related code
fd -e rs eventbus
rg -n "eventbus|EventBus" --type rust -C3

# Look up requirements 15.1, 15.3, 16.1 in requirements file
rg -n "^### Requirement (15\.1|15\.3|16\.1|15|16)" .kiro/specs/daemoneye-core-monitoring/requirements.md -A20

# Check if issue #113 gives us clues about completed work
gh issue view 113 --json title,body,state

Length of output: 97

---

🏁 Script executed:

#!/bin/bash
# Read the entire tasks.md file to understand structure
echo "=== FULL TASKS.MD ==="
cat .kiro/specs/daemoneye-core-monitoring/tasks.md

# Search for multi-process or coordination mentions
echo -e "\n=== MULTI-PROCESS COORDINATION MENTIONS ==="
rg -n "multi-process|coordination|eventbus" --type md .kiro/

# Search for any section with 2.6 or topic mentions
echo -e "\n=== SECTION 2.6 OR TOPIC MENTIONS ==="
rg -n "2\.6|topic" --type md .kiro/specs/daemoneye-core-monitoring/

# Look at requirements 15 and 16 specifically
echo -e "\n=== REQUIREMENTS 15 AND 16 ==="
rg -n "^### Requirement (15|16)" .kiro/specs/daemoneye-core-monitoring/requirements.md -A30

Length of output: 39070

---

🏁 Script executed:

#!/bin/bash
# Search for eventbus in all spec directories
echo "=== SEARCHING FOR EVENTBUS SPECS ==="
fd -t f . .kiro/specs | xargs grep -l "eventbus\|event.bus\|event-bus" 2>/dev/null || echo "No eventbus specs found"

# Search in steering documents
echo -e "\n=== STEERING DOCUMENTS ==="
fd -e md . .kiro/steering

# Check if there are any pub/sub or message broker related docs
echo -e "\n=== PUB/SUB OR MESSAGING PATTERNS ==="
rg -n "pub.?sub|publish|subscribe|broker|message.bus" --type md .kiro/ -C2 | head -50

# Look for MQTT or topic patterns
echo -e "\n=== MQTT OR TOPIC PATTERNS ==="
rg -n "mqtt|topic|wildcard" --type md .kiro/ -C2 | head -30

# Check the collector-core steering doc for coordination details
echo -e "\n=== COLLECTOR-CORE COORDINATION DETAILS ==="
cat .kiro/steering/collector-core.md | head -150

Length of output: 6536

---

@unclesp1d3r I'll enhance this issue with comprehensive context and a detailed technical proposal.

---

🏁 Script executed:

#!/bin/bash
# Update the issue with enhanced content
gh issue edit 119 --title "Implement MQTT-style topic hierarchy for daemoneye-eventbus multi-collector coordination" --body "## Overview

This issue implements the complete topic hierarchy for **daemoneye-eventbus**, a pub/sub event bus that enables multi-process collector coordination in DaemonEye. The eventbus provides MQTT-style hierarchical topics with wildcard matching for efficient event routing between multiple collector instances (procmond, netmond, fsmond, perfmond) and the daemoneye-agent orchestrator.

## Problem Statement

DaemonEye's collector-core framework enables multiple collection components to operate independently, but there's currently no mechanism for:

1. **Event Distribution**: Collectors cannot publish events to interested subscribers without direct IPC connections
2. **Control Plane**: No centralized way to manage collector lifecycle, configuration, and task distribution
3. **Health Monitoring**: No unified heartbeat and status reporting across distributed collectors  
4. **Event Filtering**: Cannot selectively subscribe to specific event types without receiving all events
5. **Multi-Collector Coordination**: No way to coordinate between multiple instances for load balancing and failover

## Proposed Solution

Implement a **lightweight, in-process pub/sub event bus** with MQTT-inspired topic hierarchy that provides:

### Core Topic Namespaces

#### 1. Process Event Topics (\`events.process.*\`)
- \`events.process.lifecycle\` - Process start/stop/exec events
- \`events.process.metadata\` - Process metadata updates (CPU, memory, etc.)
- \`events.process.tree\` - Process parent-child relationship events
- \`events.process.integrity\` - Executable hash verification events
- \`events.process.anomaly\` - Behavioral anomaly detection events  
- \`events.process.batch\` - Bulk enumeration results

#### 2. Collector Control Topics (\`control.collector.*\`)
- \`control.collector.lifecycle\` - Start/stop/restart commands
- \`control.collector.config\` - Configuration updates and hot-reload
- \`control.collector.task\` - Detection task distribution

#### 3. Health Monitoring Topics (\`control.health.*\`)
- \`control.health.heartbeat\` - Periodic keepalive signals
- \`control.health.status\` - Component status updates
- \`control.health.diagnostics\` - Diagnostic information requests

#### 4. Future Network/FS/Performance Topics (Reserved)
- \`events.network.*\` - Network monitoring events (future netmond)
- \`events.filesystem.*\` - Filesystem monitoring events (future fsmond)
- \`events.performance.*\` - Performance monitoring events (future perfmond)

### Wildcard Topic Matching

Support MQTT-style wildcards for flexible subscriptions:

- **Single-level (\`+\`)**: Matches exactly one level
  - \`events.process.+\` matches \`events.process.lifecycle\` but NOT \`events.process.tree.update\`
  
- **Multi-level (\`#\`)**: Matches zero or more levels (must be last segment)
  - \`events.#\` matches all event topics
  - \`events.process.#\` matches all process event topics
  - \`control.collector.#\` matches all collector control topics

### Topic-Based Access Control

Implement security boundaries:
- Read-only topics: \`events.*\` (collectors publish, agent subscribes)
- Control topics: \`control.*\` (agent publishes, collectors subscribe)
- Admin topics: Restricted to daemoneye-agent only
- No cross-namespace permissions by default

## Technical Implementation

### Data Structures

\`\`\`rust
use std::collections::HashMap;
use tokio::sync::mpsc;

pub struct EventBus {
    subscriptions: HashMap<TopicPattern, Vec<mpsc::Sender<Event>>>,
    topic_acl: HashMap<TopicPattern, AccessControl>,
}

pub struct TopicPattern {
    segments: Vec<TopicSegment>,
}

pub enum TopicSegment {
    Literal(String),
    SingleWildcard,  // +
    MultiWildcard,   // #
}

pub struct Event {
    topic: String,
    payload: Vec<u8>,
    timestamp: SystemTime,
    publisher_id: String,
}

pub enum AccessControl {
    ReadOnly,
    WriteOnly, 
    ReadWrite,
    Admin,
}
\`\`\`

### Core Operations

\`\`\`rust
impl EventBus {
    // Subscribe to topics with wildcard support
    pub fn subscribe(&mut self, pattern: &str) -> mpsc::Receiver<Event>;
    
    // Publish events to matching subscribers
    pub async fn publish(&self, topic: &str, payload: Vec<u8>) -> Result<()>;
    
    // Check topic access permissions
    fn check_acl(&self, topic: &str, operation: Operation) -> Result<()>;
    
    // Match topic against wildcard patterns
    fn matches(&self, pattern: &TopicPattern, topic: &str) -> bool;
}
\`\`\`

## Acceptance Criteria

- [ ] **Topic Hierarchy**: Implement all process event topics (\`events.process.*\` with 6 subtopics)
- [ ] **Control Topics**: Implement collector control topics (\`control.collector.*\` with 3 subtopics)  
- [ ] **Health Topics**: Implement health monitoring topics (\`control.health.*\` with 3 subtopics)
- [ ] **Future Topics**: Reserve namespace for \`events.network.*\`, \`events.filesystem.*\`, \`events.performance.*\`
- [ ] **Single-Level Wildcard**: Support \`+\` wildcard matching exactly one topic level
- [ ] **Multi-Level Wildcard**: Support \`#\` wildcard matching zero or more levels (only as last segment)
- [ ] **Topic ACL**: Implement read-only, write-only, and admin access control per topic namespace
- [ ] **Subscription API**: Provide \`subscribe(pattern)\` method returning event channel
- [ ] **Publishing API**: Provide \`publish(topic, payload)\` method with ACL enforcement
- [ ] **Unit Tests**: Test wildcard matching, ACL enforcement, and topic routing
- [ ] **Integration Tests**: Test multi-subscriber scenarios and cross-namespace isolation
- [ ] **Performance**: Handle 10,000+ events/second with <5% CPU overhead
- [ ] **Documentation**: Document topic hierarchy, wildcard syntax, and ACL model

## Dependencies

- Parent: #114 (Spec 2.6 Add multi-process collector coordination via daemoneye-eventbus)
- Blocked by: None (can start immediately)
- Blocks: #115 (multi-process workflows), #116 (task distribution), #117 (result aggregation)

## Testing Strategy

1. **Unit Tests**: Topic parsing, wildcard matching, ACL validation
2. **Integration Tests**: Multi-publisher/subscriber scenarios, event ordering
3. **Performance Tests**: Throughput benchmarks, memory usage under load
4. **Security Tests**: ACL enforcement, namespace isolation

## References

- MQTT 3.1.1 Topic Specification: https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/
- Collector-Core Framework: \`.kiro/steering/collector-core.md\`
- Parent Spec: Issue #114"

# Add labels
gh issue edit 119 --add-label "enhancement,pub-sub,distributed-systems"

# Assign to unclesp1d3r
gh issue edit 119 --assignee unclesp1d3r

# Show the updated issue
gh issue view 119 --json title,body,labels,assignees | jq -r '"Title: " + .title + "\n\nLabels: " + ([.labels[].name] | join(", ")) + "\n\nAssignees: " + ([.assignees[].login] | join(", "))'

Length of output: 1264

---

🏁 Script executed:

#!/bin/bash
# Fix labels - use existing labels or create valid ones
# First, list available labels to see what exists
echo "=== Available labels ==="
gh label list --limit 50

echo -e "\n=== Attempting to add enhancement label ==="
gh issue edit 119 --add-label "enhancement"

echo -e "\n=== Adding assignee with correct syntax ==="
gh issue edit 119 --add-assignee unclesp1d3r

echo -e "\n=== Final issue status ==="
gh issue view 119 --json title,body,labels,assignees

Length of output: 9897

---

@unclesp1d3r I've successfully enhanced issue #119 with comprehensive context and a detailed technical proposal. Here's what I've added:

Summary of Changes

✅ Updated Title

New: "Implement MQTT-style topic hierarchy for daemoneye-eventbus multi-collector coordination"

This clearly describes both the technical approach (MQTT-style topics) and the purpose (multi-collector coordination).

✅ Enhanced Issue Body

The issue now includes:

1. Problem Statement - Explains the 5 key gaps in current collector coordination:

   • Event distribution limitations
   • Missing control plane
   • No unified health monitoring
   • Lack of event filtering
   • No multi-collector coordination

2. Comprehensive Solution - Details the complete topic hierarchy:

   • Process Events (events.process.*) - 6 subtopics for lifecycle, metadata, tree, integrity, anomaly, and batch
   • Collector Control (control.collector.*) - 3 subtopics for lifecycle, config, and task distribution
   • Health Monitoring (control.health.*) - 3 subtopics for heartbeat, status, and diagnostics
   • Future Topics - Reserved namespaces for network, filesystem, and performance monitoring

3. MQTT-Style Wildcards:

   • + for single-level matching
   • # for multi-level matching (must be last segment)

4. Topic-Based ACL - Security boundaries between namespaces (read-only events, control topics, admin restrictions)

5. Technical Implementation - Rust code examples showing:

   • Data structures (EventBus, TopicPattern, Event, AccessControl)
   • Core operations (subscribe, publish, ACL checking, pattern matching)

6. Acceptance Criteria - 13 clear checkboxes covering topics, wildcards, ACL, APIs, testing, and performance

7. Testing Strategy - Unit, integration, performance, and security tests

✅ Metadata Updates

• Labels Added: enhancement (kept existing security and architecture)
• Assigned: unclesp1d3r
• Dependencies: Linked to parent Epic: Multi-Process Collector Coordination via Event Bus Architecture #114 and dependent issues Implement message broker for multi-collector coordination and load balancing #115, Implement Task Distribution and Capability-Based Routing via EventBus #116, Implement Result Aggregation, Correlation, and Load Balancing for Multi-Collector Coordination #117

The issue is now ready for implementation with clear technical direction and comprehensive acceptance criteria. The pub/sub eventbus will enable distributed collector coordination while maintaining DaemonEye's security-first architecture.

[traycerai]

I have updated the Implementation Plan to reflect the recent changes in the ticket.

New Plan

Observations

The codebase has solid IPC foundations (interprocess crate, protobuf codec, connection pooling) but lacks pub/sub capabilities. Issues #113 and #121 are marked "completed" but no code exists - this issue must implement everything. The architecture is strictly cross-process with daemoneye-agent hosting the broker and collectors as separate OS processes communicating via Unix sockets/named pipes.

Approach

Create a new daemoneye-eventbus workspace crate implementing MQTT-style topic hierarchy with wildcard matching for cross-process pub/sub coordination. The broker will be hosted in daemoneye-agent, while collectors use EventBusClient to publish/subscribe via existing interprocess IPC. Extend protobuf schema with EventEnvelope and CorrelationMetadata messages. Implement topic matching engine with ACL enforcement for security boundaries between processes.

Reasoning

I explored the repository structure, examined existing IPC infrastructure in daemoneye-lib/src/ipc/, reviewed protobuf definitions, analyzed the collector-core framework, and confirmed that no EventBus code exists yet despite dependencies claiming completion. I studied the current request/response IPC architecture and identified integration points for the new pub/sub layer.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

Add daemoneye-eventbus to the workspace members list after daemoneye-lib. Add dashmap dependency (version 6.1 or later) to workspace dependencies for concurrent topic routing within the broker process. This enables efficient thread-safe HashMap operations for subscription management.

📁 daemoneye-eventbus ^(NEW) 🔗

Create new workspace crate directory for the event bus implementation. This will contain the broker, client, topic matching engine, and ACL enforcement logic.

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

Create Cargo.toml for daemoneye-eventbus crate. Include workspace package metadata inheritance. Add dependencies: tokio (workspace, for async runtime), dashmap (workspace, for concurrent subscriptions), prost (workspace, for protobuf), interprocess (workspace, for IPC), anyhow (workspace, for error handling), thiserror (workspace, for custom errors), tracing (workspace, for logging), uuid (workspace, for correlation IDs), async-trait (workspace, for async traits). Set lib.bench = true for benchmarking support. Inherit workspace lints configuration.

📁 daemoneye-eventbus/src ^(NEW) 🔗

Create source directory for daemoneye-eventbus implementation.

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

Create main library file with module declarations and public API exports. Export EventBusBroker, EventBusClient, TopicPattern, TopicSegment, AccessControl, and error types. Add comprehensive module documentation describing the cross-process pub/sub architecture, MQTT-style topic hierarchy, and usage examples for both broker (daemoneye-agent) and client (collector processes) scenarios. Include architecture diagram showing process boundaries and IPC communication flow.

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

References

• 📄 daemoneye-lib/src/detection/sql_to_ipc.rs

Implement topic parsing and matching engine. Define TopicPattern struct with segments field (Vec). Define TopicSegment enum with variants: Literal(String), SingleWildcard (for +), MultiWildcard (for #). Implement TopicPattern::parse() method to parse topic strings like "events.process.lifecycle" into segments, validating format (no empty segments, # only at end). Implement TopicPattern::matches() method for MQTT-style wildcard matching: single-level wildcard (+) matches exactly one segment, multi-level wildcard (#) matches zero or more segments and must be last. Add comprehensive unit tests covering: exact matches, single-level wildcards in various positions, multi-level wildcards, invalid patterns, edge cases like "events.+.#", and performance with deeply nested topics. Follow patterns from existing validation code in daemoneye-lib/src/detection/sql_to_ipc.rs.

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

Implement topic-based access control for cross-process security. Define AccessControl enum with variants: ReadOnly (can subscribe/receive), WriteOnly (can publish/send), ReadWrite (both), Restricted (daemoneye-agent only). Define TopicAcl struct managing HashMap<TopicPattern, AccessControl> with default rules: events.* → ReadOnly for collectors, control.* → WriteOnly for agent, admin.* → Restricted. Implement TopicAcl::check_publish_permission() and TopicAcl::check_subscribe_permission() methods that validate process permissions against topic patterns. Add ProcessIdentity struct with process_id and process_type (Agent, Collector) fields for permission checks. Include unit tests for ACL enforcement across different topic namespaces and process types.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 collector-core/src/collector.rs

Implement EventBusBroker for cross-process pub/sub coordination. Define broker struct with fields: subscriptions (DashMap<TopicPattern, Vec>), topic_acl (TopicAcl), process_connections (DashMap<ProcessId, IpcConnection>), shutdown_signal (Arc). Define ProcessSubscription struct with process_id and ipc_sender (mpsc::Sender) fields. Implement EventBusBroker::new() constructor initializing ACL with default rules. Implement subscribe_from_process() async method: validate ACL permissions, register subscription in DashMap, send acknowledgment via IPC. Implement publish_from_process() async method: validate ACL, match topic against all subscription patterns using TopicPattern::matches(), route EventEnvelope to matching subscriber processes via their IPC channels, handle send failures gracefully. Implement unsubscribe_from_process() for cleanup. Implement start() method to spawn IPC listener accepting connections from collector processes, similar to InterprocessServer in daemoneye-lib/src/ipc/interprocess_transport.rs. Implement shutdown() for graceful cleanup of all connections and subscriptions. Add metrics tracking (messages routed, subscriptions active, ACL violations). Follow async patterns from collector-core/src/collector.rs for lifecycle management.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 daemoneye-agent/src/ipc_client.rs
• 📄 daemoneye-lib/src/ipc/client.rs

Implement EventBusClient for collector processes to interact with broker. Define client struct with fields: ipc_connection (connection to broker via interprocess crate), process_id (ProcessId), codec (IpcCodec for message framing), subscriptions (local tracking of active subscriptions). Implement EventBusClient::connect() async method to establish IPC connection to broker endpoint (similar to InterprocessClient::new() in daemoneye-lib/src/ipc/interprocess_transport.rs), send initial handshake with process identity. Implement subscribe() async method: send SubscribeRequest protobuf message to broker, return EventStream (mpsc::Receiver) for receiving events. Implement publish() async method: serialize event payload with prost, create EventEnvelope with topic and metadata, send PublishRequest to broker via IPC. Implement unsubscribe() for cleanup. Add automatic reconnection logic with exponential backoff similar to IpcClientManager in daemoneye-agent/src/ipc_client.rs. Implement background task to receive events from broker and dispatch to local subscribers. Add connection health monitoring and metrics. Follow error handling patterns from daemoneye-lib/src/ipc/client.rs.

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

References

• 📄 daemoneye-lib/src/ipc/codec.rs

Define error types for event bus operations using thiserror crate. Create EventBusError enum with variants: TopicParseError (invalid topic format), AclViolation (permission denied), ConnectionError (IPC failure), SerializationError (protobuf encoding/decoding), SubscriptionNotFound, BrokerShutdown, InvalidWildcard (# not at end), EmptySegment, and other domain-specific errors. Implement Display and Error traits. Add context-rich error messages for debugging cross-process issues. Follow error handling patterns from daemoneye-lib/src/ipc/codec.rs IpcError enum.

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

Define core data types for event bus. Create ProcessId type alias or newtype for process identification. Define EventStream type alias for mpsc::Receiver. Define SubscriptionId type for tracking subscriptions. Add type aliases for clarity and future extensibility. Keep types simple and focused on cross-process communication semantics.

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

References

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

Define protobuf messages for event bus cross-process communication. Create EventEnvelope message with fields: topic (string), payload (bytes - serialized event data), timestamp (int64 - Unix timestamp in milliseconds), publisher_process_id (string), correlation_metadata (optional CorrelationMetadata). Create CorrelationMetadata message with fields: correlation_id (string - UUID), parent_correlation_id (optional string), root_correlation_id (string), sequence_number (uint64), workflow_stage (optional string), source_component (string), correlation_tags (map<string, string>). Create SubscribeRequest message with fields: subscription_id (string), topic_pattern (string - with wildcards), process_identity (ProcessIdentity). Create SubscribeResponse message with fields: subscription_id (string), success (bool), error_message (optional string). Create PublishRequest message with fields: envelope (EventEnvelope). Create PublishResponse message with fields: success (bool), error_message (optional string). Create UnsubscribeRequest message with fields: subscription_id (string). Create ProcessIdentity message with fields: process_id (string), process_type (ProcessType enum - AGENT, COLLECTOR). Follow protobuf conventions from daemoneye-lib/proto/ipc.proto and daemoneye-lib/proto/common.proto.

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

References

• 📄 daemoneye-lib/proto/ipc.proto

Add proto/eventbus.proto to the list of protobuf files compiled by prost-build. Ensure the new EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, and ProcessIdentity messages are generated. Follow the existing pattern for compiling proto files and generating Rust code in the OUT_DIR.

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

Re-export the new event bus protobuf types from the generated code. Add pub use statements for EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, ProcessIdentity, and ProcessType enum. Add helper methods to EventEnvelope for creating envelopes with correlation metadata, extracting topic, and validating structure. Follow the pattern of helper methods on DetectionTask and DetectionResult in this file.

📁 daemoneye-eventbus/tests ^(NEW) 🔗

Create tests directory for integration and unit tests.

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

References

• 📄 collector-core/tests/integration_test.rs

Create comprehensive unit tests for topic matching engine. Test exact topic matches ("events.process.lifecycle" matches "events.process.lifecycle"). Test single-level wildcard: "events.process.+" matches "events.process.lifecycle" but NOT "events.process.tree.update", "events.+.lifecycle" matches "events.process.lifecycle" and "events.network.lifecycle". Test multi-level wildcard: "events.#" matches all event topics, "events.process.#" matches all process events including nested ones. Test invalid patterns: "events.#.invalid" (# not at end), "events..lifecycle" (empty segment), "" (empty topic). Test edge cases: single segment topics, deeply nested topics (10+ levels), topics with special characters. Test performance with 1000+ topic patterns. Follow test patterns from collector-core/tests/ for structure and assertions.

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

References

• 📄 daemoneye-lib/tests/ipc_security_validation.rs

Create unit tests for ACL enforcement logic. Test that collector processes can publish to "events." topics but NOT "control." topics. Test that daemoneye-agent process can publish to "control." topics but collectors cannot. Test that collectors can subscribe to "control.collector." and "control.health." topics. Test that agent can subscribe to "events." topics. Test that "admin.*" topics are restricted to agent only. Test ACL violations return appropriate errors. Test default ACL rules are correctly initialized. Follow security test patterns from daemoneye-lib/tests/ipc_security_validation.rs.

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

References

• 📄 collector-core/tests/integration_test.rs
• 📄 daemoneye-lib/tests/ipc_integration.rs

Create integration tests for EventBusBroker with multiple collector processes. Test broker startup and IPC listener initialization. Test multiple collector processes connecting to broker. Test subscription registration from different processes. Test event publishing and routing to matching subscribers across process boundaries. Test wildcard subscriptions receiving appropriate events. Test subscription cleanup when collector process disconnects. Test concurrent publish/subscribe operations from multiple processes. Test broker shutdown and cleanup of all resources. Use tempfile crate for test socket paths similar to collector-core/tests/integration_test.rs. Spawn mock collector processes using tokio::spawn. Follow integration test patterns from daemoneye-lib/tests/ipc_integration.rs for multi-process coordination.

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

References

• 📄 daemoneye-lib/tests/resilient_client_integration.rs

Create integration tests for EventBusClient. Test client connection to broker. Test subscribe() method and receiving events via EventStream. Test publish() method sending events to broker. Test multiple clients subscribing to same topic pattern. Test client reconnection after broker restart or connection failure. Test correlation metadata propagation through publish/subscribe flow. Test client cleanup and unsubscribe on shutdown. Follow client testing patterns from daemoneye-lib/tests/resilient_client_integration.rs for connection management and error scenarios.

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

References

• 📄 daemoneye-lib/tests/ipc_cross_platform_tests.rs

Create cross-platform validation tests for Windows (named pipes) and Unix (domain sockets). Test broker and client work correctly on both platforms. Test socket path generation and cleanup. Test platform-specific IPC error handling. Test concurrent connections on both platforms. Follow cross-platform test patterns from daemoneye-lib/tests/ipc_cross_platform_tests.rs with conditional compilation for platform-specific code.

📁 daemoneye-eventbus/benches ^(NEW) 🔗

Create benchmarks directory for performance testing.

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

References

• 📄 daemoneye-lib/benches/ipc_performance_comprehensive.rs

Create performance benchmarks using criterion crate. Benchmark topic matching performance with various pattern complexities (exact, single wildcard, multi-level wildcard) against 1000+ topics. Benchmark broker routing latency for single subscriber, multiple subscribers (10+), and wildcard subscriptions. Benchmark cross-process IPC throughput: messages/second with 1 publisher + 1 subscriber, 1 publisher + 10 subscribers, 10 publishers + 10 subscribers. Benchmark subscription management: add/remove subscriptions, lookup performance with 1000+ active subscriptions. Target: <1ms p99 latency for local IPC routing. Follow benchmark patterns from daemoneye-lib/benches/ipc_performance_comprehensive.rs using criterion framework and black_box for preventing optimization.

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

References

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

Create integration module for hosting EventBusBroker in daemoneye-agent process. Define EventBusService struct wrapping EventBusBroker with lifecycle management. Implement start_broker() function to initialize broker with default ACL rules, start IPC listener on dedicated endpoint (e.g., /var/run/daemoneye/eventbus.sock or \.\pipe\daemoneye\eventbus), and spawn broker runtime task. Implement shutdown_broker() for graceful cleanup. Add broker health monitoring and metrics integration with existing telemetry system in daemoneye-lib/src/telemetry.rs. Provide API for agent components to subscribe to topics (e.g., detection engine subscribing to "events.process.#"). Follow service initialization patterns from daemoneye-agent/src/main.rs main loop structure.

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

References

• 📄 daemoneye-agent/src/ipc_client.rs

Integrate EventBusBroker into daemoneye-agent startup sequence. Add module declaration for eventbus_integration. In the run() function after IPC client initialization (around line 64), add broker initialization: create EventBusService, start broker with start_broker(), store broker handle for shutdown. In the shutdown section (after line 203), add broker.shutdown() call before final cleanup. Add broker health monitoring to the main loop alongside existing telemetry health checks. Consider subscribing agent to "events.process.#" topic to receive process events from collectors as an alternative/complement to the current IPC polling approach. Maintain backward compatibility with existing IPC client for gradual migration.

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

References

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

Add daemoneye-eventbus as a dependency to enable broker integration. Add it to the dependencies section with path = "../daemoneye-eventbus" similar to how collector-core and daemoneye-lib are included.

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

References

• 📄 collector-core/src/ipc.rs
• 📄 collector-core/src/event.rs

Create optional integration module for collector-core to use EventBusClient. Define CollectorEventBusAdapter struct that wraps EventBusClient and provides methods for collectors to publish events to the broker. Implement publish_process_event() method to publish ProcessEvent to "events.process.lifecycle" or other appropriate topics based on event type. Implement subscribe_to_tasks() method to subscribe to "control.collector.task.*" topics for receiving tasks from agent. Add helper methods to convert between CollectionEvent (from collector-core/src/event.rs) and EventEnvelope payloads. Make this integration optional/feature-gated to maintain backward compatibility with direct IPC. Follow integration patterns from collector-core/src/ipc.rs for IPC server integration.

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

Add optional module declaration for eventbus_client_integration (conditionally compiled with feature flag). Re-export CollectorEventBusAdapter if the eventbus feature is enabled. This allows collectors to optionally use the event bus without breaking existing direct IPC functionality.

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

Add optional daemoneye-eventbus dependency with path = "../daemoneye-eventbus" and optional = true. Add new feature flag eventbus that enables the daemoneye-eventbus dependency. This allows collectors to opt-in to event bus functionality while maintaining backward compatibility with direct IPC.

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

References

• 📄 collector-core/README.md

Create comprehensive documentation for daemoneye-eventbus crate. Document the cross-process pub/sub architecture with process diagrams. Explain MQTT-style topic hierarchy with examples: events.process.lifecycle, events.process.metadata, control.collector.task.process, control.health.heartbeat. Document wildcard matching rules with examples of + and # usage. Explain topic-based ACL and security boundaries between processes. Provide usage examples for broker (in daemoneye-agent) and client (in collectors). Include code examples for subscribing, publishing, and handling events. Document performance characteristics and benchmarks. Explain integration with existing IPC infrastructure. Add troubleshooting section for common cross-process issues. Follow documentation style from collector-core/README.md.

📄 docs/src/technical/eventbus-architecture.md ^(NEW) 🔗

References

• 📄 docs/src/technical/ipc-implementation.md

Create technical documentation for event bus architecture in the mdBook documentation. Document the complete topic hierarchy with all namespaces: events.process.* (lifecycle, metadata, tree, integrity, anomaly, batch), events.network., events.filesystem., events.performance., control.collector. (lifecycle, config, task.), control.health. (heartbeat, status, diagnostics). Explain cross-process message flow with sequence diagrams. Document ACL rules and security model. Provide deployment guidance for multi-collector coordination. Include performance tuning recommendations. Add examples of common pub/sub patterns for multi-collector scenarios. Follow documentation structure from docs/src/technical/ipc-implementation.md.

📄 docs/src/SUMMARY.md ^(MODIFY) 🔗

Add link to the new eventbus architecture documentation under the Technical section. Add entry: "- Event Bus Architecture" after the IPC Implementation entry to maintain logical flow of technical documentation.

Diagram

sequenceDiagram
    participant Agent as daemoneye-agent<br/>(Broker Process)
    participant Broker as EventBusBroker<br/>(in Agent)
    participant Client1 as procmond-1<br/>(Collector Process)
    participant Client2 as procmond-2<br/>(Collector Process)
    
    Note over Agent,Client2: Startup & Connection Phase
    Agent->>Broker: Initialize broker with ACL rules
    Broker->>Broker: Start IPC listener on eventbus.sock
    
    Client1->>Broker: Connect via IPC (Unix socket/named pipe)
    Broker->>Client1: Connection accepted
    Client1->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Validate ACL (collectors can subscribe to control.*)
    Broker->>Broker: Register subscription in DashMap
    Broker->>Client1: SubscribeResponse(success=true)
    
    Client2->>Broker: Connect via IPC
    Broker->>Client2: Connection accepted
    Client2->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Register subscription
    Broker->>Client2: SubscribeResponse(success=true)
    
    Note over Agent,Client2: Event Publishing Phase
    Agent->>Broker: PublishRequest("control.collector.task.process", task_data)
    Broker->>Broker: Match topic against subscriptions
    Broker->>Broker: Find Client1 and Client2 subscribed
    Broker-->>Client1: EventEnvelope via IPC
    Broker-->>Client2: EventEnvelope via IPC
    
    Note over Agent,Client2: Result Publishing Phase
    Client1->>Broker: PublishRequest("events.process.lifecycle", event_data)
    Broker->>Broker: Validate ACL (collectors can publish to events.*)
    Broker->>Broker: Match topic "events.process.lifecycle"
    Broker->>Broker: Find Agent subscribed to "events.process.#"
    Broker-->>Agent: EventEnvelope via IPC
    
    Note over Agent,Client2: Shutdown Phase
    Client1->>Broker: UnsubscribeRequest(subscription_id)
    Broker->>Broker: Remove subscription from DashMap
    Client1->>Broker: Disconnect IPC
    Broker->>Broker: Cleanup Client1 resources
    
    Agent->>Broker: shutdown()
    Broker->>Broker: Signal all clients to disconnect
    Broker->>Client2: Connection closing
    Broker->>Broker: Cleanup all subscriptions and connections

Loading

Import In IDE

🤖 Prompt for AI Agents

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

### Observations

The codebase has solid IPC foundations (`interprocess` crate, protobuf codec, connection pooling) but lacks pub/sub capabilities. Issues #113 and #121 are marked "completed" but no code exists - this issue must implement everything. The architecture is strictly cross-process with daemoneye-agent hosting the broker and collectors as separate OS processes communicating via Unix sockets/named pipes.

### Approach

Create a new **daemoneye-eventbus** workspace crate implementing MQTT-style topic hierarchy with wildcard matching for cross-process pub/sub coordination. The broker will be hosted in **daemoneye-agent**, while collectors use **EventBusClient** to publish/subscribe via existing `interprocess` IPC. Extend protobuf schema with **EventEnvelope** and **CorrelationMetadata** messages. Implement topic matching engine with ACL enforcement for security boundaries between processes.

### Reasoning

I explored the repository structure, examined existing IPC infrastructure in `daemoneye-lib/src/ipc/`, reviewed protobuf definitions, analyzed the collector-core framework, and confirmed that no EventBus code exists yet despite dependencies claiming completion. I studied the current request/response IPC architecture and identified integration points for the new pub/sub layer.

## Mermaid Diagram

sequenceDiagram
    participant Agent as daemoneye-agent<br/>(Broker Process)
    participant Broker as EventBusBroker<br/>(in Agent)
    participant Client1 as procmond-1<br/>(Collector Process)
    participant Client2 as procmond-2<br/>(Collector Process)
    
    Note over Agent,Client2: Startup & Connection Phase
    Agent->>Broker: Initialize broker with ACL rules
    Broker->>Broker: Start IPC listener on eventbus.sock
    
    Client1->>Broker: Connect via IPC (Unix socket/named pipe)
    Broker->>Client1: Connection accepted
    Client1->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Validate ACL (collectors can subscribe to control.*)
    Broker->>Broker: Register subscription in DashMap
    Broker->>Client1: SubscribeResponse(success=true)
    
    Client2->>Broker: Connect via IPC
    Broker->>Client2: Connection accepted
    Client2->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Register subscription
    Broker->>Client2: SubscribeResponse(success=true)
    
    Note over Agent,Client2: Event Publishing Phase
    Agent->>Broker: PublishRequest("control.collector.task.process", task_data)
    Broker->>Broker: Match topic against subscriptions
    Broker->>Broker: Find Client1 and Client2 subscribed
    Broker-->>Client1: EventEnvelope via IPC
    Broker-->>Client2: EventEnvelope via IPC
    
    Note over Agent,Client2: Result Publishing Phase
    Client1->>Broker: PublishRequest("events.process.lifecycle", event_data)
    Broker->>Broker: Validate ACL (collectors can publish to events.*)
    Broker->>Broker: Match topic "events.process.lifecycle"
    Broker->>Broker: Find Agent subscribed to "events.process.#"
    Broker-->>Agent: EventEnvelope via IPC
    
    Note over Agent,Client2: Shutdown Phase
    Client1->>Broker: UnsubscribeRequest(subscription_id)
    Broker->>Broker: Remove subscription from DashMap
    Client1->>Broker: Disconnect IPC
    Broker->>Broker: Cleanup Client1 resources
    
    Agent->>Broker: shutdown()
    Broker->>Broker: Signal all clients to disconnect
    Broker->>Client2: Connection closing
    Broker->>Broker: Cleanup all subscriptions and connections

## Proposed File Changes

### Cargo.toml(MODIFY)

Add **daemoneye-eventbus** to the workspace members list after **daemoneye-lib**. Add **dashmap** dependency (version 6.1 or later) to workspace dependencies for concurrent topic routing within the broker process. This enables efficient thread-safe HashMap operations for subscription management.

### daemoneye-eventbus(NEW)

Create new workspace crate directory for the event bus implementation. This will contain the broker, client, topic matching engine, and ACL enforcement logic.

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

Create Cargo.toml for daemoneye-eventbus crate. Include workspace package metadata inheritance. Add dependencies: **tokio** (workspace, for async runtime), **dashmap** (workspace, for concurrent subscriptions), **prost** (workspace, for protobuf), **interprocess** (workspace, for IPC), **anyhow** (workspace, for error handling), **thiserror** (workspace, for custom errors), **tracing** (workspace, for logging), **uuid** (workspace, for correlation IDs), **async-trait** (workspace, for async traits). Set lib.bench = true for benchmarking support. Inherit workspace lints configuration.

### daemoneye-eventbus/src(NEW)

Create source directory for daemoneye-eventbus implementation.

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

Create main library file with module declarations and public API exports. Export **EventBusBroker**, **EventBusClient**, **TopicPattern**, **TopicSegment**, **AccessControl**, and error types. Add comprehensive module documentation describing the cross-process pub/sub architecture, MQTT-style topic hierarchy, and usage examples for both broker (daemoneye-agent) and client (collector processes) scenarios. Include architecture diagram showing process boundaries and IPC communication flow.

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

References: 

- daemoneye-lib/src/detection/sql_to_ipc.rs

Implement topic parsing and matching engine. Define **TopicPattern** struct with segments field (Vec<TopicSegment>). Define **TopicSegment** enum with variants: Literal(String), SingleWildcard (for +), MultiWildcard (for #). Implement **TopicPattern::parse()** method to parse topic strings like "events.process.lifecycle" into segments, validating format (no empty segments, # only at end). Implement **TopicPattern::matches()** method for MQTT-style wildcard matching: single-level wildcard (+) matches exactly one segment, multi-level wildcard (#) matches zero or more segments and must be last. Add comprehensive unit tests covering: exact matches, single-level wildcards in various positions, multi-level wildcards, invalid patterns, edge cases like "events.+.#", and performance with deeply nested topics. Follow patterns from existing validation code in `daemoneye-lib/src/detection/sql_to_ipc.rs`.

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

Implement topic-based access control for cross-process security. Define **AccessControl** enum with variants: ReadOnly (can subscribe/receive), WriteOnly (can publish/send), ReadWrite (both), Restricted (daemoneye-agent only). Define **TopicAcl** struct managing HashMap<TopicPattern, AccessControl> with default rules: events.* → ReadOnly for collectors, control.* → WriteOnly for agent, admin.* → Restricted. Implement **TopicAcl::check_publish_permission()** and **TopicAcl::check_subscribe_permission()** methods that validate process permissions against topic patterns. Add **ProcessIdentity** struct with process_id and process_type (Agent, Collector) fields for permission checks. Include unit tests for ACL enforcement across different topic namespaces and process types.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- collector-core/src/collector.rs

Implement **EventBusBroker** for cross-process pub/sub coordination. Define broker struct with fields: subscriptions (DashMap<TopicPattern, Vec<ProcessSubscription>>), topic_acl (TopicAcl), process_connections (DashMap<ProcessId, IpcConnection>), shutdown_signal (Arc<AtomicBool>). Define **ProcessSubscription** struct with process_id and ipc_sender (mpsc::Sender<EventEnvelope>) fields. Implement **EventBusBroker::new()** constructor initializing ACL with default rules. Implement **subscribe_from_process()** async method: validate ACL permissions, register subscription in DashMap, send acknowledgment via IPC. Implement **publish_from_process()** async method: validate ACL, match topic against all subscription patterns using TopicPattern::matches(), route EventEnvelope to matching subscriber processes via their IPC channels, handle send failures gracefully. Implement **unsubscribe_from_process()** for cleanup. Implement **start()** method to spawn IPC listener accepting connections from collector processes, similar to `InterprocessServer` in `daemoneye-lib/src/ipc/interprocess_transport.rs`. Implement **shutdown()** for graceful cleanup of all connections and subscriptions. Add metrics tracking (messages routed, subscriptions active, ACL violations). Follow async patterns from `collector-core/src/collector.rs` for lifecycle management.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- daemoneye-agent/src/ipc_client.rs
- daemoneye-lib/src/ipc/client.rs

Implement **EventBusClient** for collector processes to interact with broker. Define client struct with fields: ipc_connection (connection to broker via interprocess crate), process_id (ProcessId), codec (IpcCodec for message framing), subscriptions (local tracking of active subscriptions). Implement **EventBusClient::connect()** async method to establish IPC connection to broker endpoint (similar to `InterprocessClient::new()` in `daemoneye-lib/src/ipc/interprocess_transport.rs`), send initial handshake with process identity. Implement **subscribe()** async method: send SubscribeRequest protobuf message to broker, return EventStream (mpsc::Receiver<EventEnvelope>) for receiving events. Implement **publish()** async method: serialize event payload with prost, create EventEnvelope with topic and metadata, send PublishRequest to broker via IPC. Implement **unsubscribe()** for cleanup. Add automatic reconnection logic with exponential backoff similar to `IpcClientManager` in `daemoneye-agent/src/ipc_client.rs`. Implement background task to receive events from broker and dispatch to local subscribers. Add connection health monitoring and metrics. Follow error handling patterns from `daemoneye-lib/src/ipc/client.rs`.

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

References: 

- daemoneye-lib/src/ipc/codec.rs

Define error types for event bus operations using **thiserror** crate. Create **EventBusError** enum with variants: TopicParseError (invalid topic format), AclViolation (permission denied), ConnectionError (IPC failure), SerializationError (protobuf encoding/decoding), SubscriptionNotFound, BrokerShutdown, InvalidWildcard (# not at end), EmptySegment, and other domain-specific errors. Implement Display and Error traits. Add context-rich error messages for debugging cross-process issues. Follow error handling patterns from `daemoneye-lib/src/ipc/codec.rs` IpcError enum.

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

Define core data types for event bus. Create **ProcessId** type alias or newtype for process identification. Define **EventStream** type alias for mpsc::Receiver<EventEnvelope>. Define **SubscriptionId** type for tracking subscriptions. Add type aliases for clarity and future extensibility. Keep types simple and focused on cross-process communication semantics.

### daemoneye-lib/proto/eventbus.proto(NEW)

References: 

- daemoneye-lib/proto/ipc.proto
- daemoneye-lib/proto/common.proto

Define protobuf messages for event bus cross-process communication. Create **EventEnvelope** message with fields: topic (string), payload (bytes - serialized event data), timestamp (int64 - Unix timestamp in milliseconds), publisher_process_id (string), correlation_metadata (optional CorrelationMetadata). Create **CorrelationMetadata** message with fields: correlation_id (string - UUID), parent_correlation_id (optional string), root_correlation_id (string), sequence_number (uint64), workflow_stage (optional string), source_component (string), correlation_tags (map<string, string>). Create **SubscribeRequest** message with fields: subscription_id (string), topic_pattern (string - with wildcards), process_identity (ProcessIdentity). Create **SubscribeResponse** message with fields: subscription_id (string), success (bool), error_message (optional string). Create **PublishRequest** message with fields: envelope (EventEnvelope). Create **PublishResponse** message with fields: success (bool), error_message (optional string). Create **UnsubscribeRequest** message with fields: subscription_id (string). Create **ProcessIdentity** message with fields: process_id (string), process_type (ProcessType enum - AGENT, COLLECTOR). Follow protobuf conventions from `daemoneye-lib/proto/ipc.proto` and `daemoneye-lib/proto/common.proto`.

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

References: 

- daemoneye-lib/proto/ipc.proto

Add **proto/eventbus.proto** to the list of protobuf files compiled by prost-build. Ensure the new EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, and ProcessIdentity messages are generated. Follow the existing pattern for compiling proto files and generating Rust code in the OUT_DIR.

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

Re-export the new event bus protobuf types from the generated code. Add pub use statements for **EventEnvelope**, **CorrelationMetadata**, **SubscribeRequest**, **SubscribeResponse**, **PublishRequest**, **PublishResponse**, **UnsubscribeRequest**, **ProcessIdentity**, and **ProcessType** enum. Add helper methods to EventEnvelope for creating envelopes with correlation metadata, extracting topic, and validating structure. Follow the pattern of helper methods on DetectionTask and DetectionResult in this file.

### daemoneye-eventbus/tests(NEW)

Create tests directory for integration and unit tests.

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

References: 

- collector-core/tests/integration_test.rs

Create comprehensive unit tests for topic matching engine. Test exact topic matches ("events.process.lifecycle" matches "events.process.lifecycle"). Test single-level wildcard: "events.process.+" matches "events.process.lifecycle" but NOT "events.process.tree.update", "events.+.lifecycle" matches "events.process.lifecycle" and "events.network.lifecycle". Test multi-level wildcard: "events.#" matches all event topics, "events.process.#" matches all process events including nested ones. Test invalid patterns: "events.#.invalid" (# not at end), "events..lifecycle" (empty segment), "" (empty topic). Test edge cases: single segment topics, deeply nested topics (10+ levels), topics with special characters. Test performance with 1000+ topic patterns. Follow test patterns from `collector-core/tests/` for structure and assertions.

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

References: 

- daemoneye-lib/tests/ipc_security_validation.rs

Create unit tests for ACL enforcement logic. Test that collector processes can publish to "events.*" topics but NOT "control.*" topics. Test that daemoneye-agent process can publish to "control.*" topics but collectors cannot. Test that collectors can subscribe to "control.collector.*" and "control.health.*" topics. Test that agent can subscribe to "events.*" topics. Test that "admin.*" topics are restricted to agent only. Test ACL violations return appropriate errors. Test default ACL rules are correctly initialized. Follow security test patterns from `daemoneye-lib/tests/ipc_security_validation.rs`.

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

References: 

- collector-core/tests/integration_test.rs
- daemoneye-lib/tests/ipc_integration.rs

Create integration tests for EventBusBroker with multiple collector processes. Test broker startup and IPC listener initialization. Test multiple collector processes connecting to broker. Test subscription registration from different processes. Test event publishing and routing to matching subscribers across process boundaries. Test wildcard subscriptions receiving appropriate events. Test subscription cleanup when collector process disconnects. Test concurrent publish/subscribe operations from multiple processes. Test broker shutdown and cleanup of all resources. Use tempfile crate for test socket paths similar to `collector-core/tests/integration_test.rs`. Spawn mock collector processes using tokio::spawn. Follow integration test patterns from `daemoneye-lib/tests/ipc_integration.rs` for multi-process coordination.

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

References: 

- daemoneye-lib/tests/resilient_client_integration.rs

Create integration tests for EventBusClient. Test client connection to broker. Test subscribe() method and receiving events via EventStream. Test publish() method sending events to broker. Test multiple clients subscribing to same topic pattern. Test client reconnection after broker restart or connection failure. Test correlation metadata propagation through publish/subscribe flow. Test client cleanup and unsubscribe on shutdown. Follow client testing patterns from `daemoneye-lib/tests/resilient_client_integration.rs` for connection management and error scenarios.

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

References: 

- daemoneye-lib/tests/ipc_cross_platform_tests.rs

Create cross-platform validation tests for Windows (named pipes) and Unix (domain sockets). Test broker and client work correctly on both platforms. Test socket path generation and cleanup. Test platform-specific IPC error handling. Test concurrent connections on both platforms. Follow cross-platform test patterns from `daemoneye-lib/tests/ipc_cross_platform_tests.rs` with conditional compilation for platform-specific code.

### daemoneye-eventbus/benches(NEW)

Create benchmarks directory for performance testing.

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

References: 

- daemoneye-lib/benches/ipc_performance_comprehensive.rs

Create performance benchmarks using criterion crate. Benchmark topic matching performance with various pattern complexities (exact, single wildcard, multi-level wildcard) against 1000+ topics. Benchmark broker routing latency for single subscriber, multiple subscribers (10+), and wildcard subscriptions. Benchmark cross-process IPC throughput: messages/second with 1 publisher + 1 subscriber, 1 publisher + 10 subscribers, 10 publishers + 10 subscribers. Benchmark subscription management: add/remove subscriptions, lookup performance with 1000+ active subscriptions. Target: <1ms p99 latency for local IPC routing. Follow benchmark patterns from `daemoneye-lib/benches/ipc_performance_comprehensive.rs` using criterion framework and black_box for preventing optimization.

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

References: 

- daemoneye-agent/src/main.rs(MODIFY)
- daemoneye-lib/src/telemetry.rs

Create integration module for hosting EventBusBroker in daemoneye-agent process. Define **EventBusService** struct wrapping EventBusBroker with lifecycle management. Implement **start_broker()** function to initialize broker with default ACL rules, start IPC listener on dedicated endpoint (e.g., /var/run/daemoneye/eventbus.sock or \\.\pipe\daemoneye\eventbus), and spawn broker runtime task. Implement **shutdown_broker()** for graceful cleanup. Add broker health monitoring and metrics integration with existing telemetry system in `daemoneye-lib/src/telemetry.rs`. Provide API for agent components to subscribe to topics (e.g., detection engine subscribing to "events.process.#"). Follow service initialization patterns from `daemoneye-agent/src/main.rs` main loop structure.

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

References: 

- daemoneye-agent/src/ipc_client.rs

Integrate EventBusBroker into daemoneye-agent startup sequence. Add module declaration for **eventbus_integration**. In the run() function after IPC client initialization (around line 64), add broker initialization: create EventBusService, start broker with start_broker(), store broker handle for shutdown. In the shutdown section (after line 203), add broker.shutdown() call before final cleanup. Add broker health monitoring to the main loop alongside existing telemetry health checks. Consider subscribing agent to "events.process.#" topic to receive process events from collectors as an alternative/complement to the current IPC polling approach. Maintain backward compatibility with existing IPC client for gradual migration.

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

References: 

- collector-core/Cargo.toml(MODIFY)

Add **daemoneye-eventbus** as a dependency to enable broker integration. Add it to the dependencies section with path = "../daemoneye-eventbus" similar to how **collector-core** and **daemoneye-lib** are included.

### collector-core/src/eventbus_client_integration.rs(NEW)

References: 

- collector-core/src/ipc.rs
- collector-core/src/event.rs

Create optional integration module for collector-core to use EventBusClient. Define **CollectorEventBusAdapter** struct that wraps EventBusClient and provides methods for collectors to publish events to the broker. Implement **publish_process_event()** method to publish ProcessEvent to "events.process.lifecycle" or other appropriate topics based on event type. Implement **subscribe_to_tasks()** method to subscribe to "control.collector.task.*" topics for receiving tasks from agent. Add helper methods to convert between CollectionEvent (from `collector-core/src/event.rs`) and EventEnvelope payloads. Make this integration optional/feature-gated to maintain backward compatibility with direct IPC. Follow integration patterns from `collector-core/src/ipc.rs` for IPC server integration.

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

Add optional module declaration for **eventbus_client_integration** (conditionally compiled with feature flag). Re-export CollectorEventBusAdapter if the eventbus feature is enabled. This allows collectors to optionally use the event bus without breaking existing direct IPC functionality.

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

Add optional **daemoneye-eventbus** dependency with path = "../daemoneye-eventbus" and optional = true. Add new feature flag **eventbus** that enables the daemoneye-eventbus dependency. This allows collectors to opt-in to event bus functionality while maintaining backward compatibility with direct IPC.

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

References: 

- collector-core/README.md

Create comprehensive documentation for daemoneye-eventbus crate. Document the cross-process pub/sub architecture with process diagrams. Explain MQTT-style topic hierarchy with examples: events.process.lifecycle, events.process.metadata, control.collector.task.process, control.health.heartbeat. Document wildcard matching rules with examples of + and # usage. Explain topic-based ACL and security boundaries between processes. Provide usage examples for broker (in daemoneye-agent) and client (in collectors). Include code examples for subscribing, publishing, and handling events. Document performance characteristics and benchmarks. Explain integration with existing IPC infrastructure. Add troubleshooting section for common cross-process issues. Follow documentation style from `collector-core/README.md`.

### docs/src/technical/eventbus-architecture.md(NEW)

References: 

- docs/src/technical/ipc-implementation.md

Create technical documentation for event bus architecture in the mdBook documentation. Document the complete topic hierarchy with all namespaces: events.process.* (lifecycle, metadata, tree, integrity, anomaly, batch), events.network.*, events.filesystem.*, events.performance.*, control.collector.* (lifecycle, config, task.*), control.health.* (heartbeat, status, diagnostics). Explain cross-process message flow with sequence diagrams. Document ACL rules and security model. Provide deployment guidance for multi-collector coordination. Include performance tuning recommendations. Add examples of common pub/sub patterns for multi-collector scenarios. Follow documentation structure from `docs/src/technical/ipc-implementation.md`.

### docs/src/SUMMARY.md(MODIFY)

Add link to the new eventbus architecture documentation under the Technical section. Add entry: "- [Event Bus Architecture](technical/eventbus-architecture.md)" after the IPC Implementation entry to maintain logical flow of technical documentation.

---

Developer Humor

🎭 A broker walked into a pub/sub bar,\n> Wildcards matching topics near and far.\n> \Events dot hash!\ cried the collector with glee,\n> \I'll subscribe to them all, just route them to me!\n> \n> The agent said \Control dot star is mine,\n> \Your ACL says 'ReadOnly' - that's by design!\n> With topics and patterns, they danced through the night,\n> Cross-process IPC never felt so right! 🎪

---

Execution Information

Branch: main
Commit: a169b68

[traycerai]

I have updated the Implementation Plan to reflect the recent changes in the ticket.

Previous Plan

Observations

The codebase has solid IPC foundations (interprocess crate, protobuf codec, connection pooling) but lacks pub/sub capabilities. Issues #113 and #121 are marked "completed" but no code exists - this issue must implement everything. The architecture is strictly cross-process with daemoneye-agent hosting the broker and collectors as separate OS processes communicating via Unix sockets/named pipes.

Approach

Create a new daemoneye-eventbus workspace crate implementing MQTT-style topic hierarchy with wildcard matching for cross-process pub/sub coordination. The broker will be hosted in daemoneye-agent, while collectors use EventBusClient to publish/subscribe via existing interprocess IPC. Extend protobuf schema with EventEnvelope and CorrelationMetadata messages. Implement topic matching engine with ACL enforcement for security boundaries between processes.

Reasoning

I explored the repository structure, examined existing IPC infrastructure in daemoneye-lib/src/ipc/, reviewed protobuf definitions, analyzed the collector-core framework, and confirmed that no EventBus code exists yet despite dependencies claiming completion. I studied the current request/response IPC architecture and identified integration points for the new pub/sub layer.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

Add daemoneye-eventbus to the workspace members list after daemoneye-lib. Add dashmap dependency (version 6.1 or later) to workspace dependencies for concurrent topic routing within the broker process. This enables efficient thread-safe HashMap operations for subscription management.

📁 daemoneye-eventbus ^(NEW) 🔗

Create new workspace crate directory for the event bus implementation. This will contain the broker, client, topic matching engine, and ACL enforcement logic.

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

Create Cargo.toml for daemoneye-eventbus crate. Include workspace package metadata inheritance. Add dependencies: tokio (workspace, for async runtime), dashmap (workspace, for concurrent subscriptions), prost (workspace, for protobuf), interprocess (workspace, for IPC), anyhow (workspace, for error handling), thiserror (workspace, for custom errors), tracing (workspace, for logging), uuid (workspace, for correlation IDs), async-trait (workspace, for async traits). Set lib.bench = true for benchmarking support. Inherit workspace lints configuration.

📁 daemoneye-eventbus/src ^(NEW) 🔗

Create source directory for daemoneye-eventbus implementation.

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

Create main library file with module declarations and public API exports. Export EventBusBroker, EventBusClient, TopicPattern, TopicSegment, AccessControl, and error types. Add comprehensive module documentation describing the cross-process pub/sub architecture, MQTT-style topic hierarchy, and usage examples for both broker (daemoneye-agent) and client (collector processes) scenarios. Include architecture diagram showing process boundaries and IPC communication flow.

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

References

• 📄 daemoneye-lib/src/detection/sql_to_ipc.rs

Implement topic parsing and matching engine. Define TopicPattern struct with segments field (Vec). Define TopicSegment enum with variants: Literal(String), SingleWildcard (for +), MultiWildcard (for #). Implement TopicPattern::parse() method to parse topic strings like "events.process.lifecycle" into segments, validating format (no empty segments, # only at end). Implement TopicPattern::matches() method for MQTT-style wildcard matching: single-level wildcard (+) matches exactly one segment, multi-level wildcard (#) matches zero or more segments and must be last. Add comprehensive unit tests covering: exact matches, single-level wildcards in various positions, multi-level wildcards, invalid patterns, edge cases like "events.+.#", and performance with deeply nested topics. Follow patterns from existing validation code in daemoneye-lib/src/detection/sql_to_ipc.rs.

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

Implement topic-based access control for cross-process security. Define AccessControl enum with variants: ReadOnly (can subscribe/receive), WriteOnly (can publish/send), ReadWrite (both), Restricted (daemoneye-agent only). Define TopicAcl struct managing HashMap<TopicPattern, AccessControl> with default rules: events.* → ReadOnly for collectors, control.* → WriteOnly for agent, admin.* → Restricted. Implement TopicAcl::check_publish_permission() and TopicAcl::check_subscribe_permission() methods that validate process permissions against topic patterns. Add ProcessIdentity struct with process_id and process_type (Agent, Collector) fields for permission checks. Include unit tests for ACL enforcement across different topic namespaces and process types.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 collector-core/src/collector.rs

Implement EventBusBroker for cross-process pub/sub coordination. Define broker struct with fields: subscriptions (DashMap<TopicPattern, Vec>), topic_acl (TopicAcl), process_connections (DashMap<ProcessId, IpcConnection>), shutdown_signal (Arc). Define ProcessSubscription struct with process_id and ipc_sender (mpsc::Sender) fields. Implement EventBusBroker::new() constructor initializing ACL with default rules. Implement subscribe_from_process() async method: validate ACL permissions, register subscription in DashMap, send acknowledgment via IPC. Implement publish_from_process() async method: validate ACL, match topic against all subscription patterns using TopicPattern::matches(), route EventEnvelope to matching subscriber processes via their IPC channels, handle send failures gracefully. Implement unsubscribe_from_process() for cleanup. Implement start() method to spawn IPC listener accepting connections from collector processes, similar to InterprocessServer in daemoneye-lib/src/ipc/interprocess_transport.rs. Implement shutdown() for graceful cleanup of all connections and subscriptions. Add metrics tracking (messages routed, subscriptions active, ACL violations). Follow async patterns from collector-core/src/collector.rs for lifecycle management.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 daemoneye-agent/src/ipc_client.rs
• 📄 daemoneye-lib/src/ipc/client.rs

Implement EventBusClient for collector processes to interact with broker. Define client struct with fields: ipc_connection (connection to broker via interprocess crate), process_id (ProcessId), codec (IpcCodec for message framing), subscriptions (local tracking of active subscriptions). Implement EventBusClient::connect() async method to establish IPC connection to broker endpoint (similar to InterprocessClient::new() in daemoneye-lib/src/ipc/interprocess_transport.rs), send initial handshake with process identity. Implement subscribe() async method: send SubscribeRequest protobuf message to broker, return EventStream (mpsc::Receiver) for receiving events. Implement publish() async method: serialize event payload with prost, create EventEnvelope with topic and metadata, send PublishRequest to broker via IPC. Implement unsubscribe() for cleanup. Add automatic reconnection logic with exponential backoff similar to IpcClientManager in daemoneye-agent/src/ipc_client.rs. Implement background task to receive events from broker and dispatch to local subscribers. Add connection health monitoring and metrics. Follow error handling patterns from daemoneye-lib/src/ipc/client.rs.

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

References

• 📄 daemoneye-lib/src/ipc/codec.rs

Define error types for event bus operations using thiserror crate. Create EventBusError enum with variants: TopicParseError (invalid topic format), AclViolation (permission denied), ConnectionError (IPC failure), SerializationError (protobuf encoding/decoding), SubscriptionNotFound, BrokerShutdown, InvalidWildcard (# not at end), EmptySegment, and other domain-specific errors. Implement Display and Error traits. Add context-rich error messages for debugging cross-process issues. Follow error handling patterns from daemoneye-lib/src/ipc/codec.rs IpcError enum.

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

Define core data types for event bus. Create ProcessId type alias or newtype for process identification. Define EventStream type alias for mpsc::Receiver. Define SubscriptionId type for tracking subscriptions. Add type aliases for clarity and future extensibility. Keep types simple and focused on cross-process communication semantics.

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

References

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

Define protobuf messages for event bus cross-process communication. Create EventEnvelope message with fields: topic (string), payload (bytes - serialized event data), timestamp (int64 - Unix timestamp in milliseconds), publisher_process_id (string), correlation_metadata (optional CorrelationMetadata). Create CorrelationMetadata message with fields: correlation_id (string - UUID), parent_correlation_id (optional string), root_correlation_id (string), sequence_number (uint64), workflow_stage (optional string), source_component (string), correlation_tags (map<string, string>). Create SubscribeRequest message with fields: subscription_id (string), topic_pattern (string - with wildcards), process_identity (ProcessIdentity). Create SubscribeResponse message with fields: subscription_id (string), success (bool), error_message (optional string). Create PublishRequest message with fields: envelope (EventEnvelope). Create PublishResponse message with fields: success (bool), error_message (optional string). Create UnsubscribeRequest message with fields: subscription_id (string). Create ProcessIdentity message with fields: process_id (string), process_type (ProcessType enum - AGENT, COLLECTOR). Follow protobuf conventions from daemoneye-lib/proto/ipc.proto and daemoneye-lib/proto/common.proto.

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

References

• 📄 daemoneye-lib/proto/ipc.proto

Add proto/eventbus.proto to the list of protobuf files compiled by prost-build. Ensure the new EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, and ProcessIdentity messages are generated. Follow the existing pattern for compiling proto files and generating Rust code in the OUT_DIR.

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

Re-export the new event bus protobuf types from the generated code. Add pub use statements for EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, ProcessIdentity, and ProcessType enum. Add helper methods to EventEnvelope for creating envelopes with correlation metadata, extracting topic, and validating structure. Follow the pattern of helper methods on DetectionTask and DetectionResult in this file.

📁 daemoneye-eventbus/tests ^(NEW) 🔗

Create tests directory for integration and unit tests.

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

References

• 📄 collector-core/tests/integration_test.rs

Create comprehensive unit tests for topic matching engine. Test exact topic matches ("events.process.lifecycle" matches "events.process.lifecycle"). Test single-level wildcard: "events.process.+" matches "events.process.lifecycle" but NOT "events.process.tree.update", "events.+.lifecycle" matches "events.process.lifecycle" and "events.network.lifecycle". Test multi-level wildcard: "events.#" matches all event topics, "events.process.#" matches all process events including nested ones. Test invalid patterns: "events.#.invalid" (# not at end), "events..lifecycle" (empty segment), "" (empty topic). Test edge cases: single segment topics, deeply nested topics (10+ levels), topics with special characters. Test performance with 1000+ topic patterns. Follow test patterns from collector-core/tests/ for structure and assertions.

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

References

• 📄 daemoneye-lib/tests/ipc_security_validation.rs

Create unit tests for ACL enforcement logic. Test that collector processes can publish to "events." topics but NOT "control." topics. Test that daemoneye-agent process can publish to "control." topics but collectors cannot. Test that collectors can subscribe to "control.collector." and "control.health." topics. Test that agent can subscribe to "events." topics. Test that "admin.*" topics are restricted to agent only. Test ACL violations return appropriate errors. Test default ACL rules are correctly initialized. Follow security test patterns from daemoneye-lib/tests/ipc_security_validation.rs.

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

References

• 📄 collector-core/tests/integration_test.rs
• 📄 daemoneye-lib/tests/ipc_integration.rs

Create integration tests for EventBusBroker with multiple collector processes. Test broker startup and IPC listener initialization. Test multiple collector processes connecting to broker. Test subscription registration from different processes. Test event publishing and routing to matching subscribers across process boundaries. Test wildcard subscriptions receiving appropriate events. Test subscription cleanup when collector process disconnects. Test concurrent publish/subscribe operations from multiple processes. Test broker shutdown and cleanup of all resources. Use tempfile crate for test socket paths similar to collector-core/tests/integration_test.rs. Spawn mock collector processes using tokio::spawn. Follow integration test patterns from daemoneye-lib/tests/ipc_integration.rs for multi-process coordination.

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

References

• 📄 daemoneye-lib/tests/resilient_client_integration.rs

Create integration tests for EventBusClient. Test client connection to broker. Test subscribe() method and receiving events via EventStream. Test publish() method sending events to broker. Test multiple clients subscribing to same topic pattern. Test client reconnection after broker restart or connection failure. Test correlation metadata propagation through publish/subscribe flow. Test client cleanup and unsubscribe on shutdown. Follow client testing patterns from daemoneye-lib/tests/resilient_client_integration.rs for connection management and error scenarios.

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

References

• 📄 daemoneye-lib/tests/ipc_cross_platform_tests.rs

Create cross-platform validation tests for Windows (named pipes) and Unix (domain sockets). Test broker and client work correctly on both platforms. Test socket path generation and cleanup. Test platform-specific IPC error handling. Test concurrent connections on both platforms. Follow cross-platform test patterns from daemoneye-lib/tests/ipc_cross_platform_tests.rs with conditional compilation for platform-specific code.

📁 daemoneye-eventbus/benches ^(NEW) 🔗

Create benchmarks directory for performance testing.

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

References

• 📄 daemoneye-lib/benches/ipc_performance_comprehensive.rs

Create performance benchmarks using criterion crate. Benchmark topic matching performance with various pattern complexities (exact, single wildcard, multi-level wildcard) against 1000+ topics. Benchmark broker routing latency for single subscriber, multiple subscribers (10+), and wildcard subscriptions. Benchmark cross-process IPC throughput: messages/second with 1 publisher + 1 subscriber, 1 publisher + 10 subscribers, 10 publishers + 10 subscribers. Benchmark subscription management: add/remove subscriptions, lookup performance with 1000+ active subscriptions. Target: <1ms p99 latency for local IPC routing. Follow benchmark patterns from daemoneye-lib/benches/ipc_performance_comprehensive.rs using criterion framework and black_box for preventing optimization.

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

References

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

Create integration module for hosting EventBusBroker in daemoneye-agent process. Define EventBusService struct wrapping EventBusBroker with lifecycle management. Implement start_broker() function to initialize broker with default ACL rules, start IPC listener on dedicated endpoint (e.g., /var/run/daemoneye/eventbus.sock or \.\pipe\daemoneye\eventbus), and spawn broker runtime task. Implement shutdown_broker() for graceful cleanup. Add broker health monitoring and metrics integration with existing telemetry system in daemoneye-lib/src/telemetry.rs. Provide API for agent components to subscribe to topics (e.g., detection engine subscribing to "events.process.#"). Follow service initialization patterns from daemoneye-agent/src/main.rs main loop structure.

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

References

• 📄 daemoneye-agent/src/ipc_client.rs

Integrate EventBusBroker into daemoneye-agent startup sequence. Add module declaration for eventbus_integration. In the run() function after IPC client initialization (around line 64), add broker initialization: create EventBusService, start broker with start_broker(), store broker handle for shutdown. In the shutdown section (after line 203), add broker.shutdown() call before final cleanup. Add broker health monitoring to the main loop alongside existing telemetry health checks. Consider subscribing agent to "events.process.#" topic to receive process events from collectors as an alternative/complement to the current IPC polling approach. Maintain backward compatibility with existing IPC client for gradual migration.

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

References

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

Add daemoneye-eventbus as a dependency to enable broker integration. Add it to the dependencies section with path = "../daemoneye-eventbus" similar to how collector-core and daemoneye-lib are included.

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

References

• 📄 collector-core/src/ipc.rs
• 📄 collector-core/src/event.rs

Create optional integration module for collector-core to use EventBusClient. Define CollectorEventBusAdapter struct that wraps EventBusClient and provides methods for collectors to publish events to the broker. Implement publish_process_event() method to publish ProcessEvent to "events.process.lifecycle" or other appropriate topics based on event type. Implement subscribe_to_tasks() method to subscribe to "control.collector.task.*" topics for receiving tasks from agent. Add helper methods to convert between CollectionEvent (from collector-core/src/event.rs) and EventEnvelope payloads. Make this integration optional/feature-gated to maintain backward compatibility with direct IPC. Follow integration patterns from collector-core/src/ipc.rs for IPC server integration.

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

Add optional module declaration for eventbus_client_integration (conditionally compiled with feature flag). Re-export CollectorEventBusAdapter if the eventbus feature is enabled. This allows collectors to optionally use the event bus without breaking existing direct IPC functionality.

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

Add optional daemoneye-eventbus dependency with path = "../daemoneye-eventbus" and optional = true. Add new feature flag eventbus that enables the daemoneye-eventbus dependency. This allows collectors to opt-in to event bus functionality while maintaining backward compatibility with direct IPC.

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

References

• 📄 collector-core/README.md

Create comprehensive documentation for daemoneye-eventbus crate. Document the cross-process pub/sub architecture with process diagrams. Explain MQTT-style topic hierarchy with examples: events.process.lifecycle, events.process.metadata, control.collector.task.process, control.health.heartbeat. Document wildcard matching rules with examples of + and # usage. Explain topic-based ACL and security boundaries between processes. Provide usage examples for broker (in daemoneye-agent) and client (in collectors). Include code examples for subscribing, publishing, and handling events. Document performance characteristics and benchmarks. Explain integration with existing IPC infrastructure. Add troubleshooting section for common cross-process issues. Follow documentation style from collector-core/README.md.

📄 docs/src/technical/eventbus-architecture.md ^(NEW) 🔗

References

• 📄 docs/src/technical/ipc-implementation.md

Create technical documentation for event bus architecture in the mdBook documentation. Document the complete topic hierarchy with all namespaces: events.process.* (lifecycle, metadata, tree, integrity, anomaly, batch), events.network., events.filesystem., events.performance., control.collector. (lifecycle, config, task.), control.health. (heartbeat, status, diagnostics). Explain cross-process message flow with sequence diagrams. Document ACL rules and security model. Provide deployment guidance for multi-collector coordination. Include performance tuning recommendations. Add examples of common pub/sub patterns for multi-collector scenarios. Follow documentation structure from docs/src/technical/ipc-implementation.md.

📄 docs/src/SUMMARY.md ^(MODIFY) 🔗

Add link to the new eventbus architecture documentation under the Technical section. Add entry: "- Event Bus Architecture" after the IPC Implementation entry to maintain logical flow of technical documentation.

Diagram

sequenceDiagram
    participant Agent as daemoneye-agent<br/>(Broker Process)
    participant Broker as EventBusBroker<br/>(in Agent)
    participant Client1 as procmond-1<br/>(Collector Process)
    participant Client2 as procmond-2<br/>(Collector Process)
    
    Note over Agent,Client2: Startup & Connection Phase
    Agent->>Broker: Initialize broker with ACL rules
    Broker->>Broker: Start IPC listener on eventbus.sock
    
    Client1->>Broker: Connect via IPC (Unix socket/named pipe)
    Broker->>Client1: Connection accepted
    Client1->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Validate ACL (collectors can subscribe to control.*)
    Broker->>Broker: Register subscription in DashMap
    Broker->>Client1: SubscribeResponse(success=true)
    
    Client2->>Broker: Connect via IPC
    Broker->>Client2: Connection accepted
    Client2->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Register subscription
    Broker->>Client2: SubscribeResponse(success=true)
    
    Note over Agent,Client2: Event Publishing Phase
    Agent->>Broker: PublishRequest("control.collector.task.process", task_data)
    Broker->>Broker: Match topic against subscriptions
    Broker->>Broker: Find Client1 and Client2 subscribed
    Broker-->>Client1: EventEnvelope via IPC
    Broker-->>Client2: EventEnvelope via IPC
    
    Note over Agent,Client2: Result Publishing Phase
    Client1->>Broker: PublishRequest("events.process.lifecycle", event_data)
    Broker->>Broker: Validate ACL (collectors can publish to events.*)
    Broker->>Broker: Match topic "events.process.lifecycle"
    Broker->>Broker: Find Agent subscribed to "events.process.#"
    Broker-->>Agent: EventEnvelope via IPC
    
    Note over Agent,Client2: Shutdown Phase
    Client1->>Broker: UnsubscribeRequest(subscription_id)
    Broker->>Broker: Remove subscription from DashMap
    Client1->>Broker: Disconnect IPC
    Broker->>Broker: Cleanup Client1 resources
    
    Agent->>Broker: shutdown()
    Broker->>Broker: Signal all clients to disconnect
    Broker->>Client2: Connection closing
    Broker->>Broker: Cleanup all subscriptions and connections

Loading

Import In IDE

🤖 Prompt for AI Agents

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

### Observations

The codebase has solid IPC foundations (`interprocess` crate, protobuf codec, connection pooling) but lacks pub/sub capabilities. Issues #113 and #121 are marked "completed" but no code exists - this issue must implement everything. The architecture is strictly cross-process with daemoneye-agent hosting the broker and collectors as separate OS processes communicating via Unix sockets/named pipes.

### Approach

Create a new **daemoneye-eventbus** workspace crate implementing MQTT-style topic hierarchy with wildcard matching for cross-process pub/sub coordination. The broker will be hosted in **daemoneye-agent**, while collectors use **EventBusClient** to publish/subscribe via existing `interprocess` IPC. Extend protobuf schema with **EventEnvelope** and **CorrelationMetadata** messages. Implement topic matching engine with ACL enforcement for security boundaries between processes.

### Reasoning

I explored the repository structure, examined existing IPC infrastructure in `daemoneye-lib/src/ipc/`, reviewed protobuf definitions, analyzed the collector-core framework, and confirmed that no EventBus code exists yet despite dependencies claiming completion. I studied the current request/response IPC architecture and identified integration points for the new pub/sub layer.

## Mermaid Diagram

sequenceDiagram
    participant Agent as daemoneye-agent<br/>(Broker Process)
    participant Broker as EventBusBroker<br/>(in Agent)
    participant Client1 as procmond-1<br/>(Collector Process)
    participant Client2 as procmond-2<br/>(Collector Process)
    
    Note over Agent,Client2: Startup & Connection Phase
    Agent->>Broker: Initialize broker with ACL rules
    Broker->>Broker: Start IPC listener on eventbus.sock
    
    Client1->>Broker: Connect via IPC (Unix socket/named pipe)
    Broker->>Client1: Connection accepted
    Client1->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Validate ACL (collectors can subscribe to control.*)
    Broker->>Broker: Register subscription in DashMap
    Broker->>Client1: SubscribeResponse(success=true)
    
    Client2->>Broker: Connect via IPC
    Broker->>Client2: Connection accepted
    Client2->>Broker: SubscribeRequest("control.collector.task.process")
    Broker->>Broker: Register subscription
    Broker->>Client2: SubscribeResponse(success=true)
    
    Note over Agent,Client2: Event Publishing Phase
    Agent->>Broker: PublishRequest("control.collector.task.process", task_data)
    Broker->>Broker: Match topic against subscriptions
    Broker->>Broker: Find Client1 and Client2 subscribed
    Broker-->>Client1: EventEnvelope via IPC
    Broker-->>Client2: EventEnvelope via IPC
    
    Note over Agent,Client2: Result Publishing Phase
    Client1->>Broker: PublishRequest("events.process.lifecycle", event_data)
    Broker->>Broker: Validate ACL (collectors can publish to events.*)
    Broker->>Broker: Match topic "events.process.lifecycle"
    Broker->>Broker: Find Agent subscribed to "events.process.#"
    Broker-->>Agent: EventEnvelope via IPC
    
    Note over Agent,Client2: Shutdown Phase
    Client1->>Broker: UnsubscribeRequest(subscription_id)
    Broker->>Broker: Remove subscription from DashMap
    Client1->>Broker: Disconnect IPC
    Broker->>Broker: Cleanup Client1 resources
    
    Agent->>Broker: shutdown()
    Broker->>Broker: Signal all clients to disconnect
    Broker->>Client2: Connection closing
    Broker->>Broker: Cleanup all subscriptions and connections

## Proposed File Changes

### Cargo.toml(MODIFY)

Add **daemoneye-eventbus** to the workspace members list after **daemoneye-lib**. Add **dashmap** dependency (version 6.1 or later) to workspace dependencies for concurrent topic routing within the broker process. This enables efficient thread-safe HashMap operations for subscription management.

### daemoneye-eventbus(NEW)

Create new workspace crate directory for the event bus implementation. This will contain the broker, client, topic matching engine, and ACL enforcement logic.

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

Create Cargo.toml for daemoneye-eventbus crate. Include workspace package metadata inheritance. Add dependencies: **tokio** (workspace, for async runtime), **dashmap** (workspace, for concurrent subscriptions), **prost** (workspace, for protobuf), **interprocess** (workspace, for IPC), **anyhow** (workspace, for error handling), **thiserror** (workspace, for custom errors), **tracing** (workspace, for logging), **uuid** (workspace, for correlation IDs), **async-trait** (workspace, for async traits). Set lib.bench = true for benchmarking support. Inherit workspace lints configuration.

### daemoneye-eventbus/src(NEW)

Create source directory for daemoneye-eventbus implementation.

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

Create main library file with module declarations and public API exports. Export **EventBusBroker**, **EventBusClient**, **TopicPattern**, **TopicSegment**, **AccessControl**, and error types. Add comprehensive module documentation describing the cross-process pub/sub architecture, MQTT-style topic hierarchy, and usage examples for both broker (daemoneye-agent) and client (collector processes) scenarios. Include architecture diagram showing process boundaries and IPC communication flow.

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

References: 

- daemoneye-lib/src/detection/sql_to_ipc.rs

Implement topic parsing and matching engine. Define **TopicPattern** struct with segments field (Vec<TopicSegment>). Define **TopicSegment** enum with variants: Literal(String), SingleWildcard (for +), MultiWildcard (for #). Implement **TopicPattern::parse()** method to parse topic strings like "events.process.lifecycle" into segments, validating format (no empty segments, # only at end). Implement **TopicPattern::matches()** method for MQTT-style wildcard matching: single-level wildcard (+) matches exactly one segment, multi-level wildcard (#) matches zero or more segments and must be last. Add comprehensive unit tests covering: exact matches, single-level wildcards in various positions, multi-level wildcards, invalid patterns, edge cases like "events.+.#", and performance with deeply nested topics. Follow patterns from existing validation code in `daemoneye-lib/src/detection/sql_to_ipc.rs`.

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

Implement topic-based access control for cross-process security. Define **AccessControl** enum with variants: ReadOnly (can subscribe/receive), WriteOnly (can publish/send), ReadWrite (both), Restricted (daemoneye-agent only). Define **TopicAcl** struct managing HashMap<TopicPattern, AccessControl> with default rules: events.* → ReadOnly for collectors, control.* → WriteOnly for agent, admin.* → Restricted. Implement **TopicAcl::check_publish_permission()** and **TopicAcl::check_subscribe_permission()** methods that validate process permissions against topic patterns. Add **ProcessIdentity** struct with process_id and process_type (Agent, Collector) fields for permission checks. Include unit tests for ACL enforcement across different topic namespaces and process types.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- collector-core/src/collector.rs

Implement **EventBusBroker** for cross-process pub/sub coordination. Define broker struct with fields: subscriptions (DashMap<TopicPattern, Vec<ProcessSubscription>>), topic_acl (TopicAcl), process_connections (DashMap<ProcessId, IpcConnection>), shutdown_signal (Arc<AtomicBool>). Define **ProcessSubscription** struct with process_id and ipc_sender (mpsc::Sender<EventEnvelope>) fields. Implement **EventBusBroker::new()** constructor initializing ACL with default rules. Implement **subscribe_from_process()** async method: validate ACL permissions, register subscription in DashMap, send acknowledgment via IPC. Implement **publish_from_process()** async method: validate ACL, match topic against all subscription patterns using TopicPattern::matches(), route EventEnvelope to matching subscriber processes via their IPC channels, handle send failures gracefully. Implement **unsubscribe_from_process()** for cleanup. Implement **start()** method to spawn IPC listener accepting connections from collector processes, similar to `InterprocessServer` in `daemoneye-lib/src/ipc/interprocess_transport.rs`. Implement **shutdown()** for graceful cleanup of all connections and subscriptions. Add metrics tracking (messages routed, subscriptions active, ACL violations). Follow async patterns from `collector-core/src/collector.rs` for lifecycle management.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- daemoneye-agent/src/ipc_client.rs
- daemoneye-lib/src/ipc/client.rs

Implement **EventBusClient** for collector processes to interact with broker. Define client struct with fields: ipc_connection (connection to broker via interprocess crate), process_id (ProcessId), codec (IpcCodec for message framing), subscriptions (local tracking of active subscriptions). Implement **EventBusClient::connect()** async method to establish IPC connection to broker endpoint (similar to `InterprocessClient::new()` in `daemoneye-lib/src/ipc/interprocess_transport.rs`), send initial handshake with process identity. Implement **subscribe()** async method: send SubscribeRequest protobuf message to broker, return EventStream (mpsc::Receiver<EventEnvelope>) for receiving events. Implement **publish()** async method: serialize event payload with prost, create EventEnvelope with topic and metadata, send PublishRequest to broker via IPC. Implement **unsubscribe()** for cleanup. Add automatic reconnection logic with exponential backoff similar to `IpcClientManager` in `daemoneye-agent/src/ipc_client.rs`. Implement background task to receive events from broker and dispatch to local subscribers. Add connection health monitoring and metrics. Follow error handling patterns from `daemoneye-lib/src/ipc/client.rs`.

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

References: 

- daemoneye-lib/src/ipc/codec.rs

Define error types for event bus operations using **thiserror** crate. Create **EventBusError** enum with variants: TopicParseError (invalid topic format), AclViolation (permission denied), ConnectionError (IPC failure), SerializationError (protobuf encoding/decoding), SubscriptionNotFound, BrokerShutdown, InvalidWildcard (# not at end), EmptySegment, and other domain-specific errors. Implement Display and Error traits. Add context-rich error messages for debugging cross-process issues. Follow error handling patterns from `daemoneye-lib/src/ipc/codec.rs` IpcError enum.

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

Define core data types for event bus. Create **ProcessId** type alias or newtype for process identification. Define **EventStream** type alias for mpsc::Receiver<EventEnvelope>. Define **SubscriptionId** type for tracking subscriptions. Add type aliases for clarity and future extensibility. Keep types simple and focused on cross-process communication semantics.

### daemoneye-lib/proto/eventbus.proto(NEW)

References: 

- daemoneye-lib/proto/ipc.proto
- daemoneye-lib/proto/common.proto

Define protobuf messages for event bus cross-process communication. Create **EventEnvelope** message with fields: topic (string), payload (bytes - serialized event data), timestamp (int64 - Unix timestamp in milliseconds), publisher_process_id (string), correlation_metadata (optional CorrelationMetadata). Create **CorrelationMetadata** message with fields: correlation_id (string - UUID), parent_correlation_id (optional string), root_correlation_id (string), sequence_number (uint64), workflow_stage (optional string), source_component (string), correlation_tags (map<string, string>). Create **SubscribeRequest** message with fields: subscription_id (string), topic_pattern (string - with wildcards), process_identity (ProcessIdentity). Create **SubscribeResponse** message with fields: subscription_id (string), success (bool), error_message (optional string). Create **PublishRequest** message with fields: envelope (EventEnvelope). Create **PublishResponse** message with fields: success (bool), error_message (optional string). Create **UnsubscribeRequest** message with fields: subscription_id (string). Create **ProcessIdentity** message with fields: process_id (string), process_type (ProcessType enum - AGENT, COLLECTOR). Follow protobuf conventions from `daemoneye-lib/proto/ipc.proto` and `daemoneye-lib/proto/common.proto`.

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

References: 

- daemoneye-lib/proto/ipc.proto

Add **proto/eventbus.proto** to the list of protobuf files compiled by prost-build. Ensure the new EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, and ProcessIdentity messages are generated. Follow the existing pattern for compiling proto files and generating Rust code in the OUT_DIR.

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

Re-export the new event bus protobuf types from the generated code. Add pub use statements for **EventEnvelope**, **CorrelationMetadata**, **SubscribeRequest**, **SubscribeResponse**, **PublishRequest**, **PublishResponse**, **UnsubscribeRequest**, **ProcessIdentity**, and **ProcessType** enum. Add helper methods to EventEnvelope for creating envelopes with correlation metadata, extracting topic, and validating structure. Follow the pattern of helper methods on DetectionTask and DetectionResult in this file.

### daemoneye-eventbus/tests(NEW)

Create tests directory for integration and unit tests.

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

References: 

- collector-core/tests/integration_test.rs

Create comprehensive unit tests for topic matching engine. Test exact topic matches ("events.process.lifecycle" matches "events.process.lifecycle"). Test single-level wildcard: "events.process.+" matches "events.process.lifecycle" but NOT "events.process.tree.update", "events.+.lifecycle" matches "events.process.lifecycle" and "events.network.lifecycle". Test multi-level wildcard: "events.#" matches all event topics, "events.process.#" matches all process events including nested ones. Test invalid patterns: "events.#.invalid" (# not at end), "events..lifecycle" (empty segment), "" (empty topic). Test edge cases: single segment topics, deeply nested topics (10+ levels), topics with special characters. Test performance with 1000+ topic patterns. Follow test patterns from `collector-core/tests/` for structure and assertions.

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

References: 

- daemoneye-lib/tests/ipc_security_validation.rs

Create unit tests for ACL enforcement logic. Test that collector processes can publish to "events.*" topics but NOT "control.*" topics. Test that daemoneye-agent process can publish to "control.*" topics but collectors cannot. Test that collectors can subscribe to "control.collector.*" and "control.health.*" topics. Test that agent can subscribe to "events.*" topics. Test that "admin.*" topics are restricted to agent only. Test ACL violations return appropriate errors. Test default ACL rules are correctly initialized. Follow security test patterns from `daemoneye-lib/tests/ipc_security_validation.rs`.

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

References: 

- collector-core/tests/integration_test.rs
- daemoneye-lib/tests/ipc_integration.rs

Create integration tests for EventBusBroker with multiple collector processes. Test broker startup and IPC listener initialization. Test multiple collector processes connecting to broker. Test subscription registration from different processes. Test event publishing and routing to matching subscribers across process boundaries. Test wildcard subscriptions receiving appropriate events. Test subscription cleanup when collector process disconnects. Test concurrent publish/subscribe operations from multiple processes. Test broker shutdown and cleanup of all resources. Use tempfile crate for test socket paths similar to `collector-core/tests/integration_test.rs`. Spawn mock collector processes using tokio::spawn. Follow integration test patterns from `daemoneye-lib/tests/ipc_integration.rs` for multi-process coordination.

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

References: 

- daemoneye-lib/tests/resilient_client_integration.rs

Create integration tests for EventBusClient. Test client connection to broker. Test subscribe() method and receiving events via EventStream. Test publish() method sending events to broker. Test multiple clients subscribing to same topic pattern. Test client reconnection after broker restart or connection failure. Test correlation metadata propagation through publish/subscribe flow. Test client cleanup and unsubscribe on shutdown. Follow client testing patterns from `daemoneye-lib/tests/resilient_client_integration.rs` for connection management and error scenarios.

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

References: 

- daemoneye-lib/tests/ipc_cross_platform_tests.rs

Create cross-platform validation tests for Windows (named pipes) and Unix (domain sockets). Test broker and client work correctly on both platforms. Test socket path generation and cleanup. Test platform-specific IPC error handling. Test concurrent connections on both platforms. Follow cross-platform test patterns from `daemoneye-lib/tests/ipc_cross_platform_tests.rs` with conditional compilation for platform-specific code.

### daemoneye-eventbus/benches(NEW)

Create benchmarks directory for performance testing.

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

References: 

- daemoneye-lib/benches/ipc_performance_comprehensive.rs

Create performance benchmarks using criterion crate. Benchmark topic matching performance with various pattern complexities (exact, single wildcard, multi-level wildcard) against 1000+ topics. Benchmark broker routing latency for single subscriber, multiple subscribers (10+), and wildcard subscriptions. Benchmark cross-process IPC throughput: messages/second with 1 publisher + 1 subscriber, 1 publisher + 10 subscribers, 10 publishers + 10 subscribers. Benchmark subscription management: add/remove subscriptions, lookup performance with 1000+ active subscriptions. Target: <1ms p99 latency for local IPC routing. Follow benchmark patterns from `daemoneye-lib/benches/ipc_performance_comprehensive.rs` using criterion framework and black_box for preventing optimization.

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

References: 

- daemoneye-agent/src/main.rs(MODIFY)
- daemoneye-lib/src/telemetry.rs

Create integration module for hosting EventBusBroker in daemoneye-agent process. Define **EventBusService** struct wrapping EventBusBroker with lifecycle management. Implement **start_broker()** function to initialize broker with default ACL rules, start IPC listener on dedicated endpoint (e.g., /var/run/daemoneye/eventbus.sock or \\.\pipe\daemoneye\eventbus), and spawn broker runtime task. Implement **shutdown_broker()** for graceful cleanup. Add broker health monitoring and metrics integration with existing telemetry system in `daemoneye-lib/src/telemetry.rs`. Provide API for agent components to subscribe to topics (e.g., detection engine subscribing to "events.process.#"). Follow service initialization patterns from `daemoneye-agent/src/main.rs` main loop structure.

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

References: 

- daemoneye-agent/src/ipc_client.rs

Integrate EventBusBroker into daemoneye-agent startup sequence. Add module declaration for **eventbus_integration**. In the run() function after IPC client initialization (around line 64), add broker initialization: create EventBusService, start broker with start_broker(), store broker handle for shutdown. In the shutdown section (after line 203), add broker.shutdown() call before final cleanup. Add broker health monitoring to the main loop alongside existing telemetry health checks. Consider subscribing agent to "events.process.#" topic to receive process events from collectors as an alternative/complement to the current IPC polling approach. Maintain backward compatibility with existing IPC client for gradual migration.

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

References: 

- collector-core/Cargo.toml(MODIFY)

Add **daemoneye-eventbus** as a dependency to enable broker integration. Add it to the dependencies section with path = "../daemoneye-eventbus" similar to how **collector-core** and **daemoneye-lib** are included.

### collector-core/src/eventbus_client_integration.rs(NEW)

References: 

- collector-core/src/ipc.rs
- collector-core/src/event.rs

Create optional integration module for collector-core to use EventBusClient. Define **CollectorEventBusAdapter** struct that wraps EventBusClient and provides methods for collectors to publish events to the broker. Implement **publish_process_event()** method to publish ProcessEvent to "events.process.lifecycle" or other appropriate topics based on event type. Implement **subscribe_to_tasks()** method to subscribe to "control.collector.task.*" topics for receiving tasks from agent. Add helper methods to convert between CollectionEvent (from `collector-core/src/event.rs`) and EventEnvelope payloads. Make this integration optional/feature-gated to maintain backward compatibility with direct IPC. Follow integration patterns from `collector-core/src/ipc.rs` for IPC server integration.

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

Add optional module declaration for **eventbus_client_integration** (conditionally compiled with feature flag). Re-export CollectorEventBusAdapter if the eventbus feature is enabled. This allows collectors to optionally use the event bus without breaking existing direct IPC functionality.

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

Add optional **daemoneye-eventbus** dependency with path = "../daemoneye-eventbus" and optional = true. Add new feature flag **eventbus** that enables the daemoneye-eventbus dependency. This allows collectors to opt-in to event bus functionality while maintaining backward compatibility with direct IPC.

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

References: 

- collector-core/README.md

Create comprehensive documentation for daemoneye-eventbus crate. Document the cross-process pub/sub architecture with process diagrams. Explain MQTT-style topic hierarchy with examples: events.process.lifecycle, events.process.metadata, control.collector.task.process, control.health.heartbeat. Document wildcard matching rules with examples of + and # usage. Explain topic-based ACL and security boundaries between processes. Provide usage examples for broker (in daemoneye-agent) and client (in collectors). Include code examples for subscribing, publishing, and handling events. Document performance characteristics and benchmarks. Explain integration with existing IPC infrastructure. Add troubleshooting section for common cross-process issues. Follow documentation style from `collector-core/README.md`.

### docs/src/technical/eventbus-architecture.md(NEW)

References: 

- docs/src/technical/ipc-implementation.md

Create technical documentation for event bus architecture in the mdBook documentation. Document the complete topic hierarchy with all namespaces: events.process.* (lifecycle, metadata, tree, integrity, anomaly, batch), events.network.*, events.filesystem.*, events.performance.*, control.collector.* (lifecycle, config, task.*), control.health.* (heartbeat, status, diagnostics). Explain cross-process message flow with sequence diagrams. Document ACL rules and security model. Provide deployment guidance for multi-collector coordination. Include performance tuning recommendations. Add examples of common pub/sub patterns for multi-collector scenarios. Follow documentation structure from `docs/src/technical/ipc-implementation.md`.

### docs/src/SUMMARY.md(MODIFY)

Add link to the new eventbus architecture documentation under the Technical section. Add entry: "- [Event Bus Architecture](technical/eventbus-architecture.md)" after the IPC Implementation entry to maintain logical flow of technical documentation.

---

Developer Humor

🎭 A broker walked into a pub/sub bar,\n> Wildcards matching topics near and far.\n> \Events dot hash!\ cried the collector with glee,\n> \I'll subscribe to them all, just route them to me!\n> \n> The agent said \Control dot star is mine,\n> \Your ACL says 'ReadOnly' - that's by design!\n> With topics and patterns, they danced through the night,\n> Cross-process IPC never felt so right! 🎪

---

Execution Information

Branch: main
Commit: a169b68

New Plan

Observations

The codebase features robust IPC infrastructure in daemoneye-lib/src/ipc/ using interprocess for cross-process communication, protobuf codec, and connection management. Collector-core handles events via event.rs and IPC server in ipc.rs. Daemoneye-agent uses ipc_client.rs for polling. No daemoneye-eventbus crate exists, and related issues #113/#121 show no implementation despite "completed" status. Protobuf schema in proto/ lacks event bus messages. Architecture emphasizes separate OS processes for agent and collectors with Unix sockets/named pipes.

Approach

Develop daemoneye-eventbus as a new workspace crate providing cross-process pub/sub with MQTT-style topics. Host EventBusBroker in daemoneye-agent on dedicated IPC endpoint. Collectors use EventBusClient for publish/subscribe. Extend protobuf with EventEnvelope and control messages. Implement wildcard matching engine and topic ACL for security. Ensure <1ms routing latency via efficient DashMap routing and prost serialization. Integrate optionally in collector-core and agent for gradual adoption.

Reasoning

I examined the workspace Cargo.toml for members and dependencies, explored daemoneye-lib/src/ipc for interprocess transport and codec patterns, reviewed proto files for existing messages, analyzed collector-core/src for event and IPC handling, inspected daemoneye-agent/src/main.rs for startup flow, and confirmed no eventbus code exists across the repository.

Proposed File Changes

📄 Cargo.toml ^(MODIFY) 🔗

Add daemoneye-eventbus to the workspace members array after daemoneye-lib. Add dashmap =

📁 daemoneye-eventbus ^(NEW) 🔗

Create new directory for the daemoneye-eventbus workspace crate containing broker, client, topic engine, and ACL implementations.

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

Initialize Cargo.toml for daemoneye-eventbus crate. Set package name, version from workspace, edition 2021. Add dependencies: tokio (workspace), prost (workspace), interprocess (workspace), dashmap (workspace), anyhow (workspace), thiserror (workspace), tracing (workspace), uuid (workspace), async-trait (workspace), chrono (workspace for timestamps). Enable lib crate-type. Inherit workspace lints.

📁 daemoneye-eventbus/src ^(NEW) 🔗

Create src directory for daemoneye-eventbus source code.

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

Create lib.rs with public mod declarations: pub mod broker; pub mod client; pub mod topic; pub mod acl; pub mod error; pub mod types; pub mod proto; Re-export key types: pub use broker::EventBusBroker; pub use client::EventBusClient; pub use topic::{TopicPattern, TopicSegment}; pub use acl::{AccessControl, TopicAcl}; pub use error::EventBusError; pub use types::*; Add crate-level docs explaining cross-process pub/sub architecture, MQTT topics, and integration with existing IPC.

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

References

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

Implement TopicPattern and TopicSegment for MQTT-style hierarchy. TopicPattern holds Vec. Parse topics like "events.process.lifecycle" into segments, handling / as separator, escaping. TopicSegment: Literal(String), SingleLevelWildcard, MultiLevelWildcard. Implement parse_from_str with validation (no consecutive /, # only at end). Implement matches method: recursive segment matching, + matches one level, # matches remainder. Add tests for exact, +, # patterns, invalid inputs. Ensure <100μs matching time.

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

Define AccessControl enum: ReadOnly, WriteOnly, ReadWrite, Restricted. TopicAcl struct with HashMap<TopicPattern, AccessControl> and default_fallback. Implement new with defaults: events.# -> ReadOnly (collectors publish, agent reads), control.# -> WriteOnly (agent publishes, collectors read), admin.# -> Restricted (agent only). Methods: can_publish(process_type: ProcessType, topic: &str) -> bool, can_subscribe(process_type, pattern: &str) -> bool. ProcessType enum: Agent, Collector. Use topic.matches for permission checks. Tests for namespace permissions.

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

References

• 📄 daemoneye-lib/src/ipc/interprocess_transport.rs
• 📄 daemoneye-lib/src/ipc/codec.rs

Implement EventBusBroker struct: subscriptions: DashMap<TopicPattern, Vec>, acl: TopicAcl, connections: DashMap<ProcessId, Arc<Mutex>>, runtime: tokio::runtime::Handle. ProcessSubscription { process_id: ProcessId, sender: mpsc::UnboundedSender }. new() initializes acl. async start(endpoint: &str) spawns IPC server using interprocess::local_socket::LocalSocketListener, accepts connections, authenticates process_type. subscribe(process_id, pattern, sender) validates acl, inserts to dashmap. publish(process_id, envelope) validates acl, iterates subscriptions, matches topic, sends to matching senders. handle_connection loop deserializes requests (SubscribeRequest, PublishRequest), dispatches. shutdown() closes listener, drains connections. Integrate tracing for metrics.

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

References

• 📄 daemoneye-lib/src/ipc/client.rs
• 📄 daemoneye-agent/src/ipc_client.rs

Implement EventBusClient struct: connection: IpcConnection, process_id: ProcessId, process_type: ProcessType, codec: IpcCodec. async connect(endpoint: &str, process_type) establishes LocalSocketStream, sends handshake ProcessIdentity, receives ack. async subscribe(pattern: &str) -> mpsc::Receiver sends SubscribeRequest, creates local channel, spawns receive task deserializing EventEnvelope from stream, forwards to receiver. async publish(topic: &str, payload: Vec) creates EventEnvelope, serializes PublishRequest, sends via codec. async unsubscribe(sub_id). Handle disconnects with retry (exponential backoff). Use patterns from daemoneye-lib/src/ipc/client.rs.

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

References

• 📄 daemoneye-lib/src/ipc/codec.rs

#[derive(thiserror::Error, Debug)] pub enum EventBusError { #[error("IPC connection failed: {0}")] Connection(anyhow::Error), #[error("ACL violation for topic '{topic}' by process {process}")] AclViolation { topic: String, process: ProcessId }, #[error("Invalid topic pattern: {0}")] InvalidTopic(String), #[error("Subscription not found: {0}")] NoSubscription(ProcessId), #[error("Serialization error: {0}")] Serde(prost::EncodeError), #[error("Deserialization error: {0}")] De( prost::DecodeError), #[error("Wildcard mismatch")] WildcardError, #[error("Process authentication failed")] AuthFailed, } Implement From for anyhow::Error, prost errors. Context-aware messages for cross-process debugging.

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

References

• 📄 daemoneye-lib/proto/common.proto

pub type ProcessId = uuid::Uuid; pub type SubscriptionId = u64; pub type EventStream = mpsc::Receiver; #[derive(Debug, Clone, PartialEq)] pub enum ProcessType { Agent, Collector, } #[derive(Debug, Clone)] pub struct ProcessIdentity { pub process_id: ProcessId, pub process_type: ProcessType, pub name: Option, } Add derive for Serialize/Deserialize if needed for future extensions.

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

References

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

syntax = "proto3"; package daemoneye.eventbus; import "common.proto"; message EventEnvelope { string topic = 1; bytes payload = 2; int64 timestamp_ms = 3; string publisher_id = 4; CorrelationMetadata correlation = 5; } message CorrelationMetadata { string correlation_id = 1; optional string parent_id = 2; string root_id = 3; uint64 sequence = 4; optional string stage = 5; string source = 6; map<string, string> tags = 7; } message SubscribeRequest { string id = 1; string pattern = 2; ProcessIdentity identity = 3; } message SubscribeResponse { string id = 1; bool success = 2; optional string error = 3; } message PublishRequest { EventEnvelope envelope = 1; } message PublishResponse { bool success = 1; optional string error = 2; } message UnsubscribeRequest { string id = 1; } message HandshakeRequest { ProcessIdentity identity = 1; } message HandshakeResponse { bool accepted = 1; optional string error = 2; } message ProcessIdentity { string id = 1; ProcessType type = 2; optional string name = 3; } enum ProcessType { AGENT = 0; COLLECTOR = 1; } Follow existing proto style from ipc.proto.

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

In prost_build::Config, add .file("proto/eventbus.proto") to compile_protos. Ensure generated code includes EventEnvelope, etc. Output to OUT_DIR as before. Add include!("$OUT_DIR/eventbus.rs"); in proto.rs if needed.

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

Add pub use statements for eventbus types: pub use prost::eventbus::{EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, HandshakeRequest, HandshakeResponse, ProcessIdentity, ProcessType}; Add impl helpers, e.g., impl EventEnvelope { pub fn new(topic: &str, payload: Vec) -> Self { ... } } Similar to existing ipc types.

📁 daemoneye-eventbus/tests ^(NEW) 🔗

Create tests directory for unit and integration tests.

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

#[cfg(test)] mod topic_tests { use super::*; #[test] fn exact_match() { assert!(TopicPattern::parse("events.process.lifecycle").unwrap().matches("events.process.lifecycle")); } #[test] fn single_wildcard() { let pat = TopicPattern::parse("events.process.+").unwrap(); assert!(pat.matches("events.process.lifecycle")); assert!(!pat.matches("events.process.tree.update")); } #[test] fn multi_wildcard() { let pat = TopicPattern::parse("events.#").unwrap(); assert!(pat.matches("events.process.anomaly")); } #[test] fn invalid_pattern() { assert!(TopicPattern::parse("events.#.invalid").is_err()); } // Add 20+ tests for edge cases, performance. }

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

#[cfg(test)] mod acl_tests { use super::*; let acl = TopicAcl::default(); #[test] fn collector_publish_events() { assert!(acl.can_publish(ProcessType::Collector, "events.process.lifecycle")); assert!(!acl.can_publish(ProcessType::Collector, "control.collector.task")); } #[test] fn agent_subscribe_events() { assert!(acl.can_subscribe(ProcessType::Agent, "events.#")); } // Tests for restricted, fallbacks. }

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

References

• 📄 collector-core/tests/integration_test.rs

#[tokio::test] async fn multi_process_pubsub() { use tempfile::tempdir; let dir = tempdir().unwrap(); let endpoint = format!("{}eventbus.sock", dir.path().display()); // Spawn broker task let (broker_tx, mut broker_rx) = mpsc::channel(1); tokio::spawn(async move { let mut broker = EventBusBroker::new(); broker.start(&endpoint).await.unwrap(); }); // Spawn mock collector tokio::spawn(async move { let client = EventBusClient::connect(&endpoint, ProcessType::Collector).await.unwrap(); let mut stream = client.subscribe("control.#").await.unwrap(); client.publish("events.process.spawn", vec![]).await.unwrap(); // Wait and assert receipt }); // Assert events routed. Use tokio::time::sleep. } // Add tests for disconnect, reconnect, multiple clients.

📁 daemoneye-eventbus/benches ^(NEW) 🔗

Create benches directory for criterion benchmarks.

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

References

• 📄 daemoneye-lib/benches

use criterion::{criterion_group, criterion_main, Criterion}; fn bench_matching(c: &mut Criterion) { let patterns = vec!["events.process.+".to_string(), "events.#".to_string()]; c.bench_function("wildcard_match", |b| { b.iter(|| { for pat in &patterns { let _ = TopicPattern::parse(pat).unwrap().matches("events.process.lifecycle.anomaly"); } }) }); } criterion_group!(benches, bench_matching); criterion_main!(benches); // Benchmarks for routing, serialization. Target <1ms p99.

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

References

• 📄 daemoneye-lib/src/telemetry.rs

pub struct AgentEventBus { broker: EventBusBroker, handle: JoinHandle<()>, } impl AgentEventBus { pub async fn start(endpoint: &str) -> Result { let mut broker = EventBusBroker::new(); let handle = tokio::spawn(async move { broker.start(endpoint).await }); // Subscribe agent to events.# Ok(Self { broker, handle }) } pub async fn publish(&self, envelope: EventEnvelope) -> Result<()> { self.broker.publish(ProcessId::nil(), &envelope).await } pub async fn subscribe(&mut self, pattern: &str) -> EventStream { // Internal subscription since broker is local } pub async fn shutdown(self) { self.broker.shutdown().await; self.handle.abort(); } } Add tracing integration with daemoneye-lib/src/telemetry.rs.

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

References

• 📄 daemoneye-agent/src/ipc_client.rs

Add mod eventbus; In async fn run(), after let mut ipc_client = IpcClientManager::new(...).await?, add: let endpoint = env::var("DAEMON_EYE_EVENTBUS_ENDPOINT").unwrap_or_else(|_| "/tmp/daemoneye-eventbus.sock".to_string()); let mut eventbus = AgentEventBus::start(&endpoint).await?; // Subscribe to events broker.subscribe("events.#").await; In shutdown: eventbus.shutdown().await; Ensure graceful handling.

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

Add daemoneye-eventbus = { path = "../daemoneye-eventbus" } to dependencies.

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

References

• 📄 collector-core/src/event.rs
• 📄 collector-core/src/ipc.rs

#[cfg(feature = "eventbus")] pub struct CollectorEventBus { client: EventBusClient, } impl CollectorEventBus { pub async fn connect(endpoint: &str) -> Result { let client = EventBusClient::connect(endpoint, ProcessType::Collector).await?; // Subscribe to control.collector.# let mut stream = client.subscribe("control.collector.#").await?; // Spawn handler task to process control messages Ok(Self { client }) } pub async fn publish_event(&self, event: &CollectionEvent) -> Result<()> { let topic = match event { CollectionEvent::ProcessSpawn(_) => "events.process.lifecycle", // etc. }; let payload = bincode::serialize(event)?; // Or prost if needed self.client.publish(topic, payload).await } }

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

#[cfg(feature = "eventbus")] pub mod eventbus; #[cfg(feature = "eventbus")] pub use eventbus::CollectorEventBus; Add to existing mod declarations.

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

Add [features] default = [] eventbus = ["daemoneye-eventbus"] Add daemoneye-eventbus = { path = "../daemoneye-eventbus", optional = true } to dependencies.

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

References

• 📄 collector-core/README.md

daemoneye-eventbus

Cross-process pub/sub event bus for DaemonEye multi-collector coordination.

Architecture

• Broker hosted in daemoneye-agent
• Clients in separate collector processes
• IPC via interprocess (sockets/pipes)
• MQTT-style topics with + and # wildcards

Topics

• events.process.*: Process events from procmond
• control.collector.*: Tasks/config from agent
• control.health.*: Heartbeats/status

Usage

let broker = EventBusBroker::new();
broker.start("/tmp/eventbus.sock").await;

let client = EventBusClient::connect("/tmp/eventbus.sock", ProcessType::Collector).await;
let mut stream = client.subscribe("events.#").await;

See docs for ACL, performance.

📄 docs/src/technical/eventbus.md ^(NEW) 🔗

References

• 📄 docs/src/technical/ipc-implementation.md

Event Bus Architecture

Topic Hierarchy

Events

• events.process.lifecycle: Process start/stop
• events.process.metadata: Resource usage
• etc.

Control

• control.collector.task.process: Detection tasks

Wildcards

• events.process.+: Specific subtopic
• events.#: All events

Security

ACL enforces: collectors publish events., subscribe control.; agent full access.

Cross-Process Flow

See sequence diagram in plan.

Performance

<1ms latency target via DashMap routing.

📄 docs/src/SUMMARY.md ^(MODIFY) 🔗

Under Technical Documentation, add - Event Bus after IPC Implementation.

Diagram

sequenceDiagram
    participant A as daemoneye-agent (Broker)
    participant C1 as procmond (Client)
    participant C2 as netmond (Client)
    
    A->>A: EventBusBroker::start(\"/tmp/eventbus.sock\")
    
    C1->>A: connect() via IPC
    A->>C1: accept, handshake
    C1->>A: subscribe(\"control.collector.task.process\")
    A->>A: validate ACL, register subscription
    
    C2->>A: connect() via IPC
    C2->>A: subscribe(\"events.process.#\")
    A->>A: register subscription
    
    A->>C1: publish(\"control.collector.task.process\", task)
    A->>A: match subscriptions, route to C1
    A-->>C1: EventEnvelope via IPC
    
    C1->>A: publish(\"events.process.lifecycle\", event)
    A->>A: match, route to C2
    A-->>C2: EventEnvelope via IPC
    
    Note over A,C2: Wildcard matching: events.process.# receives lifecycle

Loading

Import In IDE

🤖 Prompt for AI Agents

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

### Observations

The codebase features robust IPC infrastructure in `daemoneye-lib/src/ipc/` using `interprocess` for cross-process communication, protobuf codec, and connection management. Collector-core handles events via `event.rs` and IPC server in `ipc.rs`. Daemoneye-agent uses `ipc_client.rs` for polling. No `daemoneye-eventbus` crate exists, and related issues #113/#121 show no implementation despite "completed" status. Protobuf schema in `proto/` lacks event bus messages. Architecture emphasizes separate OS processes for agent and collectors with Unix sockets/named pipes.

### Approach

Develop **daemoneye-eventbus** as a new workspace crate providing cross-process pub/sub with MQTT-style topics. Host **EventBusBroker** in **daemoneye-agent** on dedicated IPC endpoint. Collectors use **EventBusClient** for publish/subscribe. Extend protobuf with **EventEnvelope** and control messages. Implement wildcard matching engine and topic ACL for security. Ensure <1ms routing latency via efficient DashMap routing and prost serialization. Integrate optionally in collector-core and agent for gradual adoption.

### Reasoning

I examined the workspace Cargo.toml for members and dependencies, explored daemoneye-lib/src/ipc for interprocess transport and codec patterns, reviewed proto files for existing messages, analyzed collector-core/src for event and IPC handling, inspected daemoneye-agent/src/main.rs for startup flow, and confirmed no eventbus code exists across the repository.

## Mermaid Diagram

sequenceDiagram
    participant A as daemoneye-agent (Broker)
    participant C1 as procmond (Client)
    participant C2 as netmond (Client)
    
    A->>A: EventBusBroker::start(\"/tmp/eventbus.sock\")
    
    C1->>A: connect() via IPC
    A->>C1: accept, handshake
    C1->>A: subscribe(\"control.collector.task.process\")
    A->>A: validate ACL, register subscription
    
    C2->>A: connect() via IPC
    C2->>A: subscribe(\"events.process.#\")
    A->>A: register subscription
    
    A->>C1: publish(\"control.collector.task.process\", task)
    A->>A: match subscriptions, route to C1
    A-->>C1: EventEnvelope via IPC
    
    C1->>A: publish(\"events.process.lifecycle\", event)
    A->>A: match, route to C2
    A-->>C2: EventEnvelope via IPC
    
    Note over A,C2: Wildcard matching: events.process.# receives lifecycle

## Proposed File Changes

### Cargo.toml(MODIFY)

Add **daemoneye-eventbus** to the workspace members array after **daemoneye-lib**. Add **dashmap** =

### daemoneye-eventbus(NEW)

Create new directory for the daemoneye-eventbus workspace crate containing broker, client, topic engine, and ACL implementations.

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

Initialize Cargo.toml for daemoneye-eventbus crate. Set package name, version from workspace, edition 2021. Add dependencies: tokio (workspace), prost (workspace), interprocess (workspace), dashmap (workspace), anyhow (workspace), thiserror (workspace), tracing (workspace), uuid (workspace), async-trait (workspace), chrono (workspace for timestamps). Enable lib crate-type. Inherit workspace lints.

### daemoneye-eventbus/src(NEW)

Create src directory for daemoneye-eventbus source code.

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

Create lib.rs with public mod declarations: pub mod broker; pub mod client; pub mod topic; pub mod acl; pub mod error; pub mod types; pub mod proto; Re-export key types: pub use broker::EventBusBroker; pub use client::EventBusClient; pub use topic::{TopicPattern, TopicSegment}; pub use acl::{AccessControl, TopicAcl}; pub use error::EventBusError; pub use types::*; Add crate-level docs explaining cross-process pub/sub architecture, MQTT topics, and integration with existing IPC.

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

References: 

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

Implement TopicPattern and TopicSegment for MQTT-style hierarchy. TopicPattern holds Vec<TopicSegment>. Parse topics like "events.process.lifecycle" into segments, handling / as separator, escaping. TopicSegment: Literal(String), SingleLevelWildcard, MultiLevelWildcard. Implement parse_from_str with validation (no consecutive /, # only at end). Implement matches method: recursive segment matching, + matches one level, # matches remainder. Add tests for exact, +, # patterns, invalid inputs. Ensure <100μs matching time.

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

Define AccessControl enum: ReadOnly, WriteOnly, ReadWrite, Restricted. TopicAcl struct with HashMap<TopicPattern, AccessControl> and default_fallback. Implement new with defaults: events.# -> ReadOnly (collectors publish, agent reads), control.# -> WriteOnly (agent publishes, collectors read), admin.# -> Restricted (agent only). Methods: can_publish(process_type: ProcessType, topic: &str) -> bool, can_subscribe(process_type, pattern: &str) -> bool. ProcessType enum: Agent, Collector. Use topic.matches for permission checks. Tests for namespace permissions.

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

References: 

- daemoneye-lib/src/ipc/interprocess_transport.rs
- daemoneye-lib/src/ipc/codec.rs

Implement EventBusBroker struct: subscriptions: DashMap<TopicPattern, Vec<ProcessSubscription>>, acl: TopicAcl, connections: DashMap<ProcessId, Arc<Mutex<IpcConnection>>>, runtime: tokio::runtime::Handle. ProcessSubscription { process_id: ProcessId, sender: mpsc::UnboundedSender<EventEnvelope> }. new() initializes acl. async start(endpoint: &str) spawns IPC server using interprocess::local_socket::LocalSocketListener, accepts connections, authenticates process_type. subscribe(process_id, pattern, sender) validates acl, inserts to dashmap. publish(process_id, envelope) validates acl, iterates subscriptions, matches topic, sends to matching senders. handle_connection loop deserializes requests (SubscribeRequest, PublishRequest), dispatches. shutdown() closes listener, drains connections. Integrate tracing for metrics.

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

References: 

- daemoneye-lib/src/ipc/client.rs
- daemoneye-agent/src/ipc_client.rs

Implement EventBusClient struct: connection: IpcConnection, process_id: ProcessId, process_type: ProcessType, codec: IpcCodec. async connect(endpoint: &str, process_type) establishes LocalSocketStream, sends handshake ProcessIdentity, receives ack. async subscribe(pattern: &str) -> mpsc::Receiver<EventEnvelope> sends SubscribeRequest, creates local channel, spawns receive task deserializing EventEnvelope from stream, forwards to receiver. async publish(topic: &str, payload: Vec<u8>) creates EventEnvelope, serializes PublishRequest, sends via codec. async unsubscribe(sub_id). Handle disconnects with retry (exponential backoff). Use patterns from daemoneye-lib/src/ipc/client.rs.

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

References: 

- daemoneye-lib/src/ipc/codec.rs

#[derive(thiserror::Error, Debug)] pub enum EventBusError { #[error("IPC connection failed: {0}")] Connection(anyhow::Error), #[error("ACL violation for topic '{topic}' by process {process}")] AclViolation { topic: String, process: ProcessId }, #[error("Invalid topic pattern: {0}")] InvalidTopic(String), #[error("Subscription not found: {0}")] NoSubscription(ProcessId), #[error("Serialization error: {0}")] Serde(prost::EncodeError), #[error("Deserialization error: {0}")] De( prost::DecodeError), #[error("Wildcard mismatch")] WildcardError, #[error("Process authentication failed")] AuthFailed, } Implement From for anyhow::Error, prost errors. Context-aware messages for cross-process debugging.

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

References: 

- daemoneye-lib/proto/common.proto

pub type ProcessId = uuid::Uuid; pub type SubscriptionId = u64; pub type EventStream = mpsc::Receiver<EventEnvelope>; #[derive(Debug, Clone, PartialEq)] pub enum ProcessType { Agent, Collector, } #[derive(Debug, Clone)] pub struct ProcessIdentity { pub process_id: ProcessId, pub process_type: ProcessType, pub name: Option<String>, } Add derive for Serialize/Deserialize if needed for future extensions.

### daemoneye-lib/proto/eventbus.proto(NEW)

References: 

- daemoneye-lib/proto/ipc.proto
- daemoneye-lib/proto/common.proto

syntax = "proto3"; package daemoneye.eventbus; import "common.proto"; message EventEnvelope { string topic = 1; bytes payload = 2; int64 timestamp_ms = 3; string publisher_id = 4; CorrelationMetadata correlation = 5; } message CorrelationMetadata { string correlation_id = 1; optional string parent_id = 2; string root_id = 3; uint64 sequence = 4; optional string stage = 5; string source = 6; map<string, string> tags = 7; } message SubscribeRequest { string id = 1; string pattern = 2; ProcessIdentity identity = 3; } message SubscribeResponse { string id = 1; bool success = 2; optional string error = 3; } message PublishRequest { EventEnvelope envelope = 1; } message PublishResponse { bool success = 1; optional string error = 2; } message UnsubscribeRequest { string id = 1; } message HandshakeRequest { ProcessIdentity identity = 1; } message HandshakeResponse { bool accepted = 1; optional string error = 2; } message ProcessIdentity { string id = 1; ProcessType type = 2; optional string name = 3; } enum ProcessType { AGENT = 0; COLLECTOR = 1; } Follow existing proto style from ipc.proto.

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

In prost_build::Config, add .file("proto/eventbus.proto") to compile_protos. Ensure generated code includes EventEnvelope, etc. Output to OUT_DIR as before. Add include!("$OUT_DIR/eventbus.rs"); in proto.rs if needed.

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

Add pub use statements for eventbus types: pub use prost::eventbus::{EventEnvelope, CorrelationMetadata, SubscribeRequest, SubscribeResponse, PublishRequest, PublishResponse, UnsubscribeRequest, HandshakeRequest, HandshakeResponse, ProcessIdentity, ProcessType}; Add impl helpers, e.g., impl EventEnvelope { pub fn new(topic: &str, payload: Vec<u8>) -> Self { ... } } Similar to existing ipc types.

### daemoneye-eventbus/tests(NEW)

Create tests directory for unit and integration tests.

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

#[cfg(test)] mod topic_tests { use super::*; #[test] fn exact_match() { assert!(TopicPattern::parse("events.process.lifecycle").unwrap().matches("events.process.lifecycle")); } #[test] fn single_wildcard() { let pat = TopicPattern::parse("events.process.+").unwrap(); assert!(pat.matches("events.process.lifecycle")); assert!(!pat.matches("events.process.tree.update")); } #[test] fn multi_wildcard() { let pat = TopicPattern::parse("events.#").unwrap(); assert!(pat.matches("events.process.anomaly")); } #[test] fn invalid_pattern() { assert!(TopicPattern::parse("events.#.invalid").is_err()); } // Add 20+ tests for edge cases, performance. }

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

#[cfg(test)] mod acl_tests { use super::*; let acl = TopicAcl::default(); #[test] fn collector_publish_events() { assert!(acl.can_publish(ProcessType::Collector, "events.process.lifecycle")); assert!(!acl.can_publish(ProcessType::Collector, "control.collector.task")); } #[test] fn agent_subscribe_events() { assert!(acl.can_subscribe(ProcessType::Agent, "events.#")); } // Tests for restricted, fallbacks. }

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

References: 

- collector-core/tests/integration_test.rs

#[tokio::test] async fn multi_process_pubsub() { use tempfile::tempdir; let dir = tempdir().unwrap(); let endpoint = format!("{}eventbus.sock", dir.path().display()); // Spawn broker task let (broker_tx, mut broker_rx) = mpsc::channel(1); tokio::spawn(async move { let mut broker = EventBusBroker::new(); broker.start(&endpoint).await.unwrap(); }); // Spawn mock collector tokio::spawn(async move { let client = EventBusClient::connect(&endpoint, ProcessType::Collector).await.unwrap(); let mut stream = client.subscribe("control.#").await.unwrap(); client.publish("events.process.spawn", vec![]).await.unwrap(); // Wait and assert receipt }); // Assert events routed. Use tokio::time::sleep. } // Add tests for disconnect, reconnect, multiple clients.

### daemoneye-eventbus/benches(NEW)

Create benches directory for criterion benchmarks.

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

References: 

- daemoneye-lib/benches

use criterion::{criterion_group, criterion_main, Criterion}; fn bench_matching(c: &mut Criterion) { let patterns = vec!["events.process.+".to_string(), "events.#".to_string()]; c.bench_function("wildcard_match", |b| { b.iter(|| { for pat in &patterns { let _ = TopicPattern::parse(pat).unwrap().matches("events.process.lifecycle.anomaly"); } }) }); } criterion_group!(benches, bench_matching); criterion_main!(benches); // Benchmarks for routing, serialization. Target <1ms p99.

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

References: 

- daemoneye-lib/src/telemetry.rs

pub struct AgentEventBus { broker: EventBusBroker, handle: JoinHandle<()>, } impl AgentEventBus { pub async fn start(endpoint: &str) -> Result<Self> { let mut broker = EventBusBroker::new(); let handle = tokio::spawn(async move { broker.start(endpoint).await }); // Subscribe agent to events.# Ok(Self { broker, handle }) } pub async fn publish(&self, envelope: EventEnvelope) -> Result<()> { self.broker.publish(ProcessId::nil(), &envelope).await } pub async fn subscribe(&mut self, pattern: &str) -> EventStream { // Internal subscription since broker is local } pub async fn shutdown(self) { self.broker.shutdown().await; self.handle.abort(); } } Add tracing integration with daemoneye-lib/src/telemetry.rs.

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

References: 

- daemoneye-agent/src/ipc_client.rs

Add mod eventbus; In async fn run(), after let mut ipc_client = IpcClientManager::new(...).await?, add: let endpoint = env::var("DAEMON_EYE_EVENTBUS_ENDPOINT").unwrap_or_else(|_| "/tmp/daemoneye-eventbus.sock".to_string()); let mut eventbus = AgentEventBus::start(&endpoint).await?; // Subscribe to events broker.subscribe("events.#").await; In shutdown: eventbus.shutdown().await; Ensure graceful handling.

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

Add daemoneye-eventbus = { path = "../daemoneye-eventbus" } to dependencies.

### collector-core/src/eventbus.rs(NEW)

References: 

- collector-core/src/event.rs
- collector-core/src/ipc.rs

#[cfg(feature = "eventbus")] pub struct CollectorEventBus { client: EventBusClient, } impl CollectorEventBus { pub async fn connect(endpoint: &str) -> Result<Self> { let client = EventBusClient::connect(endpoint, ProcessType::Collector).await?; // Subscribe to control.collector.# let mut stream = client.subscribe("control.collector.#").await?; // Spawn handler task to process control messages Ok(Self { client }) } pub async fn publish_event(&self, event: &CollectionEvent) -> Result<()> { let topic = match event { CollectionEvent::ProcessSpawn(_) => "events.process.lifecycle", // etc. }; let payload = bincode::serialize(event)?; // Or prost if needed self.client.publish(topic, payload).await } }

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

#[cfg(feature = "eventbus")] pub mod eventbus; #[cfg(feature = "eventbus")] pub use eventbus::CollectorEventBus; Add to existing mod declarations.

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

Add [features] default = [] eventbus = ["daemoneye-eventbus"] Add daemoneye-eventbus = { path = "../daemoneye-eventbus", optional = true } to dependencies.

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

References: 

- collector-core/README.md

# daemoneye-eventbus

Cross-process pub/sub event bus for DaemonEye multi-collector coordination.

## Architecture

- Broker hosted in daemoneye-agent
- Clients in separate collector processes
- IPC via interprocess (sockets/pipes)
- MQTT-style topics with + and # wildcards

## Topics

- `events.process.*`: Process events from procmond
- `control.collector.*`: Tasks/config from agent
- `control.health.*`: Heartbeats/status

## Usage

```rust
let broker = EventBusBroker::new();
broker.start("/tmp/eventbus.sock").await;

let client = EventBusClient::connect("/tmp/eventbus.sock", ProcessType::Collector).await;
let mut stream = client.subscribe("events.#").await;
```

See docs for ACL, performance.

### docs/src/technical/eventbus.md(NEW)

References: 

- docs/src/technical/ipc-implementation.md

# Event Bus Architecture

## Topic Hierarchy

### Events
- `events.process.lifecycle`: Process start/stop
- `events.process.metadata`: Resource usage
- etc.

### Control
- `control.collector.task.process`: Detection tasks

## Wildcards

- `events.process.+`: Specific subtopic
- `events.#`: All events

## Security

ACL enforces: collectors publish events.*, subscribe control.*; agent full access.

## Cross-Process Flow

See sequence diagram in plan.

## Performance

<1ms latency target via DashMap routing.

### docs/src/SUMMARY.md(MODIFY)

Under Technical Documentation, add - [Event Bus](technical/eventbus.md) after IPC Implementation.

---

Developer Humor

🐛 Why did the collector subscribe to \events.process.+?\n> Because it wanted to catch every spawn without the drama of full # wildcards!\n> The broker chuckled: \That's +hilarious, but remember ACL— no control topics for you!\n> In the IPC bar, topics flowed freely, but only with proper permissions. 🍻

---

Execution Information

Branch: main
Commit: a169b68