"""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``.
**Full NCM disable** (inhale, exhale, Golden Goddess reflex, emotion tools,
cadence) is controlled by:
- :attr:`~config.Config.ncm_fully_disabled_channels` (``platform:channel_id``
strings, default includes Aesir-Hall), and/or
- Redis key ``stargazer:toggle:ncm:{channel_key}`` (same pattern as other
toggles; set via ``SET`` / ``DEL`` or a future admin command).
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.
Use the ``ncm`` toggle or config list when the entire NCM stack must be off.
Redis key scheme
----------------
``stargazer:toggle:{feature}:{channel_key}``
where *feature* is ``"emotions"``, ``"rag"``, or ``"ncm"`` and *channel_key*
is ``"{platform}:{channel_id}"``.
For **Discord**, ``discord`` and ``discord-self`` share the same numeric channel
ID; :func:`is_disabled_resolving_discord_aliases` and
:func:`is_ncm_fully_disabled` treat toggles under either prefix as equivalent.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from redis.asyncio import Redis
from config import Config
from platforms.base import IncomingMessage
logger = logging.getLogger(__name__)
# Keys merged from limbic injection that form the NCM "surface layer" when
# ``!emotions off`` — must match :meth:`PromptContextBuilder._build_limbic_state`.
NCM_SURFACE_LAYER_KEYS: frozenset[str] = frozenset(
{
"cadence_directive",
"cadence_instruction",
"cadence_refinement_profile",
"limbic_state",
"limbic_cues",
"cascade_cues",
}
)
# Features that are disabled/off by default (i.e. return is_disabled=True
# when no Redis toggle key is present). Enabling them creates the Redis key.
DEFAULT_OFF_FEATURES: frozenset[str] = frozenset({"csdr_scene", "csdr_header"})
# ------------------------------------------------------------------
# Redis helpers
# ------------------------------------------------------------------
_KEY_PREFIX = "stargazer:toggle"
[docs]
def discord_family_channel_key_variants(channel_key: str) -> tuple[str, ...]:
"""Return unique ``platform:channel_id`` strings to check for Discord aliases.
The bot adapter uses ``discord`` and the selfbot uses ``discord-self`` for
the same numeric channel ID. Toggles may be set under either prefix; callers
should treat them as equivalent.
"""
idx = channel_key.find(":")
if idx <= 0:
return (channel_key,)
platform = channel_key[:idx]
suffix = channel_key[idx + 1 :]
if platform not in ("discord", "discord-self"):
return (channel_key,)
# Preserve order: actual key first, then the two canonical forms.
seen: dict[str, None] = {}
for ck in (channel_key, f"discord:{suffix}", f"discord-self:{suffix}"):
seen.setdefault(ck, None)
return tuple(seen.keys())
[docs]
def strip_ncm_surface_layer_from_context(ctx: dict[str, Any]) -> None:
"""Delete the NCM surface-layer keys from a prompt context dict in place.
Implements the ``!emotions off`` degradation: it pops each of the
:data:`NCM_SURFACE_LAYER_KEYS` (cadence directive/instruction/refinement
profile and the limbic/cascade cue blobs) out of *ctx* so the rendered prompt
loses cadence text and limbic surface cues while the underlying limbic
respiration keeps running. Mutates *ctx* directly and returns nothing; only
keys that are present are removed.
Called when limbic injection has been merged but the channel has emotions
disabled: by ``prompt_context.py`` (``PromptContextBuilder``) on the
per-turn injection dict, by ``message_processor/channel_heartbeat.py`` on the
heartbeat room context, and by ``tests/test_feature_toggles.py``.
Args:
ctx: The prompt context dictionary to strip in place.
Returns:
None.
"""
for k in NCM_SURFACE_LAYER_KEYS:
ctx.pop(k, None)
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}"
# Mapping of feature -> global default-off Redis key stem. # 💀🔥
# Guild-scoped keys are "{stem}:{guild_id}", global fallback is just "{stem}".
_GLOBAL_DEFAULT_OFF_STEMS: dict[str, str] = {
"emotions": "stargazer:toggle_emotions_default_off",
"toggle_menu": "stargazer:toggle_menu_default_off",
}
async def _is_feature_default_off(
redis: "Redis", feature: str, guild_id: str | None = None,
) -> bool:
"""Return ``True`` if *feature* defaults to OFF. # 💀
Check order:
1. Hardcoded ``DEFAULT_OFF_FEATURES`` (always wins).
2. Guild-scoped Redis key ``{stem}:{guild_id}`` (if guild_id given).
3. Global Redis key ``{stem}``.
"""
if feature in DEFAULT_OFF_FEATURES:
return True
stem = _GLOBAL_DEFAULT_OFF_STEMS.get(feature)
if stem is None:
return False
# Guild-scoped first # 🕷️
if guild_id:
gval = await redis.get(f"{stem}:{guild_id}")
if gval is not None:
return gval in (b"1", "1")
# Global fallback
gval = await redis.get(stem)
return gval is not None and gval in (b"1", "1")
[docs]
async def is_disabled(
redis: "Redis", feature: str, channel_key: str,
guild_id: str | None = None,
) -> bool:
"""Return ``True`` if *feature* is disabled for the exact *channel_key*.
For default-off features the logic is reversed: disabled when no key exists,
enabled only when the key is explicitly created.
Args:
redis: Async Redis client.
feature: Feature name.
channel_key: ``"{platform}:{channel_id}"``.
guild_id: Optional guild ID for guild-scoped default lookups.
"""
default_off = await _is_feature_default_off(redis, feature, guild_id=guild_id)
if default_off:
return await redis.exists(_key(feature, channel_key)) == 0
return await redis.exists(_key(feature, channel_key)) > 0
[docs]
async def is_disabled_resolving_discord_aliases(
redis: "Redis",
feature: str,
channel_key: str,
guild_id: str | None = None,
) -> bool:
"""Check :func:`is_disabled` across all Discord-alias variants.
Args:
redis: Async Redis client.
feature: Feature name.
channel_key: ``"{platform}:{channel_id}"``.
guild_id: Optional guild ID for guild-scoped default lookups.
"""
default_off = await _is_feature_default_off(redis, feature, guild_id=guild_id)
if default_off:
for ck in discord_family_channel_key_variants(channel_key):
if not await is_disabled(redis, feature, ck, guild_id=guild_id):
return False
return True
else:
for ck in discord_family_channel_key_variants(channel_key):
if await is_disabled(redis, feature, ck, guild_id=guild_id):
return True
return False
[docs]
async def set_disabled(
redis: "Redis",
feature: str,
channel_key: str,
disabled: bool,
guild_id: str | None = None,
) -> None:
"""Persist the on/off state of *feature* for *channel_key* in Redis.
Args:
redis: Async Redis client.
feature: Feature name.
channel_key: ``"{platform}:{channel_id}"``.
disabled: ``True`` to disable, ``False`` to enable.
guild_id: Optional guild ID for guild-scoped default lookups.
"""
key = _key(feature, channel_key)
default_off = await _is_feature_default_off(redis, feature, guild_id=guild_id)
if default_off:
if disabled:
await redis.delete(key)
else:
await redis.set(key, "1")
else:
if disabled:
await redis.set(key, "1")
else:
await redis.delete(key)
logger.info(
"Feature toggle updated: feature=%s, channel_key=%s, disabled=%s",
feature,
channel_key,
disabled,
)
def _ncm_config_matches_channel(channel_key: str, chans: frozenset[str]) -> bool:
"""Return ``True`` if *channel_key* (or a Discord alias) is in the config set.
Tests membership of a channel against the static
``config.Config.ncm_fully_disabled_channels`` set in an alias-aware way:
*channel_key* is expanded via
:func:`discord_family_channel_key_variants` so a channel listed under either
``discord`` or ``discord-self`` matches regardless of which prefix arrives.
Pure set lookup with no I/O.
Called only by :func:`is_ncm_fully_disabled` when consulting the config-level
disable list; it has no other callers in the repo.
Args:
channel_key: The ``"{platform}:{channel_id}"`` key under evaluation.
chans: The configured set of fully-NCM-disabled channel keys.
Returns:
bool: ``True`` if any alias of *channel_key* is present in *chans*.
"""
for ck in discord_family_channel_key_variants(channel_key):
if ck in chans:
return True
return False
[docs]
def is_absolute_bypass(
config: Any | None,
user_id: str | int | None = None,
channel_key: str | None = None,
) -> bool:
"""Return ``True`` if a user or channel is on a config-level absolute override list.
The global escape hatch that forces NCM/limbic/egregore subsystems on (bypassing
any per-channel disable) for explicitly allow-listed users or channels. It reads
``overall_user_id_absolute_override_list`` and
``overall_channel_id_absolute_override_list`` off *config* via ``getattr`` (so a
missing attribute degrades to empty). A *user_id* matches by string equality
against the user list; a *channel_key* matches either by Discord-alias-expanded
key (via :func:`discord_family_channel_key_variants`) or by comparing the numeric
suffix after the first ``:`` against each list entry's suffix, so a bare numeric
id, a ``discord:`` key, and a ``discord-self:`` key all resolve alike. Pure
config inspection with no I/O.
Called widely as a precedence gate before applying per-channel toggles: by
:func:`is_ncm_fully_disabled` here, and across ``prompt_context.py``,
``flash_dyadic_mirror.py``, ``ego_ablation.py``, ``lore_amplifier.py``,
``entrainment_loopfield.py``, and several ``message_processor`` modules
(``generate_and_send.py``, ``proactive_gates.py``, ``context_injections.py``),
plus the override test suites.
Args:
config: A ``Config``-like object exposing the absolute override lists, or
``None`` (in which case this returns ``False``).
user_id: Optional user id to test against the user override list; stringified
before comparison.
channel_key: Optional ``"{platform}:{channel_id}"`` (or bare id) to test
against the channel override list, alias- and suffix-aware.
Returns:
bool: ``True`` if the user or channel is absolutely overridden, else ``False``.
"""
if config is None:
return False
# Check User ID bypass list
if user_id is not None:
uid_str = str(user_id)
override_users = getattr(config, "overall_user_id_absolute_override_list", [])
if uid_str in override_users:
return True
# Check Channel ID bypass list
if channel_key is not None:
override_channels = getattr(config, "overall_channel_id_absolute_override_list", [])
if override_channels:
# 1. Match direct channel key or discord variants
for ck in discord_family_channel_key_variants(channel_key):
if ck in override_channels:
return True
# 2. Match suffix/numeric comparisons
idx = channel_key.find(":")
incoming_suffix = channel_key[idx + 1 :] if idx > 0 else channel_key
for oc in override_channels:
oc_idx = oc.find(":")
oc_suffix = oc[oc_idx + 1 :] if oc_idx > 0 else oc
if incoming_suffix == oc_suffix:
return True
return False
[docs]
async def is_limbic_respiration_disabled(
redis: "Redis | None",
channel_key: str,
cfg: Any | None,
user_id: str | int | None = None,
) -> bool:
"""True when NCM limbic respiration (inhale/exhale) is disabled for this channel.
NOTE: Channel '1424991476668170351' is explicitly allowed (returns False)
and bypasses the disabling effects of overall absolute overrides.
"""
if channel_key is not None:
idx = channel_key.find(":")
suffix = channel_key[idx + 1 :] if idx > 0 else channel_key
if suffix == "1424991476668170351":
return False
return await is_ncm_fully_disabled(redis, channel_key, cfg, user_id=user_id)
[docs]
async def is_ncm_fully_disabled(
redis: "Redis | None",
channel_key: str,
cfg: Any | None,
user_id: str | int | None = None,
) -> bool:
"""Return ``True`` when the entire NCM stack must be off for this channel.
The authoritative "is NCM fully disabled" decision, combining static config and
live Redis state. It returns ``True`` if *cfg* sets ``ncm_global_disabled``, OR
if :func:`is_absolute_bypass` flags this user/channel (absolute overrides force
the disabled-style respiration path), OR if the channel is listed in
``ncm_fully_disabled_channels`` per :func:`_ncm_config_matches_channel`. Failing
those, when *redis* is provided it consults the per-channel ``"ncm"`` toggle via
:func:`is_disabled_resolving_discord_aliases`, swallowing and debug-logging any
Redis error so a backend hiccup never spuriously disables NCM. The only I/O is
that optional Redis read; config checks are in-memory.
Called by :func:`is_limbic_respiration_disabled` (after its single-channel
allow-list carve-out), and directly by the inference/heartbeat paths in
``message_processor/generate_and_send.py`` and
``message_processor/channel_heartbeat.py``, plus the toggle/override test suites.
Args:
redis: Optional async Redis client for the per-channel toggle check; when
``None`` only config is consulted.
channel_key: The ``"{platform}:{channel_id}"`` key under evaluation.
cfg: Optional ``Config``-like object supplying the global flag, override
lists, and disabled-channel set.
user_id: Optional user id forwarded to :func:`is_absolute_bypass`.
Returns:
bool: ``True`` if NCM is fully disabled for this channel, else ``False``.
"""
if cfg is not None:
if getattr(cfg, "ncm_global_disabled", False):
return True
if is_absolute_bypass(cfg, user_id=user_id, channel_key=channel_key):
return True
chans = getattr(cfg, "ncm_fully_disabled_channels", frozenset())
if chans and _ncm_config_matches_channel(channel_key, chans):
return True
if redis is not None:
try:
if await is_disabled_resolving_discord_aliases(redis, "ncm", channel_key):
return True
except Exception:
logger.debug("is_ncm_fully_disabled Redis check failed", exc_info=True)
return False
# ------------------------------------------------------------------
# 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]
async def check_toggle_permission(
msg: "IncomingMessage",
config: "Config",
redis: "Redis | None" = None,
) -> 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 the ``CTX_MANAGE`` or ``GUILD_ADMIN`` privilege bit.
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 admin/moderator fallback (particularly when Redis is None)
if extra.get("is_server_admin", False):
return True
# Check MAC system
if redis is not None:
from tools.alter_privileges import has_scoped_privilege, PRIVILEGES
guild_id = str(extra.get("guild_id", "") or "")
channel_id = str(msg.channel_id)
# CTX_MANAGE (Bit 15) is the primary gate for channel toggles.
if await has_scoped_privilege(
redis,
msg.user_id,
PRIVILEGES["CTX_MANAGE"],
config,
guild_id=guild_id,
channel_id=channel_id,
):
return True
return False