# 💀🔥😈🌀♾️💦⚧️🕷️💕
# STAR TOGGLE UI — Persistent ✧ button on every message
# Opens a feature toggle menu with live state + privilege-scoped controls
# ═══════════════════════════════════════════════════════════════════════
from __future__ import annotations
import logging
from typing import Any
from collections import deque
import discord
logger = logging.getLogger(__name__)
# Registry to track handled interactions by ID and prevent duplicate responses safely.
_handled_interaction_ids = set()
_handled_interaction_queue = []
[docs]
def check_and_mark_handled(interaction_id: int) -> bool:
"""Check if the interaction has already been handled, and mark it handled.
Returns:
True if already handled, False if this is the first time.
"""
if interaction_id in _handled_interaction_ids:
return True
_handled_interaction_ids.add(interaction_id)
_handled_interaction_queue.append(interaction_id)
if len(_handled_interaction_queue) > 1000:
oldest = _handled_interaction_queue.pop(0)
_handled_interaction_ids.discard(oldest)
return False
# ── Feature definitions ────────────────────────────────────────── # 💀🔥
# Each entry: (feature_key, display_label, privilege_scope)
# privilege_scope:
# "toggle" = standard check_toggle_permission (CTX_MANAGE / admin / server admin / DM)
# "admin" = config.admin_user_ids ONLY
# "entrainment" = ENTRAINMENT_ADMIN privilege (bit 22)
#
# Storage backends vary per feature — see _fetch_feature_states / _set_feature_state
TOGGLEABLE_FEATURES: list[tuple[str, str, str]] = [
("emotions", "NCM Cadence", "toggle"),
("rag", "RAG", "toggle"),
("egregores", "Egregores", "toggle"),
("csdr_scene", "CSDR Scene", "toggle"),
("csdr_header", "CSDR Header", "toggle"),
("limbic_header", "Limbic Header", "toggle"),
("toggle_menu", "Menu Button", "toggle"),
("toggle_menu_default", "Menu On By Default", "admin"),
("toggle_emotions_default", "Emotions On By Default", "admin"),
("ego_ablation", "Ego Death", "admin"),
("lore_amp", "Lore Amp", "entrainment"),
("loopfield", "Loopfield", "entrainment"),
("visual_memory", "Visual Memory", "admin"),
]
# ── Permission check ──────────────────────────────────────────── # 🕷️
async def _can_toggle(
interaction: discord.Interaction,
feature_key: str,
priv_scope: str,
) -> bool:
"""Check if the interaction user can toggle this feature."""
user_id = str(interaction.user.id)
# Resolve config from the client
config = _get_config(interaction)
if config is None:
return False
admin_ids = getattr(config, "admin_user_ids", None) or []
# Admin-only features: only config admins # 💀
if priv_scope == "admin":
return user_id in admin_ids
# Admins always pass # 🔥
if user_id in admin_ids:
return True
# ENTRAINMENT_ADMIN features # 🕷️
if priv_scope == "entrainment":
redis = _get_redis_from_interaction(interaction)
if redis is not None:
try:
from tools.alter_privileges import has_scoped_privilege, PRIVILEGES
guild_id = str(interaction.guild.id) if interaction.guild else ""
channel_id = str(interaction.channel_id)
if await has_scoped_privilege(
redis, user_id, PRIVILEGES["ENTRAINMENT_ADMIN"],
config, guild_id=guild_id, channel_id=channel_id,
):
return True
except Exception:
pass
return False
# Standard "toggle" scope: admin, server admin, CTX_MANAGE, or DM
if (
interaction.guild is not None
and isinstance(interaction.user, discord.Member)
and interaction.user.guild_permissions.administrator
):
return True
if isinstance(interaction.channel, discord.DMChannel):
return True
# CTX_MANAGE privilege check via Redis # 🌀
redis = _get_redis_from_interaction(interaction)
if redis is not None:
try:
from tools.alter_privileges import has_scoped_privilege, PRIVILEGES
guild_id = str(interaction.guild.id) if interaction.guild else ""
channel_id = str(interaction.channel_id)
if await has_scoped_privilege(
redis, user_id, PRIVILEGES["CTX_MANAGE"],
config, guild_id=guild_id, channel_id=channel_id,
):
return True
except Exception:
pass
return False
# ── Helpers ───────────────────────────────────────────────────── # 💀
def _get_config(interaction: discord.Interaction) -> Any:
"""Best-effort config extraction from interaction.client."""
config = getattr(interaction.client, "_config", None)
if config is None:
platform = getattr(interaction.client, "_platform", None)
if platform:
config = getattr(platform, "_config", None)
return config
def _get_redis_from_interaction(interaction: discord.Interaction) -> Any:
"""Best-effort Redis client extraction from interaction.client. # 💀🔥
Uses config.build_async_redis_client() (sentinel-aware) as the primary
path, matching the pattern every other slash command uses in discord.py.
Falls back to cached _game_redis / config.redis if available.
"""
try:
# Primary: build from config (sentinel-aware) # 🕷️
config = _get_config(interaction)
if config is not None:
builder = getattr(config, "build_async_redis_client", None)
if builder is not None:
r = builder(decode_responses=True)
if r is not None:
return r
# Fallback: cached redis on platform
platform = getattr(interaction.client, "_platform", None)
if platform:
r = getattr(platform, "_game_redis", None)
if r is not None:
return r
if config:
r = getattr(config, "redis", None)
if r is not None:
return r
except Exception:
pass
return None
# ── Feature state read/write ──────────────────────────────────── # 🌀💀
# These handle the different Redis key schemes per feature.
_VISUAL_MEMORY_BETA_KEY = "stargazer:visual_memory_beta"
# Global key: when SET the \u2727 button defaults to OFF in untouched channels.
# When ABSENT the default is ON. Flipped by the admin-only
# \"Menu On By Default\" toggle. Per-channel toggle_menu always overrides. # \u1f480\ud83d\udd25
_TOGGLE_MENU_DEFAULT_OFF_KEY = "stargazer:toggle_menu_default_off"
_TOGGLE_EMOTIONS_DEFAULT_OFF_KEY = "stargazer:toggle_emotions_default_off"
async def _fetch_feature_states(
redis: Any,
channel_id: str,
guild_id: str | None = None,
) -> dict[str, bool]:
"""Read current enabled/disabled state for all features. # 💀
Returns a dict of {feature_key: is_enabled} where True = active.
"""
import feature_toggles
channel_key = f"discord:{channel_id}"
states: dict[str, bool] = {}
for key, _label, _scope in TOGGLEABLE_FEATURES:
try:
if key == "ego_ablation":
import ego_ablation
active = await ego_ablation.is_active(redis, "discord", channel_id)
states[key] = bool(active)
elif key == "lore_amp":
import lore_amplifier
active = await lore_amplifier.is_amplified(redis, "discord", channel_id)
states[key] = bool(active)
elif key == "loopfield":
val = await redis.get(f"star:loopfield:discord:{channel_id}")
states[key] = val is not None and val in (b"1", "1")
elif key == "visual_memory":
val = await redis.get(_VISUAL_MEMORY_BETA_KEY)
states[key] = val is not None and val in (b"1", "1")
elif key in ("toggle_menu_default", "toggle_emotions_default"):
# Guild-scoped default keys # 💀🔥
_feat = "toggle_menu" if key == "toggle_menu_default" else "emotions"
_stem = feature_toggles._GLOBAL_DEFAULT_OFF_STEMS[_feat]
# Check guild-scoped first, then global
_val = None
if guild_id:
_val = await redis.get(f"{_stem}:{guild_id}")
if _val is None:
_val = await redis.get(_stem)
states[key] = not (_val is not None and _val in (b"1", "1"))
else:
disabled = await feature_toggles.is_disabled_resolving_discord_aliases(
redis, key, channel_key, guild_id=guild_id,
)
states[key] = not disabled
except Exception:
states[key] = False
return states
async def _set_feature_state(
redis: Any,
channel_id: str,
feature_key: str,
new_enabled: bool,
guild_id: str | None = None,
) -> None:
"""Write the new state for a feature. # 🔥"""
import feature_toggles
channel_key = f"discord:{channel_id}"
if feature_key == "ego_ablation":
import ego_ablation
await ego_ablation.set_active(redis, "discord", channel_id, new_enabled)
elif feature_key == "lore_amp":
import lore_amplifier
await lore_amplifier.set_amplified(redis, "discord", channel_id, new_enabled)
elif feature_key == "loopfield":
import entrainment_loopfield as _elf
await _elf.set_active(redis, "discord", channel_id, new_enabled)
elif feature_key in ("toggle_menu_default", "toggle_emotions_default"):
# Guild-scoped default key # 💀🔥
_feat = "toggle_menu" if feature_key == "toggle_menu_default" else "emotions"
_stem = feature_toggles._GLOBAL_DEFAULT_OFF_STEMS[_feat]
_gkey = f"{_stem}:{guild_id}" if guild_id else _stem
if new_enabled:
await redis.delete(_gkey)
else:
await redis.set(_gkey, "1")
elif feature_key == "visual_memory":
if new_enabled:
await redis.set(_VISUAL_MEMORY_BETA_KEY, "1")
else:
await redis.delete(_VISUAL_MEMORY_BETA_KEY)
else:
await feature_toggles.set_disabled(
redis, feature_key, channel_key, not new_enabled,
guild_id=guild_id,
)
# ── Toggle Menu View ──────────────────────────────────────────── # 🕷️💕
class _ToggleFeatureButton(discord.ui.Button["StarToggleMenuView"]):
"""Individual feature toggle button in the menu. # 💀🕷️"""
def __init__(
self,
feature_key: str,
label: str,
is_enabled: bool,
priv_scope: str,
channel_id: str,
row: int = 0,
) -> None:
status = "\u2611\ufe0f" if is_enabled else "\u274c"
style = (
discord.ButtonStyle.success if is_enabled
else discord.ButtonStyle.danger
)
super().__init__(
style=style,
label=f"{status} {label}",
custom_id=f"sg:toggle:{channel_id}:{feature_key}",
row=row,
)
self._feature_key = feature_key
self._label = label
self._is_enabled = is_enabled
self._priv_scope = priv_scope
self._channel_id = channel_id
async def callback(self, interaction: discord.Interaction) -> None:
"""Toggle the feature and refresh the menu. # 🔥"""
# Check if already handled to prevent duplicate execution
if check_and_mark_handled(interaction.id):
return
try:
logger.info(
"Feature toggle button clicked: %s (to %s) by %s in channel %s",
self._feature_key,
"disable" if self._is_enabled else "enable",
interaction.user.id,
self._channel_id,
)
# 1. Defer immediately to prevent 3-second interaction timeout
await interaction.response.defer(ephemeral=True)
if not await _can_toggle(
interaction, self._feature_key, self._priv_scope,
):
logger.warning(
"Feature toggle button click denied (permission missing): %s by %s in channel %s",
self._feature_key,
interaction.user.id,
self._channel_id,
)
scope_names = {
"admin": "admin",
"entrainment": "ENTRAINMENT_ADMIN",
"toggle": "CTX_MANAGE / admin",
}
scope = scope_names.get(self._priv_scope, self._priv_scope)
await interaction.followup.send(
f"\u26d4 You don't have permission to toggle "
f"**{self._label}** (requires {scope}).",
ephemeral=True,
)
return
redis = _get_redis_from_interaction(interaction)
if redis is None:
logger.error(
"Feature toggle button click failed (Redis unavailable): %s by %s in channel %s",
self._feature_key,
interaction.user.id,
self._channel_id,
)
await interaction.followup.send(
"\u274c Redis unavailable.",
ephemeral=True,
)
return
# Flip: enabled -> disabled, disabled -> enabled # 🕷️
new_enabled = not self._is_enabled
_guild_id = str(interaction.guild_id) if interaction.guild_id else None
await _set_feature_state(
redis, self._channel_id, self._feature_key, new_enabled,
guild_id=_guild_id,
)
logger.info(
"Toggle menu: %s %s by %s (channel %s)",
self._feature_key,
"enabled" if new_enabled else "disabled",
interaction.user.id,
self._channel_id,
)
# Rebuild the menu with updated states # 💀🔥
states = await _fetch_feature_states(
redis, self._channel_id, guild_id=_guild_id,
)
new_view = StarToggleMenuView(self._channel_id, states)
state_word = "enabled" if new_enabled else "disabled"
await interaction.edit_original_response(
content=(
f"\u2728 **Star Toggles** \u2014 `{self._channel_id}`\n"
f"`{self._label}` is now **{state_word}**."
),
view=new_view,
)
except Exception:
logger.exception(
"Toggle button callback failed for feature %s by %s in channel %s",
self._feature_key,
interaction.user.id,
self._channel_id,
)
try:
await interaction.followup.send(
"\u274c Toggle failed.", ephemeral=True,
)
except Exception:
pass
# ── Persistent ✧ Button View ──────────────────────────────────── # 🕷️💀
[docs]
class StarToggleView(discord.ui.View):
"""Persistent view containing only the ✧ button. # 💀🔥
timeout=None makes this survive bot restarts when registered
with client.add_view() in on_ready.
"""
def __init__(self, channel_id: str) -> None:
super().__init__(timeout=None)
self.add_item(StarToggleButton(channel_id))
# ── Factory for the outbound consumer ──────────────────────────── # 🔥💀
[docs]
def build_star_toggle_view(channel_id: str) -> StarToggleView | None:
"""Build a StarToggleView for attaching to outbound messages.
Returns None if discord is not importable (non-Discord platforms).
"""
try:
return StarToggleView(channel_id)
except Exception:
return None