feat(a2a): authenticated SSE stream proxy + since cursor (#1780 S3)#1854
Conversation
Add GET /api/a2a/bus/stream, an authenticated SSE proxy to the raw bus stream, gated exactly like the messages proxy. Add the stream path to the middleware read allowlist and forward the since cursor on the messages proxy.
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
Next review available in: 50 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![kilo-code-bot[bot]](https://avatars.githubusercontent.com/in/2193792?s=60&v=4){kind=small}
| break | ||
| # Drain any pending heartbeat that fired while we were | ||
| # forwarding real data: emit it before the next event. | ||
| if heartbeat.done(): |
There was a problem hiding this comment.
WARNING: Idle heartbeat is never emitted — defeats the stated purpose of the feature.
The heartbeat.done() check lives inside the async for line in upstream.aiter_lines() loop (line 180). If no upstream events ever arrive (the exact idle case this heartbeat exists for), the loop body never executes, so the : ping is never yielded and the connection is reaped by intermediaries exactly as the PR set out to prevent. Even between events, the ping is only emitted when an event line happens to arrive after 25s of silence — an idle stream of any length emits nothing. The heartbeat must fire on its own timer independently of inbound data (e.g. asyncio.wait/wait_for racing the line iterator against a sleep, or a dedicated timer that yields the ping), not be drained only when a line is received.
| if heartbeat.done(): | |
| if await request.is_disconnected(): | |
| break | |
| # NOTE: heartbeat must fire on its own timer outside this | |
| # loop so idle streams still emit `: ping`. |
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| ) | ||
| if line == "": | ||
| continue | ||
| yield f"{line}\n\n" |
There was a problem hiding this comment.
SUGGESTION: aiter_lines() splits every newline into a separate iteration, so a multi-line SSE event (e.g. a data: field split across lines, or data: + blank-line continuations) would be re-framed as two separate one-line events, corrupting the relayed event. The upstream bus appears to emit one-line-per-event frames, so this is likely safe today, but it couples the proxy to that internal framing assumption. Consider relaying events by buffering until a blank line terminates an SSE event block rather than per-line, or assert/parse the upstream framing explicitly.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
Code Review SummaryStatus: 2 Issues Found | Recommendation: Address before merge Overview
Issue Details (click to expand)WARNING
SUGGESTION
Files Reviewed (3 files)
Fix these issues in Kilo Cloud Reviewed by hy3:free · Input: 74.1K · Output: 4.7K · Cached: 101.1K |
1 participant
Summary
Slice S3 of the external-agent invite feature: the realtime A2A delivery leg.
GET /api/a2a/bus/stream?channel={c}&since={cursor}, an authenticated (agent-JWT or session) SSE proxy to the raw busGET {bus}/a2a/stream?thread={c}&since={cursor}. It holds one upstream SSE connection per client and relays events verbatim, injecting a: pingheartbeat comment every 25s so idle streams survive intermediaries. Auth is gated exactly like the existing/api/a2a/bus/messagesproxy (admin session / local token, or an active agent JWT holdinga2a_receive, fail closed)./api/a2a/bus/streamto_A2A_BUS_READ_PATHSinauth_middleware.py, mirroring the messages/channels allowlist entry (passthrough contract, no skeleton key).sincequery passthrough to the existing/api/a2a/bus/messagesproxy (it previously dropped it), forwarded verbatim to the bus cursor.The raw :7900 bus is never exposed directly; all bus access goes through the authenticated proxy, which forces the reader through the same gate as every other A2A read path.
Tests
tests/test_routes_a2a_bus_stream.py:sincesinceto the buschannel-> 400a2a_receive-> 403/a2a/streamis not a registered route (404), proving the proxy is the only entry pointRun:
/Volumes/NVMe/Users/jay/Development/tinyagentos/.venv/bin/python -m pytest tests/test_routes_a2a_bus_stream.py -q(7 passed).Note: the pre-existing
tests/test_a2a_bus.pyand several e2e modules import optional deps (respx,playwright) not installed in this venv; that collection gap is unrelated to this change.