fix(runner): use unbounded SSE tap queue

[markturansky]

Summary

• SSE tap queues were bounded at maxsize=100, causing silent event drops when no consumer was connected (CLI sessions, background agent runs, slow frontends)
• Each queue is ephemeral — created per session, drained or GC'd when the run ends — so unbounded buffering is safe
• Worst case: transient memory growth during a long Claude turn with no SSE consumer; acceptable within the 4Gi pod memory limit

Test plan

• Start a session via CLI (no SSE consumer) and verify no "SSE tap queue full" warnings in runner logs
• Confirm session messages are delivered correctly end-to-end

🤖 Generated with Claude Code

Summary by CodeRabbit

• Bug Fixes
  • Enhanced reliability of real-time event streaming by removing queue capacity limitations, preventing potential event loss during high-volume scenarios.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 9b44db35-5739-41ea-a582-1a1549e9ed91

📥 Commits

Reviewing files that changed from the base of the PR and between 8329cad and 71aa365.

📒 Files selected for processing (2)

• components/runners/ambient-runner/ambient_runner/app.py
• components/runners/ambient-runner/ambient_runner/endpoints/events.py

---

📝 Walkthrough

Walkthrough

These changes remove the 100-item capacity limit from async queues in SSE (Server-Sent Events) stream initialization, allowing unbounded queue sizes in both the bridge pre-registration and per-thread stream endpoints.

Changes

Cohort / File(s)	Summary
SSE Queue Initialization components/runners/ambient-runner/ambient_runner/app.py, components/runners/ambient-runner/ambient_runner/endpoints/events.py	Changed from bounded asyncio.Queue(maxsize=100) to unbounded asyncio.Queue() for pre-registered SSE queues and per-thread stream initialization.

---

Important

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

❌ Failed checks (1 error, 1 warning)

Check name	Status	Explanation	Resolution
Security And Secret Handling	❌ Error	The /events/{thread_id} endpoint lacks authentication and authorization checks, exposing sensitive SSE streams to any caller knowing the thread_id.	Implement auth via FastAPI Depends(), bearer token validation, or permission checks before queue access to verify caller identity and thread_id authorization.
Docstring Coverage	⚠️ Warning	Docstring coverage is 66.67% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title follows Conventional Commits format (fix(runner): description) and directly describes the core change: replacing bounded SSE queues with unbounded ones.
Performance And Algorithmic Complexity	✅ Passed	Unbounded asyncio queue change intentionally fixes silent event drops with explicit per-session cleanup in finally blocks, scoped memory growth within 4Gi pod limit.
Kubernetes Resource Safety	✅ Passed	PR modifies only Python application code (SSE queue initialization) without altering Kubernetes manifests, RBAC policies, resource definitions, or pod security configurations.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/sse-tap-unbounded-queue

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch fix/sse-tap-unbounded-queue

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}