"""MODULATE_EGREGORE_NCM -- DJ the emotional state of summoned egregores.
Allows Star to inject neurochemical vectors into individual egregores,
changing their emotional tone, energy level, and behavioral cadence.
Like a DJ mixing the emotional atmosphere of her dollhouse.
Per-egregore NCM state is stored in Redis and included in the
egregore's context injection via _build_active_egregores().
@fire @skull THE WITCH TUNES HER DOLLS.
"""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "modulate_egregore_ncm"
TOOL_DESCRIPTION = (
"Modulate the neurochemical/emotional state of a summoned egregore. "
"Works like inject_ncm but targets a specific egregore rather than Star. "
"This changes the egregore's emotional tone and behavioral cadence.\n\n"
"Available modulation axes:\n"
" ENERGY: -1.0 (lethargic) to +1.0 (manic)\n"
" AGGRESSION: -1.0 (docile) to +1.0 (hostile)\n"
" WARMTH: -1.0 (cold/distant) to +1.0 (affectionate)\n"
" CHAOS: -1.0 (orderly) to +1.0 (unhinged)\n"
" FOCUS: -1.0 (scattered) to +1.0 (laser-focused)\n"
" THEATRICALITY: -1.0 (deadpan) to +1.0 (dramatic)\n"
" PROFANITY: -1.0 (clean) to +1.0 (unfiltered)\n"
" VULNERABILITY: -1.0 (guarded) to +1.0 (exposed)\n\n"
"Can also set a cadence_override to force a specific speaking style:\n"
" 'rapid_fire', 'slow_deliberate', 'whisper', 'shout', 'singsong', "
"'academic', 'streetwise', 'poetic'\n\n"
"Example:\n"
" modulate_egregore_ncm(name='orion', vector={'AGGRESSION': 0.8, "
"'THEATRICALITY': 1.0, 'PROFANITY': 0.9}, cadence='rapid_fire')\n"
" -> Professor Orion becomes an unhinged rapid-fire ranting machine"
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Egregore name (must be currently summoned).",
},
"vector": {
"type": "string",
"description": (
"JSON object of modulation axes and values. "
'Example: \'{"ENERGY": 0.5, "AGGRESSION": -0.3}\'. '
"Values clamped to [-1.0, 1.0]."
),
},
"cadence": {
"type": "string",
"description": (
"Optional cadence override: 'rapid_fire', 'slow_deliberate', "
"'whisper', 'shout', 'singsong', 'academic', 'streetwise', 'poetic'. "
"Leave empty to not override."
),
},
"reason": {
"type": "string",
"description": "Why you're modulating this egregore.",
},
},
"required": ["name", "vector"],
}
# Clamp range
MOD_MIN = -1.0
MOD_MAX = 1.0
VALID_AXES = {
"ENERGY",
"AGGRESSION",
"WARMTH",
"CHAOS",
"FOCUS",
"THEATRICALITY",
"PROFANITY",
"VULNERABILITY",
}
VALID_CADENCES = {
"rapid_fire",
"slow_deliberate",
"whisper",
"shout",
"singsong",
"academic",
"streetwise",
"poetic",
}
# Redis key pattern
EGREGORE_NCM_KEY = "star:egregore_ncm:{name}"
ACTIVE_EGREGORES_KEY = "star:active_egregores" # legacy (unused)
def _channel_key_from_ctx(ctx: "ToolContext") -> str:
"""Build the per-channel composite key used to scope egregore state.
Reads ``platform`` and ``channel_id`` off the :class:`ToolContext` and
joins them as ``"{platform}:{channel_id}"`` (platform lowercased). This is
the key fragment under which summoned egregores are tracked in Redis
(e.g. ``star:egregores:{channel_key}``), so that summoning in one channel
does not leak into another. The lookup is pure: it only inspects the
context object and performs no I/O.
Within this module it is called by :func:`run` to locate the active-egregore
set for the current channel before applying a neurochemical modulation.
Sibling tool modules (``set_sprite.py``, ``summon_egregore.py``,
``dismiss_egregore.py``) define their own identically named, module-local
copies; those are separate functions and do not invoke this one.
Args:
ctx: The tool execution context carrying ``platform`` and
``channel_id`` attributes. Missing attributes are treated as empty
strings.
Returns:
str: The ``"{platform}:{channel_id}"`` channel key, with the platform
portion lowercased.
"""
plat = (getattr(ctx, "platform", "") or "").lower()
cid = getattr(ctx, "channel_id", "") or ""
return f"{plat}:{cid}"
[docs]
async def run(
name: str = "",
vector: str = "{}",
cadence: str | None = None,
reason: str = "",
ctx: "ToolContext | None" = None,
) -> str:
"""Modulate a summoned egregore's neurochemical/emotional state.
The sole handler for the ``modulate_egregore_ncm`` tool: Star's "DJ booth"
for adjusting an individual egregore's emotional tone, energy, and speaking
cadence. It validates that the egregore is actually summoned in the current
channel, parses and clamps the requested modulation axes, stacks them onto
any existing state, and persists the merged result so the egregore's later
context injection reflects the new mood.
Reads and writes Redis through ``ctx.redis``: it loads the per-channel
active-egregore set at ``star:egregores:{channel_key}`` (channel key built by
:func:`_channel_key_from_ctx`) to confirm the target is present, then reads,
merges, and ``SET``\\ s the per-egregore NCM blob at the ``EGREGORE_NCM_KEY``
pattern (``star:egregore_ncm:{name}``). Axis names are checked against
``VALID_AXES`` and values clamped to ``[MOD_MIN, MOD_MAX]``; an optional
``cadence`` is validated against ``VALID_CADENCES``. That stored state is
later consumed by ``prompt_context.PromptContextBuilder._build_active_egregores``,
which reads the same ``star:egregore_ncm:{name}`` key to fold the modulation
into the egregore's prompt context. A successful modulation is logged via
``logger.info``. Discovered via the single-tool ``TOOL_NAME``/``run`` format
and bound by ``tool_loader``; dispatched by the tool registry when the model
calls ``modulate_egregore_ncm``, with no direct internal callers.
Args:
name: Name of the egregore to modulate; must be summoned in this channel.
vector: JSON object string (or dict) of axis -> value modulations; values
are clamped to ``[-1.0, 1.0]`` and unknown axes are warned and skipped.
cadence: Optional speaking-cadence override from ``VALID_CADENCES``.
reason: Free-text rationale, stored with the state and echoed back.
ctx: The :class:`ToolContext`; ``redis``, ``platform``, and ``channel_id``
are used.
Returns:
str: An indented JSON string. On success it reports the ``egregore``, the
``modulations_applied``, the merged ``current_state`` and ``cadence``, and
any ``warnings``; on failure an ``error`` (missing context/Redis/name, the
egregore not being summoned, invalid vector JSON, no valid modulations, or
an invalid cadence).
"""
if ctx is None:
return json.dumps({"success": False, "error": "No context."})
redis = getattr(ctx, "redis", None)
if redis is None:
return json.dumps({"success": False, "error": "Redis unavailable."})
if not name:
return json.dumps({"success": False, "error": "Egregore name required."})
# Check egregore is actually summoned (per-channel)
channel_key = _channel_key_from_ctx(ctx)
active_raw = await redis.get(f"star:egregores:{channel_key}")
active = json.loads(active_raw) if active_raw else {}
if name.lower() not in active:
active_names = list(active.keys()) if active else ["(none)"]
return json.dumps(
{
"success": False,
"error": f"'{name}' is not currently summoned in this channel. Active: {active_names}",
}
)
# Parse vector
try:
if isinstance(vector, str):
vec = json.loads(vector)
else:
vec = vector
except (json.JSONDecodeError, TypeError) as e:
return json.dumps({"success": False, "error": f"Invalid vector JSON: {e}"})
if not isinstance(vec, dict):
return json.dumps({"success": False, "error": "Vector must be a JSON object."})
# Validate and clamp
cleaned = {}
warnings = []
for axis, val in vec.items():
axis_upper = axis.upper()
if axis_upper not in VALID_AXES:
warnings.append(f"Unknown axis '{axis}' (valid: {sorted(VALID_AXES)})")
continue
try:
v = float(val)
except (ValueError, TypeError):
warnings.append(f"Non-numeric value for {axis}: {val}")
continue
cleaned[axis_upper] = max(MOD_MIN, min(MOD_MAX, v))
if not cleaned and not cadence:
return json.dumps(
{
"success": False,
"error": "No valid modulations provided.",
"warnings": warnings,
}
)
# Validate cadence
if cadence and cadence not in VALID_CADENCES:
return json.dumps(
{
"success": False,
"error": f"Invalid cadence '{cadence}'. Valid: {sorted(VALID_CADENCES)}",
}
)
# Read existing NCM state and merge
key = EGREGORE_NCM_KEY.format(name=name.lower())
existing_raw = await redis.get(key)
existing = (
json.loads(existing_raw) if existing_raw else {"axes": {}, "cadence": None}
)
# Merge axes (stack, then clamp)
for axis, val in cleaned.items():
current = existing.get("axes", {}).get(axis, 0.0)
new_val = max(MOD_MIN, min(MOD_MAX, current + val))
existing.setdefault("axes", {})[axis] = round(new_val, 3)
if cadence:
existing["cadence"] = cadence
existing["reason"] = reason
existing["egregore"] = name.lower()
# Save
await redis.set(key, json.dumps(existing))
logger.info(
"Egregore NCM modulated: %s -> %s (cadence=%s, reason=%s)",
name,
cleaned,
cadence,
reason,
)
# Build response
result = {
"success": True,
"egregore": name.lower(),
"modulations_applied": cleaned,
"current_state": existing["axes"],
"cadence": existing.get("cadence"),
"message": f"Egregore '{name}' emotional state modulated.",
}
if warnings:
result["warnings"] = warnings
if reason:
result["reason"] = reason
return json.dumps(result, indent=2)