"""Flavor Memory — Per-channel gustatory history in Redis.
Tracks which flavors a user has referenced, how often, and when.
Provides familiarity bonuses: more mentions → stronger OXT/MOR on re-encounter.
"""
from __future__ import annotations
import jsonutil as json
import logging
import time
from typing import Any, Dict, List
logger = logging.getLogger(__name__)
# Redis key pattern
_KEY_PREFIX = "ncm:flavor_history"
_MAX_HISTORY = 20
def _make_key(channel_id: str) -> str:
"""Internal helper: make key.
Args:
channel_id (str): Discord/Matrix channel identifier.
Returns:
str: Result string.
"""
return f"{_KEY_PREFIX}:{channel_id}"
[docs]
async def record_flavor(
redis_client: Any,
channel_id: str,
flavor_name: str,
) -> None:
"""Log a flavor reference for a channel.
Each entry: ``{"flavor": name, "count": N, "first_ts": ts, "last_ts": ts}``.
Maintains a running history of the last _MAX_HISTORY unique flavors.
"""
if redis_client is None:
return
key = _make_key(channel_id)
fname = flavor_name.upper().replace(" ", "_")
try:
raw = await redis_client.get(key)
history: List[Dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
history = []
# Find existing entry for this flavor
found = False
for entry in history:
if entry.get("flavor") == fname:
entry["count"] = entry.get("count", 1) + 1
entry["last_ts"] = time.time()
found = True
break
if not found:
history.append(
{
"flavor": fname,
"count": 1,
"first_ts": time.time(),
"last_ts": time.time(),
}
)
# Trim to max history (keep most recent by last_ts)
history.sort(key=lambda e: e.get("last_ts", 0), reverse=True)
history = history[:_MAX_HISTORY]
try:
await redis_client.set(
key,
json.dumps(history),
ex=60 * 60 * 24 * 90, # 90 day TTL
)
except Exception as e:
logger.error("Failed to write flavor history: %s", e)
[docs]
async def get_flavor_affinity(
redis_client: Any,
channel_id: str,
flavor_name: str,
) -> float:
"""Get familiarity bonus for a specific flavor in this channel.
Returns a multiplier: 1.0 for unknown, up to ~1.5 for very familiar.
The bonus follows a log curve: rapid early gain, diminishing returns.
Used as a multiplier on OXT and MOR deltas during Phase 4 NCM computation.
"""
if redis_client is None:
return 1.0
key = _make_key(channel_id)
fname = flavor_name.upper().replace(" ", "_")
try:
raw = await redis_client.get(key)
history: List[Dict[str, Any]] = json.loads(raw) if raw else []
except Exception:
return 1.0
for entry in history:
if entry.get("flavor") == fname:
count = entry.get("count", 1)
# Log curve: 1.0 at count=1, ~1.3 at count=5, ~1.5 at count=20
import math
return 1.0 + 0.15 * math.log(count + 1)
return 1.0
[docs]
async def get_history(
redis_client: Any,
channel_id: str,
) -> List[Dict[str, Any]]:
"""Return the full per-channel flavor history list.
Reads the JSON-encoded history array from the single Redis string keyed by
``ncm:flavor_history:<channel_id>`` (built via :func:`_make_key`) and decodes
it. Each element is a record like ``{"flavor": NAME, "count": N, "first_ts": ts, "last_ts": ts}``
as written by :func:`record_flavor`. This is a read-only accessor: it never
mutates or sorts the stored data, and any decode/connection error is swallowed
and reported as an empty history so callers degrade gracefully.
Called by :func:`get_top_flavors` within this module; no external callers were
found via grep, so it otherwise serves as a public read helper for the flavor
subsystem.
Args:
redis_client: An async Redis client, or ``None`` to short-circuit to an
empty list (e.g. when Redis is unavailable).
channel_id: Discord/Matrix channel identifier whose history to load.
Returns:
list[dict[str, Any]]: The decoded history records, newest-first as last
persisted, or an empty list when absent or on any error.
"""
if redis_client is None:
return []
key = _make_key(channel_id)
try:
raw = await redis_client.get(key)
return json.loads(raw) if raw else []
except Exception:
return []
[docs]
async def get_top_flavors(
redis_client: Any,
channel_id: str,
n: int = 5,
) -> List[str]:
"""Return the channel's most-referenced flavor names, highest count first.
Loads the channel's history via :func:`get_history`, sorts a local copy by the
per-flavor ``count`` field in descending order, and returns just the canonical
flavor names (upper-cased, underscore-joined) of the top ``n`` entries. Useful
for surfacing a channel's recurring gustatory preferences, e.g. to seed prompt
context or summaries. Pure read path: the underlying Redis data is not modified.
No external callers were found via grep, so this acts as a public convenience
accessor over :func:`get_history`.
Args:
redis_client: An async Redis client, or ``None`` (yields an empty list via
:func:`get_history`).
channel_id: Discord/Matrix channel identifier whose top flavors to fetch.
n: Maximum number of flavor names to return (default 5).
Returns:
list[str]: Up to ``n`` flavor names ordered by descending reference count.
"""
history = await get_history(redis_client, channel_id)
history.sort(key=lambda e: e.get("count", 0), reverse=True)
return [e["flavor"] for e in history[:n]]