Claude/issue 1471 20260420 0633#1479
Conversation
Adds core capabilities for agents to learn and persist knowledge as skills: Core SDK (praisonaiagents): - Add SkillMutatorProtocol for agent-managed skill CRUD operations - Extend LearnConfig with nudge_interval, propose_skills fields - Add _maybe_emit_nudge() method for periodic knowledge persistence prompts - Wire improvements auto-extraction in LearnManager.process_conversation() Wrapper (praisonai): - Create skill_manage tool for agent skill management - Implement BasicSkillMutator with safe propose/approve workflow - Add integration test demonstrating end-to-end self-improving loop This enables the core self-improving claim: agents can learn procedures and persist them as skills for reuse across sessions. Fixes #1471 🤖 Generated with [Claude Code](https://claude.ai/code) Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>
- Add comprehensive path validation to prevent directory traversal attacks
- Fix broken propose/approve workflow for non-create actions
- Add proper validation for all skill name parameters
- Fix integration test to use isolated storage instead of global state
- Fix performance issue: lazy initialization of skill mutator
- Ensure edit action properly stages proposals instead of creating {name}_edit
Security improvements:
- Validate file_path parameters in write_file, remove_file, patch
- Reject dangerous patterns: .., /, \, ~
- Use resolved path checks to prevent escape attempts
- Add name validation to all CRUD operations
Fixes issues identified in Copilot security review.
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>
- Wire _maybe_emit_nudge() into post-turn flow in tool_execution.py (both sync and async paths). Previously defined but never invoked, so LearnConfig(nudge_interval>0) had no effect. - Nudge is appended to chat_history as a system message so it is visible on the next LLM call. Zero overhead when nudge_interval=0. - Remove root-level SELF_IMPROVING_AGENT_FEATURES.md (scope creep; not requested by issue #1471).
…o claude/issue-1471-20260420-0633
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
MervinPraison has reached the 50-review limit for trial accounts. To continue receiving code reviews, upgrade your plan.
📝 WalkthroughWalkthroughAdds a push/real-time subsystem: new push configs and protocol types, a PushClient SDK with WebSocket and polling transports, models, transport implementations, gateway export updates, and unit tests covering the new functionality. Changes
Sequence DiagramsequenceDiagram
participant App as Application
participant Client as PushClient
participant Transport as Transport<br/>(WebSocket / Polling)
participant Gateway as Gateway Server
App->>Client: connect()
activate Client
Client->>Transport: connect()
activate Transport
Transport->>Gateway: establish connection
Gateway-->>Transport: ack connection
Transport-->>Client: connected
deactivate Transport
Client->>Client: start receive_loop & heartbeat_loop
deactivate Client
App->>Client: subscribe(channel)
activate Client
Client->>Transport: send({type: "channel.subscribe", channel})
Transport->>Gateway: forward subscribe
deactivate Client
Gateway->>Transport: send({type: "channel_message", event_id, channel, data})
activate Transport
Transport->>Client: deliver message
deactivate Transport
activate Client
Client->>Client: dispatch to callbacks
Client->>Transport: send({type: "message_ack", event_id})
Transport->>Gateway: forward ack
deactivate Client
App->>Client: publish(channel, data)
activate Client
Client->>Transport: send({type: "channel.publish", channel, data})
Transport->>Gateway: forward publish
deactivate Client
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested labels
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 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 |
|
@copilot Do a thorough review of this PR. Read ALL existing reviewer comments above from Qodo, Coderabbit, and Gemini first — incorporate their findings. Review areas:
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a push notification system for PraisonAI Agents, featuring gateway protocols, configuration structures, and a client SDK with WebSocket and HTTP long-polling support. The implementation includes presence tracking and delivery guarantees. Feedback focuses on addressing security risks by moving authentication tokens from URLs to headers, ensuring the receive loop remains non-blocking during message acknowledgments and user callbacks, and improving the robustness of URL derivation and HTTP error handling in the polling transport.
| url = self._url | ||
| if self._auth_token: | ||
| sep = "&" if "?" in url else "?" | ||
| url = f"{url}{sep}token={self._auth_token}" |
There was a problem hiding this comment.
Passing authentication tokens in the URL query parameters is a security risk as they can be leaked in server logs, proxy logs, or browser history. The websockets library supports sending custom headers via the extra_headers argument. It is highly recommended to use the Authorization header instead.
| url = self._url | |
| if self._auth_token: | |
| sep = "&" if "?" in url else "?" | |
| url = f"{url}{sep}token={self._auth_token}" | |
| url = self._url | |
| headers = {} | |
| if self._auth_token: | |
| headers["Authorization"] = f"Bearer {self._auth_token}" | |
| self._ws = await websockets.connect(url, extra_headers=headers) |
| "channel.list", "presence.heartbeat_ack", "presence.list", | ||
| "joined", "left", "error", | ||
| ): | ||
| await self._send({"type": "message_ack", "event_id": event_id}) |
There was a problem hiding this comment.
In PollingTransport, _send performs an HTTP request which is awaited here. This blocks the entire receive loop, preventing the client from processing subsequent messages until the acknowledgment is sent. Consider using asyncio.create_task to send ACKs in the background to maintain throughput.
| await self._send({"type": "message_ack", "event_id": event_id}) | |
| asyncio.create_task(self._send({"type": "message_ack", "event_id": event_id})) |
| for cb in self._channel_callbacks.get(channel, []): | ||
| try: | ||
| await cb(msg) | ||
| except Exception as e: | ||
| logger.error("Channel callback error: %s", e) |
There was a problem hiding this comment.
User callbacks are currently awaited sequentially within the receive loop. A slow or blocking callback will prevent the client from receiving further messages or sending heartbeats. Additionally, this implementation assumes all callbacks are coroutines, which will fail with a TypeError if a user provides a synchronous function (unlike the event handler logic implemented below).
| for cb in self._channel_callbacks.get(channel, []): | |
| try: | |
| await cb(msg) | |
| except Exception as e: | |
| logger.error("Channel callback error: %s", e) | |
| for cb in self._channel_callbacks.get(channel, []): | |
| try: | |
| if asyncio.iscoroutinefunction(cb): | |
| asyncio.create_task(cb(msg)) | |
| else: | |
| cb(msg) | |
| except Exception as e: | |
| logger.error("Channel callback error: %s", e) |
| async def get_presence( | ||
| self, channel: Optional[str] = None, | ||
| ) -> List[PresenceInfo]: | ||
| """Query presence for a channel or all clients.""" | ||
| await self._send({ | ||
| "type": "presence.query", | ||
| "channel": channel, | ||
| }) | ||
| # The response will come via the receive loop | ||
| # For a synchronous result, we'd need a request-response pattern | ||
| # For now, presence info arrives via event handlers | ||
| return [] |
There was a problem hiding this comment.
The get_presence method is misleading as it always returns an empty list while the actual data arrives asynchronously via event handlers. The return type hint List[PresenceInfo] suggests a synchronous result. Consider changing the return type to None or implementing a request-response pattern using a Future to return the actual data.
| async with self._session.post( | ||
| f"{self._base_url}/api/push/poll/register", | ||
| json={}, | ||
| headers=headers, | ||
| ) as resp: | ||
| data = await resp.json() | ||
| self._poll_token = data["poll_token"] |
There was a problem hiding this comment.
The polling transport does not check the HTTP response status code. If the server returns an error (e.g., 401 Unauthorized or 500 Internal Server Error), resp.json() might fail or return an unexpected structure, leading to a KeyError when accessing poll_token. Use resp.raise_for_status() to ensure the request was successful.
| async with self._session.post( | |
| f"{self._base_url}/api/push/poll/register", | |
| json={}, | |
| headers=headers, | |
| ) as resp: | |
| data = await resp.json() | |
| self._poll_token = data["poll_token"] | |
| async with self._session.post( | |
| f"{self._base_url}/api/push/poll/register", | |
| json={}, | |
| headers=headers, | |
| ) as resp: | |
| resp.raise_for_status() | |
| data = await resp.json() | |
| self._poll_token = data["poll_token"] | |
| self._client_id = data["client_id"] |
| http_url = self._url.replace("ws://", "http://").replace("wss://", "https://") | ||
| # Remove /ws path suffix if present | ||
| if http_url.endswith("/ws"): | ||
| http_url = http_url[:-3] |
There was a problem hiding this comment.
Deriving the HTTP URL from the WebSocket URL using simple string replacement and suffix checking is fragile. It may fail if the URL structure differs slightly (e.g., trailing slashes or 'ws' appearing in the hostname). Using urllib.parse provides a more robust way to manipulate URLs.
from urllib.parse import urlparse, urlunparse
parsed = urlparse(self._url)
scheme = "https" if parsed.scheme == "wss" else "http"
path = parsed.path.rstrip("/")
if path.endswith("/ws"):
path = path[:-3]
http_url = urlunparse(parsed._replace(scheme=scheme, path=path))
Reviewed at commit Current external reviewer state: CodeRabbit has only a review in progress note; there are no actionable Qodo/Gemini findings posted yet. Findings:
# safer pattern (when server supports it)
await websockets.connect(self._url, additional_headers={"Authorization": f"******"})
async def _send(self, data):
if not (self._transport and self._transport.is_connected):
raise ConnectionError("Push client is not connected")
await self._transport.send(data)
Suggested concrete improvements:
UI screenshot: N/A (no UI changes in this PR). |
|
@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 |
|
Claude finished @MervinPraison's task —— View job Final Architecture Review & Fixes ✅All critical issues have been addressed and fixes implemented. Todo List:
Architecture Review Results✅ COMPLIANT with AGENTS.md guidelines:
Fixes Implemented (Commit: 58bf8bc)🔒 Security Fixes
⚡ Performance Fixes
🛡️ Error Handling Improvements
🧹 Code Quality
Final Verdict: ✅ APPROVEThis PR is ready for merge after addressing all critical reviewer feedback:
The push notification system provides a solid foundation for real-time communication in the PraisonAI ecosystem while maintaining the framework's core design principles. 🔗 Job Run: View Actions |
There was a problem hiding this comment.
Pull request overview
Adds initial push-notification client SDK support to praisonaiagents, alongside gateway-side protocol/config dataclasses and unit tests to validate the new interfaces.
Changes:
- Introduces push client SDK (
PushClient) with WebSocket transport and HTTP long-polling fallback transport. - Extends gateway protocols/config to include push channels, presence, delivery guarantees, and push configuration.
- Adds unit tests covering new dataclasses/protocol shapes and PushClient behavior with a mock transport.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| src/praisonai-agents/tests/unit/test_push_protocols.py | Tests new gateway push-related enums/dataclasses/protocols (currently contains duplicated module content). |
| src/praisonai-agents/tests/unit/test_push_client.py | Unit tests for PushClient using a mock transport. |
| src/praisonai-agents/praisonaiagents/push/transports.py | Implements WebSocketTransport and PollingTransport plus a transport protocol. |
| src/praisonai-agents/praisonaiagents/push/models.py | Adds ChannelMessage model wrapper for channel events. |
| src/praisonai-agents/praisonaiagents/push/client.py | Adds PushClient with subscribe/publish, presence heartbeat, reconnect + polling fallback. |
| src/praisonai-agents/praisonaiagents/push/init.py | Adds lazy-loading exports for push client/transports. |
| src/praisonai-agents/praisonaiagents/gateway/protocols.py | Extends EventType and adds push/presence/delivery protocol definitions + dataclasses. |
| src/praisonai-agents/praisonaiagents/gateway/config.py | Adds push-related config dataclasses and includes push in GatewayConfig. |
| src/praisonai-agents/praisonaiagents/gateway/init.py | Re-exports new push protocols/configs from the gateway module. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| async with self._session.post( | ||
| f"{self._base_url}/api/push/poll/register", | ||
| json={}, | ||
| headers=headers, | ||
| ) as resp: | ||
| data = await resp.json() | ||
| self._poll_token = data["poll_token"] | ||
| self._client_id = data["client_id"] | ||
| self._connected = True | ||
| logger.debug("Polling transport registered: %s", self._client_id) |
There was a problem hiding this comment.
PollingTransport.connect() creates an aiohttp.ClientSession() and then performs the register request, but if the request/JSON decode fails an exception will escape and the session will remain open. Wrap the register call in a try/except and ensure the session is closed (or use an async context manager / resp.raise_for_status()) on failure.
| async with self._session.post( | |
| f"{self._base_url}/api/push/poll/register", | |
| json={}, | |
| headers=headers, | |
| ) as resp: | |
| data = await resp.json() | |
| self._poll_token = data["poll_token"] | |
| self._client_id = data["client_id"] | |
| self._connected = True | |
| logger.debug("Polling transport registered: %s", self._client_id) | |
| try: | |
| async with self._session.post( | |
| f"{self._base_url}/api/push/poll/register", | |
| json={}, | |
| headers=headers, | |
| ) as resp: | |
| resp.raise_for_status() | |
| data = await resp.json() | |
| self._poll_token = data["poll_token"] | |
| self._client_id = data["client_id"] | |
| self._connected = True | |
| logger.debug("Polling transport registered: %s", self._client_id) | |
| except Exception: | |
| self._connected = False | |
| self._poll_token = None | |
| self._client_id = None | |
| if self._session is not None: | |
| await self._session.close() | |
| self._session = None | |
| raise |
| async def send(self, data: Dict[str, Any]) -> None: | ||
| if not self._connected or self._session is None: | ||
| raise ConnectionError("Polling transport not connected") | ||
|
|
||
| msg_type = data.get("type", "") | ||
| headers = {} | ||
| if self._auth_token: | ||
| headers["Authorization"] = f"Bearer {self._auth_token}" | ||
|
|
||
| # Map message types to polling endpoints | ||
| if msg_type == "channel.subscribe": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/subscribe", | ||
| json={"poll_token": self._poll_token, "channel": data.get("channel", "")}, | ||
| headers=headers, | ||
| ) | ||
| elif msg_type == "channel.unsubscribe": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/unsubscribe", | ||
| json={"poll_token": self._poll_token, "channel": data.get("channel", "")}, | ||
| headers=headers, | ||
| ) | ||
| elif msg_type == "message_ack": |
There was a problem hiding this comment.
PollingTransport.send() ignores unsupported message types (e.g., channel.publish / channel.create) with no error, so PushClient.publish() becomes a silent no-op when the client is using polling fallback. Either implement the missing endpoints or raise an explicit error for unknown type values so callers can detect unsupported operations.
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/heartbeat", | ||
| json={"poll_token": self._poll_token, "status": data.get("status", "online")}, |
There was a problem hiding this comment.
For polling fallback, presence.heartbeat drops the metadata field (only status is forwarded). This makes PushClient.set_status(..., metadata=...) behave differently between WebSocket and polling modes. Include metadata in the heartbeat POST body (or document that metadata is unsupported in polling).
| await self._session.post( | |
| f"{self._base_url}/api/push/poll/heartbeat", | |
| json={"poll_token": self._poll_token, "status": data.get("status", "online")}, | |
| heartbeat_payload = { | |
| "poll_token": self._poll_token, | |
| "status": data.get("status", "online"), | |
| } | |
| if "metadata" in data: | |
| heartbeat_payload["metadata"] = data.get("metadata") | |
| await self._session.post( | |
| f"{self._base_url}/api/push/poll/heartbeat", | |
| json=heartbeat_payload, |
| """Tests for push notification protocols and dataclasses.""" | ||
|
|
||
| import time | ||
| import pytest | ||
|
|
||
| from praisonaiagents.gateway.protocols import ( | ||
| ChannelInfo, | ||
| DeliveryGuaranteeProtocol, | ||
| EventType, | ||
| GatewayEvent, | ||
| PresenceInfo, | ||
| PresenceProtocol, | ||
| PushChannelProtocol, | ||
| ) | ||
| from praisonaiagents.gateway.config import ( | ||
| DeliveryConfig, | ||
| GatewayConfig, | ||
| PollingConfig, | ||
| PresenceConfig, | ||
| PushConfig, | ||
| RedisConfig, | ||
| ) | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # EventType additions | ||
| # --------------------------------------------------------------------------- |
There was a problem hiding this comment.
The module content is duplicated starting at this second top-level docstring/import block. This redefines the same test classes twice and makes the file much harder to maintain (and only the second set of class definitions will be collected). Remove the duplicated second half of the file.
| """Tests for push notification protocols and dataclasses.""" | |
| import time | |
| import pytest | |
| from praisonaiagents.gateway.protocols import ( | |
| ChannelInfo, | |
| DeliveryGuaranteeProtocol, | |
| EventType, | |
| GatewayEvent, | |
| PresenceInfo, | |
| PresenceProtocol, | |
| PushChannelProtocol, | |
| ) | |
| from praisonaiagents.gateway.config import ( | |
| DeliveryConfig, | |
| GatewayConfig, | |
| PollingConfig, | |
| PresenceConfig, | |
| PushConfig, | |
| RedisConfig, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # EventType additions | |
| # --------------------------------------------------------------------------- | |
| # --------------------------------------------------------------------------- |
|
|
||
| @pytest.fixture | ||
| def client(mock_transport): | ||
| c = PushClient("ws://test:8765/ws", auto_reconnect=False) |
There was a problem hiding this comment.
The client fixture marks PushClient as connected, but the MockTransport remains disconnected (connected=False). Because PushClient._send() checks self._transport.is_connected, none of the subscribe/publish/unsubscribe messages will be sent and the receive loop will exit early. Update the fixture to connect the transport (or set mock_transport.connected = True) before running these tests.
| c = PushClient("ws://test:8765/ws", auto_reconnect=False) | |
| c = PushClient("ws://test:8765/ws", auto_reconnect=False) | |
| mock_transport.connected = True |
| except Exception as e: | ||
| logger.warning("WebSocket connect failed: %s", e) | ||
| if self._fallback_to_polling: | ||
| await self._switch_to_polling() | ||
| else: | ||
| raise | ||
|
|
||
| self._closed_event.clear() | ||
| self._receive_task = asyncio.ensure_future(self._receive_loop()) | ||
| self._heartbeat_task = asyncio.ensure_future(self._heartbeat_loop()) |
There was a problem hiding this comment.
If the WebSocket connect fails and _switch_to_polling() also fails, connect() will still clear _closed_event and start the background tasks without raising an error. This can make await client.connect() appear successful even though the client is not connected. Consider re-raising on total connection failure (or return a success boolean) and only start tasks after a transport has connected.
| if self._transport and self._transport.is_connected: | ||
| await self._transport.send(data) |
There was a problem hiding this comment.
_send() silently drops messages when the transport is missing or disconnected. That makes higher-level APIs like subscribe()/publish() succeed locally (they update internal state) while not actually sending anything to the gateway. Consider raising a ConnectionError (or queueing messages until connected) so callers can handle the failure explicitly.
| if self._transport and self._transport.is_connected: | |
| await self._transport.send(data) | |
| if self._transport is None: | |
| raise ConnectionError("Push client transport is not available.") | |
| if not self._transport.is_connected: | |
| raise ConnectionError("Push client transport is disconnected.") | |
| await self._transport.send(data) |
| url = self._url | ||
| if self._auth_token: | ||
| sep = "&" if "?" in url else "?" | ||
| url = f"{url}{sep}token={self._auth_token}" | ||
|
|
||
| self._ws = await websockets.connect(url) |
There was a problem hiding this comment.
The WebSocket auth token is appended to the URL query string (...&token=...). Tokens in URLs are prone to leaking via logs, proxies, browser history, etc. Prefer passing the token via websockets.connect(..., extra_headers={"Authorization": "Bearer ..."}) or a subprotocol/header mechanism consistent with other WebSocket code in this repo.
| url = self._url | |
| if self._auth_token: | |
| sep = "&" if "?" in url else "?" | |
| url = f"{url}{sep}token={self._auth_token}" | |
| self._ws = await websockets.connect(url) | |
| extra_headers = None | |
| if self._auth_token: | |
| extra_headers = {"Authorization": f"Bearer {self._auth_token}"} | |
| self._ws = await websockets.connect(self._url, extra_headers=extra_headers) |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/praisonai-agents/praisonaiagents/gateway/config.py (1)
222-237:⚠️ Potential issue | 🟠 MajorParse
pushwhen loading gateway config.
GatewayConfig.to_dict()now emitspush, butMultiChannelGatewayConfig.from_dict()still buildsGatewayConfigwithout readinggw_data["push"]. A YAML config withgateway.push.enabled: truewill round-trip back to disabled/defaults.🐛 Suggested direction
+ push_data = gw_data.get("push", {}) + redis_data = push_data.get("redis") gateway_config = GatewayConfig( host=gw_data.get("host", "127.0.0.1"), port=gw_data.get("port", 8765), cors_origins=gw_data.get("cors_origins", []), auth_token=gw_data.get("auth_token"), max_connections=gw_data.get("max_connections", 1000), + push=PushConfig( + enabled=push_data.get("enabled", False), + redis=RedisConfig(**redis_data) if isinstance(redis_data, dict) else None, + presence=PresenceConfig(**push_data.get("presence", {})), + delivery=DeliveryConfig(**push_data.get("delivery", {})), + polling=PollingConfig(**push_data.get("polling", {})), + ) if isinstance(push_data, dict) else PushConfig(), )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/praisonai-agents/praisonaiagents/gateway/config.py` around lines 222 - 237, GatewayConfig.to_dict now includes "push", but MultiChannelGatewayConfig.from_dict doesn't read gw_data["push"], so rehydrate the PushConfig when building GatewayConfig: in MultiChannelGatewayConfig.from_dict (look for the gw_data usage and GatewayConfig(...) construction) add push=PushConfig.from_dict(gw_data.get("push", {})) or equivalent (falling back to PushConfig() when absent) so the GatewayConfig created mirrors the to_dict output; ensure you import or reference PushConfig and use its from_dict or constructor to populate the push field.
🧹 Nitpick comments (2)
src/praisonai-agents/praisonaiagents/push/client.py (2)
95-96: Preferasyncio.create_taskoverasyncio.ensure_futurefor coroutines.
create_taskis the documented, idiomatic API for scheduling coroutines on the running loop (since 3.7) and expresses intent more clearly thanensure_future, which is a general-purpose future coercer.- self._receive_task = asyncio.ensure_future(self._receive_loop()) - self._heartbeat_task = asyncio.ensure_future(self._heartbeat_loop()) + self._receive_task = asyncio.create_task(self._receive_loop()) + self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 95 - 96, Replace use of asyncio.ensure_future when scheduling the two coroutine loops with asyncio.create_task to follow the modern idiom: change the assignments to self._receive_task and self._heartbeat_task so they call asyncio.create_task(self._receive_loop()) and asyncio.create_task(self._heartbeat_loop()) respectively; locate where asyncio.ensure_future is used to schedule _receive_loop and _heartbeat_loop and swap to asyncio.create_task without changing the coroutine functions themselves.
285-285: Redundant single-element tuple membership check.
msg_type in ("channel_message",)is equivalent tomsg_type == "channel_message". Use the direct comparison (or extend the tuple if more types are intended later).- if channel and msg_type in ("channel_message",): + if channel and msg_type == "channel_message":🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/praisonai-agents/praisonaiagents/push/client.py` at line 285, The condition uses a redundant single-element tuple membership check; in the if statement that reads something like `if channel and msg_type in ("channel_message",)`, replace the membership check with a direct equality check `msg_type == "channel_message"` (or, if you actually intend multiple message types later, expand the tuple/list to include them) so locate the conditional in the push client code (around the handler that inspects `channel` and `msg_type`) and update that expression accordingly.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/praisonai-agents/praisonaiagents/gateway/protocols.py`:
- Around line 516-686: The protocol interfaces mix synchronous channel/presence
methods (PushChannelProtocol methods like add_channel, get_channel,
list_channels, subscribe_client, get_subscribers, get_client_channels;
PresenceProtocol methods like get_presence, get_all_presence, get_online_count)
with purely async delivery methods (DeliveryGuaranteeProtocol async methods),
which risks blocking the event loop; introduce explicit sync/async counterparts
or split into separate in-memory vs I/O-backed protocols: either (A) add async
versions for all sync-read methods (e.g., async_get_channel,
async_list_channels, async_get_presence, async_get_all_presence,
async_get_online_count) and/or add sync wrappers for DeliveryGuaranteeProtocol
if needed, or (B) split into PushChannelSync/PushChannelAsync and
PresenceSync/PresenceAsync (keeping DeliveryGuaranteeProtocol as async-only) and
update any adapters to implement the appropriate interface so I/O-backed
adapters implement async variants and in-memory adapters implement sync
variants—do not perform blocking I/O in any async method and ensure method names
above are used to locate and update implementations.
In `@src/praisonai-agents/praisonaiagents/push/__init__.py`:
- Around line 45-51: The __all__ list in the push package is unsorted and
triggers Ruff RUF022; sort the entries in the __all__ list alphabetically (e.g.,
"ChannelMessage", "PollingTransport", "PushClient", "PushTransportProtocol",
"WebSocketTransport") so the module-level export order is deterministic and
lint-clean; update the __all__ assignment in
src/praisonai-agents/praisonaiagents/push/__init__.py accordingly.
In `@src/praisonai-agents/praisonaiagents/push/client.py`:
- Around line 179-190: The get_presence method currently just sends a
presence.query via self._send and always returns an empty list; implement a
request-response pattern instead: generate a unique request id (e.g., event_id),
attach it to the message sent in get_presence, create a Future/asyncio.Event
mapped by that id on the client (e.g., self._pending_requests), await that
future in get_presence, and resolve the future inside the receive loop (the
method handling incoming messages, e.g., _receive_loop or the handler for
"presence.list") when a response with the matching event_id arrives;
alternatively, if you prefer fire-and-forget, rename get_presence to
request_presence and return None, clarifying in the docstring that results come
via the on("presence.list") event handler—pick one approach and update
get_presence, the message payload (include event_id), and the response handling
in _receive_loop/on("presence.list") accordingly.
- Around line 242-245: The _send method currently drops messages when there's no
active connection; update async def _send(self, data: Dict[str, Any]) in
client.py to detect a disconnected transport (self._transport is None or not
self._transport.is_connected) and raise a clear, specific exception (e.g.,
ConnectionError or a custom DisconnectedError) with a remediation hint (suggest
reconnect() or queueing for retry) instead of silently returning; keep the
existing path that awaits self._transport.send(data) when connected, and ensure
callers of subscribe/unsubscribe/publish/create_channel/get_presence/set_status
will receive the error so they can handle reconnect/queue logic.
- Around line 23-43: The PushClient class is a heavy concrete implementation
that violates core SDK protocol-driven rules; extract a lightweight protocol and
config in this package and move the implementation into the wrapper: define a
PushClientProtocol (typing.Protocol) and a PushClientConfig dataclass here to
describe connect(), disconnect(), subscribe(), publish(), on(), wait_closed(),
and properties used by callers, then remove the concrete logic from PushClient
and replace it with a thin adapter that implements the protocol by delegating to
an implementation injected from the praisonai wrapper; move all heavy methods
and internals (_receive_loop, _heartbeat_loop, _reconnect, _switch_to_polling,
transport selection, URL rewriting, reconnection/heartbeat state) into the
wrapper’s concrete PushClient implementation and ensure the core package only
exports the protocol and config types.
- Around line 319-332: The _heartbeat_loop currently swallows all exceptions
with "except Exception: pass", hiding real errors; update async def
_heartbeat_loop to catch exceptions the same way _receive_loop does: log the
exception (include exception info) via the same logger or processLogger used in
the module, and set self._connected = False or trigger the same
reconnect/cleanup path after logging so failures in _send (or
serialization/transport errors) are visible and handled; reference the
_heartbeat_loop coroutine, the _send call inside it, and mirror the exception
handling/logging pattern used by _receive_loop.
- Around line 220-236: In wait_for(), replace the explicit event-loop call
asyncio.get_event_loop().create_future() with asyncio.Future() to create the
awaiting future in the running async context; keep the rest of the logic (the
inner _one_shot callback, appending to self._channel_callbacks, awaiting with
asyncio.wait_for, and the finally cleanup that removes _one_shot) unchanged and
ensure the created future is typed as an asyncio.Future (e.g., representing
ChannelMessage) if you want static typing.
In `@src/praisonai-agents/praisonaiagents/push/models.py`:
- Around line 31-40: The from_event_dict in ChannelMessage currently reads
top-level keys but GatewayEvent.to_dict nests event-specific fields under
"data", so add nested fallbacks: for channel use event_dict.get("channel") or
event_dict.get("data", {}).get("channel", "") (similarly for source if needed),
and for event_id/timestamp consider falling back to event_dict.get("data",
{}).get("event_id", ...) and .get("timestamp", ...) so
ChannelMessage.from_event_dict returns the correct values when the payload is
shaped like {"type":"channel_message","data":{...}}.
In `@src/praisonai-agents/praisonaiagents/push/transports.py`:
- Around line 68-73: Current code appends the raw _auth_token into the WebSocket
URL which risks leaking credentials; instead, when calling
websockets.connect(...) pass the token in an Authorization header (e.g.,
"Authorization: Bearer <token>") via the headers argument so the token is not in
the URL, falling back to URL-encoding the token with urllib.parse.quote_plus
only if the gateway absolutely requires query auth; update the block that builds
url and calls websockets.connect to use headers when _auth_token is set
(referencing _auth_token, _url, websockets.connect, and self._ws).
- Around line 17-40: PushTransportProtocol (the interface defining is_connected,
connect, disconnect, send, receive) currently resides with concrete transport
implementations; extract that Protocol into a new module protocols.py and export
PushTransportProtocol from there, then import and use it in transports.py so
transports implement the interface rather than declare it inline; ensure the
Protocol retains the same method/property names (is_connected, connect,
disconnect, send, receive) and async signatures and update any imports/usages to
reference the moved symbol.
- Around line 130-139: The connect() method opens a ClientSession and performs
registration but does not close the session on failure, causing a resource leak;
update connect() so that the registration block (the POST to
f"{self._base_url}/api/push/poll/register" and the JSON parsing that sets
self._poll_token, self._client_id, self._connected) is wrapped in
try/except/finally: on exception, await self._session.close() in the finally (or
in except before re-raising) and clear any partially-set state, and only set
self._connected and log success after the POST/JSON succeed; ensure errors from
malformed JSON or missing keys are caught and the session is always closed when
registration fails.
- Around line 151-180: The send() method currently ignores "channel.publish" and
leaks aiohttp connections by calling self._session.post(...) without context
managers; update send() to handle msg_type == "channel.publish" (messages coming
from PushClient.publish()) by POSTing to the polling publish endpoint (e.g.,
"/api/push/poll/publish") with the appropriate json (poll_token, channel,
message/body) and headers, and for all POST calls replace direct await
self._session.post(...) with the async with pattern used in register() and
receive() (async with self._session.post(...) as resp:) and call
resp.raise_for_status() to surface server errors; mirror the request payload
structure used elsewhere and reuse headers construction at the top of send().
In `@src/praisonai-agents/tests/unit/test_push_protocols.py`:
- Around line 147-151: The test test_redis_config_hides_secrets currently
includes hard-coded password/url literals that trigger Ruff S105/S106; change
the fixture to build those values dynamically (e.g., compose the password and
url from smaller pieces or variables) before passing them into
RedisConfig(password=..., url=...) so no explicit "password"/credential literal
appears in the source, then keep the same assertions against
RedisConfig.to_dict() to verify redaction. Alternatively, if you prefer
suppression, add a targeted noqa (e.g., # noqa: S105,S106) on the test function
definition to silence Ruff for this intentional test case.
- Around line 227-447: The file contains a duplicated copy of the test module
(repeat of imports and all test classes such as TestEventTypeAdditions,
TestChannelInfo, TestPresenceInfo, TestPushConfig, TestGatewayConfigPushField,
TestDeliveryConfig, TestPresenceConfig, TestPollingConfig, TestProtocolStructure
and their test methods), so remove the entire duplicated block leaving a single
canonical copy; locate the duplicate by finding repeated definitions of those
class names or repeated imports from praisonaiagents.gateway.protocols and
praisonaiagents.gateway.config and delete the second occurrence to keep only one
set of tests.
---
Outside diff comments:
In `@src/praisonai-agents/praisonaiagents/gateway/config.py`:
- Around line 222-237: GatewayConfig.to_dict now includes "push", but
MultiChannelGatewayConfig.from_dict doesn't read gw_data["push"], so rehydrate
the PushConfig when building GatewayConfig: in
MultiChannelGatewayConfig.from_dict (look for the gw_data usage and
GatewayConfig(...) construction) add
push=PushConfig.from_dict(gw_data.get("push", {})) or equivalent (falling back
to PushConfig() when absent) so the GatewayConfig created mirrors the to_dict
output; ensure you import or reference PushConfig and use its from_dict or
constructor to populate the push field.
---
Nitpick comments:
In `@src/praisonai-agents/praisonaiagents/push/client.py`:
- Around line 95-96: Replace use of asyncio.ensure_future when scheduling the
two coroutine loops with asyncio.create_task to follow the modern idiom: change
the assignments to self._receive_task and self._heartbeat_task so they call
asyncio.create_task(self._receive_loop()) and
asyncio.create_task(self._heartbeat_loop()) respectively; locate where
asyncio.ensure_future is used to schedule _receive_loop and _heartbeat_loop and
swap to asyncio.create_task without changing the coroutine functions themselves.
- Line 285: The condition uses a redundant single-element tuple membership
check; in the if statement that reads something like `if channel and msg_type in
("channel_message",)`, replace the membership check with a direct equality check
`msg_type == "channel_message"` (or, if you actually intend multiple message
types later, expand the tuple/list to include them) so locate the conditional in
the push client code (around the handler that inspects `channel` and `msg_type`)
and update that expression accordingly.
🪄 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
Run ID: 5fb536f7-b548-4ed8-ba91-e56cf55338d9
📒 Files selected for processing (9)
src/praisonai-agents/praisonaiagents/gateway/__init__.pysrc/praisonai-agents/praisonaiagents/gateway/config.pysrc/praisonai-agents/praisonaiagents/gateway/protocols.pysrc/praisonai-agents/praisonaiagents/push/__init__.pysrc/praisonai-agents/praisonaiagents/push/client.pysrc/praisonai-agents/praisonaiagents/push/models.pysrc/praisonai-agents/praisonaiagents/push/transports.pysrc/praisonai-agents/tests/unit/test_push_client.pysrc/praisonai-agents/tests/unit/test_push_protocols.py
| @runtime_checkable | ||
| class PushChannelProtocol(Protocol): | ||
| """Protocol for channel/topic-based push messaging. | ||
|
|
||
| Channels group clients by interest. Messages published to a channel | ||
| are delivered to all subscribed clients. | ||
| """ | ||
|
|
||
| def add_channel( | ||
| self, channel_name: str, metadata: Optional[Dict[str, Any]] = None, | ||
| ) -> bool: | ||
| """Create a named channel. | ||
|
|
||
| Args: | ||
| channel_name: Unique channel name | ||
| metadata: Optional channel metadata | ||
|
|
||
| Returns: | ||
| True if created, False if already exists | ||
| """ | ||
| ... | ||
|
|
||
| def remove_channel(self, channel_name: str) -> bool: | ||
| """Delete a channel and unsubscribe all clients. | ||
|
|
||
| Returns: | ||
| True if removed, False if not found | ||
| """ | ||
| ... | ||
|
|
||
| def get_channel(self, channel_name: str) -> Optional[ChannelInfo]: | ||
| """Get channel metadata.""" | ||
| ... | ||
|
|
||
| def list_channels(self) -> List[str]: | ||
| """List all active channel names.""" | ||
| ... | ||
|
|
||
| def subscribe_client(self, client_id: str, channel_name: str) -> bool: | ||
| """Subscribe a client to a channel. | ||
|
|
||
| Returns: | ||
| True if subscribed, False if already subscribed or channel missing | ||
| """ | ||
| ... | ||
|
|
||
| def unsubscribe_client(self, client_id: str, channel_name: str) -> bool: | ||
| """Unsubscribe a client from a channel. | ||
|
|
||
| Returns: | ||
| True if unsubscribed, False if not subscribed | ||
| """ | ||
| ... | ||
|
|
||
| def get_subscribers(self, channel_name: str) -> List[str]: | ||
| """List client IDs subscribed to a channel.""" | ||
| ... | ||
|
|
||
| def get_client_channels(self, client_id: str) -> List[str]: | ||
| """List channels a client is subscribed to.""" | ||
| ... | ||
|
|
||
| async def publish_to_channel( | ||
| self, | ||
| channel_name: str, | ||
| event: GatewayEvent, | ||
| exclude: Optional[List[str]] = None, | ||
| ) -> int: | ||
| """Publish an event to all subscribers of a channel. | ||
|
|
||
| Args: | ||
| channel_name: Target channel | ||
| event: The event to deliver | ||
| exclude: Optional client IDs to skip | ||
|
|
||
| Returns: | ||
| Number of clients the event was sent to | ||
| """ | ||
| ... | ||
|
|
||
|
|
||
| @runtime_checkable | ||
| class PresenceProtocol(Protocol): | ||
| """Protocol for tracking client presence (online/idle/offline).""" | ||
|
|
||
| async def track_presence( | ||
| self, | ||
| client_id: str, | ||
| status: str = "online", | ||
| metadata: Optional[Dict[str, Any]] = None, | ||
| ) -> None: | ||
| """Set or update a client's presence status. | ||
|
|
||
| Args: | ||
| client_id: Client identifier | ||
| status: Presence status ("online", "idle", "offline") | ||
| metadata: Optional client metadata | ||
| """ | ||
| ... | ||
|
|
||
| async def remove_presence(self, client_id: str) -> None: | ||
| """Remove a client's presence (on disconnect).""" | ||
| ... | ||
|
|
||
| def get_presence(self, client_id: str) -> Optional[PresenceInfo]: | ||
| """Get a single client's presence info.""" | ||
| ... | ||
|
|
||
| def get_all_presence( | ||
| self, channel_name: Optional[str] = None, | ||
| ) -> List[PresenceInfo]: | ||
| """Get presence info, optionally filtered by channel. | ||
|
|
||
| Args: | ||
| channel_name: If provided, only return presence for channel members | ||
| """ | ||
| ... | ||
|
|
||
| def get_online_count(self, channel_name: Optional[str] = None) -> int: | ||
| """Count online clients, optionally filtered by channel.""" | ||
| ... | ||
|
|
||
|
|
||
| @runtime_checkable | ||
| class DeliveryGuaranteeProtocol(Protocol): | ||
| """Protocol for at-least-once message delivery. | ||
|
|
||
| Messages are stored, tracked, and retried until acknowledged. | ||
| """ | ||
|
|
||
| async def store_message(self, event: GatewayEvent) -> str: | ||
| """Persist a message to the store. | ||
|
|
||
| Returns: | ||
| The event_id of the stored message | ||
| """ | ||
| ... | ||
|
|
||
| async def acknowledge(self, client_id: str, event_id: str) -> bool: | ||
| """Mark a message as acknowledged by a client. | ||
|
|
||
| Returns: | ||
| True if found and acknowledged, False if not found | ||
| """ | ||
| ... | ||
|
|
||
| async def nack(self, client_id: str, event_id: str) -> None: | ||
| """Negative acknowledge - request redelivery.""" | ||
| ... | ||
|
|
||
| async def get_unacknowledged( | ||
| self, client_id: str, limit: int = 100, | ||
| ) -> List[GatewayEvent]: | ||
| """Get pending unacknowledged messages for a client.""" | ||
| ... | ||
|
|
||
| async def retry_unacknowledged(self, client_id: str) -> int: | ||
| """Redeliver all unacknowledged messages to a client. | ||
|
|
||
| Returns: | ||
| Number of messages redelivered | ||
| """ | ||
| ... | ||
|
|
||
| async def purge_acknowledged(self, max_age_seconds: int = 86400) -> int: | ||
| """Remove old acknowledged messages from the store. | ||
|
|
||
| Returns: | ||
| Number of messages purged | ||
| """ | ||
| ... |
There was a problem hiding this comment.
Add sync/async counterparts before freezing these protocol APIs.
These operations can be backed by Redis or another remote store, but the protocol currently mixes sync-only channel/presence reads and async-only delivery operations. That forces either blocking I/O in async gateway paths or no sync API for sync callers. Please add explicit sync and async variants, or split pure in-memory APIs from I/O-backed adapters before this becomes public API. As per coding guidelines, “All I/O operations must have both sync and async variants; never block the event loop with sync I/O in async context.”
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/gateway/protocols.py` around lines 516 -
686, The protocol interfaces mix synchronous channel/presence methods
(PushChannelProtocol methods like add_channel, get_channel, list_channels,
subscribe_client, get_subscribers, get_client_channels; PresenceProtocol methods
like get_presence, get_all_presence, get_online_count) with purely async
delivery methods (DeliveryGuaranteeProtocol async methods), which risks blocking
the event loop; introduce explicit sync/async counterparts or split into
separate in-memory vs I/O-backed protocols: either (A) add async versions for
all sync-read methods (e.g., async_get_channel, async_list_channels,
async_get_presence, async_get_all_presence, async_get_online_count) and/or add
sync wrappers for DeliveryGuaranteeProtocol if needed, or (B) split into
PushChannelSync/PushChannelAsync and PresenceSync/PresenceAsync (keeping
DeliveryGuaranteeProtocol as async-only) and update any adapters to implement
the appropriate interface so I/O-backed adapters implement async variants and
in-memory adapters implement sync variants—do not perform blocking I/O in any
async method and ensure method names above are used to locate and update
implementations.
| __all__ = [ | ||
| "PushClient", | ||
| "ChannelMessage", | ||
| "WebSocketTransport", | ||
| "PollingTransport", | ||
| "PushTransportProtocol", | ||
| ] |
There was a problem hiding this comment.
Sort __all__ to satisfy Ruff RUF022.
Static analysis flags this list; sorting it avoids a lint-only failure.
🧹 Proposed fix
__all__ = [
- "PushClient",
"ChannelMessage",
- "WebSocketTransport",
"PollingTransport",
+ "PushClient",
"PushTransportProtocol",
+ "WebSocketTransport",
]🧰 Tools
🪛 Ruff (0.15.10)
[warning] 45-51: __all__ is not sorted
Apply an isort-style sorting to __all__
(RUF022)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/__init__.py` around lines 45 - 51,
The __all__ list in the push package is unsorted and triggers Ruff RUF022; sort
the entries in the __all__ list alphabetically (e.g., "ChannelMessage",
"PollingTransport", "PushClient", "PushTransportProtocol", "WebSocketTransport")
so the module-level export order is deterministic and lint-clean; update the
__all__ assignment in src/praisonai-agents/praisonaiagents/push/__init__.py
accordingly.
| class PushClient: | ||
| """Client SDK for receiving push notifications from a PraisonAI gateway. | ||
|
|
||
| Supports WebSocket (primary) with automatic fallback to HTTP polling. | ||
| Handles reconnections, channel subscriptions, and message ACKs transparently. | ||
|
|
||
| Example:: | ||
|
|
||
| client = PushClient("ws://localhost:8765/ws", auth_token="my-token") | ||
| await client.connect() | ||
|
|
||
| @client.on("channel_message") | ||
| async def on_msg(msg: ChannelMessage): | ||
| print(f"Got message on {msg.channel}: {msg.data}") | ||
|
|
||
| await client.subscribe("alerts", on_msg) | ||
| await client.publish("alerts", {"level": "info", "text": "hello"}) | ||
|
|
||
| # Block until disconnected | ||
| await client.wait_closed() | ||
| """ |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Heavy implementation in core SDK violates protocol-driven design guidelines.
This module contains a ~380-line concrete push client (reconnect loop, heartbeat, URL rewriting, transport selection, callback dispatch) living directly in praisonaiagents/. Per the project's Core SDK rules, praisonaiagents should contain only protocols, hooks, adapters, base classes, and decorators; heavy implementations belong in the praisonai wrapper package.
Consider splitting this into:
- A
PushClientProtocol(and any needed config dataclass likePushClientConfig) defined here in core SDK. - The concrete
PushClientimplementation (including_receive_loop,_heartbeat_loop,_reconnect,_switch_to_polling) moved to thepraisonaiwrapper, with core SDK only exposing the protocol and config.
As per coding guidelines: "Core SDK (praisonaiagents) must use protocol-driven design with typing.Protocol for all extension points, not heavy implementations" and the retrieved learning "Core SDK (praisonaiagents) must contain only protocols, hooks, adapters, base classes, and decorators — never heavy implementations".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 23 - 43,
The PushClient class is a heavy concrete implementation that violates core SDK
protocol-driven rules; extract a lightweight protocol and config in this package
and move the implementation into the wrapper: define a PushClientProtocol
(typing.Protocol) and a PushClientConfig dataclass here to describe connect(),
disconnect(), subscribe(), publish(), on(), wait_closed(), and properties used
by callers, then remove the concrete logic from PushClient and replace it with a
thin adapter that implements the protocol by delegating to an implementation
injected from the praisonai wrapper; move all heavy methods and internals
(_receive_loop, _heartbeat_loop, _reconnect, _switch_to_polling, transport
selection, URL rewriting, reconnection/heartbeat state) into the wrapper’s
concrete PushClient implementation and ensure the core package only exports the
protocol and config types.
| async def get_presence( | ||
| self, channel: Optional[str] = None, | ||
| ) -> List[PresenceInfo]: | ||
| """Query presence for a channel or all clients.""" | ||
| await self._send({ | ||
| "type": "presence.query", | ||
| "channel": channel, | ||
| }) | ||
| # The response will come via the receive loop | ||
| # For a synchronous result, we'd need a request-response pattern | ||
| # For now, presence info arrives via event handlers | ||
| return [] |
There was a problem hiding this comment.
get_presence returns empty list unconditionally — misleading API.
The signature (-> List[PresenceInfo]) and name imply a query result, but the method only sends a presence.query and always returns []; the actual response arrives asynchronously via on("presence.list"). Callers using await client.get_presence(...) will get no data and no error, which is a footgun.
Either implement a real request/response pattern (correlate via an event_id and await a future resolved in _receive_loop), or rename/reshape this to clearly be a fire-and-forget (e.g., request_presence() returning None) and document that results come via event handlers.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 179 - 190,
The get_presence method currently just sends a presence.query via self._send and
always returns an empty list; implement a request-response pattern instead:
generate a unique request id (e.g., event_id), attach it to the message sent in
get_presence, create a Future/asyncio.Event mapped by that id on the client
(e.g., self._pending_requests), await that future in get_presence, and resolve
the future inside the receive loop (the method handling incoming messages, e.g.,
_receive_loop or the handler for "presence.list") when a response with the
matching event_id arrives; alternatively, if you prefer fire-and-forget, rename
get_presence to request_presence and return None, clarifying in the docstring
that results come via the on("presence.list") event handler—pick one approach
and update get_presence, the message payload (include event_id), and the
response handling in _receive_loop/on("presence.list") accordingly.
| async def wait_for( | ||
| self, channel: str, timeout: float = 30.0, | ||
| ) -> ChannelMessage: | ||
| """Block until the next message on a channel, or raise TimeoutError.""" | ||
| future: asyncio.Future = asyncio.get_event_loop().create_future() | ||
|
|
||
| async def _one_shot(msg: ChannelMessage) -> None: | ||
| if not future.done(): | ||
| future.set_result(msg) | ||
|
|
||
| self._channel_callbacks.setdefault(channel, []).append(_one_shot) | ||
| try: | ||
| return await asyncio.wait_for(future, timeout=timeout) | ||
| finally: | ||
| cbs = self._channel_callbacks.get(channel, []) | ||
| if _one_shot in cbs: | ||
| cbs.remove(_one_shot) |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Python 3.12 asyncio.get_event_loop deprecation for running loop
💡 Result:
In Python 3.12, asyncio.get_event_loop is deprecated when called without a current event loop set in the thread. It emits a DeprecationWarning in such cases, and this behavior will become an error in a future Python release. When there is a running event loop (i.e., called from a coroutine or callback), it returns the running loop, equivalent to asyncio.get_running_loop, which is the preferred function in those contexts. The deprecation specifically targets the case with no current event loop, where previously it would create one via the policy; now it warns instead. For code needing the loop when it's running, replace with asyncio.get_running_loop, which raises RuntimeError if no loop is running. For top-level synchronous code needing a loop without one running, use asyncio.new_event_loop instead, then set and run it manually (or prefer asyncio.run for high-level entry points). Official docs recommend asyncio.run over manual loop management where possible.
Citations:
- 1: https://docs.python.org/3.12/library/asyncio-eventloop.html
- 2: https://docs.python.org/release/3.12.0/library/asyncio-eventloop.html
- 3: https://docs.python.org/3/library/asyncio-eventloop.html
- 4: https://docs.python.org/3.12/library/asyncio-policy.html
- 5: bug: asyncio.get_event_loop() deprecated in single_generate_async (openai_model.py & siliconflow_model.py) EvoAgentX/EvoAgentX#233
- 6: Finish deprecation in asyncio.get_event_loop() python/cpython#93453
- 7: https://stackoverflow.com/questions/78577538/replacement-for-deprecated-asyncio-get-event-loop
- 8: https://stackoverflow.com/questions/73361664/asyncio-get-event-loop-deprecationwarning-there-is-no-current-event-loop/73367187
🏁 Script executed:
# Check the actual file and method signature
fd -t f 'client.py' src/praisonai-agents/praisonaiagents/push/ | head -5Repository: MervinPraison/PraisonAI
Length of output: 118
🏁 Script executed:
# Check Python version requirements for the project
fd -t f 'setup.py|setup.cfg|pyproject.toml|requirements.txt' | head -10 | xargs grep -l 'python' 2>/dev/null | head -3Repository: MervinPraison/PraisonAI
Length of output: 173
🏁 Script executed:
# Verify the exact code in the file
if [ -f "src/praisonai-agents/praisonaiagents/push/client.py" ]; then
sed -n '215,240p' src/praisonai-agents/praisonaiagents/push/client.py
fiRepository: MervinPraison/PraisonAI
Length of output: 1090
🏁 Script executed:
# Check Python version requirements in the project's main pyproject.toml
cat src/praisonai-agents/pyproject.toml 2>/dev/null | grep -A 5 -B 5 "python" | head -20Repository: MervinPraison/PraisonAI
Length of output: 509
🏁 Script executed:
# Also check if there's a pyproject.toml in the praisonaiagents directory
find src/praisonai-agents -name "pyproject.toml" -exec cat {} \; | grep -A 5 -B 5 "python" | head -30Repository: MervinPraison/PraisonAI
Length of output: 605
🏁 Script executed:
# Check setup.py or setup.cfg if they exist for Python version
find src/praisonai-agents -name "setup.py" -o -name "setup.cfg" | xargs cat 2>/dev/null | grep -i "python" | head -10Repository: MervinPraison/PraisonAI
Length of output: 49
🌐 Web query:
Python 3.10 asyncio.get_event_loop deprecation async context
💡 Result:
In Python 3.10, asyncio.get_event_loop emits a DeprecationWarning when called without a running event loop or current loop set in the thread. This behavior started in Python 3.10.0–3.10.8 (warning even if policy loop set), refined in 3.10.9+ (warning only if no running or policy loop). The warning indicates "There is no current event loop". In future versions (post-3.14, fully by 3.16 with policy removal), it will raise RuntimeError if no suitable loop exists. The preferred replacement depends on context: - Inside coroutines/async functions (async context): Use asyncio.get_running_loop, which requires a running loop and raises RuntimeError otherwise. - At top-level (no running loop): Avoid get_event_loop; use high-level asyncio.run(coro) for simple cases, which creates, runs, and closes a loop automatically. - For manual loop control: Use loop = asyncio.new_event_loop; asyncio.set_event_loop(loop); then loop.run_until_complete or similar. There is no built-in synchronous context manager for event loops. asyncio.Runner (since 3.11) is a synchronous context manager for running multiple coros in one loop: with asyncio.Runner as runner: runner.run(coro1) runner.run(coro2) Custom async context managers exist for servers (e.g., async with srv:), but for the loop itself, manual management or Runner is recommended. Custom classes can wrap new_event_loop in enter/exit as shown in community examples, but official docs favor high-level APIs over low-level loop handling.
Citations:
- 1: https://docs.python.org/release/3.10.10/library/asyncio-eventloop.html
- 2: https://docs.python.org/3.10/library/asyncio-eventloop.html
- 3: https://docs.python.org/3/library/asyncio-eventloop.html
- 4: https://docs.python.org/3.11/library/asyncio-eventloop.html
- 5: https://docs.python.org/library/asyncio-runner.html
- 6: emulated_hue: asyncio.get_event_loop() deprecated in async_create_upnp_datagram_endpoint home-assistant/core#167755
- 7: bug: asyncio.get_event_loop() deprecated in single_generate_async (openai_model.py & siliconflow_model.py) EvoAgentX/EvoAgentX#233
- 8: https://stackoverflow.com/questions/73361664/asyncio-get-event-loop-deprecationwarning-there-is-no-current-event-loop/73367187
- 9: https://stackoverflow.com/questions/72806119/python-asyncio-eventloop-in-a-contextmanager
- 10: https://docs.python.org/3/library/asyncio-runner.html
Use asyncio.Future() instead of asyncio.get_event_loop().create_future().
Since wait_for() is an async method, a running loop is guaranteed. asyncio.Future() is the standard pattern for creating futures in async contexts and is simpler than explicitly referencing the event loop. While asyncio.get_event_loop() doesn't currently emit a deprecation warning when called from within an async function with a running loop, using asyncio.Future() is more direct and aligns with modern asyncio best practices.
Proposed fix
- future: asyncio.Future = asyncio.get_event_loop().create_future()
+ future: asyncio.Future = asyncio.Future()📝 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.
| async def wait_for( | |
| self, channel: str, timeout: float = 30.0, | |
| ) -> ChannelMessage: | |
| """Block until the next message on a channel, or raise TimeoutError.""" | |
| future: asyncio.Future = asyncio.get_event_loop().create_future() | |
| async def _one_shot(msg: ChannelMessage) -> None: | |
| if not future.done(): | |
| future.set_result(msg) | |
| self._channel_callbacks.setdefault(channel, []).append(_one_shot) | |
| try: | |
| return await asyncio.wait_for(future, timeout=timeout) | |
| finally: | |
| cbs = self._channel_callbacks.get(channel, []) | |
| if _one_shot in cbs: | |
| cbs.remove(_one_shot) | |
| async def wait_for( | |
| self, channel: str, timeout: float = 30.0, | |
| ) -> ChannelMessage: | |
| """Block until the next message on a channel, or raise TimeoutError.""" | |
| future: asyncio.Future = asyncio.Future() | |
| async def _one_shot(msg: ChannelMessage) -> None: | |
| if not future.done(): | |
| future.set_result(msg) | |
| self._channel_callbacks.setdefault(channel, []).append(_one_shot) | |
| try: | |
| return await asyncio.wait_for(future, timeout=timeout) | |
| finally: | |
| cbs = self._channel_callbacks.get(channel, []) | |
| if _one_shot in cbs: | |
| cbs.remove(_one_shot) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 220 - 236,
In wait_for(), replace the explicit event-loop call
asyncio.get_event_loop().create_future() with asyncio.Future() to create the
awaiting future in the running async context; keep the rest of the logic (the
inner _one_shot callback, appending to self._channel_callbacks, awaiting with
asyncio.wait_for, and the finally cleanup that removes _one_shot) unchanged and
ensure the created future is typed as an asyncio.Future (e.g., representing
ChannelMessage) if you want static typing.
| async with self._session.post( | ||
| f"{self._base_url}/api/push/poll/register", | ||
| json={}, | ||
| headers=headers, | ||
| ) as resp: | ||
| data = await resp.json() | ||
| self._poll_token = data["poll_token"] | ||
| self._client_id = data["client_id"] | ||
| self._connected = True | ||
| logger.debug("Polling transport registered: %s", self._client_id) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd src/praisonai-agents && wc -l praisonaiagents/push/transports.pyRepository: MervinPraison/PraisonAI
Length of output: 105
🏁 Script executed:
cd src/praisonai-agents && sed -n '100,160p' praisonaiagents/push/transports.pyRepository: MervinPraison/PraisonAI
Length of output: 2193
🏁 Script executed:
cd src/praisonai-agents && head -50 praisonaiagents/push/transports.pyRepository: MervinPraison/PraisonAI
Length of output: 1207
🏁 Script executed:
cd src/praisonai-agents && sed -n '110,145p' praisonaiagents/push/transports.pyRepository: MervinPraison/PraisonAI
Length of output: 1205
Close the session if registration fails.
After creating the ClientSession at the start of connect(), wrap the registration logic in a try-except-finally block to ensure the session is closed if any exception occurs (malformed JSON, missing keys, network errors, etc.). Without this, the session remains open and causes a resource leak.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/transports.py` around lines 130 -
139, The connect() method opens a ClientSession and performs registration but
does not close the session on failure, causing a resource leak; update connect()
so that the registration block (the POST to
f"{self._base_url}/api/push/poll/register" and the JSON parsing that sets
self._poll_token, self._client_id, self._connected) is wrapped in
try/except/finally: on exception, await self._session.close() in the finally (or
in except before re-raising) and clear any partially-set state, and only set
self._connected and log success after the POST/JSON succeed; ensure errors from
malformed JSON or missing keys are caught and the session is always closed when
registration fails.
| msg_type = data.get("type", "") | ||
| headers = {} | ||
| if self._auth_token: | ||
| headers["Authorization"] = f"Bearer {self._auth_token}" | ||
|
|
||
| # Map message types to polling endpoints | ||
| if msg_type == "channel.subscribe": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/subscribe", | ||
| json={"poll_token": self._poll_token, "channel": data.get("channel", "")}, | ||
| headers=headers, | ||
| ) | ||
| elif msg_type == "channel.unsubscribe": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/unsubscribe", | ||
| json={"poll_token": self._poll_token, "channel": data.get("channel", "")}, | ||
| headers=headers, | ||
| ) | ||
| elif msg_type == "message_ack": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/ack", | ||
| json={"poll_token": self._poll_token, "event_id": data.get("event_id", "")}, | ||
| headers=headers, | ||
| ) | ||
| elif msg_type == "presence.heartbeat": | ||
| await self._session.post( | ||
| f"{self._base_url}/api/push/poll/heartbeat", | ||
| json={"poll_token": self._poll_token, "status": data.get("status", "online")}, | ||
| headers=headers, | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify polling send paths and whether aiohttp responses are context-managed.
rg -n -C3 'channel\.publish|self\._session\.post|raise_for_status|async with self\._session\.post' --iglob 'transports.py'Repository: MervinPraison/PraisonAI
Length of output: 4108
🏁 Script executed:
rg -n "def publish" src/praisonai-agents/praisonaiagents/push/Repository: MervinPraison/PraisonAI
Length of output: 195
🏁 Script executed:
cat -n src/praisonai-agents/praisonaiagents/push/transports.py | head -200 | tail -100Repository: MervinPraison/PraisonAI
Length of output: 4482
🏁 Script executed:
cat -n src/praisonai-agents/praisonaiagents/push/client.py | sed -n '150,180p'Repository: MervinPraison/PraisonAI
Length of output: 1368
🏁 Script executed:
rg -n "channel\.publish" src/praisonai-agents/praisonaiagents/push/Repository: MervinPraison/PraisonAI
Length of output: 301
Add channel.publish support and fix aiohttp resource leaks in polling transport.
The send() method silently ignores channel.publish messages (sent by PushClient.publish()), and its HTTP calls don't use context managers—leaking aiohttp connections and masking server errors. The register() and receive() methods in the same class demonstrate the correct async with pattern.
🐛 Suggested structure
msg_type = data.get("type", "")
headers = {}
if self._auth_token:
headers["Authorization"] = f"Bearer {self._auth_token}"
+ async def post(path: str, payload: Dict[str, Any]) -> None:
+ async with self._session.post(
+ f"{self._base_url}{path}",
+ json=payload,
+ headers=headers,
+ ) as resp:
+ resp.raise_for_status()
+
# Map message types to polling endpoints
if msg_type == "channel.subscribe":
- await self._session.post(
- f"{self._base_url}/api/push/poll/subscribe",
- json={"poll_token": self._poll_token, "channel": data.get("channel", "")},
- headers=headers,
- )
+ await post("/api/push/poll/subscribe", {
+ "poll_token": self._poll_token,
+ "channel": data.get("channel", ""),
+ })
elif msg_type == "channel.unsubscribe":
- await self._session.post(
- f"{self._base_url}/api/push/poll/unsubscribe",
- json={"poll_token": self._poll_token, "channel": data.get("channel", "")},
- headers=headers,
- )
+ await post("/api/push/poll/unsubscribe", {
+ "poll_token": self._poll_token,
+ "channel": data.get("channel", ""),
+ })
+ elif msg_type == "channel.publish":
+ await post("/api/push/poll/publish", {
+ "poll_token": self._poll_token,
+ "channel": data.get("channel", ""),
+ "data": data.get("data", {}),
+ })
elif msg_type == "message_ack":
- await self._session.post(
- f"{self._base_url}/api/push/poll/ack",
- json={"poll_token": self._poll_token, "event_id": data.get("event_id", "")},
- headers=headers,
- )
+ await post("/api/push/poll/ack", {
+ "poll_token": self._poll_token,
+ "event_id": data.get("event_id", ""),
+ })
elif msg_type == "presence.heartbeat":
- await self._session.post(
- f"{self._base_url}/api/push/poll/heartbeat",
- json={"poll_token": self._poll_token, "status": data.get("status", "online")},
- headers=headers,
- )
+ await post("/api/push/poll/heartbeat", {
+ "poll_token": self._poll_token,
+ "status": data.get("status", "online"),
+ })
+ else:
+ raise ValueError(f"Unsupported polling message type: {msg_type}")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/transports.py` around lines 151 -
180, The send() method currently ignores "channel.publish" and leaks aiohttp
connections by calling self._session.post(...) without context managers; update
send() to handle msg_type == "channel.publish" (messages coming from
PushClient.publish()) by POSTing to the polling publish endpoint (e.g.,
"/api/push/poll/publish") with the appropriate json (poll_token, channel,
message/body) and headers, and for all POST calls replace direct await
self._session.post(...) with the async with pattern used in register() and
receive() (async with self._session.post(...) as resp:) and call
resp.raise_for_status() to surface server errors; mirror the request payload
structure used elsewhere and reuse headers construction at the top of send().
- SECURITY: Move auth tokens from URL query params to Authorization headers - PERFORMANCE: Make receive loop non-blocking for ACKs and callbacks - ERROR HANDLING: Add HTTP status checks in polling transport - ERROR HANDLING: Use robust URL parsing for WebSocket to HTTP conversion - ERROR HANDLING: Fail fast when sending on disconnected transport - CODE QUALITY: Remove duplicate test file content Addresses issues identified by Gemini, CodeRabbit, and Copilot reviewers. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-authored-by: Claude <noreply@anthropic.com>
There was a problem hiding this comment.
MervinPraison has reached the 50-review limit for trial accounts. To continue receiving code reviews, upgrade your plan.
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/praisonai-agents/praisonaiagents/push/client.py`:
- Around line 77-96: The connect() method currently schedules _receive_loop and
_heartbeat_loop even when both WebSocket connection and the fallback
_switch_to_polling() fail, allowing connect() to appear to succeed while
self._connected is False; fix this by either modifying _switch_to_polling to
re-raise its exception on failure or, inside connect() after the except block,
check self._connected and if False raise a ConnectionError with a clear
remediation hint (e.g., suggest verifying network/auth or enabling polling)
before creating _receive_task/_heartbeat_task; reference the connect,
_switch_to_polling, _connected, _receive_loop, and _heartbeat_loop symbols when
making the change.
- Around line 274-295: The code uses fire-and-forget asyncio.create_task calls
in the message handling block (notably when calling
self._send({"type":"message_ack",...}) and when dispatching channel callbacks
via cb(msg) from ChannelMessage.from_event_dict) which can be garbage-collected
and cancelled; fix by maintaining strong references to these tasks: add a set
(e.g. self._pending_tasks) on the client class, create tasks for the ACK and for
each coroutine callback via asyncio.create_task, add the task to
self._pending_tasks, and attach a done callback that removes the task from the
set when complete and logs exceptions; ensure non-coroutine callbacks are
invoked synchronously as before and that the _send, _channel_callbacks and
ChannelMessage.from_event_dict code paths use this task-tracking approach.
In `@src/praisonai-agents/praisonaiagents/push/transports.py`:
- Around line 55-57: The is_connected property currently assumes self._ws has an
.open attribute (legacy websockets API); update it to first check for the modern
API by inspecting self._ws.state (the State enum) and consider connected when
state is websockets.protocol.State.OPEN (or the equivalent OPEN member), and if
.state is not present fall back to the legacy checks using self._ws.open or not
self._ws.closed; modify the is_connected property (referencing is_connected and
self._ws) to handle both APIs safely without raising AttributeError.
🪄 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
Run ID: 2962f29f-86b2-4644-b188-c5dc12367c9d
📒 Files selected for processing (3)
src/praisonai-agents/praisonaiagents/push/client.pysrc/praisonai-agents/praisonaiagents/push/transports.pysrc/praisonai-agents/tests/unit/test_push_protocols.py
| async def connect(self) -> None: | ||
| """Establish connection to the gateway.""" | ||
| from .transports import WebSocketTransport | ||
|
|
||
| self._transport = WebSocketTransport(self._url, self._auth_token) | ||
| try: | ||
| await self._transport.connect() | ||
| self._connected = True | ||
| self._using_polling = False | ||
| logger.info("PushClient connected via WebSocket") | ||
| except Exception as e: | ||
| logger.warning("WebSocket connect failed: %s", e) | ||
| if self._fallback_to_polling: | ||
| await self._switch_to_polling() | ||
| else: | ||
| raise | ||
|
|
||
| self._closed_event.clear() | ||
| self._receive_task = asyncio.ensure_future(self._receive_loop()) | ||
| self._heartbeat_task = asyncio.ensure_future(self._heartbeat_loop()) |
There was a problem hiding this comment.
connect() silently succeeds when both WebSocket and polling fallback fail.
If the WS connect() raises and _switch_to_polling() also fails, _switch_to_polling logs the error and sets self._connected = False but does not re-raise. Control falls through to lines 94–96, where _receive_loop / _heartbeat_loop are scheduled against a disconnected transport. Callers get no exception and await client.connect() appears to succeed, violating the fail-fast guideline.
Have _switch_to_polling re-raise on failure (or have connect() check self._connected and raise a ConnectionError with a remediation hint) before starting the background tasks.
As per coding guidelines: "Fail fast with clear error messages; include remediation hints in exceptions".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 77 - 96,
The connect() method currently schedules _receive_loop and _heartbeat_loop even
when both WebSocket connection and the fallback _switch_to_polling() fail,
allowing connect() to appear to succeed while self._connected is False; fix this
by either modifying _switch_to_polling to re-raise its exception on failure or,
inside connect() after the except block, check self._connected and if False
raise a ConnectionError with a clear remediation hint (e.g., suggest verifying
network/auth or enabling polling) before creating _receive_task/_heartbeat_task;
reference the connect, _switch_to_polling, _connected, _receive_loop, and
_heartbeat_loop symbols when making the change.
| # Auto-ACK for delivery guarantees (non-blocking) | ||
| event_id = data.get("event_id") | ||
| if event_id and msg_type not in ( | ||
| "ack_received", "nack_received", "channel.subscribed", | ||
| "channel.unsubscribed", "channel.published", "channel.created", | ||
| "channel.list", "presence.heartbeat_ack", "presence.list", | ||
| "joined", "left", "error", | ||
| ): | ||
| asyncio.create_task(self._send({"type": "message_ack", "event_id": event_id})) | ||
|
|
||
| # Dispatch channel messages (non-blocking) | ||
| channel = data.get("channel") | ||
| if channel and msg_type in ("channel_message",): | ||
| msg = ChannelMessage.from_event_dict(data) | ||
| for cb in self._channel_callbacks.get(channel, []): | ||
| try: | ||
| if asyncio.iscoroutinefunction(cb): | ||
| asyncio.create_task(cb(msg)) | ||
| else: | ||
| cb(msg) | ||
| except Exception as e: | ||
| logger.error("Channel callback error: %s", e) |
There was a problem hiding this comment.
Fire-and-forget create_task can drop ACKs and channel-message dispatches.
asyncio.create_task(...) only keeps a weak reference to the task; if the event loop GC's it before it completes, the coroutine is silently cancelled and the ACK or callback is lost. This is the exact failure mode Ruff RUF006 warns about, and it's especially damaging on line 282 because losing an ACK defeats the delivery-guarantee feature this PR is introducing.
Keep strong references in a set and discard on completion:
🔧 Suggested fix
+ # in __init__:
+ # self._background_tasks: set[asyncio.Task] = set()
@@
- asyncio.create_task(self._send({"type": "message_ack", "event_id": event_id}))
+ task = asyncio.create_task(
+ self._send({"type": "message_ack", "event_id": event_id})
+ )
+ self._background_tasks.add(task)
+ task.add_done_callback(self._background_tasks.discard)
@@
- if asyncio.iscoroutinefunction(cb):
- asyncio.create_task(cb(msg))
- else:
- cb(msg)
+ if asyncio.iscoroutinefunction(cb):
+ task = asyncio.create_task(cb(msg))
+ self._background_tasks.add(task)
+ task.add_done_callback(self._background_tasks.discard)
+ else:
+ cb(msg)🧰 Tools
🪛 Ruff (0.15.10)
[warning] 282-282: Store a reference to the return value of asyncio.create_task
(RUF006)
[warning] 291-291: Store a reference to the return value of asyncio.create_task
(RUF006)
[warning] 294-294: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/client.py` around lines 274 - 295,
The code uses fire-and-forget asyncio.create_task calls in the message handling
block (notably when calling self._send({"type":"message_ack",...}) and when
dispatching channel callbacks via cb(msg) from ChannelMessage.from_event_dict)
which can be garbage-collected and cancelled; fix by maintaining strong
references to these tasks: add a set (e.g. self._pending_tasks) on the client
class, create tasks for the ACK and for each coroutine callback via
asyncio.create_task, add the task to self._pending_tasks, and attach a done
callback that removes the task from the set when complete and logs exceptions;
ensure non-coroutine callbacks are invoked synchronously as before and that the
_send, _channel_callbacks and ChannelMessage.from_event_dict code paths use this
task-tracking approach.
| @property | ||
| def is_connected(self) -> bool: | ||
| return self._ws is not None and self._ws.open |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Does websockets Python library version 13 or later still expose the WebSocketClientProtocol.open attribute?
💡 Result:
Yes, the websockets Python library version 13 and later still exposes the WebSocketClientProtocol.open attribute in the legacy asyncio implementation.
Citations:
- 1: https://websockets.readthedocs.io/en/13.1/reference/legacy/client.html
- 2: https://github.com/python-websockets/websockets/blob/13.0.1/src/websockets/legacy/protocol.py
- 3: https://github.com/python-websockets/websockets/blob/16.0/src/websockets/legacy/protocol.py
- 4: https://github.com/python-websockets/websockets/blob/13.1/src/websockets/legacy/protocol.py
🏁 Script executed:
# Find websockets version requirement
fd -t f '\.(py|txt|toml|cfg)$' | xargs rg -l "websockets" | head -20Repository: MervinPraison/PraisonAI
Length of output: 1236
🏁 Script executed:
# Look at the specific file and surrounding context
cat -n src/praisonai-agents/praisonaiagents/push/transports.py | sed -n '40,160p'Repository: MervinPraison/PraisonAI
Length of output: 4695
🏁 Script executed:
# Check imports and how self._ws is initialized
rg "import websockets|from websockets|self\._ws\s*=" src/praisonai-agents/praisonaiagents/push/transports.py -B 2 -A 2Repository: MervinPraison/PraisonAI
Length of output: 560
🏁 Script executed:
cat -n src/praisonai-agents/pyproject.toml | grep -A 20 -B 5 websocketsRepository: MervinPraison/PraisonAI
Length of output: 687
🏁 Script executed:
# Check if there are any version constraints documented
rg "websockets\s*[>=<]" src/praisonai-agents/Repository: MervinPraison/PraisonAI
Length of output: 125
🌐 Web query:
What object type does websockets.connect() return in websockets library version 13+ (modern asyncio API)?
💡 Result:
In the websockets library version 13+ modern asyncio API, websockets.connect (imported as from websockets.asyncio.client import connect) returns a websockets.asyncio.client.ClientConnection instance. This is an asyncio-compatible WebSocket client connection object used for sending and receiving messages, supporting async context management and iteration.
Citations:
- 1: https://websockets.readthedocs.io/en/13.1/reference/asyncio/client.html
- 2: https://websockets.readthedocs.io/en/13.1/howto/quickstart.html
- 3: https://websockets.readthedocs.io/en/13.0/howto/upgrade.html
🌐 Web query:
Does websockets.asyncio.client.ClientConnection (modern asyncio API in websockets 13+) expose an .open attribute?
💡 Result:
No, websockets.asyncio.client.ClientConnection (the modern asyncio API in websockets 13+) does not expose an .open attribute.
Citations:
- 1: https://websockets.readthedocs.io/en/13.1/reference/asyncio/client.html
- 2: https://websockets.readthedocs.io/en/stable/reference/asyncio/client.html
- 3: https://websockets.readthedocs.io/en/15.0.1/reference/asyncio/client.html
- 4: https://github.com/python-websockets/websockets/blob/16.0/src/websockets/asyncio/connection.py
- 5: https://github.com/python-websockets/websockets/blob/ea164d2/src/websockets/asyncio/connection.py
🏁 Script executed:
# Also check if there's specific websockets version usage elsewhere in the codebase
rg "websockets\." src/praisonai-agents/praisonaiagents/push/transports.py -A 2Repository: MervinPraison/PraisonAI
Length of output: 210
🌐 Web query:
What attributes or properties does websockets.asyncio.client.ClientConnection expose to check connection state (open, closed, state)?
💡 Result:
The websockets.asyncio.client.ClientConnection exposes the following attributes to check connection state: - state: A property of type State (from websockets.protocol.State), representing the WebSocket connection state as defined in RFC 6455 (CONNECTING, OPEN, CLOSING, CLOSED). This is the primary attribute for detailed state checking. In older versions (e.g., 12.0 and earlier), it also provided: - open: Boolean indicating if the connection is open. - closed: Boolean indicating if the connection is closed. The official documentation recommends against directly checking these attributes. Instead, call recv or send and handle ConnectionClosed exceptions for typical applications. After the connection is closed, close_code and close_reason provide additional details.
Citations:
- 1: https://websockets.readthedocs.io/en/stable/reference/asyncio/client.html
- 2: https://websockets.readthedocs.io/en/15.0.1/reference/asyncio/client.html
- 3: https://websockets.readthedocs.io/en/stable/genindex.html
🏁 Script executed:
# Let's also check how websockets.connect is imported in the code
rg "import websockets|from websockets" src/praisonai-agents/praisonaiagents/push/transports.py -B 2 -A 5Repository: MervinPraison/PraisonAI
Length of output: 341
self._ws.open does not exist in websockets ≥ 13 (modern asyncio API).
The project supports websockets>=12.0. In websockets 13+, the modern asyncio API (websockets.asyncio.client.ClientConnection) exposes .state (a State enum) instead of .open. Accessing .open on line 57 will raise AttributeError at runtime.
To maintain compatibility with both 12.0 and 13+, check .state first (modern API), then fall back to .open/.closed (legacy API):
🔧 Suggested fix
`@property`
def is_connected(self) -> bool:
- return self._ws is not None and self._ws.open
+ if self._ws is None:
+ return False
+ # websockets >=13 exposes .state (State enum); <=12 exposes .open
+ state = getattr(self._ws, "state", None)
+ if state is not None:
+ return getattr(state, "name", str(state)) == "OPEN"
+ return bool(getattr(self._ws, "open", False))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/praisonai-agents/praisonaiagents/push/transports.py` around lines 55 -
57, The is_connected property currently assumes self._ws has an .open attribute
(legacy websockets API); update it to first check for the modern API by
inspecting self._ws.state (the State enum) and consider connected when state is
websockets.protocol.State.OPEN (or the equivalent OPEN member), and if .state is
not present fall back to the legacy checks using self._ws.open or not
self._ws.closed; modify the is_connected property (referencing is_connected and
self._ws) to handle both APIs safely without raising AttributeError.
Summary by CodeRabbit
New Features
Tests