feat: add --llm mode for LLM-friendly stdout filtering

Split output when running with --llm: addressed messages from owners
go to stdout, everything else (chatter, logs, plugin loads) goes to
info.log. Adds owner privilege level (superset of admin) for gating
LLM access. Status lines (connect, ping, disconnect, reconnect) and
bot replies also appear on stdout for session awareness.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-19 20:11:23 +01:00
parent 0c18ba8e3a
commit ea6f07914e
4 changed files with 147 additions and 12 deletions

View File

@@ -7,6 +7,7 @@ import base64
import fnmatch
import logging
import random
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
@@ -77,9 +78,10 @@ class _TokenBucket:
class Bot:
"""IRC bot: ties connection, config, and plugins together."""
def __init__(self, config: dict, registry: PluginRegistry) -> None:
def __init__(self, config: dict, registry: PluginRegistry, *, llm: bool = False) -> None:
self.config = config
self.registry = registry
self._llm = llm
self.conn = IRCConnection(
host=config["server"]["host"],
port=config["server"]["port"],
@@ -92,6 +94,7 @@ class Bot:
self._started: float = time.monotonic()
self._tasks: set[asyncio.Task] = set()
self._reconnect_delay: float = 5.0
self._owner: list[str] = config.get("bot", {}).get("owner", [])
self._admins: list[str] = config.get("bot", {}).get("admins", [])
self._opers: set[str] = set() # hostmasks of known IRC operators
self._caps: set[str] = set() # negotiated IRCv3 caps
@@ -117,10 +120,18 @@ class Bot:
self._reconnect_delay = 5.0 # reset on clean run
except (OSError, ConnectionError) as exc:
log.error("connection lost: %s", exc)
if self._llm:
ts = time.strftime("%H:%M")
sys.stdout.write(f"{ts} --- disconnected: {exc}\n")
sys.stdout.flush()
if self._running:
jitter = self._reconnect_delay * 0.25 * (2 * random.random() - 1)
delay = self._reconnect_delay + jitter
log.info("reconnecting in %.0fs...", delay)
if self._llm:
ts = time.strftime("%H:%M")
sys.stdout.write(f"{ts} --- reconnecting in {delay:.0f}s\n")
sys.stdout.flush()
await asyncio.sleep(delay)
self._reconnect_delay = min(self._reconnect_delay * 2, 300.0)
@@ -259,11 +270,19 @@ class Bot:
# Protocol-level PING/PONG
if msg.command == "PING":
await self.conn.send(format_msg("PONG", msg.params[0] if msg.params else ""))
if self._llm:
ts = time.strftime("%H:%M")
sys.stdout.write(f"{ts} --- ping\n")
sys.stdout.flush()
return
# RPL_WELCOME (001) — join channels and WHO for oper detection
if msg.command == "001":
self.nick = msg.params[0] if msg.params else self.nick
if self._llm:
ts = time.strftime("%H:%M")
sys.stdout.write(f"{ts} --- connected as {self.nick}\n")
sys.stdout.flush()
for channel in self.config["bot"]["channels"]:
await self.join(channel)
await self.conn.send(format_msg("WHO", channel))
@@ -300,6 +319,16 @@ class Bot:
self._spawn(self._handle_ctcp(msg), name="ctcp")
return
# LLM mode: route PRIVMSG to stdout or info.log
if self._llm and msg.command == "PRIVMSG" and msg.text:
ts = time.strftime("%H:%M")
if self._is_addressed(msg) and self._is_owner(msg):
prefix = f"{msg.target} " if msg.is_channel else ""
sys.stdout.write(f"{ts} {prefix}<{msg.nick}> {msg.text}\n")
sys.stdout.flush()
else:
log.info("%s <%s> %s", msg.target, msg.nick, msg.text)
# Dispatch to event handlers (fire-and-forget)
channel = msg.target if msg.is_channel else None
event_type = msg.command
@@ -359,14 +388,25 @@ class Bot:
return True
return plugin_name in allowed
def _is_owner(self, msg: Message) -> bool:
"""Check if the sender matches a configured owner hostmask pattern."""
if not msg.prefix:
return False
for pattern in self._owner:
if fnmatch.fnmatch(msg.prefix, pattern):
return True
return False
def _is_admin(self, msg: Message) -> bool:
"""Check if the message sender is a bot admin.
Returns True if the sender is a known IRC operator or matches
a configured hostmask pattern (fnmatch-style).
Returns True if the sender is an owner, a known IRC operator,
or matches a configured admin hostmask pattern (fnmatch-style).
"""
if not msg.prefix:
return False
if self._is_owner(msg):
return True
if msg.prefix in self._opers:
return True
for pattern in self._admins:
@@ -374,6 +414,18 @@ class Bot:
return True
return False
def _is_addressed(self, msg: Message) -> bool:
"""Check if a message is addressed to the bot (DM or nick-prefixed)."""
if not msg.is_channel:
return True
text = (msg.text or "").lstrip()
nick_lower = self.nick.lower()
if text.lower().startswith(nick_lower):
rest = text[len(nick_lower):]
if rest and rest[0] in ":, ":
return True
return False
def _dispatch_command(self, msg: Message) -> None:
"""Check if a PRIVMSG is a bot command and spawn it."""
text = msg.text
@@ -453,6 +505,16 @@ class Bot:
for chunk in _split_utf8(line, max_text):
await self._bucket.acquire()
await self.conn.send(format_msg("PRIVMSG", target, chunk))
if self._llm:
ts = time.strftime("%H:%M")
display = chunk
if display.startswith("\x01ACTION ") and display.endswith("\x01"):
display = display[8:-1]
out = f"{ts} {target} * {self.nick} {display}"
else:
out = f"{ts} {target} <{self.nick}> {display}"
sys.stdout.write(out + "\n")
sys.stdout.flush()
async def reply(self, msg: Message, text: str) -> None:
"""Reply to the source of a message (channel or PM)."""

View File

@@ -33,6 +33,11 @@ def build_parser() -> argparse.ArgumentParser:
action="store_true",
help="enable debug logging",
)
p.add_argument(
"--llm",
action="store_true",
help="LLM mode: addressed messages to stdout, rest to info.log",
)
p.add_argument(
"--cprofile",
metavar="PATH",
@@ -111,7 +116,14 @@ def main(argv: list[str] | None = None) -> int:
level = logging.DEBUG if args.verbose else logging.INFO
log_fmt = config.get("logging", {}).get("format", "text")
if log_fmt == "json":
if args.llm:
handler = logging.FileHandler("info.log")
if log_fmt == "json":
handler.setFormatter(JsonFormatter())
else:
handler.setFormatter(logging.Formatter(LOG_FORMAT, datefmt=LOG_DATE))
logging.basicConfig(handlers=[handler], level=level)
elif log_fmt == "json":
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(handlers=[handler], level=level)
@@ -122,7 +134,7 @@ def main(argv: list[str] | None = None) -> int:
log.info("derp %s starting", __version__)
registry = PluginRegistry()
bot = Bot(config, registry)
bot = Bot(config, registry, llm=args.llm)
bot.load_plugins()
if args.tracemalloc: