feat: Gen 5 improvements — stream subscription check + auto-subscribe
Adds two improvements from pi Zulip extension lessons: 1. _ensure_stream_subscriptions() — checks at connect time if the bot is subscribed to the primary stream and general-chat; auto-subscribes if missing. Without this, bots with 0 subscriptions can't receive stream events (DMs only). 2. selftest check #9 — verifies stream subscriptions and reports count + names. Catches the 'zero subscriptions' failure mode that was a major debugging bottleneck in the pi extension.
This commit is contained in:
@@ -302,6 +302,9 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
# Gen 2: Try to resolve @all-bots user ID dynamically
|
# Gen 2: Try to resolve @all-bots user ID dynamically
|
||||||
await self._resolve_all_bots_user_id()
|
await self._resolve_all_bots_user_id()
|
||||||
|
|
||||||
|
# Gen 5: Ensure bot is subscribed to primary stream
|
||||||
|
await self._ensure_stream_subscriptions()
|
||||||
|
|
||||||
self._mark_connected()
|
self._mark_connected()
|
||||||
logger.info(
|
logger.info(
|
||||||
"[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)",
|
"[%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
|
# 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:
|
async def _resolve_all_bots_user_id(self) -> None:
|
||||||
"""Try to resolve the @all-bots user ID from the Zulip server.
|
"""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}",
|
"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
|
# Overall verdict
|
||||||
critical = ["connected", "queue_registered", "http_client", "poll_loop"]
|
critical = ["connected", "queue_registered", "http_client", "poll_loop"]
|
||||||
passed = sum(1 for c in checks.values() if c["status"])
|
passed = sum(1 for c in checks.values() if c["status"])
|
||||||
|
|||||||
Reference in New Issue
Block a user