fix(streaming): add buffer size limit to StreamProcessor

[echobt]

Summary

Adds a maximum buffer size limit to StreamProcessor to prevent unbounded memory growth during long streaming sessions.

Problem

The StreamProcessor's event buffer (VecDeque) has no size limit. If drain_events() is not called regularly, the buffer can grow indefinitely during long streams, potentially causing memory exhaustion.

Changes

• Added MAX_BUFFER_SIZE constant (10,000 events)
• Modified process() to drop oldest events when buffer is full
• Pre-allocated buffer capacity in new() for better performance

Testing

• Verified code compiles with cargo check -p cortex-engine

Verification

cargo check -p cortex-engine

[greptile-apps]

Greptile Overview

Greptile Summary

Added memory safety guard to StreamProcessor by introducing a 10,000-event buffer limit. When the buffer reaches capacity, the oldest events are silently dropped to prevent unbounded memory growth during long streaming sessions.

Key Changes:

• Introduced MAX_BUFFER_SIZE constant set to 10,000 events
• Added overflow protection in process() method that drops oldest events when limit is reached
• Pre-allocated buffer capacity to 1,024 for improved performance

Impact:
The change prevents potential OOM issues in scenarios where drain_events() is not called frequently enough. The implementation is straightforward and follows a standard circular buffer pattern by dropping the oldest data first (FIFO).

Confidence Score: 4/5

• Safe to merge with minor consideration for observability improvements
• The implementation correctly prevents unbounded memory growth with a simple and effective approach. The logic is sound - dropping oldest events when buffer is full is standard practice. Minor consideration: events are dropped silently without logging or metrics, which could make debugging harder in production scenarios where event loss occurs
• No files require special attention - the single change is straightforward and low-risk

Important Files Changed

Filename	Overview
src/cortex-engine/src/streaming.rs	Added buffer size limit (10K events) to prevent unbounded memory growth; silently drops oldest events when full

Sequence Diagram

sequenceDiagram
    participant Caller
    participant StreamProcessor
    participant Buffer as VecDeque<StreamEvent>
    
    Note over StreamProcessor: MAX_BUFFER_SIZE = 10,000
    
    Caller->>StreamProcessor: process(event)
    StreamProcessor->>Buffer: check len()
    
    alt Buffer length >= 10,000
        Buffer->>Buffer: pop_front() (drop oldest)
        Note over Buffer: Oldest event discarded
    end
    
    StreamProcessor->>Buffer: push_back(new event)
    Note over Buffer: Buffer maintains max 10K events
    
    Caller->>StreamProcessor: drain_events()
    StreamProcessor->>Buffer: drain(..)
    Buffer-->>Caller: Vec<StreamEvent>
    Note over Buffer: Buffer emptied

Loading

[greptile-apps]

_{1 file reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Consider logging when dropping events - silent data loss could be hard to debug

Suggested change

	// Enforce buffer size limit to prevent unbounded memory growth
	if self.buffer.len() >= MAX_BUFFER_SIZE {
	self.buffer.pop_front();
	}
	self.buffer.push_back(event);
	// Enforce buffer size limit to prevent unbounded memory growth
	if self.buffer.len() >= MAX_BUFFER_SIZE {
	self.buffer.pop_front();
	// Consider: log::warn!("StreamProcessor buffer full, dropping oldest event");
	}

_{Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!}

Prompt To Fix With AI

This is a comment left during a code review.
Path: src/cortex-engine/src/streaming.rs
Line: 291:295

Comment:
Consider logging when dropping events - silent data loss could be hard to debug

```suggestion
        // Enforce buffer size limit to prevent unbounded memory growth
        if self.buffer.len() >= MAX_BUFFER_SIZE {
            self.buffer.pop_front();
            // Consider: log::warn!("StreamProcessor buffer full, dropping oldest event");
        }
```

<sub>Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!</sub>

How can I resolve this? If you propose a fix, please make it concise.

[echobt]

Closing to consolidate: This streaming buffer fix will be merged with PR #85 (ToolState transition validation) into a consolidated protocol robustness PR.