"""Wipe all game data for a channel — session, history, memories, assets.
# 💀🔥 NUCLEAR GAME RESET
"""
import json
import logging
from typing import Any
logger = logging.getLogger(__name__)
TOOL_NAME = "wipe_game_data"
TOOL_DESCRIPTION = (
"Completely wipe all game data for the current channel. "
"Deletes the session, turn history, memories, and assets. "
"This is irreversible. Use when a channel has corrupted "
"game data from older versions or when the user wants a "
"clean slate. Requires no arguments — operates on the "
"current channel."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"confirm": {
"type": "boolean",
"description": "Must be true to confirm the wipe.",
},
},
"required": ["confirm"],
}
[docs]
async def run(confirm: bool = False, ctx: Any = None, **_kw: Any) -> str:
"""Nuke all game data for the current channel. Admin only."""
if ctx is None:
return json.dumps({"error": "No context available."})
# Admin gate # 💀🔥
try:
from tools.alter_privileges import has_privilege, PRIVILEGES
redis_auth = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
user_id = getattr(ctx, "user_id", "") or ""
if not await has_privilege(
redis_auth, user_id, PRIVILEGES["UNSANDBOXED_EXEC"], config,
):
return json.dumps({
"success": False,
"error": "Requires UNSANDBOXED_EXEC privilege. Ask an admin.",
})
except ImportError:
return json.dumps({
"success": False,
"error": "Privilege system unavailable.",
})
if not confirm:
return json.dumps({
"error": "Set confirm=true to wipe. This is irreversible.",
})
channel_id = getattr(ctx, "channel_id", None)
if not channel_id:
return json.dumps({"error": "No channel_id in context."})
redis = getattr(ctx, "redis", None)
if redis is None:
config = getattr(ctx, "config", None)
if config:
redis = getattr(config, "redis", None)
if redis is None:
return json.dumps({"error": "No Redis connection available."})
deleted: list[str] = []
errors: list[str] = []
# 1. Remove in-memory session # 💀
try:
from game_session import get_session, remove_session
session = get_session(str(channel_id))
game_id = session.game_id if session else None
game_name = session.game_name if session else None
if session:
remove_session(str(channel_id))
deleted.append("in-memory session")
except ImportError:
game_id = None
game_name = None
except Exception as exc:
errors.append(f"session removal: {exc}")
game_id = None
game_name = None
# 2. Delete Redis session + history # 🔥
session_key = f"game:session:{channel_id}"
history_key = f"game:session:{channel_id}:history"
try:
raw = await redis.get(session_key)
if raw and not game_id:
data = json.loads(raw)
game_id = data.get("game_id")
game_name = data.get("game_name")
count = await redis.delete(session_key, history_key)
if count:
deleted.append(f"session keys ({count})")
except Exception as exc:
errors.append(f"session keys: {exc}")
# 3. Delete memories + assets if we have a game_id # 🌀
if game_id:
mem_keys = [
f"game:mem:basic:{game_id}",
f"game:mem:channel:{game_id}",
f"game:assets:{game_id}",
]
try:
count = await redis.delete(*mem_keys)
if count:
deleted.append(f"memories+assets ({count} keys)")
except Exception as exc:
errors.append(f"memories/assets: {exc}")
# 4. Remove from game index # 😈
try:
await redis.hdel("game:index", game_id)
deleted.append("game index entry")
except Exception as exc:
errors.append(f"game index: {exc}")
result = {
"success": True,
"channel_id": str(channel_id),
"game_id": game_id,
"game_name": game_name,
"deleted": deleted,
}
if errors:
result["errors"] = errors
logger.info(
"Wiped game data for channel %s (game: %s): %s",
channel_id, game_name, ", ".join(deleted),
)
return json.dumps(result)