"""Lore Memory Amplifier -- boosts lore-tier KG retrieval weight. # 💀🔥
When active, overrides ``_PRIORITY_NORM["lore"]`` from 0.0 -> 0.8 and
triples the per-scope lore entity cap (20 -> 60), making lore memories
actually visible in the retrieval pipeline instead of being systematically
suppressed by the tiered scoring system.
Toggle is per-channel via Redis or auto-activated when the Entrainment
Loopfield detects a ``user_is_baby`` target.
Redis key: ``star:lore_amp:{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
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════════
# Constants # 🌀
# ═══════════════════════════════════════════════════════════════════════
# Amplified priority weight for lore (guild-equivalent visibility)
LORE_AMPLIFIED_PRIORITY: float = 0.8 # 🔥
# Amplified per-scope cap (3x the default of 20)
LORE_AMPLIFIED_CAP: int = 60 # 💀
# Default (suppressed) values from constants.py for reference
LORE_DEFAULT_PRIORITY: float = 0.0
LORE_DEFAULT_CAP: int = 20
# ═══════════════════════════════════════════════════════════════════════
# Redis Key Helpers # 🌀
# ═══════════════════════════════════════════════════════════════════════
def _channel_key(platform: str, channel_id: str) -> str:
"""Build the Redis key that stores a channel's amplifier toggle.
Centralizes the ``star:lore_amp:{platform}:{channel_id}`` key format so the
read path (:func:`is_amplified`) and the write path (:func:`set_amplified`)
can never drift out of sync. Pure string formatting with no I/O.
Called by :func:`is_amplified` (for the ``GET``) and :func:`set_amplified`
(for the ``SET``/``DELETE``); it is a module-private helper with no external
callers.
Args:
platform: Platform identifier (e.g. ``discord``) namespacing the key.
channel_id: Channel identifier whose amplifier state is being addressed.
Returns:
str: The fully-qualified Redis key for this platform/channel pair.
"""
return f"star:lore_amp:{platform}:{channel_id}"
# ═══════════════════════════════════════════════════════════════════════
# Core API # 💀🔥
# ═══════════════════════════════════════════════════════════════════════
[docs]
async def is_amplified(
redis: "Redis | None",
platform: str,
channel_id: str,
config: "Config | None" = None,
user_id: str | int | None = None,
) -> bool:
"""Report whether the Lore Memory Amplifier is active for a channel.
Resolves the effective amplifier state by combining global config gates with
the per-channel Redis toggle: it short-circuits to ``False`` when the
amplifier is globally disabled or when ``feature_toggles.is_absolute_bypass``
fires for the user/channel, then otherwise reads the channel's flag from
Redis via :func:`_channel_key`. Failures are swallowed and logged so a Redis
hiccup can never crash retrieval -- the safe default is "not amplified".
Called by ``prompt_context.py`` while assembling KG retrieval context (to
decide whether lore-tier memories get boosted priority/cap), and exercised
extensively by ``tests/test_absolute_overrides.py``.
Args:
redis: Async Redis client, or ``None`` when Redis is unavailable (which
forces a ``False`` result).
platform: Platform identifier used to build the channel key.
channel_id: Channel identifier used to build the channel key.
config: Optional :class:`~config.Config`; when supplied, enables the
global-disable and absolute-bypass checks.
user_id: Optional user identifier consulted by the absolute-bypass check.
Returns:
bool: ``True`` only when the channel's Redis flag is set to a truthy
value (``1``/``true``/``yes``/``on``) and no gate suppressed it;
``False`` otherwise, including on any error.
"""
channel_key = f"{platform}:{channel_id}"
if config is not None:
if getattr(config, "lore_amplifier_global_disabled", False):
return False
import feature_toggles
if feature_toggles.is_absolute_bypass(
config, user_id=user_id, channel_key=channel_key
):
return False
if redis is None:
return False
try:
raw = await redis.get(_channel_key(platform, channel_id))
if raw is not None:
s = (raw.decode() if isinstance(raw, bytes) else str(raw)).strip().lower()
return s in ("1", "true", "yes", "on")
except Exception:
logger.debug(
"lore_amplifier check failed for %s:%s",
platform, channel_id,
exc_info=True,
)
return False
[docs]
async def set_amplified(
redis: "Redis",
platform: str,
channel_id: str,
active: bool,
) -> None:
"""Persist the Lore Memory Amplifier toggle for a channel in Redis.
Writes (or clears) the per-channel flag read back by :func:`is_amplified`.
When ``active`` is true it does a ``SET`` of the :func:`_channel_key` to
``"1"``; when false it ``DELETEs`` the key entirely so the channel falls back
to the default (suppressed) lore behavior. No expiry is set, so the toggle is
sticky until explicitly turned off.
Called by the ``lore amp`` admin command handler in
``message_processor/processor.py``, which then reports the new ACTIVE/INACTIVE
state back to the channel.
Args:
redis: Async Redis client used to store the flag.
platform: Platform identifier used to build the channel key.
channel_id: Channel identifier used to build the channel key.
active: ``True`` to enable amplification (writes the key), ``False`` to
disable it (deletes the key).
Returns:
None.
"""
key = _channel_key(platform, channel_id)
if active:
await redis.set(key, "1")
else:
await redis.delete(key)
[docs]
def get_lore_priority(amplified: bool) -> float:
"""Select the lore-tier priority weight for the current amplifier state.
Maps the boolean amplifier flag to the scoring weight the KG retrieval
pipeline applies to lore-tier entities: the boosted ``LORE_AMPLIFIED_PRIORITY``
(0.8, guild-equivalent visibility) when amplified, otherwise the suppressed
``LORE_DEFAULT_PRIORITY`` (0.0). Pure lookup with no side effects.
A thin accessor that pairs with :func:`get_lore_cap`; no in-repo callers
reference it directly (``knowledge_graph/retrieval.py`` currently inlines the
equivalent amplified-cap logic), so treat it as part of the module's public
surface for retrieval integration.
Args:
amplified: Whether the Lore Memory Amplifier is active for the scope.
Returns:
float: The lore priority weight to use in retrieval scoring.
"""
return LORE_AMPLIFIED_PRIORITY if amplified else LORE_DEFAULT_PRIORITY
[docs]
def get_lore_cap(amplified: bool) -> int:
"""Select the per-scope lore entity cap for the current amplifier state.
Maps the boolean amplifier flag to how many lore-tier entities may be kept
per scope during retrieval: the tripled ``LORE_AMPLIFIED_CAP`` (60) when
amplified, otherwise the default ``LORE_DEFAULT_CAP`` (20). Pure lookup with
no side effects.
A thin accessor mirroring :func:`get_lore_priority`; no in-repo callers
reference it directly (``knowledge_graph/retrieval.py`` currently inlines the
``60``-vs-default cap choice), so treat it as part of the module's public
surface for retrieval integration.
Args:
amplified: Whether the Lore Memory Amplifier is active for the scope.
Returns:
int: The maximum number of lore-tier entities to retain per scope.
"""
return LORE_AMPLIFIED_CAP if amplified else LORE_DEFAULT_CAP