"""Redis key namespace + typed helpers per docs/Specs.md §9.2. Every key carries the subreddit_id where applicable (invariant I-7). Public functions take `subreddit_id` as a mandatory positional arg. Key patterns owned by this module: profile:{sub}:{user} cached UserMemory (1h TTL) summary:{thread} thread summary blob (24h TTL) velocity:{sub}:{target} sliding-window report timestamps (1h TTL) verdict:{correlation_id} full verdict for "Explain last call" (7d TTL) embedding:{rule} precomputed rule embedding (30d TTL) budget:{sub}:{day} daily spend counter (24h TTL) """ from __future__ import annotations import json import math import time from typing import TYPE_CHECKING from store.types import UserMemoryRow if TYPE_CHECKING: from redis.asyncio import Redis # === TTL constants (seconds) =========================================== _TTL_PROFILE = 60 * 60 _TTL_SUMMARY = 60 * 60 * 24 _TTL_VELOCITY = 60 * 60 _TTL_VERDICT = 60 * 60 * 24 * 7 _TTL_EMBEDDING = 60 * 60 * 24 * 30 _TTL_BUDGET = 60 * 60 * 24 # === Key builders ====================================================== def k_profile(subreddit_id: str, user_id: str) -> str: return f"profile:{subreddit_id}:{user_id}" def k_summary(thread_id: str) -> str: return f"summary:{thread_id}" def k_velocity(subreddit_id: str, target_id: str) -> str: return f"velocity:{subreddit_id}:{target_id}" def k_verdict(correlation_id: str) -> str: return f"verdict:{correlation_id}" def k_embedding(rule_id: str) -> str: return f"embedding:{rule_id}" def k_rules_embed(subreddit_id: str) -> str: """All rule embeddings for a subreddit, stored as a single JSON blob.""" return f"rules_embed:{subreddit_id}" def k_budget(subreddit_id: str, day: str) -> str: """day is ISO YYYY-MM-DD (UTC).""" return f"budget:{subreddit_id}:{day}" # === Profile cache ===================================================== async def get_profile_cache( client: Redis[str], *, subreddit_id: str, user_id: str ) -> UserMemoryRow | None: raw = await client.get(k_profile(subreddit_id, user_id)) if not raw: return None return UserMemoryRow.model_validate(json.loads(raw)) async def set_profile_cache( client: Redis[str], *, subreddit_id: str, user_id: str, row: UserMemoryRow ) -> None: await client.set( k_profile(subreddit_id, user_id), row.model_dump_json(), ex=_TTL_PROFILE, ) # === Thread summary cache ============================================== async def get_thread_summary(client: Redis[str], *, thread_id: str) -> dict[str, object] | None: raw = await client.get(k_summary(thread_id)) return json.loads(raw) if raw else None async def set_thread_summary( client: Redis[str], *, thread_id: str, summary: dict[str, object] ) -> None: await client.set(k_summary(thread_id), json.dumps(summary), ex=_TTL_SUMMARY) # === Rule embeddings (policy_match) ==================================== # One blob per subreddit: list of {id, text, embedding}. Invalidated when # settings change (post-MVP: settings handler calls invalidate_rule_embeddings). async def get_rule_embeddings( client: Redis[str], *, subreddit_id: str ) -> list[dict[str, object]] | None: """Return cached rule embeddings or None if cold.""" raw = await client.get(k_rules_embed(subreddit_id)) return json.loads(raw) if raw else None async def set_rule_embeddings( client: Redis[str], *, subreddit_id: str, rules: list[dict[str, object]], ) -> None: await client.set(k_rules_embed(subreddit_id), json.dumps(rules), ex=_TTL_EMBEDDING) async def invalidate_rule_embeddings( client: Redis[str], *, subreddit_id: str ) -> None: """Called when subreddit settings change.""" await client.delete(k_rules_embed(subreddit_id)) # === Report velocity (sliding window) ================================== # Implementation: a Redis sorted set per target, scored by epoch-seconds. # `ZADD` on each report, `ZREMRANGEBYSCORE` to evict outside the window, # `ZCARD` to read the current count. The z-score is computed against a # rolling per-subreddit baseline (a separate key, post-MVP). async def record_report( client: Redis[str], *, subreddit_id: str, target_id: str, timestamp: float | None = None, ) -> int: """Append a report event to the sliding window. Returns the current count.""" now = timestamp if timestamp is not None else time.time() key = k_velocity(subreddit_id, target_id) await client.zadd(key, {f"{now}-{int(now * 1000)}": now}) cutoff = now - _TTL_VELOCITY await client.zremrangebyscore(key, "-inf", cutoff) await client.expire(key, _TTL_VELOCITY) return int(await client.zcard(key)) async def velocity_count( client: Redis[str], *, subreddit_id: str, target_id: str, window_seconds: int = _TTL_VELOCITY ) -> int: now = time.time() key = k_velocity(subreddit_id, target_id) cutoff = now - window_seconds return int(await client.zcount(key, cutoff, "+inf")) def velocity_zscore(count: int, baseline_mean: float, baseline_stddev: float) -> float: """Tiny pure helper — used by the report_velocity tool. Treats stddev=0 as 1.""" sd = baseline_stddev if baseline_stddev > 0 else 1.0 z = (count - baseline_mean) / sd # Cap to avoid pathological values for prompt tokens / UI: return float(max(-9.0, min(9.0, z))) # === Verdict cache (for "Explain ModPilot's last call" menu) =========== async def get_cached_verdict( client: Redis[str], *, correlation_id: str ) -> dict[str, object] | None: raw = await client.get(k_verdict(correlation_id)) return json.loads(raw) if raw else None async def set_cached_verdict( client: Redis[str], *, correlation_id: str, verdict: dict[str, object] ) -> None: await client.set(k_verdict(correlation_id), json.dumps(verdict), ex=_TTL_VERDICT) # === Daily budget tracking ============================================= def _today_utc() -> str: return time.strftime("%Y-%m-%d", time.gmtime()) async def add_spend( client: Redis[str], *, subreddit_id: str, cents: int, day: str | None = None ) -> int: """Atomically increment today's spend (integer cents). Returns the new total. The Calibrator + budget gate (post-MVP) read this to enforce daily_spend_cap_per_sub_usd from Settings. """ key = k_budget(subreddit_id, day or _today_utc()) new_total = int(await client.incrby(key, cents)) await client.expire(key, _TTL_BUDGET) return new_total async def todays_spend_cents( client: Redis[str], *, subreddit_id: str, day: str | None = None ) -> int: key = k_budget(subreddit_id, day or _today_utc()) raw = await client.get(key) return int(raw) if raw else 0 def cents(usd: float) -> int: """Convert a USD float to integer cents (rounded). Avoids float drift in incrby.""" return math.ceil(usd * 100)