136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
"""Admin and debug commands: wipe history, debug menu, chat history."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from discord.ext import commands
|
|
|
|
from vibe_bot.commands._state import require_app, set_app
|
|
from vibe_bot.prompts import get_user_info
|
|
from vibe_bot.textutil import split_message
|
|
from vibe_bot.tools import get_channel_members, get_channel_members_impl
|
|
|
|
if TYPE_CHECKING:
|
|
from discord.ext.commands import Bot
|
|
from discord.ext.commands import Context as CommandsContext
|
|
|
|
from vibe_bot.app import App
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def register(bot: commands.Bot, app: App) -> None:
|
|
"""Register the admin and debug commands."""
|
|
set_app(app)
|
|
bot.command(name="lobotomize")(lobotomize)
|
|
bot.command(name="debug")(debug)
|
|
bot.command(name="history")(history)
|
|
|
|
|
|
async def lobotomize(ctx: CommandsContext[Bot]) -> None:
|
|
"""Clear all conversation history and memory for all bots."""
|
|
app = require_app()
|
|
logger.info("Lobotomize command triggered by user %s", ctx.author.id)
|
|
await asyncio.to_thread(app.db.clear_all_messages)
|
|
await ctx.send("All conversation history and memory has been cleared. 🧠✨")
|
|
|
|
|
|
async def debug(
|
|
ctx: CommandsContext[Bot],
|
|
*,
|
|
subcommand: str | None = None,
|
|
) -> None:
|
|
"""Debug menu for various debugging sub-commands.
|
|
|
|
Usage: !debug <subcommand>
|
|
Available sub-commands:
|
|
- members: List all members in the current channel
|
|
- whoami: Show all information the bot has about you
|
|
- tools: Show the LLM's available tools
|
|
"""
|
|
logger.info(
|
|
"Debug command triggered by user %s with subcommand %r",
|
|
ctx.author.id,
|
|
subcommand,
|
|
)
|
|
|
|
if not subcommand:
|
|
menu = "Debug Menu:\n\n"
|
|
menu += "Available sub-commands:\n"
|
|
menu += "- `members` - List all members in the current channel\n"
|
|
menu += "- `whoami` - Show all information the bot has about you\n"
|
|
menu += "- `tools` - Show the LLM's available tools"
|
|
await ctx.send(menu)
|
|
return
|
|
|
|
if subcommand == "members":
|
|
result = get_channel_members_impl(ctx.channel)
|
|
for chunk in split_message(result):
|
|
await ctx.send(chunk)
|
|
return
|
|
|
|
if subcommand == "whoami":
|
|
user_info = get_user_info(ctx.author)
|
|
for chunk in split_message(user_info):
|
|
await ctx.send(chunk)
|
|
return
|
|
|
|
if subcommand == "tools":
|
|
tool_list = "LLM Tools:\n\n"
|
|
tool_list += f"- `{get_channel_members.name}`\n"
|
|
tool_list += f" Description: {get_channel_members.description}\n"
|
|
args_schema = get_channel_members.args_schema
|
|
if isinstance(args_schema, type):
|
|
tool_list += f" Parameters: {args_schema.model_json_schema()}"
|
|
for chunk in split_message(tool_list):
|
|
await ctx.send(chunk)
|
|
return
|
|
|
|
await ctx.send(
|
|
f"Unknown debug sub-command: `{subcommand}`\n\n"
|
|
f"Use `!debug` to see available sub-commands.",
|
|
)
|
|
|
|
|
|
async def history(ctx: CommandsContext[Bot], bot_name: str) -> None:
|
|
"""View the chat history of a custom bot.
|
|
|
|
Usage: !history <bot_name>
|
|
"""
|
|
app = require_app()
|
|
logger.info(
|
|
"History command triggered by user %s for bot %r",
|
|
ctx.author.id,
|
|
bot_name,
|
|
)
|
|
|
|
bot_info = await asyncio.to_thread(app.manager.get_custom_bot, bot_name)
|
|
|
|
if not bot_info:
|
|
await ctx.send(f"Custom bot '{bot_name}' not found.")
|
|
return
|
|
|
|
history = await asyncio.to_thread(
|
|
app.db.get_bot_history, bot_name=bot_name, limit=20
|
|
)
|
|
|
|
if not history:
|
|
await ctx.send(f"No chat history found for **{bot_name}**. ")
|
|
return
|
|
|
|
history.reverse()
|
|
|
|
formatted_history: list[str] = []
|
|
for user_msg, bot_resp in history:
|
|
formatted_history.append(user_msg)
|
|
formatted_history.append(f"{bot_name}: {bot_resp}")
|
|
|
|
header = f"Chat History for **{bot_name}**:\n\n"
|
|
full_text = header + "\n---\n".join(formatted_history)
|
|
|
|
for chunk in split_message(full_text):
|
|
await ctx.send(chunk)
|