diff options
Diffstat (limited to 'main.py')
| -rw-r--r-- | main.py | 173 |
1 files changed, 56 insertions, 117 deletions
@@ -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 agentd sessions. +and delegates @mention responses to a one-shot agentd agent per message. Usage: python main.py @@ -13,9 +13,8 @@ import html import logging import re import subprocess -import threading import time -from datetime import datetime +from datetime import datetime, UTC from pathlib import Path import yfinance as yf @@ -34,7 +33,6 @@ 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 @@ -48,10 +46,6 @@ 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) @@ -105,140 +99,87 @@ def fetch_topic_history(msg: dict, limit: int = HISTORY_LIMIT) -> str: # --------------------------------------------------------------------------- -# agentd session management +# agentd one-shot helper # --------------------------------------------------------------------------- -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) + 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. +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}" - Returns True if running (or successfully restarted after stale/idle). - Returns False if the session does not exist. + +def ask_agentd(msg: dict, content: str) -> str: """ - 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.""" + 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", - "persistent", - "--model", - AGENTD_MODEL, - "--cwd", - str(Path(__file__).parent), + "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" - -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") + # 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, "--mode", "persistent") + rc, _, err = _agentd("send", name, user_msg) 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" + 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), - "--mode", - "persistent", + "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" - with _sessions_lock: - _sessions[name] = time.time() - return reply @@ -278,8 +219,6 @@ def handle_event(event: dict) -> None: 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) |