Add correlation metadata and workflow tracking for multi-collector coordination

[unclesp1d3r]

Add correlation metadata and multi-collector workflow support across processes

Parent Epic: #114 - Multi-Process Collector Coordination via Event Bus Architecture
Related Issues: #115, #116, #117
Specification: .kiro/specs/daemoneye-core-monitoring/tasks.md Section 2.6.8

Design Goals (from Parent #114)

Scope: Single-system correlation tracking (NOT distributed tracing)

1. Performance: Correlation overhead <5% of base local IPC latency
2. Cross-Platform: Windows, Linux, macOS, FreeBSD
3. Security: Local process tracking only (no network exposure)
4. Code Safety: Zero unsafe code in correlation logic
5. Dependencies: Only use interprocess, tokio, prost, dashmap

Critical Architectural Context

Cross-process correlation tracking:

• daemoneye-agent (separate process): hosts broker and correlates events from local processes
• Collector processes (procmond, netmond, etc.): separate OS processes on same machine publishing events via IPC
• Correlation metadata tracks events across local process boundaries
• EventBus envelopes carry correlation data through local cross-process IPC

This is for local process coordination, NOT distributed tracing across networks

Overview

This task implements correlation metadata and multi-collector workflow support within daemoneye-eventbus, enabling the agent to track and correlate events across multiple local collector processes for complex detection workflows and forensic investigations.

Correlation Architecture

Correlation Metadata (Carried Across Local IPC)

/// Correlation metadata for tracking workflows across local processes
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CorrelationMetadata {
    /// Unique workflow identifier
    pub correlation_id: Uuid,
    
    /// Parent correlation for hierarchical workflows
    pub parent_correlation_id: Option<Uuid>,
    
    /// Root workflow identifier
    pub root_correlation_id: Uuid,
    
    /// Sequence number within workflow
    pub sequence_number: u64,
    
    /// Current workflow stage
    pub workflow_stage: String,
    
    /// Originating local collector process
    pub source_process: String,
    
    /// Flexible tagging for forensic correlation
    pub correlation_tags: HashMap<String, String>,
    
    /// Timestamp when correlation started
    pub started_at: SystemTime,
}

EventBus Envelope (Transmitted via Local IPC)

/// Event envelope carrying payload and correlation metadata via local IPC
#[derive(Debug, Serialize, Deserialize)]
pub struct EventEnvelope {
    /// Topic this event was published to
    pub topic: String,
    
    /// Serialized event payload
    pub payload: Vec<u8>,
    
    /// Timestamp when published
    pub timestamp: SystemTime,
    
    /// Local publisher process ID
    pub publisher_process_id: ProcessId,
    
    /// Correlation metadata (optional)
    pub correlation_metadata: Option<CorrelationMetadata>,
}

Correlation Tracker (In Broker Process)

/// Tracks correlation workflows across local collector processes
pub struct CorrelationTracker {
    // Active workflows indexed by correlation ID
    active_workflows: Arc<DashMap<Uuid, WorkflowState>>,
    
    // Event history for forensic analysis
    event_history: Arc<RwLock<VecDeque<CorrelatedEvent>>>,
    
    // Maximum history size
    max_history_size: usize,
}

pub struct WorkflowState {
    correlation_id: Uuid,
    started_at: Instant,
    stages: HashMap<String, StageInfo>,
    participating_processes: HashSet<ProcessId>,
    total_events: usize,
}

pub struct StageInfo {
    stage_name: String,
    started_at: Instant,
    completed_at: Option<Instant>,
    events_count: usize,
}

pub struct CorrelatedEvent {
    correlation_id: Uuid,
    event: EventEnvelope,
    received_at: Instant,
}

Workflow Scenarios

1. Simple Single-Stage Workflow (Across Local Processes)

// Agent initiates workflow
let correlation_id = Uuid::new_v4();
let metadata = CorrelationMetadata {
    correlation_id,
    root_correlation_id: correlation_id,
    workflow_stage: "enumeration".to_string(),
    source_process: "daemoneye-agent".to_string(),
    ..Default::default()
};

// Publish task with correlation to local collectors
broker.publish_with_correlation(
    "control.collector.task.process",
    task,
    metadata.clone()
).await?;

// Collector processes respond with same correlation_id
// Results automatically grouped by correlation tracker

2. Multi-Stage Workflow (Across Local Processes)

// Stage 1: Enumeration (local procmond processes)
let root_id = Uuid::new_v4();
let stage1 = CorrelationMetadata {
    correlation_id: Uuid::new_v4(),
    root_correlation_id: root_id,
    workflow_stage: "enumeration".to_string(),
    ..Default::default()
};

broker.publish_with_correlation("control.collector.task.process", enum_task, stage1).await?;

// Wait for stage 1 results from local processes
let enum_results = broker.wait_for_correlation_results(stage1.correlation_id).await?;

// Stage 2: Integrity checks on enumerated processes (local procmond)
let stage2 = CorrelationMetadata {
    correlation_id: Uuid::new_v4(),
    parent_correlation_id: Some(stage1.correlation_id),
    root_correlation_id: root_id,
    workflow_stage: "integrity".to_string(),
    correlation_tags: hashmap! {
        "parent_stage".to_string() => "enumeration".to_string(),
    },
    ..Default::default()
};

broker.publish_with_correlation("control.collector.task.process", integrity_task, stage2).await?;

// Stage 3: Network connections from suspicious processes (local netmond)
let stage3 = CorrelationMetadata {
    correlation_id: Uuid::new_v4(),
    parent_correlation_id: Some(stage2.correlation_id),
    root_correlation_id: root_id,
    workflow_stage: "network_analysis".to_string(),
    ..Default::default()
};

broker.publish_with_correlation("control.collector.task.network", network_task, stage3).await?;

3. Forensic Correlation (Local Process Investigation)

// Tag events for forensic investigation of local processes
let forensic_metadata = CorrelationMetadata {
    correlation_id: incident_id,
    root_correlation_id: incident_id,
    workflow_stage: "forensic_collection".to_string(),
    correlation_tags: hashmap! {
        "investigation_id".to_string() => "INC-12345".to_string(),
        "suspicious_pid".to_string() => "4892".to_string(),
        "analyst".to_string() => "alice".to_string(),
    },
    ..Default::default()
};

// Collect all events related to PID 4892 from local collectors
broker.publish_with_correlation("control.collector.task.process", forensic_task, forensic_metadata).await?;

// Later: Query all events for this investigation from local event history
let related_events = correlation_tracker
    .find_events_by_tag("investigation_id", "INC-12345")
    .await?;

Implementation Steps

1. Correlation Metadata Propagation

impl EventBusBroker {
    /// Publish event with correlation metadata to local subscribers
    pub async fn publish_with_correlation(
        &self,
        topic: &str,
        payload: Vec<u8>,
        correlation: CorrelationMetadata,
    ) -> Result<()> {
        let envelope = EventEnvelope {
            topic: topic.to_string(),
            payload,
            timestamp: SystemTime::now(),
            publisher_process_id: self.process_id(),
            correlation_metadata: Some(correlation.clone()),
        };
        
        // Track workflow
        self.correlation_tracker.track_event(&envelope).await?;
        
        // Route to local subscribers
        self.route_envelope(envelope).await?;
        
        Ok(())
    }
}

2. Correlation Tracking

impl CorrelationTracker {
    /// Track event in correlation workflow
    pub async fn track_event(&self, envelope: &EventEnvelope) -> Result<()> {
        if let Some(correlation) = &envelope.correlation_metadata {
            // Update workflow state
            let mut workflow = self.active_workflows
                .entry(correlation.correlation_id)
                .or_insert_with(|| WorkflowState::new(correlation.correlation_id));
            
            workflow.total_events += 1;
            workflow.participating_processes.insert(envelope.publisher_process_id);
            
            // Track stage
            let stage = workflow.stages
                .entry(correlation.workflow_stage.clone())
                .or_insert_with(|| StageInfo::new(&correlation.workflow_stage));
            
            stage.events_count += 1;
            
            // Add to event history
            let mut history = self.event_history.write().await;
            history.push_back(CorrelatedEvent {
                correlation_id: correlation.correlation_id,
                event: envelope.clone(),
                received_at: Instant::now(),
            });
            
            // Trim history if needed
            if history.len() > self.max_history_size {
                history.pop_front();
            }
        }
        
        Ok(())
    }
    
    /// Wait for all results from a correlation workflow
    pub async fn wait_for_correlation_results(
        &self,
        correlation_id: Uuid,
        timeout: Duration,
    ) -> Result<Vec<EventEnvelope>> {
        let deadline = Instant::now() + timeout;
        let mut results = Vec::new();
        
        loop {
            // Check for completion
            if let Some(workflow) = self.active_workflows.get(&correlation_id) {
                if workflow.is_complete() {
                    // Collect all events for this correlation from local history
                    let history = self.event_history.read().await;
                    for event in history.iter() {
                        if event.correlation_id == correlation_id {
                            results.push(event.event.clone());
                        }
                    }
                    return Ok(results);
                }
            }
            
            // Check timeout
            if Instant::now() > deadline {
                return Err(Error::CorrelationTimeout(correlation_id));
            }
            
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    }
}

3. Forensic Query API

impl CorrelationTracker {
    /// Find events by correlation tag (for forensic investigations)
    pub async fn find_events_by_tag(
        &self,
        tag_key: &str,
        tag_value: &str,
    ) -> Result<Vec<EventEnvelope>> {
        let history = self.event_history.read().await;
        
        Ok(history
            .iter()
            .filter(|event| {
                event.event.correlation_metadata
                    .as_ref()
                    .and_then(|meta| meta.correlation_tags.get(tag_key))
                    .map(|v| v == tag_value)
                    .unwrap_or(false)
            })
            .map(|event| event.event.clone())
            .collect())
    }
    
    /// Get workflow timeline (for forensic analysis)
    pub async fn get_workflow_timeline(&self, root_correlation_id: Uuid) -> Result<WorkflowTimeline> {
        let history = self.event_history.read().await;
        
        let events: Vec<_> = history
            .iter()
            .filter(|event| {
                event.event.correlation_metadata
                    .as_ref()
                    .map(|meta| meta.root_correlation_id == root_correlation_id)
                    .unwrap_or(false)
            })
            .map(|event| TimelineEvent {
                timestamp: event.received_at,
                stage: event.event.correlation_metadata.as_ref().unwrap().workflow_stage.clone(),
                source_process: event.event.publisher_process_id,
                event: event.event.clone(),
            })
            .collect();
        
        Ok(WorkflowTimeline {
            root_correlation_id,
            events,
        })
    }
}

Acceptance Criteria

• CorrelationMetadata implemented and serializable
• EventEnvelope carries correlation metadata across local IPC
• CorrelationTracker tracks workflows across local processes
• Multi-stage workflow support with parent/child relationships
• Forensic tagging and querying capabilities
• Workflow timeline reconstruction from local event history
• Correlation overhead <5% of base local IPC latency
• Event history with configurable retention
• Zero unsafe code in correlation logic
• Unit tests for correlation tracking
• Integration tests with multi-stage workflows across local processes
• Documentation of correlation patterns

Testing Strategy

Unit Tests

• Correlation ID generation and propagation
• Workflow state tracking
• Event history management
• Tag-based querying

Integration Tests

• Simple single-stage workflow across local processes
• Multi-stage workflow with dependencies
• Forensic tagging and retrieval
• Timeline reconstruction

Performance Requirements (from Parent #114)

• Correlation Overhead: <5% of base local IPC latency (<50μs additional per message)
• Event History Query: <10ms for tag-based search in 10K events
• Timeline Reconstruction: <100ms for workflows with 1K events
• Memory Overhead: <10MB for 10K correlated events in history

Code Safety

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

Use Cases

1. Threat Hunting (Local System)

• Correlate process execution, network activity, and file access from local collectors
• Tag suspicious patterns for investigation
• Reconstruct attack timeline from local event history

2. Compliance Auditing (Local System)

• Track multi-stage data access workflows across local processes
• Correlate user actions with system events
• Generate audit trails from local event history

3. Performance Monitoring (Local System)

• Correlate resource usage across local collector domains
• Track performance impact of specific workflows
• Identify bottlenecks in multi-stage processing on local system

---

Status: Not Started
Priority: Medium
Estimated Effort: 1 week
Depends On: #115, #116, #117

[traycerai]

Plan

Observations

The DaemonEye codebase currently features a protobuf-based IPC system for communication between the daemoneye-agent and collector processes like procmond, but lacks an event bus architecture for pub/sub coordination. No existing EventBusBroker, CorrelationTracker, or related structures were found. The ticket specifies implementing local cross-process correlation tracking using Rust structs with serde serialization, building on dependencies like interprocess, tokio, prost, and dashmap. Key integration points include wrapping protobuf payloads in EventEnvelopes for correlation propagation, adding workflow state management in the broker, and enabling forensic querying via event history. Performance targets (<5% IPC overhead) and zero unsafe code must be met. The design supports multi-stage workflows and tagging for investigations within a single system.</parameter name>\nImplement correlation metadata and event bus support for multi-collector coordination:\n\n1. Add data models: Create CorrelationMetadata, EventEnvelope, WorkflowState, StageInfo, and CorrelatedEvent structs in new model files.\n2. Implement CorrelationTracker: Add workflow tracking, event history management, wait_for_correlation_results, find_events_by_tag, and get_workflow_timeline methods.\n3. Implement EventBusBroker: Extend with publish_with_correlation, track_event integration, and basic routing over existing IPC.\n4. Support workflows: Enable single-stage and multi-stage scenarios with parent/child correlations and tags.\n5. Forensic capabilities: Add querying and timeline reconstruction from local history.\n6. Integration: Initialize broker in daemoneye-agent and handle envelopes in collectors.\n7. Testing & Docs: Add unit/integration tests and comprehensive documentation.\n8. Dependencies: Ensure uuid, serde, dashmap are available.</parameter name>\nI listed the repository directories to understand the structure, focusing on daemoneye-lib/src (models, ipc), collector-core/src (ipc, event), and daemoneye-agent/src. I searched for eventbus-related terms across Rust files but found no matches, confirming the need for new implementations. I reviewed existing protobuf IPC in daemoneye-lib/src/ipc and collector-core/src/ipc to plan envelope wrapping. This informed the placement of new modules in daemoneye-lib for shared use across agent and collectors.

Approach

Implement correlation metadata and event bus support for multi-collector coordination:

1. Add data models: Create CorrelationMetadata, EventEnvelope, WorkflowState, StageInfo, and CorrelatedEvent structs in new model files.
2. Implement CorrelationTracker: Add workflow tracking, event history management, wait_for_correlation_results, find_events_by_tag, and get_workflow_timeline methods.
3. Implement EventBusBroker: Extend with publish_with_correlation, track_event integration, and basic routing over existing IPC.
4. Support workflows: Enable single-stage and multi-stage scenarios with parent/child correlations and tags.
5. Forensic capabilities: Add querying and timeline reconstruction from local history.
6. Integration: Initialize broker in daemoneye-agent and handle envelopes in collectors.
7. Testing & Docs: Add unit/integration tests and comprehensive documentation.
8. Dependencies: Ensure uuid, serde, dashmap are available.

Reasoning

I listed the repository directories to understand the structure, focusing on daemoneye-lib/src (models, ipc), collector-core/src (ipc, event), and daemoneye-agent/src. I searched for eventbus-related terms across Rust files but found no matches, confirming the need for new implementations. I reviewed existing protobuf IPC in daemoneye-lib/src/ipc and collector-core/src/ipc to plan envelope wrapping. This informed the placement of new modules in daemoneye-lib for shared use across agent and collectors.

Proposed File Changes

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

References

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

Create new module for correlation metadata structures as specified in the ticket:

• Add necessary imports: use std::collections::HashMap; use std::time::{SystemTime, Duration, Instant}; use uuid::Uuid; use serde::{Serialize, Deserialize}; use sysinfo::Pid as ProcessId; (note: add sysinfo dependency if needed for ProcessId)

• Define pub struct CorrelationMetadata with fields exactly as in ticket: correlation_id: Uuid, parent_correlation_id: Option, root_correlation_id: Uuid, sequence_number: u64, workflow_stage: String, source_process: String, correlation_tags: HashMap<String, String>, started_at: SystemTime

• Implement Debug, Clone, Serialize, Deserialize derives

• Add Default impl if appropriate

• Define pub struct WorkflowState with: correlation_id: Uuid, started_at: Instant, stages: HashMap<String, StageInfo>, participating_processes: std::collections::HashSet, total_events: usize

• Define pub struct StageInfo with: stage_name: String, started_at: Instant, completed_at: Option, events_count: usize

• Add new() method for StageInfo

• Define pub struct CorrelatedEvent with: correlation_id: Uuid, event: EventEnvelope (forward ref), received_at: Instant

• Add comprehensive documentation comments matching ticket examples

• Include unit tests for struct creation, serialization, and default values

• Follow existing model patterns from daemoneye-lib/src/models/process.rs and daemoneye-lib/src/models/alert.rs

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

References

• 📄 daemoneye-lib/src/models/correlation.rs ^(NEW)

Create new module for event envelope structure:

• Add imports: use std::time::SystemTime; use uuid::Uuid; (wait, no, ProcessId from sysinfo); use serde::{Serialize, Deserialize}; use crate::models::correlation::CorrelationMetadata;

• Define pub struct EventEnvelope with fields: topic: String, payload: Vec, timestamp: SystemTime, publisher_process_id: ProcessId, correlation_metadata: Option

• Implement Debug, Serialize, Deserialize derives

• Add documentation explaining usage in local IPC for carrying protobuf-serialized payloads with correlation

• Add builder methods if needed, similar to other models

• Include unit tests for envelope creation, serialization (with sample protobuf payload), and correlation attachment

• Ensure compatibility with prost for payload (Vec can hold serialized DetectionEvent or similar)

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

References

• 📄 daemoneye-lib/src/models/correlation.rs ^(NEW)
• 📄 daemoneye-lib/src/models/event_envelope.rs ^(NEW)

Implement the CorrelationTracker as per ticket:

• Imports: use std::sync::{Arc, RwLock}; use std::collections::{HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant}; use dashmap::DashMap; use tokio::time; use crate::models::{correlation::{CorrelationMetadata, WorkflowState, StageInfo, CorrelatedEvent}, event_envelope::EventEnvelope}; use sysinfo::Pid;

• Define pub struct CorrelationTracker { active_workflows: Arc<DashMap<Uuid, WorkflowState>>, event_history: Arc<RwLock<VecDeque>>, max_history_size: usize }

• Add new() method with default max_history_size = 10000

• Implement async fn track_event(&self, envelope: &EventEnvelope) -> Result<()> as in ticket: update workflow state, add to history, trim if needed. Use anyhow::Result for errors.

• Add WorkflowState::new(correlation_id: Uuid) -> Self and is_complete(&self) -> bool (e.g., based on total_events > threshold or custom logic)

• Implement StageInfo::new(stage_name: &str) -> Self

• Implement async fn wait_for_correlation_results(&self, correlation_id: Uuid, timeout: Duration) -> Result<Vec> as in ticket, with polling loop

• Implement async fn find_events_by_tag(&self, tag_key: &str, tag_value: &str) -> Result<Vec> filtering history

• Define aux structs: pub struct WorkflowTimeline { root_correlation_id: Uuid, events: Vec }
  pub struct TimelineEvent { timestamp: Instant, stage: String, source_process: Pid, event: EventEnvelope }

• Implement async fn get_workflow_timeline(&self, root_correlation_id: Uuid) -> Result as in ticket

• Add documentation with usage examples from ticket

• Include unit tests for tracking, waiting, querying, and timeline generation

• Ensure zero unsafe code and performance considerations (e.g., lock contention minimization)

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

References

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

Create eventbus module declaration:

• pub mod broker;

• pub use broker::EventBusBroker;

• Add documentation: 'Event bus for local multi-collector coordination via IPC'

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

References

• 📄 daemoneye-lib/src/correlation_tracker.rs ^(NEW)
• 📄 daemoneye-lib/src/models/event_envelope.rs ^(NEW)
• 📄 daemoneye-lib/src/ipc/client.rs

Implement EventBusBroker for local IPC pub/sub:

• Imports: use std::time::SystemTime; use sysinfo::Pid; use crate::models::event_envelope::EventEnvelope; use crate::correlation_tracker::CorrelationTracker; use crate::ipc::client::IpcClient; (integrate with existing IPC)

• Define pub struct EventBusBroker { process_id: Pid, correlation_tracker: Arc, ipc_client: IpcClient, subscribers: HashMap<String, Vec> } // basic subscriber map

• Add new(process_id: Pid) -> Self initializing tracker and ipc_client

• Implement async fn publish_with_correlation(&self, topic: &str, payload: Vec, correlation: CorrelationMetadata) -> anyhow::Result<()> as in ticket: create envelope, track_event, route_envelope (send via IPC to subscribers on topic)

• Implement async fn route_envelope(&self, envelope: EventEnvelope) -> Result<()>: Use existing IPC to send to collector processes subscribed to the topic (e.g., serialize envelope with serde and send over interprocess channel)

• Add fn process_id(&self) -> Pid

• For integration with existing protobuf: payload is prost-serialized event, envelope wraps it with serde for correlation

• Add documentation and examples from ticket scenarios

• Include tests for publishing, routing, and correlation integration

• Ensure <5% overhead by minimizing serialization (serde for envelope only)

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

References

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

Export new modules for event bus and correlation:

• After existing mod declarations (e.g., after mod models;), add:
  pub mod correlation_tracker;
  pub mod eventbus;

• In pub use section, add:
  pub use correlation_tracker::CorrelationTracker;
  pub use eventbus::EventBusBroker;

• Update models mod to include new correlation and event_envelope if placed there, or add mod correlation; mod event_envelope;

• Add documentation comments for the new features

• Ensure no breaking changes to existing exports

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

References

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

Integrate EventBusBroker into the agent:

• Import: use daemoneye_lib::{eventbus::EventBusBroker, correlation_tracker::CorrelationTracker}; use sysinfo::{System, Pid};

• In main(), after initializing sysinfo or process info:
  let process_id = Pid::from(std::process::id() as usize);
  let broker = EventBusBroker::new(process_id);
  let tracker = Arc::clone(&broker.correlation_tracker);

• Use broker for task publishing: Instead of direct IPC, use broker.publish_with_correlation for control topics like "control.collector.task.process"

• Example integration with existing ipc_client: Wrap ipc_client calls with broker methods for correlation

• Add error handling for broker operations

• Update to support workflow scenarios from ticket (e.g., multi-stage publishing)

• Add logging for correlation events

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

References

• 📄 daemoneye-lib/src/models/event_envelope.rs ^(NEW)
• 📄 daemoneye-lib/src/ipc/client.rs
• 📄 collector-core/src/event.rs

Extend IPC server to handle EventEnvelopes:

• Import: use daemoneye_lib::models::event_envelope::EventEnvelope; use serde::{Deserialize}; use crate::event::CollectionEvent;

• In the IPC server handler (start() method), after receiving protobuf message, deserialize incoming data as EventEnvelope if topic indicates (or add new message type for envelopes)

• For received envelopes: Extract payload (Vec), deserialize to DetectionTask or event using prost, process as before, then create response envelope with correlation preserved or updated

• Add method to publish response: Use a client-side broker or direct IPC to send back EventEnvelope

• Update task processing to propagate correlation_metadata to generated events (e.g., add to CollectionEvent if extended)

• Add logging for received correlations

• Update tests to include envelope handling and correlation propagation

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

Add required dependencies for correlation and eventbus:

• In [dependencies]:
  uuid = { version = "1.3", features = ["v4", "serde"] }
  dashmap = "5.5"
  sysinfo = "0.29" # for Pid
  serde = { version = "1.0", features = ["derive"] }

• Ensure interprocess, tokio, prost are already present or add if missing

• Add to existing dependencies without version conflicts

• For performance, use fast serialization if needed

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

References

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

Add unit and integration tests for correlation features:

• Imports: all relevant from new modules

• Unit tests:

  • test_correlation_metadata_serialization(): Serialize/deserialize CorrelationMetadata
  • test_workflow_state_updates(): Verify WorkflowState updates in track_event
  • test_event_history_management(): Test history push/pop and trimming
  • test_tag_filtering(): Test find_events_by_tag
  • test_timeline_reconstruction(): Test get_workflow_timeline sorting by time

• Integration tests:

  • test_single_stage_workflow(): Mock broker publish and verify results collection
  • test_multi_stage_correlation(): Simulate stages with parent/child, verify inheritance
  • test_forensic_query(): Tag events and query by tag
  • test_wait_for_results_timeout(): Test timeout handling

• Mock IPC for integration without real processes

• Measure performance: Assert overhead <5% in mock IPC roundtrips

• Follow patterns from existing tests in daemoneye-lib/tests

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

References

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

Create documentation for event bus and correlation tracking:

• Introduction: Explain local multi-collector coordination and correlation purpose

• Architecture: Describe components (Broker, Tracker, Envelope, Metadata)

• Usage: Include ticket's workflow scenarios 1-3 with code examples

• API Reference: Document key methods (publish_with_correlation, track_event, wait_for_correlation_results, find_events_by_tag, get_workflow_timeline)

• Forensic Use Cases: Threat hunting, compliance, performance monitoring as in ticket

• Performance: Note requirements and measurements

• Integration: How to use in agent and collectors

• Use markdown with code blocks matching ticket

• Follow style from existing docs/src/technical/ipc-implementation.md

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

References

• 📄 docs/src/technical/eventbus-correlation.md ^(NEW)

Add new documentation entry:

• In Technical section, after IPC Implementation:

  • Event Bus and Correlation Tracking

• Maintain formatting consistency

Import In IDE

🤖 Prompt for AI Agents

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

### Observations

The DaemonEye codebase currently features a protobuf-based IPC system for communication between the daemoneye-agent and collector processes like procmond, but lacks an event bus architecture for pub/sub coordination. No existing EventBusBroker, CorrelationTracker, or related structures were found. The ticket specifies implementing local cross-process correlation tracking using Rust structs with serde serialization, building on dependencies like interprocess, tokio, prost, and dashmap. Key integration points include wrapping protobuf payloads in EventEnvelopes for correlation propagation, adding workflow state management in the broker, and enabling forensic querying via event history. Performance targets (<5% IPC overhead) and zero unsafe code must be met. The design supports multi-stage workflows and tagging for investigations within a single system.</parameter name>\n<parameter name="approach">Implement correlation metadata and event bus support for multi-collector coordination:\n\n1. **Add data models**: Create CorrelationMetadata, EventEnvelope, WorkflowState, StageInfo, and CorrelatedEvent structs in new model files.\n2. **Implement CorrelationTracker**: Add workflow tracking, event history management, wait_for_correlation_results, find_events_by_tag, and get_workflow_timeline methods.\n3. **Implement EventBusBroker**: Extend with publish_with_correlation, track_event integration, and basic routing over existing IPC.\n4. **Support workflows**: Enable single-stage and multi-stage scenarios with parent/child correlations and tags.\n5. **Forensic capabilities**: Add querying and timeline reconstruction from local history.\n6. **Integration**: Initialize broker in daemoneye-agent and handle envelopes in collectors.\n7. **Testing & Docs**: Add unit/integration tests and comprehensive documentation.\n8. **Dependencies**: Ensure uuid, serde, dashmap are available.</parameter name>\n<parameter name="howDidIGetHere">I listed the repository directories to understand the structure, focusing on daemoneye-lib/src (models, ipc), collector-core/src (ipc, event), and daemoneye-agent/src. I searched for eventbus-related terms across Rust files but found no matches, confirming the need for new implementations. I reviewed existing protobuf IPC in daemoneye-lib/src/ipc and collector-core/src/ipc to plan envelope wrapping. This informed the placement of new modules in daemoneye-lib for shared use across agent and collectors.

### Approach

Implement correlation metadata and event bus support for multi-collector coordination:

1. **Add data models**: Create CorrelationMetadata, EventEnvelope, WorkflowState, StageInfo, and CorrelatedEvent structs in new model files.
2. **Implement CorrelationTracker**: Add workflow tracking, event history management, wait_for_correlation_results, find_events_by_tag, and get_workflow_timeline methods.
3. **Implement EventBusBroker**: Extend with publish_with_correlation, track_event integration, and basic routing over existing IPC.
4. **Support workflows**: Enable single-stage and multi-stage scenarios with parent/child correlations and tags.
5. **Forensic capabilities**: Add querying and timeline reconstruction from local history.
6. **Integration**: Initialize broker in daemoneye-agent and handle envelopes in collectors.
7. **Testing & Docs**: Add unit/integration tests and comprehensive documentation.
8. **Dependencies**: Ensure uuid, serde, dashmap are available.

### Reasoning

I listed the repository directories to understand the structure, focusing on daemoneye-lib/src (models, ipc), collector-core/src (ipc, event), and daemoneye-agent/src. I searched for eventbus-related terms across Rust files but found no matches, confirming the need for new implementations. I reviewed existing protobuf IPC in daemoneye-lib/src/ipc and collector-core/src/ipc to plan envelope wrapping. This informed the placement of new modules in daemoneye-lib for shared use across agent and collectors.

## Proposed File Changes

### daemoneye-lib/src/models/correlation.rs(NEW)

References: 

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

Create new module for correlation metadata structures as specified in the ticket:

- Add necessary imports: use std::collections::HashMap; use std::time::{SystemTime, Duration, Instant}; use uuid::Uuid; use serde::{Serialize, Deserialize}; use sysinfo::Pid as ProcessId; (note: add sysinfo dependency if needed for ProcessId)

- Define pub struct CorrelationMetadata with fields exactly as in ticket: correlation_id: Uuid, parent_correlation_id: Option<Uuid>, root_correlation_id: Uuid, sequence_number: u64, workflow_stage: String, source_process: String, correlation_tags: HashMap<String, String>, started_at: SystemTime
- Implement Debug, Clone, Serialize, Deserialize derives
- Add Default impl if appropriate

- Define pub struct WorkflowState with: correlation_id: Uuid, started_at: Instant, stages: HashMap<String, StageInfo>, participating_processes: std::collections::HashSet<ProcessId>, total_events: usize

- Define pub struct StageInfo with: stage_name: String, started_at: Instant, completed_at: Option<Instant>, events_count: usize
- Add new() method for StageInfo

- Define pub struct CorrelatedEvent with: correlation_id: Uuid, event: EventEnvelope (forward ref), received_at: Instant

- Add comprehensive documentation comments matching ticket examples
- Include unit tests for struct creation, serialization, and default values

- Follow existing model patterns from `daemoneye-lib/src/models/process.rs` and `daemoneye-lib/src/models/alert.rs`

### daemoneye-lib/src/models/event_envelope.rs(NEW)

References: 

- daemoneye-lib/src/models/correlation.rs(NEW)

Create new module for event envelope structure:

- Add imports: use std::time::SystemTime; use uuid::Uuid; (wait, no, ProcessId from sysinfo); use serde::{Serialize, Deserialize}; use crate::models::correlation::CorrelationMetadata;

- Define pub struct EventEnvelope with fields: topic: String, payload: Vec<u8>, timestamp: SystemTime, publisher_process_id: ProcessId, correlation_metadata: Option<CorrelationMetadata>
- Implement Debug, Serialize, Deserialize derives

- Add documentation explaining usage in local IPC for carrying protobuf-serialized payloads with correlation

- Add builder methods if needed, similar to other models

- Include unit tests for envelope creation, serialization (with sample protobuf payload), and correlation attachment

- Ensure compatibility with prost for payload (Vec<u8> can hold serialized DetectionEvent or similar)

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

References: 

- daemoneye-lib/src/models/correlation.rs(NEW)
- daemoneye-lib/src/models/event_envelope.rs(NEW)

Implement the CorrelationTracker as per ticket:

- Imports: use std::sync::{Arc, RwLock}; use std::collections::{HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant}; use dashmap::DashMap; use tokio::time; use crate::models::{correlation::{CorrelationMetadata, WorkflowState, StageInfo, CorrelatedEvent}, event_envelope::EventEnvelope}; use sysinfo::Pid;

- Define pub struct CorrelationTracker { active_workflows: Arc<DashMap<Uuid, WorkflowState>>, event_history: Arc<RwLock<VecDeque<CorrelatedEvent>>>, max_history_size: usize }
- Add new() method with default max_history_size = 10000

- Implement async fn track_event(&self, envelope: &EventEnvelope) -> Result<()> as in ticket: update workflow state, add to history, trim if needed. Use anyhow::Result for errors.
- Add WorkflowState::new(correlation_id: Uuid) -> Self and is_complete(&self) -> bool (e.g., based on total_events > threshold or custom logic)
- Implement StageInfo::new(stage_name: &str) -> Self

- Implement async fn wait_for_correlation_results(&self, correlation_id: Uuid, timeout: Duration) -> Result<Vec<EventEnvelope>> as in ticket, with polling loop

- Implement async fn find_events_by_tag(&self, tag_key: &str, tag_value: &str) -> Result<Vec<EventEnvelope>> filtering history

- Define aux structs: pub struct WorkflowTimeline { root_correlation_id: Uuid, events: Vec<TimelineEvent> }
  pub struct TimelineEvent { timestamp: Instant, stage: String, source_process: Pid, event: EventEnvelope }

- Implement async fn get_workflow_timeline(&self, root_correlation_id: Uuid) -> Result<WorkflowTimeline> as in ticket

- Add documentation with usage examples from ticket
- Include unit tests for tracking, waiting, querying, and timeline generation
- Ensure zero unsafe code and performance considerations (e.g., lock contention minimization)

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

References: 

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

Create eventbus module declaration:

- pub mod broker;
- pub use broker::EventBusBroker;

- Add documentation: 'Event bus for local multi-collector coordination via IPC'

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

References: 

- daemoneye-lib/src/correlation_tracker.rs(NEW)
- daemoneye-lib/src/models/event_envelope.rs(NEW)
- daemoneye-lib/src/ipc/client.rs

Implement EventBusBroker for local IPC pub/sub:

- Imports: use std::time::SystemTime; use sysinfo::Pid; use crate::models::event_envelope::EventEnvelope; use crate::correlation_tracker::CorrelationTracker; use crate::ipc::client::IpcClient; (integrate with existing IPC)

- Define pub struct EventBusBroker { process_id: Pid, correlation_tracker: Arc<CorrelationTracker>, ipc_client: IpcClient, subscribers: HashMap<String, Vec<Pid>> } // basic subscriber map
- Add new(process_id: Pid) -> Self initializing tracker and ipc_client

- Implement async fn publish_with_correlation(&self, topic: &str, payload: Vec<u8>, correlation: CorrelationMetadata) -> anyhow::Result<()> as in ticket: create envelope, track_event, route_envelope (send via IPC to subscribers on topic)

- Implement async fn route_envelope(&self, envelope: EventEnvelope) -> Result<()>: Use existing IPC to send to collector processes subscribed to the topic (e.g., serialize envelope with serde and send over interprocess channel)

- Add fn process_id(&self) -> Pid

- For integration with existing protobuf: payload is prost-serialized event, envelope wraps it with serde for correlation

- Add documentation and examples from ticket scenarios
- Include tests for publishing, routing, and correlation integration
- Ensure <5% overhead by minimizing serialization (serde for envelope only)

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

References: 

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

Export new modules for event bus and correlation:

- After existing mod declarations (e.g., after mod models;), add:
  pub mod correlation_tracker;
  pub mod eventbus;

- In pub use section, add:
  pub use correlation_tracker::CorrelationTracker;
  pub use eventbus::EventBusBroker;

- Update models mod to include new correlation and event_envelope if placed there, or add mod correlation; mod event_envelope;

- Add documentation comments for the new features
- Ensure no breaking changes to existing exports

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

References: 

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

Integrate EventBusBroker into the agent:

- Import: use daemoneye_lib::{eventbus::EventBusBroker, correlation_tracker::CorrelationTracker}; use sysinfo::{System, Pid};

- In main(), after initializing sysinfo or process info:
  let process_id = Pid::from(std::process::id() as usize);
  let broker = EventBusBroker::new(process_id);
  let tracker = Arc::clone(&broker.correlation_tracker);

- Use broker for task publishing: Instead of direct IPC, use broker.publish_with_correlation for control topics like "control.collector.task.process"

- Example integration with existing ipc_client: Wrap ipc_client calls with broker methods for correlation

- Add error handling for broker operations

- Update to support workflow scenarios from ticket (e.g., multi-stage publishing)

- Add logging for correlation events

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

References: 

- daemoneye-lib/src/models/event_envelope.rs(NEW)
- daemoneye-lib/src/ipc/client.rs
- collector-core/src/event.rs

Extend IPC server to handle EventEnvelopes:

- Import: use daemoneye_lib::models::event_envelope::EventEnvelope; use serde::{Deserialize}; use crate::event::CollectionEvent;

- In the IPC server handler (start() method), after receiving protobuf message, deserialize incoming data as EventEnvelope if topic indicates (or add new message type for envelopes)

- For received envelopes: Extract payload (Vec<u8>), deserialize to DetectionTask or event using prost, process as before, then create response envelope with correlation preserved or updated

- Add method to publish response: Use a client-side broker or direct IPC to send back EventEnvelope

- Update task processing to propagate correlation_metadata to generated events (e.g., add to CollectionEvent if extended)

- Add logging for received correlations

- Update tests to include envelope handling and correlation propagation

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

Add required dependencies for correlation and eventbus:

- In [dependencies]:
  uuid = { version = "1.3", features = ["v4", "serde"] }
  dashmap = "5.5"
  sysinfo = "0.29"  # for Pid
  serde = { version = "1.0", features = ["derive"] }

- Ensure interprocess, tokio, prost are already present or add if missing
- Add to existing dependencies without version conflicts
- For performance, use fast serialization if needed

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

References: 

- daemoneye-lib/src/correlation_tracker.rs(NEW)
- daemoneye-lib/src/eventbus/broker.rs(NEW)

Add unit and integration tests for correlation features:

- Imports: all relevant from new modules

- Unit tests:
  - test_correlation_metadata_serialization(): Serialize/deserialize CorrelationMetadata
  - test_workflow_state_updates(): Verify WorkflowState updates in track_event
  - test_event_history_management(): Test history push/pop and trimming
  - test_tag_filtering(): Test find_events_by_tag
  - test_timeline_reconstruction(): Test get_workflow_timeline sorting by time

- Integration tests:
  - test_single_stage_workflow(): Mock broker publish and verify results collection
  - test_multi_stage_correlation(): Simulate stages with parent/child, verify inheritance
  - test_forensic_query(): Tag events and query by tag
  - test_wait_for_results_timeout(): Test timeout handling

- Mock IPC for integration without real processes
- Measure performance: Assert overhead <5% in mock IPC roundtrips
- Follow patterns from existing tests in daemoneye-lib/tests

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

References: 

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

Create documentation for event bus and correlation tracking:

- Introduction: Explain local multi-collector coordination and correlation purpose
- Architecture: Describe components (Broker, Tracker, Envelope, Metadata)
- Usage: Include ticket's workflow scenarios 1-3 with code examples
- API Reference: Document key methods (publish_with_correlation, track_event, wait_for_correlation_results, find_events_by_tag, get_workflow_timeline)
- Forensic Use Cases: Threat hunting, compliance, performance monitoring as in ticket
- Performance: Note requirements and measurements
- Integration: How to use in agent and collectors

- Use markdown with code blocks matching ticket
- Follow style from existing docs/src/technical/ipc-implementation.md

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

References: 

- docs/src/technical/eventbus-correlation.md(NEW)

Add new documentation entry:

- In Technical section, after IPC Implementation:
  - [Event Bus and Correlation Tracking](technical/eventbus-correlation.md)

- Maintain formatting consistency

---

Developer Humor

🕵️‍♂️ Correlation chains link the processes tight,\n> From agent to collector, through IPC's flight.\n> Workflows multi-stage, forensics in sight,\n> No event left orphaned in the dead of night!\n> EventBus unites, what a delightful bite. 🔗

---

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

# Fetch the current issue details
echo "=== Current Issue Details ==="
gh issue view 120 --json title,body,labels,assignees,milestone,projectItems

echo -e "\n=== Reading spec file ==="
# Read the spec file to get more context
fd -t f "tasks.md" .kiro/specs/daemoneye-core-monitoring/ --exec cat {}

echo -e "\n=== Finding EventBus implementation ==="
# Find EventBus related files
fd -t f -e py -e rb | xargs rg -l "class EventBus" | head -5

echo -e "\n=== Finding existing metadata structures ==="
# Find metadata related files
fd -t f -e py -e rb | xargs rg -l "Metadata" | head -10

Length of output: 39564

---

🏁 Script executed:

#!/bin/bash

echo "=== Fetching parent issue #114 ==="
gh issue view 114 --json title,body,labels

echo -e "\n=== Searching for EventBus or event handling infrastructure ==="
rg -n "EventBus|event_bus" --type rust -A 3 -B 3 | head -100

echo -e "\n=== Searching for correlation patterns ==="
rg -n "correlation|Correlation" --type rust -A 3 -B 3 | head -80

echo -e "\n=== Searching for existing metadata structures ==="
ast-grep --pattern 'struct $NAME {
  $$$
  metadata: $_,
  $$$
}'

echo -e "\n=== Looking for message envelope patterns ==="
rg -n "envelope|Envelope|message.*wrapper" --type rust -A 3 -B 3 | head -80

echo -e "\n=== Checking collector-core event structures ==="
fd -e rs collector-core | xargs rg -n "CollectionEvent|Event" -A 5 | head -100

Length of output: 10733

---

🏁 Script executed:

#!/bin/bash

echo "=== Searching for daemoneye-eventbus references ==="
rg -n "eventbus|event-bus|daemoneye-eventbus" --type rust -A 5 -B 5 | head -100

echo -e "\n=== Checking for eventbus directory or crate ==="
fd -t d eventbus

echo -e "\n=== Looking at Cargo workspace members ==="
cat Cargo.toml | grep -A 20 "members"

echo -e "\n=== Searching for requirements 15.4 and 16.3 in spec files ==="
fd -e md -t f | xargs rg "15\.4|16\.3" -A 3 -B 3 | head -100

echo -e "\n=== Checking CollectionEvent structure ==="
ast-grep --pattern 'pub enum CollectionEvent {
  $$$
}'

echo -e "\n=== Looking for existing event metadata ==="
rg -n "timestamp|collected_at|source_id" collector-core/src/event.rs -A 2 -B 2

Length of output: 5003

---

🏁 Script executed:

#!/bin/bash

echo "=== Searching for requirements documents ==="
fd -e md requirements | head -10

echo -e "\n=== Searching for architecture docs with requirements ==="
fd -t f -e md docs | xargs rg -l "requirement|Requirements" | head -10

echo -e "\n=== Looking for any mention of requirement 15 or 16 ==="
fd -t f -e md | xargs rg "requirement.*1[56]|1[56]\.\d" -i | head -50

echo -e "\n=== Checking existing ProcessEvent structure ==="
cat collector-core/src/event.rs | head -120

echo -e "\n=== Looking for existing IPC message structures ==="
fd -e proto | head -10

Length of output: 4230

---

🏁 Script executed:

#!/bin/bash

echo "=== Checking IPC proto definitions ==="
cat daemoneye-lib/proto/ipc.proto

echo -e "\n=== Checking common proto definitions ==="
cat daemoneye-lib/proto/common.proto

echo -e "\n=== Searching for workflow or stage tracking ==="
rg -n "workflow|stage|sequence" --type rust -i | head -50

echo -e "\n=== Looking for existing collector coordination ==="
rg -n "coordinator|orchestrat|distribute" --type rust -i -A 3 | head -80

Length of output: 10403

---

🏁 Script executed:

#!/bin/bash

echo "=== Getting repository labels for proper categorization ==="
gh label list --limit 100

echo -e "\n=== Checking if there are any existing issues about EventBus implementation ==="
gh issue list --search "eventbus OR EventBus" --json number,title,state

echo -e "\n=== Looking at spec section 2.6 context ==="
rg -A 100 "^- \[ \] 6\. Implement" .kiro/specs/daemoneye-core-monitoring/tasks.md | head -150

Length of output: 11637

---

🏁 Script executed:

#!/bin/bash

echo "=== Checking for issue #87 about EventBus architecture ==="
gh issue view 87 --json title,body,labels

echo -e "\n=== Looking for related issues in the 2.6 series ==="
gh issue list --search "2.6" --json number,title,state --limit 20

echo -e "\n=== Checking current ProcessRecord structure for adding correlation fields ==="
cat daemoneye-lib/proto/common.proto | grep -A 30 "message ProcessRecord"

Length of output: 13198

---

🏁 Script executed:

#!/bin/bash

# Update the issue with enhanced content
gh issue edit 120 \
  --title "Add correlation metadata and workflow tracking for multi-collector coordination" \
  --body "## Overview

This issue implements correlation metadata and workflow tracking capabilities to enable sophisticated multi-collector coordination and forensic investigations across the DaemonEye monitoring ecosystem.

## Context

**Parent Issue**: #114 - Spec 2.6 Add multi-process collector coordination via daemoneye-eventbus  
**Related Architecture**: #87 - Implement Custom daemoneye-eventbus Architecture  
**Source**: .kiro/specs/daemoneye-core-monitoring/tasks.md (Section 2.6.8)  
**Requirements**: 15.4, 16.3

This task is critical for enabling:
- Multi-stage detection workflows across multiple collectors
- Forensic correlation during security investigations
- Event causality tracking in complex distributed operations
- Performance analysis of multi-collector workflows

## Problem Statement

The current event system lacks the metadata needed to:
1. **Trace event relationships** across multiple collector processes and detection stages
2. **Correlate security events** from different domains (process, network, filesystem)
3. **Track workflow progress** through complex multi-stage analysis pipelines
4. **Support forensic investigations** that require reconstructing event causality chains
5. **Enable performance profiling** of distributed detection workflows

## Proposed Solution

### 1. CorrelationMetadata Structure

Add a new protobuf message for correlation tracking:

\`\`\`protobuf
/// Correlation metadata for tracking events across multi-collector workflows
message CorrelationMetadata {
  /// Unique identifier for this specific event/operation
  string correlation_id = 1;
  
  /// Parent operation that triggered this event (for nested operations)
  optional string parent_correlation_id = 2;
  
  /// Root operation at the start of the workflow chain
  string root_correlation_id = 3;
  
  /// Sequence number within the workflow (monotonically increasing)
  uint64 sequence_number = 4;
  
  /// Current workflow stage identifier (e.g., \"initial_scan\", \"deep_analysis\", \"yara_scan\")
  optional string workflow_stage = 5;
  
  /// Flexible tags for domain-specific correlation
  map<string, string> correlation_tags = 6;
  
  /// Timestamp when correlation context was created (Unix timestamp in milliseconds)
  int64 correlation_timestamp = 7;
  
  /// Source collector/component that generated this correlation context
  string source_component = 8;
}
\`\`\`

### 2. Integration Points

#### A. Event Message Envelopes

Extend EventBus message envelopes to carry correlation metadata:

\`\`\`protobuf
/// EventBus message envelope with correlation support
message EventEnvelope {
  /// Event topic (e.g., \"events.process.spawned\")
  string topic = 1;
  
  /// Correlation metadata for workflow tracking
  CorrelationMetadata correlation = 2;
  
  /// Event payload (serialized event data)
  bytes payload = 3;
  
  /// Message priority for routing
  optional uint32 priority = 4;
}
\`\`\`

#### B. Collection Events

Add correlation metadata to existing event structures:

\`\`\`rust
pub struct ProcessEvent {
    // ... existing fields ...
    
    /// Correlation metadata for multi-collector workflows
    pub correlation: Option<CorrelationMetadata>,
}
\`\`\`

#### C. IPC Messages

Extend DetectionTask and DetectionResult with correlation context:

\`\`\`protobuf
message DetectionTask {
  // ... existing fields ...
  
  /// Correlation metadata propagated from parent operation
  optional CorrelationMetadata correlation = 20;
}

message DetectionResult {
  // ... existing fields ...
  
  /// Correlation metadata from the task, potentially updated
  optional CorrelationMetadata correlation = 20;
}
\`\`\`

### 3. Correlation ID Propagation

Implement correlation ID propagation through the event processing pipeline:

\`\`\`rust
/// Correlation ID generator and propagation utilities
pub struct CorrelationManager {
    /// Generate new root correlation ID (UUID v7 for sortability)
    pub fn generate_correlation_id() -> String;
    
    /// Create child correlation context from parent
    pub fn create_child_context(parent: &CorrelationMetadata) -> CorrelationMetadata;
    
    /// Extract correlation context from EventBus message
    pub fn extract_from_envelope(envelope: &EventEnvelope) -> Option<CorrelationMetadata>;
    
    /// Inject correlation context into outbound message
    pub fn inject_into_envelope(envelope: &mut EventEnvelope, correlation: CorrelationMetadata);
}
\`\`\`

### 4. Correlation-Based Filtering and Routing

Add EventBus routing capabilities based on correlation metadata:

\`\`\`rust
/// Correlation-based event filtering
pub struct CorrelationFilter {
    /// Filter by root correlation ID (entire workflow)
    pub root_correlation_id: Option<String>,
    
    /// Filter by workflow stage
    pub workflow_stage: Option<String>,
    
    /// Filter by correlation tags (key-value pairs)
    pub correlation_tags: HashMap<String, String>,
    
    /// Filter by source component
    pub source_component: Option<String>,
}

impl EventBus {
    /// Subscribe to events matching correlation criteria
    pub async fn subscribe_with_correlation_filter(
        &self,
        topic: &str,
        filter: CorrelationFilter,
    ) -> Result<Subscription>;
    
    /// Query historical events by correlation ID
    pub async fn query_by_correlation(
        &self,
        root_correlation_id: &str,
    ) -> Result<Vec<EventEnvelope>>;
}
\`\`\`

### 5. Forensic Correlation Tracking

Add specialized forensic tracking capabilities:

\`\`\`rust
/// Forensic correlation tracker for security investigations
pub struct ForensicTracker {
    /// Reconstruct complete event chain from root correlation ID
    pub async fn reconstruct_event_chain(
        &self,
        root_correlation_id: &str,
    ) -> Result<EventChain>;
    
    /// Find all events with matching correlation tags
    pub async fn find_correlated_events(
        &self,
        tags: HashMap<String, String>,
    ) -> Result<Vec<EventEnvelope>>;
    
    /// Generate timeline visualization of correlated events
    pub async fn generate_timeline(
        &self,
        root_correlation_id: &str,
    ) -> Result<Timeline>;
}
\`\`\`

## Implementation Tasks

- [ ] Define \`CorrelationMetadata\` protobuf message in \`daemoneye-lib/proto/common.proto\`
- [ ] Create \`EventEnvelope\` message structure for EventBus integration
- [ ] Add correlation metadata fields to \`ProcessEvent\`, \`NetworkEvent\`, etc.
- [ ] Implement \`CorrelationManager\` for ID generation and propagation
- [ ] Extend \`DetectionTask\` and \`DetectionResult\` with correlation fields
- [ ] Add correlation-based filtering to EventBus subscription API
- [ ] Implement \`CorrelationFilter\` for routing rules
- [ ] Create \`ForensicTracker\` for investigation workflows
- [ ] Add correlation metadata to audit logging system
- [ ] Write unit tests for correlation ID generation and propagation
- [ ] Create integration tests for multi-collector correlation scenarios
- [ ] Add forensic reconstruction tests with mock event chains
- [ ] Update documentation with correlation tracking patterns

## Example Use Cases

### Use Case 1: Multi-Stage Malware Analysis
\`\`\`
1. procmond detects suspicious process (correlation_id: root-123, stage: initial_scan)
2. triggers YARA scan (parent: root-123, correlation_id: child-456, stage: yara_scan)
3. triggers PE analysis if YARA matches (parent: child-456, correlation_id: child-789, stage: pe_analysis)
4. All events linked via root_correlation_id: root-123 for forensic investigation
\`\`\`

### Use Case 2: Network-Process Correlation
\`\`\`
1. netmond detects suspicious outbound connection (correlation_tag: {\"alert_type\": \"c2_communication\"})
2. procmond enriches with process details (same correlation_tag)
3. Forensic tracker reconstructs complete attack chain from tags
\`\`\`

### Use Case 3: Distributed Detection Workflow
\`\`\`
1. Initial scan across 10 collectors (sequence: 1-10)
2. Deep analysis on 3 high-risk targets (sequence: 11-13)
3. Performance profiling: measure latency from sequence 1 → 13
\`\`\`

## Dependencies

- EventBus implementation (#87, #114)
- Multi-collector coordination workflows (#115)
- Task distribution and routing (#116)
- Result aggregation (#117)

## Testing Strategy

- Unit tests for CorrelationMetadata serialization
- Integration tests for correlation ID propagation through EventBus
- Forensic reconstruction tests with complex event chains
- Performance tests for correlation-based filtering
- Cross-platform tests for UUID generation

## Success Criteria

- [ ] CorrelationMetadata integrated into all event types
- [ ] Correlation IDs automatically propagated through EventBus messages
- [ ] Forensic tracker can reconstruct complete event chains
- [ ] Correlation-based filtering works for event subscriptions
- [ ] Performance impact < 5% for correlation tracking overhead
- [ ] Documentation includes correlation tracking patterns
- [ ] Integration tests validate multi-collector correlation scenarios

## Documentation Updates

- Architecture documentation for correlation model
- API reference for CorrelationManager and ForensicTracker
- User guide for forensic investigation workflows
- Performance tuning guide for correlation-based queries

---

**Requirements**: 15.4 (Multi-collector coordination), 16.3 (Forensic correlation tracking)"

# Add appropriate labels
gh issue edit 120 --add-label "ipc,architecture,audit,eventbus"

# Assign to unclesp1d3r as per user request
gh issue edit 120 --add-assignee unclesp1d3r

echo "Issue #120 successfully enhanced with comprehensive context and solution!"

Length of output: 296

---

🏁 Script executed:

#!/bin/bash

echo "=== Attempting to add existing labels individually ==="
# Try adding labels that we know exist from the label list
gh issue edit 120 --add-label "ipc"
gh issue edit 120 --add-label "architecture"
gh issue edit 120 --add-label "audit"

echo -e "\n=== Verifying the updated issue ==="
gh issue view 120 --json title,labels,assignees | jq '.'

echo -e "\n=== Issue successfully updated! ==="

Length of output: 1374

---

@unclesp1d3r I've successfully enhanced issue #120 with comprehensive context and a detailed technical solution! Here's what was updated:

🎯 Enhanced Title

Changed from generic task number to descriptive: "Add correlation metadata and workflow tracking for multi-collector coordination"

📋 Major Enhancements

Context & Background

• Linked to parent issue Epic: Multi-Process Collector Coordination via Event Bus Architecture #114 (Spec 2.6 EventBus coordination)
• Referenced Implement Custom daemoneye-eventbus Architecture and Remove Legacy IPC Implementation #87 (EventBus architecture) for architectural context
• Clear problem statement explaining the need for correlation tracking

Comprehensive Technical Solution

1. CorrelationMetadata Structure - Protobuf message with:

   • correlation_id, parent_correlation_id, root_correlation_id for event chaining
   • sequence_number for workflow ordering
   • workflow_stage for multi-stage analysis tracking
   • correlation_tags for flexible domain-specific correlation
   • Source component and timestamp tracking

2. Integration Points:

   • EventBus EventEnvelope with correlation support
   • Extended CollectionEvent structures (ProcessEvent, NetworkEvent, etc.)
   • Enhanced IPC messages (DetectionTask, DetectionResult)

3. Implementation Components:

   • CorrelationManager for ID generation and propagation (UUID v7)
   • CorrelationFilter for subscription-based routing
   • ForensicTracker for security investigation workflows

Practical Use Cases

Included three detailed examples:

• Multi-stage malware analysis workflow
• Network-process correlation for C2 detection
• Distributed detection with performance profiling

Implementation Checklist

13 actionable tasks covering protobuf definitions, Rust implementations, testing, and documentation

🏷️ Metadata Updates

• Labels Added: ipc, architecture, audit (joins existing data-models, security)
• Assigned: unclesp1d3r
• Dependencies: Clearly linked to issues Implement Custom daemoneye-eventbus Architecture and Remove Legacy IPC Implementation #87, Epic: Multi-Process Collector Coordination via Event Bus Architecture #114, 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

📊 Success Criteria

Defined clear acceptance criteria including:

• Full integration across all event types
• Automatic propagation through EventBus
• Forensic reconstruction capabilities
• <5% performance overhead
• Comprehensive testing and documentation

The issue is now ready for implementation with clear technical direction and proper context for the multi-collector coordination architecture! 🚀