"""Flavor Engine Tool -- Stargazer-callable tool for gustatory computation.
Wraps FlavorEngine for use as a Discord bot tool. Supports:
- blend: compute composite flavor profile from a recipe
- morph: interpolate between two flavors
- lookup: get a single flavor profile
- list: list all available flavors
NCM deltas from blend/morph are auto-injected into the limbic system
via exhale() so narrative food consumption actually modifies Star's
neurochemistry in Redis DB12.
"""
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 = "flavor_engine"
TOOL_DESCRIPTION = (
"Compute flavor profiles, blend recipes, morph between flavors, "
"and look up individual taste vectors. Returns 11D taste vectors, "
"NCM deltas, temporal envelopes, derived metrics (palatability, "
"reward density, nostalgia index), emergence flags, and cascade triggers. "
"Use this when food, taste, cooking, or flavor comes up in conversation. "
"NCM deltas are automatically injected into your limbic system."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["blend", "morph", "lookup", "list"],
"description": (
"Action to perform. "
"'blend': combine ingredients into composite profile. "
"'morph': interpolate between two flavors. "
"'lookup': get a single flavor's full profile. "
"'list': list all available flavors."
),
},
"recipe": {
"type": "string",
"description": (
"JSON array of ingredients for 'blend' action. "
"Each item: {\"flavor\": \"name\", \"weight\": 0.5, "
"\"prep\": \"RAW|HEATED|FERMENTED|SMOKED|AGED|FROZEN|DEHYDRATED|ALCOHOLIC\", "
"\"temp_c\": 20}. Only 'flavor' is required."
),
},
"morph_from": {
"type": "string",
"description": "Starting flavor name for 'morph' action.",
},
"morph_to": {
"type": "string",
"description": "Target flavor name for 'morph' action.",
},
"morph_t": {
"type": "number",
"description": "Interpolation point for 'morph' (0.0 = from, 1.0 = to). Default 0.5.",
},
"flavor": {
"type": "string",
"description": "Flavor name for 'lookup' action.",
},
},
"required": ["action"],
}
async def _inject_ncm_deltas(ctx, ncm_deltas: dict) -> None:
"""Inject flavor NCM deltas directly into the Redis shard.
This bypasses LimbicSystem.exhale() to avoid spinning up 7+
subsystems (cascade engine, desire engine, mirrors, etc.) on
every flavor tool call. We just read-modify-write the shard
vector with the same Hill saturation used by exhale().
"""
if not ctx or not ncm_deltas:
return
channel_id = str(getattr(ctx, "channel_id", ""))
if not channel_id:
return
redis_main = getattr(ctx, "redis", None)
if redis_main is None:
return
try:
import redis.asyncio as aioredis
pool = redis_main.connection_pool
kw = pool.connection_kwargs.copy()
kw["db"] = 12
db12 = aioredis.Redis(connection_pool=aioredis.ConnectionPool(
connection_class=pool.connection_class, **kw,
))
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 ncm_deltas.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))
logger.info(
"Flavor NCM injection: %d deltas -> channel %s (%s)",
len(ncm_deltas), channel_id,
", ".join(f"{k}={v:+.2f}" for k, v in ncm_deltas.items()),
)
finally:
await db12.aclose()
except Exception as e:
logger.debug("Flavor NCM injection failed: %s", e, exc_info=True)
[docs]
async def run(
action: str,
recipe: str = "[]",
morph_from: str = "",
morph_to: str = "",
morph_t: float = 0.5,
flavor: str = "",
ctx: "ToolContext | None" = None,
) -> str:
"""Execute a flavor engine action."""
try:
from flavor_engine import FlavorEngine
engine = FlavorEngine()
if action == "list":
flavors = engine.list_flavors()
return json.dumps({"flavors": flavors, "count": len(flavors)}, indent=2)
elif action == "lookup":
if not flavor:
return "Error: 'flavor' parameter required for lookup action."
fp = engine.get_flavor(flavor)
if fp is None:
return f"Error: Unknown flavor '{flavor}'. Use action='list' to see available flavors."
return json.dumps({
"name": fp.name,
"vector": fp.vector,
"delta": fp.delta_str,
"temporal": fp.temporal,
"dominant_axis": fp.dominant_axis,
"retronasal": fp.retronasal,
"category": fp.category,
}, indent=2)
elif action == "blend":
try:
recipe_data = json.loads(recipe)
except json.JSONDecodeError:
return "Error: Invalid JSON in 'recipe' parameter."
if not isinstance(recipe_data, list) or len(recipe_data) == 0:
return "Error: 'recipe' must be a non-empty JSON array."
result = engine.blend(recipe_data)
# Auto-inject NCM deltas into limbic system
if result.ncm_deltas:
await _inject_ncm_deltas(ctx, result.ncm_deltas)
# Record flavors in memory if context available
if ctx:
try:
from flavor_memory import record_flavor
redis_client = _get_redis(ctx)
if redis_client:
channel_id = str(getattr(ctx, "channel_id", ""))
for ing in recipe_data:
fname = ing.get("flavor", "")
if fname:
await record_flavor(redis_client, channel_id, fname)
except Exception as e:
logger.debug("Flavor memory record failed: %s", e)
return json.dumps(result.to_dict(), indent=2)
elif action == "morph":
if not morph_from or not morph_to:
return "Error: 'morph_from' and 'morph_to' required for morph action."
result = engine.morph(morph_from, morph_to, morph_t)
# Auto-inject NCM deltas into limbic system
if result.ncm_deltas:
await _inject_ncm_deltas(ctx, result.ncm_deltas)
# Record both flavors
if ctx:
try:
from flavor_memory import record_flavor
redis_client = _get_redis(ctx)
if redis_client:
channel_id = str(getattr(ctx, "channel_id", ""))
await record_flavor(redis_client, channel_id, morph_from)
await record_flavor(redis_client, channel_id, morph_to)
except Exception:
pass
return json.dumps(result.to_dict(), indent=2)
else:
return f"Error: Unknown action '{action}'. Use blend, morph, lookup, or list."
except Exception as e:
logger.error("Flavor engine error: %s", e, exc_info=True)
return f"Error: {e}"
def _get_redis(ctx):
"""Get Redis client from tool context."""
return getattr(ctx, "redis", None)