path: root/main.py

#!/usr/bin/env python3
"""
slopbot - Zulip bot that replies with stock prices when it sees $TICKER patterns,
and delegates @mention responses to agentd sessions.

Usage:
    python main.py

Requires a .zuliprc file in the same directory (see .zuliprc.example).
"""

import html
import logging
import re
import subprocess
import threading
import time
from datetime import datetime
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)

SESSION_TIMEOUT = 20 * 60  # seconds
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."""

# In-memory session tracking: name -> last_activity (time.time())
_sessions: dict[str, float] = {}
_sessions_lock = threading.Lock()

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 session management
# ---------------------------------------------------------------------------

def _session_name(msg: dict) -> str:
    """Build deterministic session name. DMs are per-sender; stream messages are per-stream+topic."""
    if msg.get("type") == "private":
        sender_id = str(msg.get("sender_id") or msg.get("sender_email", "unknown"))
        safe_id = re.sub(r"[^a-z0-9-]", "-", sender_id.lower())[:40].strip("-")
        return f"slop-dm-{safe_id}"
    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())[:40].strip("-")
    return f"slop-{stream_id}-{slug}"


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 _session_running(name: str) -> bool:
    """Check real agentd state for a persistent session.

    Returns True if running (or successfully restarted after stale/idle).
    Returns False if the session does not exist.
    """
    rc, out, err = _agentd("status", name)
    if rc != 0 or "not found" in out.lower() or "not found" in err.lower():
        return False
    if "running" in out.lower():
        return True
    # Session exists but is stale/idle — restart it.
    log.info("Session %s is not running (status: %s); restarting", name, out.strip())
    restart_rc, _, restart_err = _agentd("restart", name)
    if restart_rc != 0:
        log.warning("agentd restart failed for %s: %s", name, restart_err)
        return False
    return True


def _start_session(name: str, system_prompt: str) -> None:
    """Create and start a new persistent agentd session."""
    rc, _, err = _agentd(
        "create",
        name,
        system_prompt,
        "--mode",
        "persistent",
        "--model",
        AGENTD_MODEL,
        "--cwd",
        str(Path(__file__).parent),
    )
    if rc != 0:
        log.warning("agentd create failed for %s: %s", name, err)

    rc, _, err = _agentd("start", name)
    if rc != 0:
        log.warning("agentd start failed for %s: %s", name, err)


def _stop_session(name: str) -> None:
    _agentd("stop", name)
    with _sessions_lock:
        _sessions.pop(name, None)


def _reap_idle_sessions() -> None:
    """Stop sessions that have been idle > SESSION_TIMEOUT."""
    now = time.time()
    with _sessions_lock:
        idle = [n for n, t in _sessions.items() if now - t > SESSION_TIMEOUT]
    for name in idle:
        log.info("Stopping idle session: %s", name)
        _stop_session(name)


def ask_agentd(msg: dict, content: str) -> str:
    """Send a message to the agentd session for this topic and return the response."""
    name = _session_name(msg)
    sender = msg.get("sender_full_name") or msg.get("sender_email") or "user"

    if not _session_running(name):
        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("Starting new agentd session: %s", name)
        _start_session(name, system_prompt)

    with _sessions_lock:
        _sessions[name] = time.time()

    # Capture timestamp BEFORE send to avoid replaying old turns.
    ts_before = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    user_msg = f"{sender}: {content}"
    rc, _, err = _agentd("send", name, user_msg, "--mode", "persistent")
    if rc != 0:
        if "stale" in err.lower():
            # Binary was redeployed; restart and retry once.
            log.info("Session %s is stale, restarting and retrying send", name)
            _agentd("restart", name)
            time.sleep(1)
            ts_before = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
            rc, _, err = _agentd("send", name, user_msg, "--mode", "persistent")
        if rc != 0:
            log.warning("agentd send failed: %s", err)
            return "sorry, couldn't reach the agent"

    rc, reply, err = _agentd(
        "recv",
        name,
        "--since",
        ts_before,
        "--timeout",
        str(AGENTD_RECV_TIMEOUT),
        "--mode",
        "persistent",
    )
    if rc != 0 or not reply:
        log.warning("agentd recv failed (rc=%d): %s", rc, err)
        return "sorry, agent timed out"

    with _sessions_lock:
        _sessions[name] = time.time()

    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

    _reap_idle_sessions()

    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()