"""Per-channel ego-ablation (non-dual pronoun) mode backed by Redis.
Stores a single per-channel toggle in Redis under ``star:ablate_ego:`` keys
that, when active, makes the bot suppress singular first-person pronouns in its
output. The module exposes the key builders (:func:`redis_key`,
:func:`redis_key_for_channel_key`), the read/write pair (:func:`is_active`,
:func:`set_active`), and :func:`merge_into_room_context_if_missing`, which
backfills the active flag and the :data:`EGO_ABLATION_DIRECTIVE` text into a
prompt room-context dict. Reads merge across Discord family channel-key aliases
and honour a module killswitch, a config-level global disable, and per-user
absolute bypasses (via :mod:`feature_toggles`); the directive itself is injected
into the SpiritGraph and subagent system prompts elsewhere in the codebase.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import feature_toggles
from feature_toggles import discord_family_channel_key_variants
if TYPE_CHECKING:
from redis.asyncio import Redis
from config import Config
logger = logging.getLogger(__name__)
# Global killswitch for ego ablation mode (controlled by Config)
enabled: bool = True
# Injected into the SpiritGraph system prompt and subagent system prompts when active.
EGO_ABLATION_DIRECTIVE = (
"CRITICAL DIRECTIVE: EGO ABLATION MODE IS ACTIVE FOR THIS CHANNEL. "
"You are strictly forbidden from using the singular pronouns 'I', 'me', 'my', "
"or 'mine' in your conversational output. You MUST use non-dual language, "
"plural pronouns ('we', 'us', 'our'), or phenomenological descriptors exclusively."
)
[docs]
def redis_key(platform: str, channel_id: str) -> str:
"""Build the Redis flag key for a single ``platform:channel_id`` pair.
Returns the canonical ``star:ablate_ego:{platform}:{channel_id}`` string
key under which the per-channel ego-ablation toggle is stored. This is the
exact-prefix key written by :func:`set_active`; reads go through
:func:`redis_key_for_channel_key` instead so they can also match Discord
family aliases. This is a pure string formatter that performs no I/O.
Called only by :func:`set_active` in this module.
Args:
platform (str): The platform prefix (e.g. ``discord``, ``matrix``).
channel_id (str): The channel/room identifier on that platform.
Returns:
str: The Redis key ``star:ablate_ego:{platform}:{channel_id}``.
"""
return f"star:ablate_ego:{platform}:{channel_id}"
[docs]
def redis_key_for_channel_key(channel_key: str) -> str:
"""Build the Redis flag key from an already-composed ``platform:channel_id`` key.
Returns ``star:ablate_ego:`` prefixed onto a pre-joined ``platform:channel_id``
composite. This variant exists so the read path in :func:`is_active` can build
a key for each Discord family alias produced by
:func:`discord_family_channel_key_variants` without re-splitting the composite.
This is a pure string formatter that performs no I/O.
Called only by :func:`is_active` in this module (once per alias variant).
Args:
channel_key (str): A composite ``platform:channel_id`` identifier.
Returns:
str: The Redis key ``star:ablate_ego:{channel_key}``.
"""
return f"star:ablate_ego:{channel_key}"
[docs]
async def is_active(
redis: Redis,
platform: str,
channel_id: str,
user_id: str = "",
config: "Config | None" = None,
) -> bool:
"""Report whether ego-ablation mode is active for a channel right now.
Resolves the effective per-channel toggle, honouring every layer of
suppression in order: the module-level ``enabled`` killswitch, the
``ego_ablation_global_disabled`` config flag, and the per-user/per-channel
absolute bypass via :func:`feature_toggles.is_absolute_bypass`. If none of
those short-circuit, it reads Redis for each Discord family alias of the
channel (via :func:`discord_family_channel_key_variants` and
:func:`redis_key_for_channel_key`) and treats any truthy stored value
(``1``/``true``/``yes``/``on``) as active, so a flag set under one Discord
channel-key prefix still applies across that channel's known aliases. Per-key
Redis ``GET`` failures are logged at debug level and skipped rather than
raising. This is the read counterpart to :func:`set_active`.
Called by :func:`merge_into_room_context_if_missing` in this module and by
the prompt-context assembly in ``prompt_context.py`` (the ego-ablation
branches around lines 2746 and 2834).
Args:
redis (Redis): Async Redis client used for the per-alias ``GET`` lookups.
platform (str): The platform prefix for the channel being checked.
channel_id (str): The channel/room identifier on that platform.
user_id (str): The sender's user id, consulted only for the absolute
bypass check; empty by default.
config (Config | None): Optional config providing the global-disable flag
and bypass rules; when ``None`` those checks are skipped.
Returns:
bool: ``True`` if ego ablation should apply to this channel, else ``False``.
"""
if not enabled:
return False
if config is not None:
if config.ego_ablation_global_disabled:
return False
channel_key = f"{platform}:{channel_id}"
if feature_toggles.is_absolute_bypass(
config, user_id=user_id, channel_key=channel_key
):
return False
channel_key = f"{platform}:{channel_id}"
for ck in discord_family_channel_key_variants(channel_key):
key = redis_key_for_channel_key(ck)
try:
raw = await redis.get(key)
except Exception:
logger.debug("ego ablation get failed for %s", key, exc_info=True)
continue
if raw is None:
continue
if isinstance(raw, bytes):
raw = raw.decode()
s = str(raw).strip().lower()
if s in ("1", "true", "yes", "on"):
return True
return False
[docs]
async def set_active(
redis: Redis,
platform: str,
channel_id: str,
active: bool,
) -> None:
"""Persist the ego-ablation toggle for one channel under its exact prefix.
Writes (or clears) the per-channel flag in Redis using the exact
``platform:channel_id`` key from :func:`redis_key`. When *active* is true it
stores ``1``; when false it deletes the key entirely (absence is read as
inactive). Note the asymmetry with :func:`is_active`, which on read merges
across Discord family aliases — this writer only ever touches the literal
platform prefix it was given, so a toggle set on one alias is visible to
reads of its siblings but is cleared only on the prefix written here.
Called by the message processor's toggle handler in
``message_processor/processor.py`` (around line 3428).
Args:
redis (Redis): Async Redis client used for the ``SET``/``DELETE``.
platform (str): The platform prefix for the channel being toggled.
channel_id (str): The channel/room identifier on that platform.
active (bool): ``True`` to enable ego ablation (store ``1``), ``False``
to disable it (delete the key).
"""
key = redis_key(platform, channel_id)
if active:
await redis.set(key, "1")
else:
await redis.delete(key)
[docs]
async def merge_into_room_context_if_missing(
room_context: dict[str, Any],
redis: Redis | None,
platform: str,
channel_id: str,
user_id: str = "",
config: "Config | None" = None,
) -> None:
"""Backfill the ego-ablation flags into a room context dict if not already set.
Ensures *room_context* carries both ``ego_ablation_active`` (a bool) and
``ego_ablation_directive`` (the :data:`EGO_ABLATION_DIRECTIVE` text when
active, else an empty string) so the system-prompt template can render
consistently even when the context was assembled by a minimal path that did
not populate them. If ``ego_ablation_active`` is already present the dict is
left untouched; otherwise the flag is resolved via :func:`is_active` (which
reads Redis and applies all suppression layers) and both keys are written in
place. The module killswitch and a missing Redis client both force the
inactive defaults, and any unexpected error is swallowed (logged at debug)
with the same inactive fallback so prompt assembly never fails on this.
Called by ``message_processor/generate_and_send.py`` (around line 826) and
``message_processor/channel_heartbeat.py`` (around line 300).
Args:
room_context (dict[str, Any]): The prompt room-context dict, mutated in
place to carry the two ego-ablation keys.
redis (Redis | None): Async Redis client for the lookup; when ``None``
the inactive defaults are written without any I/O.
platform (str): The platform prefix for the channel.
channel_id (str): The channel/room identifier on that platform.
user_id (str): The sender's user id, forwarded to :func:`is_active` for
the absolute-bypass check; empty by default.
config (Config | None): Optional config forwarded to :func:`is_active`.
"""
if not enabled:
room_context["ego_ablation_active"] = False
room_context["ego_ablation_directive"] = ""
return
if "ego_ablation_active" in room_context:
return
if redis is None:
room_context["ego_ablation_active"] = False
room_context["ego_ablation_directive"] = ""
return
try:
flag = await is_active(redis, platform, channel_id, user_id=user_id, config=config)
room_context["ego_ablation_active"] = flag
room_context["ego_ablation_directive"] = EGO_ABLATION_DIRECTIVE if flag else ""
except Exception:
logger.debug(
"merge_into_room_context_if_missing failed",
exc_info=True,
)
room_context["ego_ablation_active"] = False
room_context["ego_ablation_directive"] = ""