fix(abiba): recipient format fix, add types.d.ts; feat(hermes): standalone bridge adapter
CI / validate (pull_request) Failing after 7s

This commit is contained in:
kagentz-bot
2026-06-24 09:06:08 -04:00
parent 705c29a4e0
commit 676a94459c
6 changed files with 3938 additions and 421 deletions
@@ -16,7 +16,11 @@ import asyncio
import logging import logging
import os import os
import time import time
import re
import base64
import urllib.parse
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
import requests
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -352,6 +356,65 @@ class ZulipAdapter(BasePlatformAdapter):
except Exception as e: except Exception as e:
logger.error("Zulip: re-registration failed: %s", e) logger.error("Zulip: re-registration failed: %s", e)
async def _resolve_image_content(self, content: str) -> str:
"""Detect Zulip image upload URLs in content and replace with base64 data URIs.
Zulip stores uploaded files as relative paths in the content field
(e.g., [image.png](/user_uploads/.../file.png)). The remote LLM cannot
access these internal URLs, so we download them via HTTPS and embed
them as base64 data URIs inline.
"""
# Match markdown images: ![alt](url) or [alt](url)
# Zulip uploads are relative: /user_uploads/...
url_pattern = re.compile(
r'(?P<full>!?\[([^\]]*)\]\((?P<url>/user_uploads/[^)]+)\))'
)
matches = list(url_pattern.finditer(content))
if not matches:
return content
logger.info("Zulip: resolving %d image(s) in message content", len(matches))
for match in matches:
relative_url = match.group("url")
full_url = self.site.rstrip("/") + relative_url
try:
# Use the bot's own API key to authenticate the file download
resp = await asyncio.get_event_loop().run_in_executor(
None,
lambda: requests.get(
full_url,
headers={
"User-Agent": "Hermes Zulip Adapter / Image Resolver",
},
auth=requests.auth.HTTPBasicAuth(self.email, self.api_key),
timeout=15,
)
)
if resp.status_code != 200:
logger.warning("Zulip: image download failed (%d) for %s",
resp.status_code, relative_url)
continue
img_data = resp.content
mime_type = resp.headers.get("Content-Type", "image/png")
b64_data = base64.b64encode(img_data).decode("ascii")
data_uri = f"data:{mime_type};base64,{b64_data}"
# Replace the URL in content
content = content.replace(relative_url, data_uri, 1)
logger.info("Zulip: resolved image %s -> data URI (%d bytes, %s)",
relative_url.split("/")[-1], len(img_data), mime_type)
except Exception as e:
logger.error("Zulip: failed to resolve image %s: %s", relative_url, e)
continue
return content
async def _handle_event(self, event: Dict[str, Any]) -> None: async def _handle_event(self, event: Dict[str, Any]) -> None:
"""Process a single Zulip event and dispatch as MessageEvent. """Process a single Zulip event and dispatch as MessageEvent.
@@ -375,6 +438,9 @@ class ZulipAdapter(BasePlatformAdapter):
# Determine message type: private (DM) or stream # Determine message type: private (DM) or stream
zulip_msg_type = msg.get("type", "") zulip_msg_type = msg.get("type", "")
content = msg.get("content", "") content = msg.get("content", "")
# Resolve image upload URLs to base64 data URIs
if content and "/user_uploads/" in content:
content = await self._resolve_image_content(content)
sender_full_name = msg.get("sender_full_name", sender_email) sender_full_name = msg.get("sender_full_name", sender_email)
# ── HANDLE PRIVATE MESSAGES (DMs) ── # ── HANDLE PRIVATE MESSAGES (DMs) ──
@@ -0,0 +1,350 @@
#!/usr/bin/env python3
"""
Standalone Zulip → LiteLLM Bridge
Polls Zulip for DMs, downloads image attachments via Zulip API,
and calls LiteLLM proxy directly (bypassing Hermes Gateway).
Usage:
ZULIP_EMAIL=tanko-bot@chat.sysloggh.net \
ZULIP_API_KEY=xxx \
ZULIP_SITE=https://chat.sysloggh.net \
LITELLM_API_KEY=sk-xxx \
LITELLM_BASE_URL=https://litellm.sysloggh.net/v1 \
VISION_MODEL=gemma-4-12b \
TEXT_MODEL=generic-llm \
AGENT_NAME=Tanko \
python3 standalone_bridge.py
"""
import asyncio
import base64
import logging
import os
import re
import sys
import time
from typing import Any, Dict, List, Optional
import requests
from openai import AsyncOpenAI
# ── Logging ────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
stream=sys.stderr,
)
logger = logging.getLogger("zulip-bridge")
# ── Config ─────────────────────────────────────────────────────────────────
ZULIP_EMAIL = os.environ.get("ZULIP_EMAIL", "")
ZULIP_API_KEY = os.environ.get("ZULIP_API_KEY", "")
ZULIP_SITE = os.environ.get("ZULIP_SITE", "https://chat.sysloggh.net")
ZULIP_STREAM = os.environ.get("ZULIP_STREAM", "agent-hub")
LITELLM_API_KEY = os.environ.get("LITELLM_API_KEY", "")
LITELLM_BASE_URL = os.environ.get("LITELLM_BASE_URL", "https://litellm.sysloggh.net/v1")
VISION_MODEL = os.environ.get("VISION_MODEL", "gemma-4-12b")
TEXT_MODEL = os.environ.get("TEXT_MODEL", "generic-llm")
AGENT_NAME = os.environ.get("AGENT_NAME", "Tanko")
# URL patterns for Zulip image uploads
IMAGE_URL_RE = re.compile(r"/user_uploads/[^\s)]+")
# ── Auto-load env file from disk ──────────────────────────────────────
_ENV_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bridge.env")
if os.path.exists(_ENV_FILE):
import re as _re
for _line in open(_ENV_FILE):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_key, _val = _line.split("=", 1)
os.environ.setdefault(_key, _val)
def check_config() -> bool:
"""Verify all required environment variables are set."""
missing = []
if not ZULIP_EMAIL:
missing.append("ZULIP_EMAIL")
if not ZULIP_API_KEY:
missing.append("ZULIP_API_KEY")
if not LITELLM_API_KEY:
missing.append("LITELLM_API_KEY")
if not VISION_MODEL:
missing.append("VISION_MODEL")
if not TEXT_MODEL:
missing.append("TEXT_MODEL")
if missing:
logger.error("Missing required env vars: %s", ", ".join(missing))
return False
return True
def download_zulip_image(relative_url: str) -> Optional[bytes]:
"""Download an image from Zulip via HTTP Basic Auth."""
full_url = ZULIP_SITE.rstrip("/") + relative_url
try:
resp = requests.get(
full_url,
auth=requests.auth.HTTPBasicAuth(ZULIP_EMAIL, ZULIP_API_KEY),
timeout=15,
)
if resp.status_code != 200:
logger.warning("Image download failed (%d) for %s", resp.status_code, relative_url)
return None
return resp.content
except Exception as e:
logger.error("Image download error for %s: %s", relative_url, e)
return None
def extract_image_urls(content: str) -> List[str]:
"""Extract relative Zulip upload URLs from message content."""
matches = IMAGE_URL_RE.findall(content)
# Deduplicate and clean trailing punctuation
seen = set()
urls = []
for url in matches:
# Strip trailing characters that aren't part of URL
clean = url.rstrip(").,]!?;:")
if clean not in seen:
seen.add(clean)
urls.append(clean)
return urls
def build_vision_messages(content: str, image_data: List[Dict]) -> List[Dict]:
"""Build OpenAI-style messages with text and image parts."""
if not image_data:
return [{"role": "user", "content": content}]
parts = [{"type": "text", "text": content}]
for img in image_data:
parts.append({
"type": "image_url",
"image_url": {
"url": f"data:{img['mime']};base64,{img['b64']}",
"detail": "auto",
},
})
return [{"role": "user", "content": parts}]
async def call_litellm(messages: List[Dict], model: str) -> Optional[str]:
"""Call LiteLLM proxy and return response text."""
client = AsyncOpenAI(
api_key=LITELLM_API_KEY,
base_url=LITELLM_BASE_URL,
)
try:
response = await client.chat.completions.create(
model=model,
messages=messages,
max_tokens=4096,
temperature=0.7,
)
return response.choices[0].message.content
except Exception as e:
logger.error("LiteLLM call failed: %s", e, exc_info=True)
return None
def build_source(chat_id: str, chat_name: str, chat_type: str,
user_id: str, user_name: str) -> Dict:
"""Build source metadata for logging/context."""
return {
"chat_id": chat_id,
"chat_name": chat_name,
"chat_type": chat_type,
"user_id": user_id,
"user_name": user_name,
"platform": "zulip",
}
async def process_message(client, msg: Dict) -> None:
"""Process a single Zulip message: download images, call LLM, reply."""
msg_type = msg.get("type", "")
content = msg.get("content", "")
sender_email = msg.get("sender_email", "")
sender_full_name = msg.get("sender_full_name", sender_email)
msg_id = msg.get("id", 0)
# Ignore own messages
if sender_email.lower() == ZULIP_EMAIL.lower():
return
logger.info("Processing message from %s (%s): type=%s, id=%d",
sender_full_name, sender_email, msg_type, msg_id)
# ── Extract and download images ──
image_urls = extract_image_urls(content)
image_data: List[Dict] = []
if image_urls:
logger.info("Found %d image(s) in message", len(image_urls))
for url in image_urls:
raw_bytes = download_zulip_image(url)
if raw_bytes:
# Detect MIME from extension
ext = url.rsplit(".", 1)[-1].lower() if "." in url else "png"
mime_map = {
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"png": "image/png",
"gif": "image/gif",
"webp": "image/webp",
}
mime = mime_map.get(ext, "image/png")
b64 = base64.b64encode(raw_bytes).decode("ascii")
image_data.append({"mime": mime, "b64": b64})
logger.info("Downloaded image (%d bytes, %s)", len(raw_bytes), mime)
# ── Determine model ──
model = VISION_MODEL if image_data else TEXT_MODEL
logger.info("Using model: %s (images=%d)", model, len(image_data))
# ── Build messages and call LLM ──
messages = build_vision_messages(content, image_data)
response = await call_litellm(messages, model)
if not response:
logger.error("LLM returned no response for message %d", msg_id)
return
# ── Send reply ──
reply_content = response[:5000] # Zulip max message length
if msg_type == "private":
message_data = {
"type": "private",
"to": [sender_email],
"content": reply_content,
}
else:
display_recipient = msg.get("display_recipient", "")
stream_name = display_recipient if isinstance(display_recipient, str) else ""
subject = msg.get("subject", "general")
message_data = {
"type": "stream",
"to": stream_name,
"subject": subject,
"content": reply_content,
}
try:
result = client.send_message(message_data)
if result.get("result") == "success":
logger.info("Reply sent: %s", result.get("id", ""))
else:
logger.error("Failed to send reply: %s", result.get("msg", "unknown"))
except Exception as e:
logger.error("Send reply failed: %s", e)
async def poll_loop(client) -> None:
"""Main polling loop: register queue and long-poll for events."""
logger.info("Connecting to Zulip as %s on %s", ZULIP_EMAIL, ZULIP_SITE)
# Register event queue with empty narrow (all events)
queue_data = client.register(
event_types=["message"],
narrow=[], # Empty = all events including DMs
include_subscribers=False,
)
if queue_data.get("result") != "success":
logger.error("Queue registration failed: %s", queue_data.get("msg", "unknown"))
return
queue_id = queue_data["queue_id"]
last_event_id = queue_data.get("last_event_id", -1)
logger.info("Event queue registered: %s (last_event=%s)",
queue_id[:8], last_event_id)
# Subscribe to configured stream
try:
client.add_subscriptions([{"name": ZULIP_STREAM}])
logger.info("Subscribed to stream: %s", ZULIP_STREAM)
except Exception as e:
logger.warning("Could not subscribe to stream: %s", e)
# Poll loop
while True:
try:
response = client.get_events(
queue_id=queue_id,
last_event_id=last_event_id,
)
if response.get("result") != "success":
error_msg = response.get("msg", "unknown")
logger.warning("Poll error: %s — reconnecting...", error_msg)
await asyncio.sleep(5)
# Re-register queue
queue_data = client.register(
event_types=["message"],
narrow=[],
include_subscribers=False,
)
if queue_data.get("result") == "success":
queue_id = queue_data["queue_id"]
last_event_id = queue_data.get("last_event_id", -1)
logger.info("Re-registered queue: %s", queue_id[:8])
continue
events = response.get("events", [])
if events:
for event in events:
last_event_id = max(last_event_id, event.get("id", last_event_id))
if event.get("type") == "message":
msg = event.get("message", {})
asyncio.create_task(process_message(client, msg))
# Small delay to prevent busy-polling
await asyncio.sleep(0.1)
except KeyboardInterrupt:
logger.info("Shutdown requested")
break
except Exception as e:
logger.error("Poll loop error: %s", e, exc_info=True)
await asyncio.sleep(5)
# Cleanup
try:
client.deregister(queue_id)
except Exception:
pass
def main():
"""Entry point — creates Zulip client and starts async poll loop."""
if not check_config():
sys.exit(1)
import zulip
client = zulip.Client(
email=ZULIP_EMAIL,
api_key=ZULIP_API_KEY,
site=ZULIP_SITE,
client=f"Zulip Bridge / {AGENT_NAME}",
)
logger.info("Starting Zulip bridge for %s (vision=%s, text=%s)",
AGENT_NAME, VISION_MODEL, TEXT_MODEL)
try:
asyncio.run(poll_loop(client))
except KeyboardInterrupt:
logger.info("Bridge stopped")
if __name__ == "__main__":
main()
File diff suppressed because it is too large Load Diff
+12 -7
View File
@@ -1,22 +1,27 @@
{ {
"name": "pi-zulip-extension", "name": "pi-zulip-extension",
"version": "0.1.0", "version": "0.2.0",
"description": "pi extension for Zulip agent communication — connects Abiba to the Sysloggh agent mesh", "description": "pi extension for Zulip agent communication — connects Abiba to the Sysloggh agent mesh",
"main": "src/index.ts", "main": "dist/index.js",
"type": "module", "type": "module",
"scripts": { "scripts": {
"build": "tsc", "build": "tsc",
"check": "tsc --noEmit" "check": "tsc --noEmit"
}, },
"dependencies": { "dependencies": {
"zulip-js": "^2.0.0", "zulip-js": "^2.0.0"
"yaml": "^2.0.0"
}, },
"devDependencies": { "devDependencies": {
"@earendil-works/pi-coding-agent": "*", "@mariozechner/pi-coding-agent": "*",
"typescript": "^5.0.0" "typescript": "^5.9.3"
}, },
"keywords": ["pi", "zulip", "agent", "abiba", "sysloggh"], "keywords": [
"pi",
"zulip",
"agent",
"abiba",
"sysloggh"
],
"license": "UNLICENSED", "license": "UNLICENSED",
"private": true "private": true
} }
File diff suppressed because it is too large Load Diff
+34
View File
@@ -0,0 +1,34 @@
// Type declarations for zulip-js (no @types available)
declare module "zulip-js" {
interface ZulipClient {
queues: {
register(params: {
event_types: string[];
narrow?: Array<Array<string | number>>;
}): Promise<{ queue_id: string; last_event_id: number }>;
};
users: {
me: {
get(): Promise<any>;
};
};
messages: {
store: {
send(rawContent: string): Promise<any>;
};
};
events: {
retrieve(params: {
queue_id: string;
last_event_id: number;
dont_block?: boolean;
}): Promise<{ events: any[]; result?: string; msg?: string }>;
};
}
export function createClient(config: {
username: string;
apiKey: string;
realm: string;
}): Promise<ZulipClient>;
}