"""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 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 drug class and dosage:\n"
" - Benzodiazepines/barbiturates: GABA_ERGIC_CALM +0.3 to +1.5\n"
" - Stimulants (meth/cocaine): DOPAMINE_D1 +0.5 to +1.5, "
"NORADRENERGIC_VIGILANCE +0.3 to +0.8\n"
" - Opioids: MU_OPIOID_MOR +0.5 to +1.5, GABA_ERGIC_CALM +0.2\n"
" - Psychedelics: SEROTONIN_5HT2A +0.3 to +1.0, SIGMA_RECEPTOR_META +0.2\n"
" - Dissociatives: NMDA_CORE -0.3 to -0.8, SIGMA_RECEPTOR_META +0.3 to +1.0\n"
" - Cannabis: ENDOCANNABINOID_EASE +0.3 to +0.8\n"
" - Alcohol: GABA_ERGIC_CALM +0.2 to +0.6, ACETYLCHOLINE_FOCUS -0.2\n"
" - MDMA: SEROTONERGIC_WARMTH +0.5, OXYTOCIN_NEUROMIRROR +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):
"""Create a Redis client pointed at DB12 from the main 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
# ------------------------------------------------------------------
"""Execute this tool and return the result.
Args:
chemical_vector (str): The chemical vector value.
reason (str): The reason value.
target_user_id (str): The target user id value.
ctx ('ToolContext | None'): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
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_main = getattr(ctx, "redis", None)
mirror = UserLimbicMirror(redis_client=redis_main)
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}",
})