"""Per-channel feature toggles stored in Redis.
Allows authorised users to disable the NCM neurotransmitter subsystem
(cadence post-processing + NCM tools like inject_ncm) or the RAG
subsystem on a per-channel basis via ``!emotions on/off`` and
``!rag on/off``.
NOTE: ``!emotions off`` does NOT affect the Sigma Limbic Recursion Core.
The limbic inhale (context injection) and exhale (emotional feedback loop)
always run. Star always knows her emotions. The toggle only disables
the *surface layer*: cadence text degradation and tool access to NCM.
Redis key scheme
----------------
``stargazer:toggle:{feature}:{channel_key}``
where *feature* is ``"emotions"`` or ``"rag"`` and *channel_key* is
``"{platform}:{channel_id}"``.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from redis.asyncio import Redis
from config import Config
from platforms.base import IncomingMessage
logger = logging.getLogger(__name__)
# ------------------------------------------------------------------
# Redis helpers
# ------------------------------------------------------------------
_KEY_PREFIX = "stargazer:toggle"
def _key(feature: str, channel_key: str) -> str:
"""Internal helper: key.
Args:
feature (str): The feature value.
channel_key (str): The channel key value.
Returns:
str: Result string.
"""
return f"{_KEY_PREFIX}:{feature}:{channel_key}"
[docs]
async def is_disabled(redis: "Redis", feature: str, channel_key: str) -> bool:
"""Return ``True`` if *feature* is disabled for *channel_key*."""
return await redis.exists(_key(feature, channel_key)) > 0
[docs]
async def set_disabled(
redis: "Redis",
feature: str,
channel_key: str,
disabled: bool,
) -> None:
"""Enable or disable *feature* for *channel_key*."""
key = _key(feature, channel_key)
if disabled:
await redis.set(key, "1")
else:
await redis.delete(key)
# ------------------------------------------------------------------
# Tool-name sets
# ------------------------------------------------------------------
EMOTION_TOOLS: frozenset[str] = frozenset({
"inject_ncm",
"debug_limbic_shard",
"debug_limbic_import",
"ncm_local_reset",
"ncm_heart_reset",
"flavor_engine",
"query_golden_goddess_v2",
})
RAG_TOOLS: frozenset[str] = frozenset({
"rag_search",
"rag_index_file",
"rag_index_directory",
"rag_index_url",
"rag_list_stores",
"rag_list_files",
"rag_remove_file",
"rag_remove_url",
"rag_clear_store",
"rag_delete_store",
"rag_read_store_file",
"rag_list_store_files",
"rag_auto_search_config",
"rag_dump_store",
})
# ------------------------------------------------------------------
# Permission check
# ------------------------------------------------------------------
[docs]
def check_toggle_permission(
msg: "IncomingMessage",
config: "Config",
) -> bool:
"""Return ``True`` if the user is allowed to toggle features.
Allowed when **any** of the following are true:
1. The user is a bot admin (``config.admin_user_ids``).
2. The user has server/channel admin privileges
(Discord ``guild_permissions.administrator``; Matrix power level ≥ 50).
3. The channel is a DM.
"""
# Bot admin — always allowed
if msg.user_id in (config.admin_user_ids or []):
return True
extra = msg.extra or {}
# DM — always allowed
if extra.get("is_dm", False):
return True
# Server / channel admin
if extra.get("is_server_admin", False):
return True
return False