"""Wipe all game data for a channel — session, history, memories, assets.
# 💀🔥 NUCLEAR GAME RESET
"""
import jsonutil as 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:
"""Irreversibly wipe every trace of game state for the current channel.
Admin-only nuclear reset used to recover a channel whose game data was
corrupted by an older version or when the user explicitly wants a clean
slate. It tears down the in-memory session, the persisted Redis session and
history, and the game's memories, assets and index entry in one pass,
requiring an explicit ``confirm=True`` so it cannot fire by accident.
Interactions: gates on ``tools.alter_privileges.has_privilege`` with the
``UNSANDBOXED_EXEC`` privilege (reading ``ctx.redis``, ``ctx.config`` and
``ctx.user_id``). Resolves the channel from ``ctx.channel_id`` and a Redis
client from ``ctx.redis`` (falling back to ``ctx.config.redis``). Calls
``game_session.get_session`` / ``game_session.remove_session`` to drop the
live session, then deletes the Redis keys ``game:session:{channel_id}`` and
its ``:history`` companion, the ``game:mem:basic:{game_id}``,
``game:mem:channel:{game_id}`` and ``game:assets:{game_id}`` keys, and the
``game_id`` field of the ``game:index`` hash. Logs a summary line on
completion. Per-step errors are collected rather than aborting the wipe.
Called by the tool-dispatch layer: ``tool_loader.py`` registers this module
via its ``TOOL_NAME`` / ``run`` contract and the registry invokes ``run``
with the parsed arguments and ``ctx``; there are no direct internal callers.
Args:
confirm (bool): Must be ``True`` to proceed; any falsy value returns an
error without touching data.
ctx (Any): The tool ``ToolContext`` providing Redis, config, the user id
for the privilege check, and the target ``channel_id``.
**_kw (Any): Ignored extra keyword arguments tolerated for dispatch.
Returns:
str: A JSON string. On success ``{"success": True, "channel_id": ...,
"game_id": ..., "game_name": ..., "deleted": [...]}`` (plus an
``errors`` list when any individual step failed). On refusal or setup
problems an ``{"error": ...}`` / ``{"success": False, "error": ...}``
object (missing context, missing privilege, unset ``confirm``, missing
``channel_id``, or no Redis connection).
"""
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)