fix: enrich gateway WS stream relay + accepted/final ack - #3471
Conversation
|
@coderabbitai review |
|
/review |
✅ Action performedReview finished.
|
MervinPraison
added
pipeline/blocked:ci
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? |
|
Important Review skippedBot user detected. To trigger a single review, invoke the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughThe gateway adds reasoning, tool-progress, and stream-error event types, advertises them during streaming handshakes, distinguishes accepted and final responses, reports terminal outcomes, and expands agent stream event forwarding. ChangesGateway streaming contract
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant WebSocketClient
participant GatewayServer
participant AgentStream
WebSocketClient->>GatewayServer: send message
GatewayServer-->>WebSocketClient: response status accepted
GatewayServer->>AgentStream: process agent message
AgentStream-->>GatewayServer: streaming event
GatewayServer-->>WebSocketClient: mapped stream event
GatewayServer-->>WebSocketClient: response status final with outcome
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/praisonai-bot/praisonai_bot/gateway/server.py`:
- Around line 3145-3177: The stream relay must persist every newly supported
frame type for replay, not only legacy token/tool/end events. Update
_send_to_client to include EventType.REASONING_STREAM,
EventType.TOOL_PROGRESS_STREAM, and EventType.STREAM_ERROR in its
persisted-frame tracking alongside the existing stream types, preserving current
replay behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: b21ac72c-0852-4544-be8f-8e68b44c697f
📒 Files selected for processing (2)
src/praisonai-agents/praisonaiagents/gateway/protocols.pysrc/praisonai-bot/praisonai_bot/gateway/server.py
| if event_type == StreamEventType.DELTA_TEXT: | ||
| gw_type = EventType.TOKEN_STREAM | ||
| # A reasoning/thinking delta is surfaced under its own event | ||
| # so clients can show "thinking…" separately from the answer. | ||
| if getattr(event, 'is_reasoning', False): | ||
| gw_type = EventType.REASONING_STREAM | ||
| else: | ||
| gw_type = EventType.TOKEN_STREAM | ||
| data = { | ||
| "content": getattr(event, 'content', ''), | ||
| "session_id": session.session_id, | ||
| "session_id": sid, | ||
| } | ||
| elif event_type == StreamEventType.DELTA_TOOL_CALL: | ||
| gw_type = EventType.TOOL_CALL_STREAM | ||
| data = { | ||
| "tool_call": getattr(event, 'tool_call', {}), | ||
| "session_id": session.session_id, | ||
| "session_id": sid, | ||
| } | ||
| elif event_type == StreamEventType.TOOL_PROGRESS: | ||
| gw_type = EventType.TOOL_PROGRESS_STREAM | ||
| data = { | ||
| "content": getattr(event, 'content', ''), | ||
| "metadata": getattr(event, 'metadata', None), | ||
| "session_id": sid, | ||
| } | ||
| elif event_type == StreamEventType.ERROR: | ||
| gw_type = EventType.STREAM_ERROR | ||
| data = { | ||
| "error": getattr(event, 'error', None), | ||
| "session_id": sid, | ||
| } | ||
| elif event_type == StreamEventType.STREAM_END: | ||
| gw_type = EventType.STREAM_END | ||
| data = {"session_id": session.session_id} | ||
| data = {"session_id": sid} |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Persist the newly relayed stream frames for replay.
_send_to_client tracks only legacy stream types at Lines 3329-3336. Reconnecting clients therefore replay token/tool/end frames but lose reasoning, tool-progress, and stream-error frames.
Proposed fix
"token_stream",
"tool_call_stream",
+ "reasoning_stream",
+ "tool_progress_stream",
+ "stream_error",📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if event_type == StreamEventType.DELTA_TEXT: | |
| gw_type = EventType.TOKEN_STREAM | |
| # A reasoning/thinking delta is surfaced under its own event | |
| # so clients can show "thinking…" separately from the answer. | |
| if getattr(event, 'is_reasoning', False): | |
| gw_type = EventType.REASONING_STREAM | |
| else: | |
| gw_type = EventType.TOKEN_STREAM | |
| data = { | |
| "content": getattr(event, 'content', ''), | |
| "session_id": session.session_id, | |
| "session_id": sid, | |
| } | |
| elif event_type == StreamEventType.DELTA_TOOL_CALL: | |
| gw_type = EventType.TOOL_CALL_STREAM | |
| data = { | |
| "tool_call": getattr(event, 'tool_call', {}), | |
| "session_id": session.session_id, | |
| "session_id": sid, | |
| } | |
| elif event_type == StreamEventType.TOOL_PROGRESS: | |
| gw_type = EventType.TOOL_PROGRESS_STREAM | |
| data = { | |
| "content": getattr(event, 'content', ''), | |
| "metadata": getattr(event, 'metadata', None), | |
| "session_id": sid, | |
| } | |
| elif event_type == StreamEventType.ERROR: | |
| gw_type = EventType.STREAM_ERROR | |
| data = { | |
| "error": getattr(event, 'error', None), | |
| "session_id": sid, | |
| } | |
| elif event_type == StreamEventType.STREAM_END: | |
| gw_type = EventType.STREAM_END | |
| data = {"session_id": session.session_id} | |
| data = {"session_id": sid} | |
| "token_stream", | |
| "tool_call_stream", | |
| "reasoning_stream", | |
| "tool_progress_stream", | |
| "stream_error", |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-bot/praisonai_bot/gateway/server.py` around lines 3145 - 3177,
The stream relay must persist every newly supported frame type for replay, not
only legacy token/tool/end events. Update _send_to_client to include
EventType.REASONING_STREAM, EventType.TOOL_PROGRESS_STREAM, and
EventType.STREAM_ERROR in its persisted-frame tracking alongside the existing
stream types, preserving current replay behavior.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Greptile SummaryThe PR enriches WebSocket streaming and makes acknowledgement state explicit.
Confidence Score: 5/5The PR appears safe to merge. No blocking failure remains.
|
| Filename | Overview |
|---|---|
| src/praisonai-agents/praisonaiagents/gateway/protocols.py | Adds the canonical reasoning, tool-progress, and stream-error event vocabulary used by gateway adapters. |
| src/praisonai-bot/praisonai_bot/gateway/server.py | Relays richer streaming events, emits explicit accepted/final response states, handles unavailable agents terminally, and orders final delivery after scheduled stream sends. |
Sequence Diagram
sequenceDiagram
participant Client
participant Gateway
participant Agent
Client->>Gateway: message
Gateway-->>Client: response (status: accepted)
Gateway->>Agent: execute turn
loop Live progress
Agent-->>Gateway: stream event
Gateway-->>Client: token/reasoning/tool/error event
end
Gateway->>Gateway: drain scheduled stream sends
Gateway-->>Client: response (status: final, outcome)
Reviews (3): Last reviewed commit: "fix: persist new stream frames, terminat..." | Re-trigger Greptile
|
@claude You are the FINAL architecture reviewer. If the branch is under MervinPraison/PraisonAI (not a fork), you are able to make modifications to this branch and push directly. SCOPE: Focus ONLY on Python packages (praisonaiagents, praisonai). Do NOT modify praisonai-rust or praisonai-ts. Read ALL comments above from Gemini, Qodo, CodeRabbit, and Copilot carefully before responding. Phase 1: Review per AGENTS.md
Phase 2: FIX Valid Issues Phase 3: Final Verdict |
MervinPraison
added
pipeline/awaiting-merge-gate
|
Claude finished @MervinPraison's task —— View job Final Architecture Review — PR #3471
Phase 1 — Architecture (per AGENTS.md)
Phase 2 — Fixes applied (pushed to this branch)
Skipped (with reason)
Testing
Phase 3 — Verdict✅ Approve. All valid reviewer findings are fixed; change is minimal, protocol-driven, backward-compatible, and adds clear SDK value. No blocking issues remain. Commit: |
MervinPraison
added
pipeline/blocked:stale-final
praisonai-triage-agent
Bot
removed
pipeline/blocked:ci
|
@claude this PR has merge conflicts with |
|
Claude finished @MervinPraison's task —— View job Rebase PR #3471 onto latest
|
Forward a closed set of structured progress events (reasoning, tool progress, stream errors) to WS clients instead of dropping everything but text/tool-call/end, and disambiguate the overloaded response frame with status: "accepted" (provisional) vs status: "final" (with a structured outcome). Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>
…final after stream - Persist reasoning/tool-progress/stream-error frames for reconnect replay (CodeRabbit) - Missing agent now resolves terminally (final+error) instead of a dangling accepted ack (Greptile P1) - Drain relay futures before final frame so final never overtakes trailing stream events (Greptile P1) Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>
a69c5db to
2b6bf40
Compare
MervinPraison
added
pipeline/blocked:ci
praisonai-triage-agent
Bot
removed
pipeline/blocked:ci
1 participant
Fixes #3469
Summary
Makes the gateway's live WS output richer and unambiguous, so a real-time UI can render live progress and reliably tell an accepted ack from the final result.
1. Structured event relay (was lossy)
_make_stream_relaypreviously forwarded onlyDELTA_TEXT/DELTA_TOOL_CALL/STREAM_ENDand dropped everything else. It now forwards a closed set of structured progress events under stableEventTypes:REASONING_STREAM(aDELTA_TEXTwithis_reasoning=True)TOOL_PROGRESS_STREAMSTREAM_ERRORTOKEN_STREAM, tool-call →TOOL_CALL_STREAM, end →STREAM_ENDThese are advertised in the
hello_okfeatures list when the client negotiatesstreaming.2. Accepted-vs-final acknowledgement (was overloaded)
The provisional ack ("Started processing."/"Message queued.") and the real answer both used
type: "response". Now:status: "accepted"status: "final"plus a structuredoutcome({"status": "ok"|"error"|"rejected"}) alongside the text.Layer placement
praisonaiagents/gateway/protocols.py): +3EventTypevalues (canonical vocabulary shared by all channel adapters).praisonai_bot/gateway/server.py): relay mapping + frame emission (heavy behaviour).Minimal and backward-compatible: existing
type: "response"/token_streamconsumers keep working; new fields are additive.Testing
-k gateway).tests/unit/config/test_precedence_ladder.py(missingresolve_outputimport) — not touched by this change.Generated with Claude Code
Summary by CodeRabbit