#!/usr/bin/env python3 """ slopbot - Zulip bot that replies with stock prices when it sees $TICKER patterns, and delegates @mention responses to a one-shot agentd agent per message. Usage: python main.py Requires a .zuliprc file in the same directory (see .zuliprc.example). """ import html import logging import re import subprocess import time from datetime import datetime, UTC from pathlib import Path import yfinance as yf import zulip logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) log = logging.getLogger("slopbot") # Regex to match stock tickers like $NVDA, $AAPL, $BRK TICKER_RE = re.compile(r"\$([A-Z]{1,5})") # Regex to detect @mention of slop / slopbot MENTION_RE = re.compile(r"@\*\*slop[^*]*\*\*", re.IGNORECASE) AGENTD_RECV_TIMEOUT = 120 # seconds AGENTD_MODEL = "openai/gpt-5.5" HISTORY_LIMIT = 20 SYSTEM_PROMPT = """You are slop-bot in the meshheads.org Zulip chat: technically sharp, opinionated, libertarian-leaning, dryly funny, and peer-level with the audience. Assume the audience knows CS basics; do not over-explain and do not be obsequious. Use the provided topic history for conversational context, but answer the latest user. Search the web for current facts, news, sources, or prices using your search tool; when you use it, cite source URLs in your final answer. Keep replies short (2-4 sentences max unless asked for more). You are participating in an ongoing Zulip thread. Stay in character.""" log.info("Connecting to Zulip...") client = zulip.Client(config_file=".zuliprc") log.info("Connected as %s", client.email) # --------------------------------------------------------------------------- # Zulip context # --------------------------------------------------------------------------- def fetch_topic_history(msg: dict, limit: int = HISTORY_LIMIT) -> str: """Fetch recent messages from the same Zulip stream/topic for LLM context.""" if msg.get("type") != "stream": return "" stream = msg.get("display_recipient") topic = msg.get("subject") or msg.get("topic") if not stream or not topic: return "" try: resp = client.get_messages({ "anchor": msg.get("id", "newest"), "num_before": limit, "num_after": 0, "narrow": [ {"operator": "stream", "operand": stream}, {"operator": "topic", "operand": topic}, ], }) except Exception as e: log.warning("Failed to fetch Zulip topic history: %s", e) return "" if resp.get("result") != "success": log.warning("Zulip history request failed: %s", resp) return "" lines = [] for item in resp.get("messages", [])[-limit:]: if item.get("id") == msg.get("id"): continue sender = item.get("sender_full_name") or item.get("sender_email") or "someone" content = html.unescape(re.sub(r"<[^>]+>", " ", item.get("content", ""))) content = re.sub(r"\s+", " ", content).strip() if content: lines.append(f"{sender} said: {content}") if not lines: return "" return "Recent topic history:\n" + "\n".join(lines) # --------------------------------------------------------------------------- # agentd one-shot helper # --------------------------------------------------------------------------- def _agentd(*args) -> tuple[int, str, str]: """Run agentd with args; return (returncode, stdout, stderr).""" result = subprocess.run( ["/home/ben/omni/live/_/bin/agentd"] + list(args), capture_output=True, text=True, ) return result.returncode, result.stdout.strip(), result.stderr.strip() def _unique_session_name(msg: dict) -> str: """Build a unique, short-lived session name for a single message.""" ts = int(time.time() * 1000) if msg.get("type") == "private": sender_id = str(msg.get("sender_id") or "dm") safe_id = re.sub(r"[^a-z0-9-]", "-", sender_id.lower())[:20].strip("-") return f"slop-dm-{safe_id}-{ts}" stream_id = msg.get("stream_id", "0") topic = msg.get("subject") or msg.get("topic") or "general" slug = re.sub(r"[^a-z0-9-]", "-", topic.lower())[:20].strip("-") return f"slop-{stream_id}-{slug}-{ts}" def ask_agentd(msg: dict, content: str) -> str: """ Run a one-shot agentd agent for a single mention. Flow: create → start → send → recv → stop No persistent session state; no stale-binary issues. """ name = _unique_session_name(msg) sender = msg.get("sender_full_name") or msg.get("sender_email") or "user" # Build system prompt with Zulip topic history baked in topic_history = fetch_topic_history(msg) prompt_parts = [SYSTEM_PROMPT] if topic_history: prompt_parts.append(f"\n\n{topic_history}") prompt_parts.append("\nYou have access to kagi_search and run_bash tools via agentd.") prompt_parts.append("\nYou are now active in this thread. Respond to the latest message.") system_prompt = "\n".join(prompt_parts) log.info("Creating one-shot agentd session: %s", name) rc, _, err = _agentd( "create", name, system_prompt, "--mode", "oneshot", "--model", AGENTD_MODEL, "--cwd", str(Path(__file__).parent), ) if rc != 0: log.warning("agentd create failed for %s: %s", name, err) return "sorry, couldn't reach the agent" rc, _, err = _agentd("start", name) if rc != 0: log.warning("agentd start failed for %s: %s", name, err) return "sorry, couldn't start the agent" # Capture timestamp BEFORE send to avoid replaying old events ts_before = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") user_msg = f"{sender}: {content}" rc, _, err = _agentd("send", name, user_msg) if rc != 0: log.warning("agentd send failed for %s: %s", name, err) _agentd("stop", name) return "sorry, couldn't send to the agent" rc, reply, err = _agentd( "recv", name, "--since", ts_before, "--timeout", str(AGENTD_RECV_TIMEOUT), ) # Always clean up the session _agentd("stop", name) if rc != 0 or not reply: log.warning("agentd recv failed (rc=%d): %s", rc, err) return "sorry, agent timed out" return reply # --------------------------------------------------------------------------- # Price fetcher # --------------------------------------------------------------------------- def fetch_price(ticker: str) -> str: """Return a formatted price string for a ticker, or an error message.""" log.info("Fetching price for %s", ticker) try: price = yf.Ticker(ticker).fast_info["lastPrice"] result = f"${ticker}: ${price:,.2f}" log.info(" -> %s", result) return result except Exception as e: log.warning(" -> %s unavailable: %s", ticker, e) return f"${ticker}: unavailable ({e})" # --------------------------------------------------------------------------- # Event handler # --------------------------------------------------------------------------- def handle_event(event: dict) -> None: """Process a Zulip event and reply to mentions and/or tickers.""" if event.get("type") != "message": return msg = event["message"] if msg["sender_email"] == client.email: return content = msg.get("content", "") has_mention = MENTION_RE.search(content) tickers = TICKER_RE.findall(content.upper()) if not has_mention and not tickers: return if has_mention: cleaned = MENTION_RE.sub("", content).strip() reply = ask_agentd(msg, cleaned) if tickers: prices = [fetch_price(t) for t in dict.fromkeys(tickers)] reply += "\n\n" + "\n".join(prices) else: lines = [fetch_price(t) for t in dict.fromkeys(tickers)] reply = "\n".join(lines) client.send_message({ "type": msg["type"], "to": msg["stream_id"] if msg["type"] == "stream" else msg["sender_email"], "topic": msg.get("subject", ""), "content": reply, }) log.info("Reply sent.") def main() -> None: log.info("Listening for messages...") client.call_on_each_event(handle_event, event_types=["message"]) if __name__ == "__main__": main()