diff --git a/plugins/platforms/zulip/adapter.py b/plugins/platforms/zulip/adapter.py index 1937f8f..a6be243 100644 --- a/plugins/platforms/zulip/adapter.py +++ b/plugins/platforms/zulip/adapter.py @@ -302,6 +302,9 @@ class ZulipAdapter(BasePlatformAdapter): # Gen 2: Try to resolve @all-bots user ID dynamically await self._resolve_all_bots_user_id() + # Gen 5: Ensure bot is subscribed to primary stream + await self._ensure_stream_subscriptions() + self._mark_connected() logger.info( "[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)", @@ -681,6 +684,64 @@ class ZulipAdapter(BasePlatformAdapter): # Gen 2: Dynamic @all-bots resolution # ------------------------------------------------------------------ + async def _ensure_stream_subscriptions(self) -> None: + """Ensure the bot is subscribed to the primary streams. + + Checks current subscriptions and subscribes to the configured primary + stream (and general-chat) if missing. Without subscription, the bot + receives DMs but no stream events — @mentions in streams are invisible. + + This is a Gen 5 improvement based on lessons from the pi Zulip extension, + where the bot had zero stream subscriptions and couldn't receive + @mentions in stream topics. + """ + try: + # Check current subscriptions + resp, _ = await self._api_call( + "GET", "/api/v1/users/me/subscriptions", + ) + if not resp: + return + + subscribed = resp.get("subscriptions", []) + subscribed_names = [s.get("name", "") for s in subscribed if isinstance(s, dict)] + + # Streams the bot should be subscribed to + required_streams = [self._stream, "general chat"] + missing = [s for s in required_streams if s not in subscribed_names] + + if not missing: + logger.info( + "[%s] Stream subscriptions OK: %s", + self.name, ", ".join(subscribed_names), + ) + return + + # Subscribe to missing streams + import json as _json + payload = {"subscriptions": _json.dumps( + [{"name": s} for s in missing] + )} + sub_resp, _ = await self._api_call( + "POST", "/api/v1/users/me/subscriptions", + data=payload, + ) + if sub_resp and sub_resp.get("result") == "success": + subscribed_str = ", ".join(missing) + logger.info( + "[%s] Subscribed to: %s", self.name, subscribed_str, + ) + self._health_stats["subscriptions_fixed"] = ( + self._health_stats.get("subscriptions_fixed", 0) + 1 + ) + else: + logger.warning( + "[%s] Failed to subscribe to %s", + self.name, ", ".join(missing), + ) + except Exception as e: + logger.warning("[%s] Subscription check failed: %s", self.name, e) + async def _resolve_all_bots_user_id(self) -> None: """Try to resolve the @all-bots user ID from the Zulip server. @@ -1100,6 +1161,35 @@ class ZulipAdapter(BasePlatformAdapter): "detail": f"@all-bots user_id: {self._all_bots_user_id}", } + # 9. Stream subscriptions (verifies bot is subscribed to at least one stream) + try: + from urllib.parse import urlencode + import json as _json + key_str = f"{self._email}:{self._api_key}" + auth_b64 = base64.b64encode(key_str.encode()).decode() + resp, _ = await self._api_call( + "GET", "/api/v1/users/me/subscriptions", + ) + if resp and "subscriptions" in resp: + stream_count = len(resp["subscriptions"]) + stream_names = [s.get("name", "?") for s in resp["subscriptions"]][:5] + checks["stream_subscriptions"] = { + "status": stream_count > 0, + "detail": f"{stream_count} stream(s): {', '.join(stream_names)}" + if stream_count > 0 + else "No stream subscriptions — bot will not receive stream events", + } + else: + checks["stream_subscriptions"] = { + "status": False, + "detail": "Failed to check subscriptions", + } + except Exception as sub_err: + checks["stream_subscriptions"] = { + "status": False, + "detail": f"Subscription check failed: {sub_err}", + } + # Overall verdict critical = ["connected", "queue_registered", "http_client", "poll_loop"] passed = sum(1 for c in checks.values() if c["status"])