fix(agent-zero): queue expiry detection in adapter
CI / validate (push) Failing after 2s
CI / deploy (push) Has been skipped

Zulip event queues expire after ~600s of inactivity. The adapter now
detects 400/BAD_EVENT_QUEUE_ID responses and auto-reconnects instead of
silently returning empty results.
This commit is contained in:
Abiba (pi)
2026-06-27 19:12:47 +00:00
parent 4938bad199
commit 1f615bbb17
@@ -189,13 +189,18 @@ class AgentZeroZulipAdapter:
await asyncio.sleep(DEFAULT_POLL_INTERVAL) await asyncio.sleep(DEFAULT_POLL_INTERVAL)
async def _fetch_events(self) -> list: async def _fetch_events(self) -> list:
"""Fetch events from Zulip queue.""" """Fetch events from Zulip queue. Raises on BAD_EVENT_QUEUE_ID."""
resp = await self._api_call("GET", "/api/v1/events", params={ resp, status = await self._api_call("GET", "/api/v1/events", params={
"queue_id": self._queue_id, "queue_id": self._queue_id,
"last_event_id": str(self._last_event_id), "last_event_id": str(self._last_event_id),
"dont_block": "true", "dont_block": "true",
}) }, return_status=True)
if resp is None:
# Detect queue expiry
if status == 400:
raise RuntimeError("BAD_EVENT_QUEUE_ID: queue expired")
if not resp:
return [] return []
events = resp.get("events", []) events = resp.get("events", [])
@@ -365,10 +370,14 @@ class AgentZeroZulipAdapter:
# ── Zulip API ── # ── Zulip API ──
async def _api_call(self, method: str, path: str, async def _api_call(self, method: str, path: str,
data: dict = None, params: dict = None) -> Optional[dict]: data: dict = None, params: dict = None,
"""Make Zulip API call.""" return_status: bool = False) -> Any:
"""Make Zulip API call.
Returns: dict response body, or if return_status=True returns (dict, int)
"""
if not self._http_client: if not self._http_client:
return None return (None, 0) if return_status else None
url = f"{self._site}{path}" url = f"{self._site}{path}"
headers = {"Authorization": self._auth_header} headers = {"Authorization": self._auth_header}
@@ -381,16 +390,17 @@ class AgentZeroZulipAdapter:
elif method == "PATCH": elif method == "PATCH":
resp = await self._http_client.patch(url, data=data, headers=headers) resp = await self._http_client.patch(url, data=data, headers=headers)
else: else:
return None return (None, 0) if return_status else None
if resp.status_code >= 400: if resp.status_code >= 400:
logger.debug(f"API {method} {path}: {resp.status_code}") logger.debug(f"API {method} {path}: {resp.status_code}")
return None return (None, resp.status_code) if return_status else None
return resp.json() body = resp.json()
return (body, resp.status_code) if return_status else body
except Exception as e: except Exception as e:
logger.debug(f"API error {method} {path}: {e}") logger.debug(f"API error {method} {path}: {e}")
return None return (None, 0) if return_status else None
async def _send_msg(self, msg_type: str, to: str, content: str) -> Optional[int]: async def _send_msg(self, msg_type: str, to: str, content: str) -> Optional[int]:
"""Send a message to Zulip.""" """Send a message to Zulip."""