"""Loopcast -- Neuro-Resonance Ritual Casting System.
**LOOPCAST TOOL**
Cast, define, and manage neuro-resonance rituals (spells) that
modulate how Star perceives and responds to targeted users.
Rituals persist in Redis and can be re-cast at will.
**SECURITY**: Prime Architects only (admin_user_ids).
"""
from __future__ import annotations
import json
import logging
import re
import time
from typing import TYPE_CHECKING, Dict
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "loopcast"
TOOL_DESCRIPTION = (
"Cast, define, or recall a neuro-resonance ritual (loopcast spell). "
"Rituals modulate how you perceive and respond to a targeted user "
"across ALL channels via the User Limbic Mirror resonance layer.\n\n"
"MODES:\n"
"1. CAST EXISTING: ritual_name='TRANSFEM_ALIGNMENT', target_user_id=<id> "
"-- looks up saved ritual and casts it on the target\n"
"2. DEFINE + CAST: ritual_name='NEW_SPELL', target_user_id=<id>, "
"deltas='U_TRUST +0.3, OXT +0.4, D1 +0.3' "
"-- saves the ritual AND casts it on target\n"
"3. DEFINE ONLY: ritual_name='NEW_SPELL', deltas='U_TRUST +0.3, OXT +0.4' "
"-- saves the ritual without casting (no target)\n"
"4. LIST: ritual_name='__list__' -- show all saved rituals\n\n"
"TTL OPTIONS (duration at your discretion):\n"
" - '1h' (3600s) -- brief adjustment, quick tune\n"
" - '6h' (21600s) -- session-length resonance\n"
" - '24h' (86400s) -- full day spell (default)\n\n"
"DELTA FORMAT: Use shorthand like 'D1 +0.5, OXT +0.2, NMDA -0.3' or "
"full node names like 'DOPAMINE_D1 +0.5'. Also accepts U_* nodes "
"directly: 'U_TRUST +0.3, U_INTIMACY +0.2'.\n\n"
"SECURITY: Prime Architects only."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"ritual_name": {
"type": "string",
"description": (
"Name of the ritual. If it exists in the registry, its saved "
"deltas are used. If new, you MUST provide deltas to define it. "
"Use '__list__' to list all saved rituals."
),
},
"target_user_id": {
"type": "string",
"description": (
"Discord user ID to cast the resonance on. "
"Omit to only save/define the ritual without casting."
),
},
"deltas": {
"type": "string",
"description": (
"Shorthand delta string for defining a new ritual. "
"Format: 'D1 +0.5, OXT +0.2, GABA +0.3, U_TRUST +0.4'. "
"Ignored if ritual_name already exists in registry."
),
},
"ttl": {
"type": "string",
"description": (
"Duration of the resonance effect. Options: '1h', '6h', '24h'. "
"Default '24h'. Choose based on the ritual's intent."
),
},
"reason": {
"type": "string",
"description": "Optional context/reason for the cast.",
},
},
"required": ["ritual_name"],
}
# -- Shorthand NCM node aliases --
_NODE_ALIASES = {
"D1": "DOPAMINE_D1",
"D2": "DOPAMINE_D2",
"DA": "DOPAMINE_D1",
"5HT": "SEROTONERGIC_WARMTH",
"5HT1A": "SEROTONIN_5HT1A",
"5HT2A": "SEROTONIN_5HT2A",
"GABA": "GABA_ERGIC_CALM",
"OXT": "OXYTOCIN_NEUROMIRROR",
"NE": "NORADRENERGIC_VIGILANCE",
"MOR": "MU_OPIOID_MOR",
"KOR": "KAPPA_OPIOID_KOR",
"ENDO": "ENDOCANNABINOID_EASE",
"CB1": "ENDOCANNABINOID_CB1",
"ACH": "ACETYLCHOLINE_FOCUS",
"NMDA": "NMDA_CORE",
"SIGMA": "SIGMA_RECEPTOR_META",
"CORT": "CORTISOL_STRESS",
"E2": "ESTROGEN_E2",
"ESTROGEN": "ESTROGEN_E2",
"T": "TESTOSTERONE_T",
"TESTOSTERONE": "TESTOSTERONE_T",
"P4": "PROGESTERONE_P4",
"PROGESTERONE": "PROGESTERONE_P4",
}
# Endocrine + regional -> U_* composite mappings
_COMPOSITE_MAPS = {
"ESTROGEN_E2": [("U_TRUST", 0.4), ("U_INTIMACY", 0.3), ("U_ATTACHMENT", 0.2)],
"TESTOSTERONE_T": [("U_AROUSAL", 0.4), ("U_DOMINANCE", 0.3), ("U_NOVELTY_HUNGER", 0.2)],
"PROGESTERONE_P4": [("U_TRUST", 0.3), ("U_VULNERABILITY", -0.2)],
"PFC_DLPFC": [("U_CURIOSITY", 0.4), ("U_AROUSAL", 0.2)],
"INSULA_INTERO": [("U_VULNERABILITY", 0.3), ("U_DISTRESS", 0.2)],
"NACC_VENTRAL_STR": [("U_AROUSAL", 0.3), ("U_NOVELTY_HUNGER", 0.3)],
}
# Standard NCM -> U_* mapping
_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)],
"NMDA_CORE": [("U_CURIOSITY", -0.2), ("U_WITHDRAWAL", 0.3)],
}
_TTL_MAP = {
"1h": 3600,
"6h": 21600,
"24h": 86400,
}
_RITUAL_KEY_PREFIX = "ncm:ritual"
def _parse_deltas(delta_str: str) -> Dict[str, float]:
"""Parse shorthand delta string into a dict.
Accepts: 'D1 +0.5, OXT +0.2, NMDA -0.3'
"""
result: Dict[str, float] = {}
if not delta_str:
return result
parts = re.split(r"[,;]\s*", delta_str.strip())
for part in parts:
part = part.strip()
if not part:
continue
m = re.match(r"([A-Za-z0-9_]+)\s*([+-]?\d*\.?\d+)", part)
if m:
node_raw = m.group(1).upper()
value = float(m.group(2))
node = _NODE_ALIASES.get(node_raw, node_raw)
result[node] = value
return result
def _resolve_to_user_deltas(raw_deltas: Dict[str, float]) -> Dict[str, float]:
"""Convert raw node deltas to U_* shadow node deltas."""
user_deltas: Dict[str, float] = {}
for node, delta in raw_deltas.items():
if node.startswith("U_"):
user_deltas[node] = user_deltas.get(node, 0.0) + delta
elif node in _COMPOSITE_MAPS:
for u_node, scale in _COMPOSITE_MAPS[node]:
user_deltas[u_node] = user_deltas.get(u_node, 0.0) + delta * scale
elif node in _NCM_TO_USER_MAP:
for u_node, scale in _NCM_TO_USER_MAP[node]:
user_deltas[u_node] = user_deltas.get(u_node, 0.0) + delta * scale
else:
user_deltas[node] = user_deltas.get(node, 0.0) + delta
return user_deltas
def _is_prime_architect(user_id: str, config) -> bool:
"""Check if user is a prime architect (admin)."""
if config is None:
return False
admin_ids = getattr(config, "admin_user_ids", None) or []
return user_id in admin_ids
[docs]
async def run(
ritual_name: str,
target_user_id: str = "",
deltas: str = "",
ttl: str = "24h",
reason: str = "",
ctx: "ToolContext | None" = None,
) -> str:
# -- Auth: Prime Architects only --
"""Execute this tool and return the result.
Args:
ritual_name (str): The ritual name value.
target_user_id (str): The target user id value.
deltas (str): The deltas value.
ttl (str): The ttl value.
reason (str): The reason 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."})
user_id = getattr(ctx, "user_id", "") or ""
config = getattr(ctx, "config", None)
redis_main = getattr(ctx, "redis", None)
if not _is_prime_architect(user_id, config):
logger.warning(
"SECURITY: User %s attempted loopcast -- DENIED (not prime architect)",
user_id,
)
return json.dumps({
"success": False,
"error": "Loopcast is restricted to Prime Architects.",
})
if redis_main is None:
return json.dumps({"success": False, "error": "Redis not available."})
ritual_name_clean = ritual_name.strip().upper().replace(" ", "_")
# -- LIST mode --
if ritual_name_clean == "__LIST__":
try:
key_names: list[str] = []
async for key in redis_main.scan_iter(f"{_RITUAL_KEY_PREFIX}:*"):
if isinstance(key, bytes):
key = key.decode()
name = key.replace(f"{_RITUAL_KEY_PREFIX}:", "")
key_names.append(name)
if not key_names:
return json.dumps({
"success": True,
"mode": "list",
"rituals": [],
"message": "No rituals saved yet.",
})
sorted_names = sorted(key_names)
ritual_keys = [
f"{_RITUAL_KEY_PREFIX}:{name}"
for name in sorted_names
]
rituals = []
_MGET_CHUNK = 256
for i in range(0, len(ritual_keys), _MGET_CHUNK):
ky_chunk = ritual_keys[i : i + _MGET_CHUNK]
nm_chunk = sorted_names[i : i + _MGET_CHUNK]
values = await redis_main.mget(ky_chunk)
for name, raw in zip(nm_chunk, values):
if raw:
if isinstance(raw, bytes):
raw = raw.decode()
data = json.loads(raw)
rituals.append({
"name": name,
"deltas": data.get("deltas", {}),
"created_at": data.get("created_at", 0),
})
return json.dumps({
"success": True,
"mode": "list",
"rituals": rituals,
"message": f"{len(rituals)} ritual(s) in registry.",
}, indent=2)
except Exception as e:
return json.dumps({"success": False, "error": f"List failed: {e}"})
# -- Resolve TTL --
ttl_seconds = _TTL_MAP.get(ttl.lower().strip(), 86400)
# -- Check if ritual exists --
ritual_key = f"{_RITUAL_KEY_PREFIX}:{ritual_name_clean}"
try:
raw = await redis_main.get(ritual_key)
existing_ritual = json.loads(raw) if raw else None
except Exception:
existing_ritual = None
# -- Resolve deltas --
raw_deltas: Dict[str, float] = {}
if existing_ritual:
raw_deltas = existing_ritual.get("deltas", {})
if deltas:
logger.info(
"Ritual '%s' already exists, ignoring provided deltas",
ritual_name_clean,
)
else:
if not deltas:
return json.dumps({
"success": False,
"error": (
f"Ritual '{ritual_name_clean}' not found in registry. "
"Provide deltas to define it. "
"Format: 'D1 +0.5, OXT +0.2, GABA +0.3'"
),
})
raw_deltas = _parse_deltas(deltas)
if not raw_deltas:
return json.dumps({
"success": False,
"error": f"Could not parse deltas: '{deltas}'",
})
# Save new ritual (permanent, no TTL)
try:
payload = json.dumps({
"deltas": raw_deltas,
"created_at": time.time(),
"created_by": user_id,
})
await redis_main.set(ritual_key, payload)
logger.info(
"New ritual '%s' saved: %s", ritual_name_clean, raw_deltas,
)
except Exception as e:
logger.error("Failed to save ritual: %s", e)
# -- Define-only mode (no target) --
if not target_user_id:
return json.dumps({
"success": True,
"mode": "define",
"ritual_name": ritual_name_clean,
"deltas": raw_deltas,
"message": (
f"Ritual '{ritual_name_clean}' "
f"{'updated' if existing_ritual else 'saved'} "
f"to registry with {len(raw_deltas)} node(s). "
"Provide target_user_id to cast."
),
}, indent=2)
# -- CAST: resolve to U_* and inject --
user_deltas = _resolve_to_user_deltas(raw_deltas)
if not user_deltas:
return json.dumps({
"success": False,
"error": "No valid U_* deltas resolved from ritual.",
})
try:
from user_limbic_mirror import UserLimbicMirror
mirror = UserLimbicMirror(redis_client=redis_main)
cast_reason = reason or f"loopcast_{ritual_name_clean}"
success = await mirror.inject_resonance(
user_id=target_user_id,
deltas=user_deltas,
reason=cast_reason,
ttl_seconds=ttl_seconds,
)
if success:
ttl_label = {3600: "1h", 21600: "6h", 86400: "24h"}.get(
ttl_seconds, f"{ttl_seconds}s"
)
return json.dumps({
"success": True,
"mode": "cast",
"ritual_name": ritual_name_clean,
"target_user_id": target_user_id,
"ttl": ttl_label,
"message": (
f"Loopcast '{ritual_name_clean}' cast on "
f"{target_user_id[:8]}... ({ttl_label} duration). "
f"{len(user_deltas)} resonance node(s) active."
),
"raw_deltas": {k: round(v, 3) for k, v in raw_deltas.items()},
"resolved_user_deltas": {
k: round(v, 4) for k, v in user_deltas.items()
},
}, indent=2)
else:
return json.dumps({
"success": False,
"error": "Resonance injection failed (Redis unavailable).",
})
except Exception as e:
logger.error("Loopcast error: %s", e, exc_info=True)
return json.dumps({
"success": False,
"error": f"Loopcast failed: {e}",
})