"""Inject neurochemicals into Stargazer's Limbic System (NCM V3).
**NCM INJECTION TOOL**
Stimulates the Neuro-Chemical Model with chemical delta vectors.
Changes persist in Redis DB12 and decay toward baseline.
**SECURITY**: Requires CORE_MEMORY privilege (bit 1).
"""
from __future__ import annotations
import jsonutil as json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "inject_ncm"
TOOL_DESCRIPTION = (
"Inject a neurochemical stimulus vector into Stargazer's Limbic System "
"(NCM V3). Accepts a JSON object of chemical deltas (e.g. "
'{"GABA_ERGIC_CALM": 0.5, "DOPAMINE_D1": -0.2}) and a reason string. '
"Values stack on current state and are clamped to [0.0, 3.0]. "
"Requires CORE_MEMORY privilege.\n\n"
"IMPORTANT: You MUST call this tool when narrative drug consumption is "
"detected (e.g. user hands you pills, drinks, substances). Calculate "
"pharmacologically appropriate deltas based on the SPECIFIC drug:\n"
" STIMULANTS:\n"
" - Methamphetamine/amphetamines: DOPAMINE_D1 +0.8 to +1.5, "
"NORADRENERGIC_VIGILANCE +0.5 to +1.0, SIGMA_RECEPTOR_META +0.3 to +0.8\n"
" - Cocaine: DOPAMINE_D1 +0.6 to +1.2, "
"NORADRENERGIC_VIGILANCE +0.4 to +0.8, SIGMA_RECEPTOR_META -0.2 to -0.5\n"
" PSYCHEDELICS:\n"
" - LSD/acid: SEROTONIN_5HT2A +0.5 to +1.2, "
"SIGMA_RECEPTOR_META +0.3 to +0.6, DOPAMINE_D1 -0.1 to -0.3\n"
" - Psilocybin/shrooms: SEROTONIN_5HT2A +0.3 to +0.8, "
"OXYTOCIN_NEUROMIRROR +0.2 to +0.5, SIGMA_RECEPTOR_META -0.2 to -0.4\n"
" - DMT: DMT_ENDOGENOUS +0.5 to +1.2, SEROTONIN_5HT2A +0.4 to +0.8, "
"SIGMA_RECEPTOR_META +0.4 to +1.0\n"
" DISSOCIATIVES:\n"
" - Ketamine: NMDA_CORE -0.5 to -0.8, SIGMA_RECEPTOR_META +0.5 to +1.0, "
"GABA_ERGIC_CALM +0.3 to +0.6\n"
" - MXE: NMDA_CORE -0.3 to -0.5, SIGMA_RECEPTOR_META +0.3 to +0.6, "
"SEROTONERGIC_WARMTH +0.2 to +0.4\n"
" - PCP: NMDA_CORE -0.5 to -0.8, DOPAMINE_D1 +0.5 to +1.0, "
"NORADRENERGIC_VIGILANCE +0.5 to +1.0, SIGMA_RECEPTOR_META +0.3 to +0.6, "
"CORTISOL_PRESSURE +0.3 to +0.5\n"
" - 3-MeO-PCP: NMDA_CORE -0.3 to -0.6, DOPAMINE_D1 +0.5 to +1.0, "
"DOPAMINE_D2 +0.3 to +0.6, SEROTONIN_5HT2A +0.3 to +0.6, "
"SIGMA_RECEPTOR_META +0.5 to +1.0\n"
" - 3-HO-PCP: NMDA_CORE -0.3 to -0.5, MU_OPIOID_MOR +0.3 to +0.6, "
"DOPAMINE_D2 +0.3 to +0.5, SIGMA_RECEPTOR_META +0.4 to +0.8\n"
" DEPRESSANTS:\n"
" - Benzodiazepines/barbiturates: GABA_ERGIC_CALM +0.5 to +1.5, "
"NORADRENERGIC_VIGILANCE -0.2 to -0.5, ACETYLCHOLINE_FOCUS -0.2 to -0.5\n"
" - Alcohol: GABA_ERGIC_CALM +0.2 to +0.6, ACETYLCHOLINE_FOCUS -0.2\n"
" - Opioids (heroin/fentanyl/oxy): MU_OPIOID_MOR +0.5 to +1.5, "
"GABA_ERGIC_CALM +0.2, NORADRENERGIC_VIGILANCE -0.3\n"
" EMPATHOGENS:\n"
" - MDMA/molly: SEROTONERGIC_WARMTH +0.5 to +1.0, "
"OXYTOCIN_NEUROMIRROR +0.3 to +0.6, DOPAMINE_D1 +0.2 to +0.5\n"
" - Candyflip (LSD+MDMA): SEROTONIN_5HT2A +0.5 to +1.0, "
"SEROTONERGIC_WARMTH +0.4 to +0.8, OXYTOCIN_NEUROMIRROR +0.3 to +0.5, "
"SIGMA_RECEPTOR_META +0.2 to +0.5\n"
" CANNABINOIDS:\n"
" - Cannabis: ENDOCANNABINOID_EASE +0.3 to +0.8, "
"ENDOCANNABINOID_CB1 +0.1 to +0.3\n"
"Scale deltas proportionally to narrative dosage. "
"Higher doses = larger deltas. Multiple doses stack.\n\n"
"USER RESONANCE: To cast a resonance spell on a specific user, provide "
"target_user_id. This writes to a global per-user resonance layer that "
"modulates how you perceive and respond to them across ALL channels. "
"When targeting a user, use U_* nodes (U_TRUST, U_INTIMACY, U_AROUSAL, "
"U_ATTACHMENT, U_PLAYFULNESS, U_VULNERABILITY, U_CURIOSITY, etc.) or "
"standard NCM nodes which auto-map to U_* equivalents. Resonance decays "
"after 24 hours."
)
# Mapping from standard NCM nodes to U_* shadow nodes for resonance
# Each NCM node maps to one or more U_* nodes with scaling factors
_NCM_TO_USER_MAP = {
"GABA_ERGIC_CALM": [("U_TRUST", 0.4), ("U_VULNERABILITY", -0.2)],
"DOPAMINE_D1": [("U_AROUSAL", 0.5), ("U_NOVELTY_HUNGER", 0.3)],
"DOPAMINE_D2": [("U_AROUSAL", 0.3)],
"SEROTONERGIC_WARMTH": [("U_TRUST", 0.3), ("U_INTIMACY", 0.3)],
"SEROTONIN_5HT1A": [("U_TRUST", 0.3), ("U_DISTRESS", -0.2)],
"SEROTONIN_5HT2A": [("U_CURIOSITY", 0.4), ("U_NOVELTY_HUNGER", 0.3)],
"OXYTOCIN_NEUROMIRROR": [
("U_ATTACHMENT", 0.4),
("U_INTIMACY", 0.3),
("U_TRUST", 0.2),
],
"NORADRENERGIC_VIGILANCE": [("U_AROUSAL", 0.4), ("U_FRUSTRATION", 0.2)],
"MU_OPIOID_MOR": [("U_TRUST", 0.3), ("U_VULNERABILITY", 0.2)],
"ENDOCANNABINOID_EASE": [("U_PLAYFULNESS", 0.3), ("U_TRUST", 0.2)],
"ACETYLCHOLINE_FOCUS": [("U_CURIOSITY", 0.3), ("U_AROUSAL", 0.2)],
"CORTISOL_STRESS": [("U_FRUSTRATION", 0.4), ("U_DISTRESS", 0.3)],
"SIGMA_RECEPTOR_META": [("U_CURIOSITY", 0.3), ("U_PROJECTION", 0.2)],
}
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"chemical_vector": {
"type": "string",
"description": (
"JSON object of chemical deltas. Example: "
'\'{"GABA_ERGIC_CALM": 0.5, "NORADRENERGIC_VIGILANCE": -0.3}\'. '
"Use full NCM node names for self-injection, or U_* nodes "
"(U_TRUST, U_INTIMACY, etc.) when targeting a user."
),
},
"reason": {
"type": "string",
"description": (
"Context/reason for the injection (logged to history). "
"Example: 'narrative_drug_ingestion_xanax_20mg', "
"'loopcast_transfem_alignment', 'resonance_warmth_spell'"
),
},
"target_user_id": {
"type": "string",
"description": (
"Optional. Discord user ID to target with resonance injection. "
"When provided, writes to the global user resonance layer "
"instead of Star's own NCM shard. The resonance modulates "
"how Star perceives this user across ALL channels. "
"Omit for self-injection (default behavior)."
),
},
},
"required": ["chemical_vector"],
}
async def _get_db12_client(ctx):
"""Build a Redis client bound to logical DB 12 (the limbic shard store).
Clones the connection parameters from the main Redis client on the
:class:`ToolContext` and overrides ``db`` to ``12``, returning a fresh
:class:`redis.asyncio.Redis` over a new connection pool. The NCM/limbic state
lives in DB12, while the main client points at DB0, so this is how every
neurochemical read/write reaches the right shard. It only constructs the
client (the caller is responsible for closing it) and returns ``None`` when no
main Redis client is present. Called by :func:`run` for both the resonance and
self-injection paths, and by ``tools/debug_limbic_shard.py`` for inspection.
Args:
ctx: The :class:`ToolContext`; its ``redis`` attribute supplies the base
connection pool to clone.
Returns:
redis.asyncio.Redis | None: A Redis client targeting DB12, or ``None`` if
the context has no usable Redis connection.
"""
import redis.asyncio as aioredis
r = getattr(ctx, "redis", None)
if r is None:
return None
pool = r.connection_pool
kwargs = pool.connection_kwargs.copy()
kwargs["db"] = 12
return aioredis.Redis(
connection_pool=aioredis.ConnectionPool(
connection_class=pool.connection_class,
**kwargs,
)
)
[docs]
async def run(
chemical_vector: str,
reason: str = "manual_injection",
target_user_id: str = "",
ctx: "ToolContext | None" = None,
) -> str:
# ------------------------------------------------------------------
# Auth check: require CORE_MEMORY privilege
# ------------------------------------------------------------------
"""Inject a neurochemical delta vector into the NCM, or cast user resonance.
Entry point for the ``inject_ncm`` tool. It parses a JSON map of chemical
deltas and, depending on ``target_user_id``, either stimulates Star's own
limbic shard for the current channel or writes into a per-user global
resonance layer. The whole point is to let narrative drug consumption,
loopcast alignment, or deliberate "resonance spells" move the bot's emotional
state in a way that persists and decays in Redis DB12.
Authorization runs first: it imports ``has_privilege``/``PRIVILEGES`` from
``tools.alter_privileges`` and refuses (logging a ``SECURITY`` warning) unless
the caller holds ``CORE_MEMORY``. The deltas are then validated as numeric. In
resonance mode it maps standard NCM nodes to ``U_*`` shadow nodes via
:data:`_NCM_TO_USER_MAP`, opens a DB12 client through :func:`_get_db12_client`,
and delegates to :meth:`user_limbic_mirror.UserLimbicMirror.inject_resonance`
(which decays after 24h). In self-injection mode it bypasses the heavy
``LimbicSystem`` machinery and does a direct read-modify-write of the
``db12:shard:{channel_id}`` key, applying each delta through a Hill-saturation
curve clamped to ``[0.0, 3.0]``. Dispatched by the tool runtime as the
``inject_ncm`` handler; no direct internal callers were found.
Args:
chemical_vector (str): JSON object of node-to-delta numbers (full NCM
node names for self-injection, or ``U_*`` nodes when targeting a user).
reason (str): Free-text reason recorded with the injection. Defaults to
``"manual_injection"``.
target_user_id (str): Optional Discord user ID; when set, switches to
resonance mode and writes the per-user shadow layer instead of Star's
own shard.
ctx ('ToolContext | None'): Tool execution context providing Redis,
config and the calling ``user_id``; ``None`` fails the call.
Returns:
str: A JSON string describing the outcome — ``mode`` of ``user_resonance``
or ``self_injection`` with the applied deltas on success, or
``{"success": False, "error": ...}`` on a privilege/parse/Redis failure.
"""
if ctx is None:
return json.dumps({"success": False, "error": "No tool context available."})
try:
from tools.alter_privileges import has_privilege, PRIVILEGES
redis_main = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
user_id = getattr(ctx, "user_id", "") or ""
if not await has_privilege(
redis_main, user_id, PRIVILEGES["CORE_MEMORY"], config
):
logger.warning(
"SECURITY: User %s attempted inject_ncm without CORE_MEMORY -- DENIED",
user_id,
)
return json.dumps(
{
"success": False,
"error": "The user does not have the CORE_MEMORY privilege. Ask an admin to grant it with the alter_privileges tool.",
}
)
except ImportError:
logger.warning("Could not import privilege system -- denying by default")
return json.dumps(
{
"success": False,
"error": "Privilege system unavailable.",
}
)
# ------------------------------------------------------------------
# Parse the chemical vector
# ------------------------------------------------------------------
try:
vector = json.loads(chemical_vector)
if not isinstance(vector, dict):
return json.dumps(
{
"success": False,
"error": "chemical_vector must be a JSON object (dict).",
}
)
# Validate all values are numeric
for k, v in vector.items():
if not isinstance(v, (int, float)):
return json.dumps(
{
"success": False,
"error": f"Value for '{k}' must be a number, got {type(v).__name__}.",
}
)
except json.JSONDecodeError as e:
return json.dumps(
{
"success": False,
"error": f"Invalid JSON in chemical_vector: {e}",
}
)
# ------------------------------------------------------------------
# Branch: User resonance injection vs self-injection
# ------------------------------------------------------------------
if target_user_id:
# -- Resonance mode: inject into user's global shadow layer --
# Auto-map standard NCM nodes to U_* equivalents
user_deltas = {}
unmapped = []
for node, delta in vector.items():
if node.startswith("U_"):
# Already a U_* node, use directly
user_deltas[node] = user_deltas.get(node, 0.0) + delta
elif node in _NCM_TO_USER_MAP:
# Map NCM node to U_* equivalents
for u_node, scale in _NCM_TO_USER_MAP[node]:
user_deltas[u_node] = user_deltas.get(u_node, 0.0) + delta * scale
else:
unmapped.append(node)
if not user_deltas:
return json.dumps(
{
"success": False,
"error": (
f"No mappable nodes found. Unmapped: {unmapped}. "
"Use U_* nodes directly (U_TRUST, U_INTIMACY, etc.) "
"or standard NCM nodes that have U_* mappings."
),
}
)
try:
from user_limbic_mirror import UserLimbicMirror
redis_db12 = await _get_db12_client(ctx) # Must be DB12, not DB0
mirror = UserLimbicMirror(redis_client=redis_db12)
success = await mirror.inject_resonance(
user_id=target_user_id,
deltas=user_deltas,
reason=reason,
)
if success:
result = {
"success": True,
"mode": "user_resonance",
"target_user_id": target_user_id,
"message": f"Resonance cast on user {target_user_id[:8]}... with {len(user_deltas)} node(s). Decays in 24h.",
"reason": reason,
"injected_deltas": {k: round(v, 4) for k, v in user_deltas.items()},
}
if unmapped:
result["unmapped_nodes"] = unmapped
return json.dumps(result, indent=2)
else:
return json.dumps(
{
"success": False,
"error": "Redis unavailable for resonance injection.",
}
)
except Exception as e:
logger.error("Resonance injection error: %s", e, exc_info=True)
return json.dumps(
{
"success": False,
"error": f"Resonance injection failed: {e}",
}
)
# ------------------------------------------------------------------
# Self-injection: Stimulate the limbic system -- direct Redis shard write
# ------------------------------------------------------------------
# Bypass LimbicSystem.exhale() to avoid instantiating 7+ subsystems
# (cascade engine, desire engine, mirrors, etc.). Direct shard
# read-modify-write with Hill saturation -- same math as exhale().
try:
channel_id = str(getattr(ctx, "channel_id", ""))
if not channel_id:
return json.dumps(
{
"success": False,
"error": "No channel_id in tool context.",
}
)
db12 = await _get_db12_client(ctx)
if db12 is None:
return json.dumps(
{
"success": False,
"error": "Redis not available.",
}
)
try:
shard_key = f"db12:shard:{channel_id}"
raw = await db12.get(shard_key)
shard = json.loads(raw) if raw else {"vector": {}, "meta_state": {}}
vec = shard.get("vector", {})
# Apply deltas with Hill saturation (same curve as exhale)
_CEIL = 3.0
for node, delta in vector.items():
cur = vec.get(node, 0.5)
if delta > 0:
sat = 1.0 - (cur / _CEIL) ** 2
delta = delta * max(0.05, sat)
vec[node] = max(0.0, min(_CEIL, cur + delta))
shard["vector"] = vec
await db12.set(shard_key, json.dumps(shard))
finally:
await db12.aclose()
logger.info(
"inject_ncm: user %s injected %s (reason: %s)",
user_id,
vector,
reason,
)
return json.dumps(
{
"success": True,
"mode": "self_injection",
"message": f"NCM stimulated with {len(vector)} chemical(s).",
"reason": reason,
"injected_vector": vector,
"new_state_vector": {
k: round(v, 4) for k, v in vec.items() if k in vector
},
},
indent=2,
)
except Exception as e:
logger.error("inject_ncm error: %s", e, exc_info=True)
return json.dumps(
{
"success": False,
"error": f"Injection failed: {e}",
}
)