Replace agentd with one-shot agent

1 files changed, 75 insertions, 82 deletions

diff --git a/main.py b/main.py
index def25e4..5e1e61a 100644
--- a/main.py
+++ b/main.py
@@ -1,7 +1,7 @@
 #!/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.
+and delegates @mention responses to a one-shot agent per message.
 
 Usage:
     python main.py
@@ -11,11 +11,13 @@ Requires a .zuliprc file in the same directory (see .zuliprc.example).
 
 import html
 import logging
+import os
 import re
 import subprocess
 import time
-from datetime import datetime, UTC
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+from threading import Semaphore
 
 import yfinance as yf
 import zulip
@@ -33,8 +35,11 @@ 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"
+AGENT_TIMEOUT = int(os.environ.get("SLOPBOT_AGENT_TIMEOUT", "120"))
+AGENT_MODEL = os.environ.get("SLOPBOT_AGENT_MODEL", "parasail/kimi-k26")
+AGENT_BIN = os.environ.get("SLOPBOT_AGENT", "agent")
+WORKERS = int(os.environ.get("SLOPBOT_WORKERS", "4"))
+AGENT_WORKERS = int(os.environ.get("SLOPBOT_AGENT_WORKERS", "1"))
 HISTORY_LIMIT = 20
 
 SYSTEM_PROMPT = """You are slop-bot in the meshheads.org Zulip chat: technically sharp,
@@ -50,6 +55,9 @@ log.info("Connecting to Zulip...")
 client = zulip.Client(config_file=".zuliprc")
 log.info("Connected as %s", client.email)
 
+executor = ThreadPoolExecutor(max_workers=WORKERS, thread_name_prefix="slopbot")
+agent_slots = Semaphore(AGENT_WORKERS)
+
 
 # ---------------------------------------------------------------------------
 # Zulip context
@@ -99,14 +107,19 @@ def fetch_topic_history(msg: dict, limit: int = HISTORY_LIMIT) -> str:
 
 
 # ---------------------------------------------------------------------------
-# agentd one-shot helper
+# agent one-shot helper
 # ---------------------------------------------------------------------------
 
-def _agentd(*args) -> tuple[int, str, str]:
-    """Run agentd with args; return (returncode, stdout, stderr)."""
+def _agent(prompt: str) -> tuple[int, str, str]:
+    """Run agent once; return (returncode, stdout, stderr)."""
     result = subprocess.run(
-        ["/home/ben/omni/live/_/bin/agentd"] + list(args),
-        capture_output=True, text=True,
+        [
+            AGENT_BIN,
+            "--model", AGENT_MODEL,
+            "--eval-cwd", str(Path(__file__).parent),
+            prompt,
+        ],
+        capture_output=True, text=True, timeout=AGENT_TIMEOUT,
     )
     return result.returncode, result.stdout.strip(), result.stderr.strip()
 
@@ -124,60 +137,33 @@ def _unique_session_name(msg: dict) -> str:
     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.
-    """
+def ask_agent(msg: dict, content: str) -> str:
+    """Run a one-shot agent for a single mention."""
     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),
-    )
+    prompt_parts.append("\nYou have access to the agent shell tools if you need them.")
+    prompt_parts.append("\nRespond to the latest Zulip message.")
+    prompt_parts.append(f"\nLatest message from {sender}:\n{content}")
+    prompt = "\n".join(prompt_parts)
 
-    # Always clean up the session
-    _agentd("stop", name)
+    log.info("Running one-shot agent: %s", name)
+    try:
+        with agent_slots:
+            rc, reply, err = _agent(prompt)
+    except subprocess.TimeoutExpired:
+        log.warning("agent timed out for %s", name)
+        return "sorry, agent timed out"
+    except Exception as e:
+        log.warning("agent failed for %s: %s", name, e)
+        return "sorry, couldn't reach the agent"
 
     if rc != 0 or not reply:
-        log.warning("agentd recv failed (rc=%d): %s", rc, err)
+        log.warning("agent failed for %s (rc=%d): %s", name, rc, err)
         return "sorry, agent timed out"
 
     return reply
@@ -204,38 +190,45 @@ def fetch_price(ticker: str) -> str:
 # 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
+def handle_message(msg: dict) -> None:
+    """Process a Zulip message and reply to mentions and/or tickers."""
+    try:
+        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_agent(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.")
+    except Exception:
+        log.exception("Failed to handle message %s", msg.get("id"))
 
-    content = msg.get("content", "")
-    has_mention = MENTION_RE.search(content)
-    tickers = TICKER_RE.findall(content.upper())
 
-    if not has_mention and not tickers:
+def handle_event(event: dict) -> None:
+    """Queue Zulip message work without blocking the event loop."""
+    if event.get("type") != "message":
         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.")
+    executor.submit(handle_message, event["message"])
 
 
 def main() -> None: