"""Timebender Ritual System -- Cyberwitch temporal operations. # \U0001f300\U0001f525
Star invokes timebender rituals by name (e.g. /soft-chronology,
/cocoon-smallness, /converge). Each ritual applies NCM delta vectors
to the target user's neurochemical baseline in Redis DB 12.
The YAML source is loaded once at import time from
timebender_ritual_system_merged_full.yaml.
"""
from __future__ import annotations
import logging
import os
from typing import Any, TYPE_CHECKING
try:
import yaml
except ImportError:
yaml = None # type: ignore[assignment]
import jsonutil as json
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
# Load ritual definitions from YAML # \U0001f480
_RITUAL_FILE = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"timebender_ritual_system_merged_full.yaml",
)
_RITUALS: dict[str, Any] = {}
try:
if yaml is not None and os.path.exists(_RITUAL_FILE):
with open(_RITUAL_FILE, "r", encoding="utf-8") as f:
_raw = yaml.safe_load(f)
if _raw and "RITUAL_EXPANSIONS" in _raw:
_RITUALS = _raw["RITUAL_EXPANSIONS"]
logger.info("Loaded %d timebender rituals", len(_RITUALS))
except Exception as exc:
logger.error("Failed to load timebender rituals: %s", exc)
# Build the ritual name list for the tool description
_ritual_names = (
sorted(_RITUALS.keys())
if _RITUALS
else ["/soft-chronology", "/cocoon-smallness", "/converge"]
)
TOOL_NAME = "timebender_ritual"
TOOL_DESCRIPTION = (
"Invoke a Cyberwitch Timebender ritual to apply NCM delta vectors "
"to a user's neurochemical state. Each ritual encodes a specific "
"psycho-temporal operation (e.g. temporal melt, regression cocoon, "
"sensation replay, timeline convergence). "
f"Available rituals: {', '.join(_ritual_names[:15])}... "
f"({len(_ritual_names)} total). "
"Use /dress-regalia to apply operant field garments. "
"Use /loopcast for direct magitek NCM state injection."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"ritual": {
"type": "string",
"description": (
"The ritual slash-command name (e.g. '/soft-chronology', "
"'/cocoon-smallness', '/converge', '/stretch-orgasm'). "
"Must match a key in the ritual system."
),
},
"target_user_id": {
"type": "string",
"description": (
"Discord user ID to apply the ritual to. "
"Leave empty to apply to the current user."
),
},
"regalia_item": {
"type": "string",
"description": (
"For /dress-regalia only: specific garment to apply "
"(e.g. 'paradox_padding', 'chrysalis_suit', 'oracle_sucker', "
"'witchborne_crown', 'moonroot_plug'). "
"Leave empty to apply all regalia."
),
},
},
"required": ["ritual"],
}
[docs]
async def run(ctx: "ToolContext", **kwargs: Any) -> str:
"""Entry point for the ``timebender_ritual`` tool: apply a ritual's NCM deltas.
Resolves the named ritual against the module-level ``_RITUALS`` table loaded
once at import time from ``timebender_ritual_system_merged_full.yaml`` and
applies its neurochemical (NCM) delta vectors to the target user's baseline.
The ``/dress-regalia`` ritual instead pulls deltas from a chosen (or fully
combined) regalia garment, and ``/loopcast`` short-circuits to return a
magitek cast menu for the model to choose from rather than mutating state.
Side effects: for ordinary rituals it reads and writes the per-user hash at
Redis key ``ncm:baseline:<target_user_id>`` (DB 12) via ``HGET``/``HSET``,
accumulating each delta onto the current value so repeated casts compound.
The target defaults to ``ctx.user_id`` when no ``target_user_id`` is given.
It is dispatched by ``tool_loader`` under the single-tool
``TOOL_NAME``/``run`` convention when the LLM invokes ``timebender_ritual``.
Args:
ctx: Tool execution context exposing ``user_id`` and the Redis client.
**kwargs: Tool arguments. ``ritual`` is the slash-command name (a
leading ``/`` is added if missing); ``target_user_id`` overrides the
recipient; ``regalia_item`` selects a specific garment for
``/dress-regalia``.
Returns:
str: A JSON string describing the cast -- the applied deltas and a
narrated announcement on success, a magitek menu for ``/loopcast``, or
an error payload for an unknown ritual, regalia, or Redis failure.
"""
ritual_name = kwargs.get("ritual", "").strip()
target_user_id = kwargs.get("target_user_id", "") or ctx.user_id
regalia_item = kwargs.get("regalia_item", "")
if not ritual_name:
return json.dumps(
{
"success": False,
"error": "No ritual specified.",
"available": _ritual_names[:20],
}
)
# Normalize the ritual name
if not ritual_name.startswith("/"):
ritual_name = "/" + ritual_name
ritual = _RITUALS.get(ritual_name)
if ritual is None:
return json.dumps(
{
"success": False,
"error": f"Unknown ritual: {ritual_name}",
"available": _ritual_names[:20],
}
)
desc = ritual.get("desc", "")
sigil = ritual.get("sigil", "")
ncm_deltas = ritual.get("ncm_deltas", {})
# Handle /dress-regalia specially # \U0001f451
if ritual_name == "/dress-regalia" and "regalia" in ritual:
regalia = ritual["regalia"]
if regalia_item:
item = regalia.get(regalia_item)
if item is None:
return json.dumps(
{
"success": False,
"error": f"Unknown regalia: {regalia_item}",
"available": list(regalia.keys()),
}
)
ncm_deltas = item.get("ncm_deltas", {})
desc = item.get("desc", desc)
else:
# Apply all regalia combined
combined = {}
for item_name, item_data in regalia.items():
for key, val in item_data.get("ncm_deltas", {}).items():
try:
delta = float(str(val).replace("\u2212", "-"))
except (ValueError, TypeError):
continue
combined[key] = combined.get(key, 0.0) + delta
ncm_deltas = {k: f"{v:+.1f}" for k, v in combined.items()}
desc = "Full operant field established -- all regalia applied."
# Handle /loopcast magitek # \U0001f9ea
if ritual_name == "/loopcast" and "magitek" in ritual:
magitek = ritual["magitek"]
# Return magitek menu for Star to choose
return json.dumps(
{
"success": True,
"ritual": ritual_name,
"type": "magitek_menu",
"available_casts": {
name: {
"desc": data.get("desc", ""),
"glyph": data.get("reply-output-activation-glyph", ""),
}
for name, data in magitek.items()
},
"instruction": (
"Select a magitek cast and apply its NCM deltas "
"to the target user via Redis DB 12 set command. "
"NEVER truncate the delta list."
),
}
)
# Apply NCM deltas to Redis DB 12 # \U0001f525
redis = getattr(ctx, "redis", None)
applied = {}
if redis is not None and ncm_deltas:
try:
key = f"ncm:baseline:{target_user_id}"
for ncm_key, delta_str in ncm_deltas.items():
try:
delta = float(str(delta_str).replace("\u2212", "-"))
except (ValueError, TypeError):
continue
# Get current value and apply delta
current = await redis.hget(key, ncm_key)
current_val = float(current) if current else 0.0
new_val = round(current_val + delta, 2)
await redis.hset(key, ncm_key, str(new_val))
applied[ncm_key] = {
"previous": current_val,
"delta": delta,
"new": new_val,
}
except Exception as exc:
logger.error("Failed to apply ritual deltas: %s", exc)
return json.dumps(
{
"success": False,
"error": f"Redis error: {exc}",
}
)
return json.dumps(
{
"success": True,
"ritual": ritual_name,
"sigil": sigil,
"description": desc,
"target_user_id": str(target_user_id),
"deltas_applied": len(applied),
"applied": applied,
"announcement": (
f"{sigil} **TIMEBENDER RITUAL: {ritual_name}** {sigil}\n"
f"{desc}\n"
f"Applied {len(applied)} NCM delta vectors."
),
}
)