Bound EventBus stream queues to prevent slow-consumer memory growth

[Justinabox]

Summary

EventBus.stream() allocates an unbounded asyncio.Queue. A slow or temporarily stalled consumer can therefore accumulate every typed event indefinitely while the modem continues processing URCs. This defeats the bounded-backlog protection used by the WebSocket adapter and is a production-memory risk for long-running Raspberry Pi deployments.

Root-cause evidence

On clean origin/main (89d0fcb), EventBus.stream() creates its queue without maxsize:

• callstack/events/bus.py:81-84 — q: asyncio.Queue[Event] = asyncio.Queue()
• callstack/events/bus.py:57-58 — every matching emitted event is awaited into that queue.
• EventStream exposes no capacity, overflow policy, or dropped-event signal.

Minimal no-hardware reproduction on the clean exported origin/main tree:

PYTHONPATH=. uv run --python 3.11 --no-project python -c $'import asyncio\nfrom callstack.events.bus import EventBus\nfrom callstack.events.types import RingEvent\n\nasync def main():\n    bus = EventBus()\n    async with bus.stream(RingEvent) as stream:\n        for _ in range(10000):\n            await bus.emit(RingEvent())\n        print(f"queued={stream._queue.qsize()} maxsize={stream._queue.maxsize}")\n\nasyncio.run(main())'

Actual output:

queued=10000 maxsize=0

In asyncio.Queue, maxsize=0 means unlimited. A caller can legally open a stream and be delayed before consumption (for example, an operator/monitor path or a DTMF/SMS integration blocked on another await); a burst then retains all matching event objects with no backpressure or cleanup until the stream context exits.

The clean default-branch suite otherwise passes:

PYTHONPATH=. uv run --python 3.11 --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q
857 passed in 7.17s

Expected behavior

Typed event streams used by unattended production integrations should have an explicit bounded-buffer policy. A slow consumer must not be able to cause unbounded process-memory growth or block the modem reader indefinitely.

Suggested fix direction

Add a small, backwards-conscious bounded-stream slice in callstack/events/bus.py:

1. Let EventBus.stream() accept a validated positive maxsize and a documented overflow policy, with a conservative finite default for production-facing streams.
2. Make emit() enqueue non-blockingly for bounded stream queues. On a full queue, use a deterministic policy (for example drop oldest then enqueue newest) rather than awaiting forever.
3. Surface loss safely: add a counter/property on EventStream, or an optional public-safe overflow sentinel/event, so operators can distinguish an empty stream from a lossy one without logging payload data.
4. Preserve existing EventBus.subscribe() handler behavior; scope this issue to stream() queues.
5. Update call sites that need stronger semantics (notably DTMF collection) to use an intentionally chosen capacity/policy, with focused tests for burst behavior.

Do not solve this by storing raw modem payloads or by applying a global queue that lets one slow consumer block unrelated subscribers.

Acceptance criteria

• EventBus.stream() has a finite, validated buffer capacity by default or requires callers to provide one explicitly.
• A burst larger than capacity has deterministic, tested behavior and does not block EventBus.emit() indefinitely.
• Consumers can observe that overflow occurred without receiving raw SMS, USSD, AT, SIM, or credential data.
• Normal ordered delivery is preserved when the consumer keeps up.
• Stream cleanup still removes queues reliably after cancellation/exception.
• DTMF and SMS stream call sites retain their current no-hardware functional behavior.

Suggested focused tests

• tests/test_events.py: assert a configured small queue does not exceed its capacity after a burst; assert ordering and the chosen loss/overflow signal.
• tests/test_dtmf.py / tests/test_call_session_play_and_collect.py: verify an active collector still receives expected tones under normal traffic.
• tests/test_sms_service.py: verify normal message-stream delivery remains unchanged.

Verification gates

git diff --check
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/test_events.py tests/test_dtmf.py tests/test_call_session_play_and_collect.py tests/test_sms_service.py -q
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q

Duplicate check

Searched issues and pull requests for EventBus bounded stream queue and unbounded EventStream memory. The only related work found is #213 / #31 / #205, which bound the per-WebSocket client queue. They do not bound the core public EventBus.stream() queues used outside the WebSocket adapter, so this is intentionally a separate core reliability slice.