"""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 jsonutil as 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:
"""Dispatch one of the four flavor-engine actions and return JSON.
This is the ``generate_image``-style single-tool entry point for the
``flavor_engine`` tool: it instantiates a fresh ``FlavorEngine`` and routes
on ``action`` to ``list``, ``lookup``, ``blend``, or ``morph``, serialising
the result with ``jsonutil`` (aliased ``json``). Beyond pure computation it
has two side effects that make narrative eating actually change Star's
state: for ``blend`` and ``morph`` it calls ``_inject_ncm_deltas`` to fold
the result's NCM deltas into the per-channel limbic shard in Redis DB12, and
(when a ``ToolContext`` is present) it records each ingredient via
``flavor_memory.record_flavor`` using the Redis client from ``_get_redis``.
All other failures are caught and returned as an ``Error:`` string rather
than raised.
Dispatched by ``tool_loader.py``, which imports this module and pulls
``run`` out with ``getattr(module, "run")`` to register it under
``TOOL_NAME`` ("flavor_engine"); it is also exercised directly in the test
suite.
Args:
action: One of ``"blend"``, ``"morph"``, ``"lookup"``, or ``"list"``.
recipe: JSON array of ingredient dicts, used only for ``blend``.
morph_from: Starting flavor name for ``morph``.
morph_to: Target flavor name for ``morph``.
morph_t: Interpolation point in ``[0, 1]`` for ``morph`` (default 0.5).
flavor: Flavor name for ``lookup``.
ctx: The current ``ToolContext``; supplies ``channel_id`` and the Redis
client needed for NCM injection and flavor-memory recording. When
``None``, those side effects are skipped.
Returns:
A pretty-printed JSON string for a successful action, or an
``"Error: ..."`` message string when the action is unknown, a required
parameter is missing, the recipe JSON is invalid, or an exception is
raised.
"""
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):
"""Pull the Redis client off the ``ToolContext``, or ``None``.
Tiny accessor that returns ``ctx.redis`` (the main shared Redis connection)
via ``getattr`` so callers can tolerate a missing attribute or a ``None``
context without guarding. It does no I/O of its own.
Called within this module by ``run`` in the ``blend`` and ``morph``
branches, where the returned client is handed to
``flavor_memory.record_flavor`` to persist consumed flavors.
Args:
ctx: The current ``ToolContext`` (or ``None``).
Returns:
The Redis client bound to the context, or ``None`` if unavailable.
"""
return getattr(ctx, "redis", None)