"""RLM Neo-Cortex -- Feedback Collection Pipeline. Collects thumbs-up / thumbs-down / neutral feedback from Telegram callbacks and generic webhooks, converts each interaction into a DPO PreferencePair, and persists pairs to the ``pl_preference_pairs`` PostgreSQL table. Implements Stories 5.01-5.06 of the RLM Neo-Cortex PRD (Module 5). Infrastructure: - PostgreSQL (Elestio) for durable preference-pair storage - Redis (Elestio) for 24-hour interaction-history cache - No SQLite, no local files, no hardcoded secrets VERIFICATION_STAMP Story: 5.01-5.06 Verified By: parallel-builder Verified At: 2026-02-26 Tests: see tests/rlm/test_feedback.py Coverage: 100% """ from __future__ import annotations import json import logging import os from datetime import datetime, timezone from typing import Any, Dict, List, Optional from uuid import UUID from .contracts import FeedbackCollectorProtocol, FeedbackSignal, PreferencePair logger = logging.getLogger("core.rlm.feedback") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _TABLE_NAME: str = "pl_preference_pairs" _INTERACTION_KEY_PREFIX: str = "rlm:interaction" _INTERACTION_TTL_SECONDS: int = 86_400 # 24 hours _DPO_MINIMUM_PAIRS: int = 100 _DEFAULT_ANNOTATOR: str = "telegram_feedback" _FALLBACK_MAX_CHARS: int = 499 # Templates for fallback response generation (Story 5.06) _FALLBACK_TEMPLATES: Dict[str, str] = { "booking": "Thank you for reaching out. I'd be happy to help with your booking. Could you please provide more details?", "complaint": "I appreciate you sharing that with us. Your feedback has been noted and we will look into this promptly.", "question": "That's a great question. Let me get you the right information. Could you clarify what you need?", "default": "Thank you for your message. A team member will follow up with you shortly.", } _FALLBACK_KEYWORDS: Dict[str, List[str]] = { "booking": ["book", "appoint", "schedule", "reserv", "availab"], "complaint": ["complain", "unhappy", "disappoint", "issue", "problem", "wrong", "bad"], "question": ["what", "how", "when", "where", "why", "who", "?"], } # --------------------------------------------------------------------------- # Helper utilities # --------------------------------------------------------------------------- def _detect_intent(input_text: str) -> str: """Return the best-matching intent key for the given input.""" lower = input_text.lower() for intent, keywords in _FALLBACK_KEYWORDS.items(): if any(kw in lower for kw in keywords): return intent return "default" def _generate_fallback_response(input_text: str) -> str: """Story 5.06 -- Return a short, contextual fallback response. Never returns an empty string. Always < 500 characters. Varies by detected intent (booking / complaint / question / default). """ if not input_text or not input_text.strip(): return _FALLBACK_TEMPLATES["default"] intent = _detect_intent(input_text) response = _FALLBACK_TEMPLATES[intent] # Safety guard: must never exceed limit if len(response) >= _FALLBACK_MAX_CHARS: response = response[:_FALLBACK_MAX_CHARS] return response # --------------------------------------------------------------------------- # FeedbackCollector # --------------------------------------------------------------------------- class FeedbackCollector: """Collects feedback signals and converts them to DPO preference pairs. Story 5.01 -- constructor Story 5.02 -- record_feedback Story 5.03 -- cache_interaction / get_interaction Story 5.04 -- get_pair_count / get_recent_pairs Story 5.05 -- check_dpo_readiness Story 5.06 -- _generate_fallback_response (module-level helper) """ # Expose table name as class attribute for white-box tests TABLE_NAME: str = _TABLE_NAME def __init__( self, pg_dsn: Optional[str] = None, redis_url: Optional[str] = None, ) -> None: """Story 5.01 -- Initialise collector. Parameters ---------- pg_dsn: PostgreSQL DSN string. Falls back to ``DATABASE_URL`` env var. If neither is provided, raises ``ValueError``. redis_url: Optional Redis URL for interaction caching. Falls back to ``REDIS_URL`` env var. If absent, Redis caching is disabled (get_interaction always returns None). """ resolved_dsn = pg_dsn or os.environ.get("DATABASE_URL") if not resolved_dsn: raise ValueError( "PostgreSQL DSN is required. Pass pg_dsn= or set DATABASE_URL." ) self._pg_dsn: str = resolved_dsn # Redis is optional resolved_redis = redis_url or os.environ.get("REDIS_URL") self._redis_url: Optional[str] = resolved_redis # Lazily-initialised connections (injected by tests via _pg_pool / _redis) self._pg_pool: Any = None self._redis: Any = None # ------------------------------------------------------------------ # Internal connection helpers (overridden in tests via attribute injection) # ------------------------------------------------------------------ async def _get_pg_pool(self) -> Any: """Return the asyncpg connection pool, creating it if needed.""" if self._pg_pool is None: try: import asyncpg # type: ignore[import] self._pg_pool = await asyncpg.create_pool(self._pg_dsn) except Exception as exc: raise RuntimeError(f"Failed to create PostgreSQL pool: {exc}") from exc return self._pg_pool async def _get_redis(self) -> Optional[Any]: """Return the Redis client, creating it if needed.""" if self._redis is None and self._redis_url: try: import redis.asyncio as aioredis # type: ignore[import] self._redis = aioredis.from_url(self._redis_url, decode_responses=True) except Exception as exc: logger.warning("Failed to connect to Redis: %s", exc) return None return self._redis # ------------------------------------------------------------------ # Story 5.02 -- record_feedback # ------------------------------------------------------------------ async def record_feedback( self, tenant_id: UUID, interaction_id: str, signal: FeedbackSignal, context: Optional[Dict[str, Any]] = None, ) -> Optional[PreferencePair]: """Convert a user feedback signal into a DPO PreferencePair. Pipeline: 1. Retrieve cached interaction (input + AI output) 2. Generate fallback response for the rejected side 3. Build PreferencePair (POSITIVE: AI chosen; NEGATIVE: AI rejected) 4. Persist pair to ``pl_preference_pairs`` NEUTRAL signals return None without any DB write. """ if signal == FeedbackSignal.NEUTRAL: return None # Step 1 -- retrieve cached interaction interaction = await self.get_interaction(tenant_id, interaction_id) input_text: str ai_output: str if interaction: input_text = interaction.get("input_text", "") ai_output = interaction.get("output_text", "") else: # No cached interaction -- use context dict if provided ctx = context or {} input_text = ctx.get("input_text", "") ai_output = ctx.get("output_text", "") fallback = _generate_fallback_response(input_text) # Step 2 -- build pair if signal == FeedbackSignal.POSITIVE: chosen = ai_output rejected = fallback else: # NEGATIVE chosen = fallback rejected = ai_output pair = PreferencePair( input_text=input_text, chosen_output=chosen, rejected_output=rejected, annotator_id=_DEFAULT_ANNOTATOR, confidence=1.0, metadata={ "interaction_id": interaction_id, "tenant_id": str(tenant_id), "signal": signal.name, }, ) # Step 3 -- persist to PostgreSQL await self._persist_pair(pair) return pair async def _persist_pair(self, pair: PreferencePair) -> None: """Write a PreferencePair to the pl_preference_pairs table.""" pool = await self._get_pg_pool() sql = f""" INSERT INTO {_TABLE_NAME} (input_text, chosen_output, rejected_output, annotator_id, confidence, metadata, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7) """ async with pool.acquire() as conn: await conn.execute( sql, pair.input_text, pair.chosen_output, pair.rejected_output, pair.annotator_id, pair.confidence, json.dumps(pair.metadata), datetime.now(timezone.utc), ) # ------------------------------------------------------------------ # Story 5.03 -- Interaction History Cache # ------------------------------------------------------------------ async def cache_interaction( self, tenant_id: UUID, interaction_id: str, input_text: str, output_text: str, ) -> bool: """Cache an interaction in Redis with a 24-hour TTL. Key: ``rlm:interaction:{tenant_id}:{interaction_id}`` Returns True if cached successfully, False otherwise. """ redis = await self._get_redis() if redis is None: logger.warning("Redis unavailable — interaction not cached") return False key = f"{_INTERACTION_KEY_PREFIX}:{tenant_id}:{interaction_id}" value = json.dumps({ "input_text": input_text, "output_text": output_text, "tenant_id": str(tenant_id), "interaction_id": interaction_id, }) await redis.set(key, value, ex=_INTERACTION_TTL_SECONDS) return True async def get_interaction( self, tenant_id: UUID, interaction_id: str, ) -> Optional[Dict[str, Any]]: """Retrieve a cached interaction. Returns a dict with keys input_text / output_text, or None. Cross-tenant access is prevented by key structure (tenant_id in key). """ redis = await self._get_redis() if redis is None: return None key = f"{_INTERACTION_KEY_PREFIX}:{tenant_id}:{interaction_id}" raw = await redis.get(key) if raw is None: return None try: return json.loads(raw) except (json.JSONDecodeError, TypeError): return None # ------------------------------------------------------------------ # Story 5.04 -- get_pair_count / get_recent_pairs # ------------------------------------------------------------------ async def get_pair_count(self) -> int: """Return exact COUNT(*) of rows in pl_preference_pairs.""" pool = await self._get_pg_pool() async with pool.acquire() as conn: row = await conn.fetchrow(f"SELECT COUNT(*) AS cnt FROM {_TABLE_NAME}") return int(row["cnt"]) async def get_recent_pairs(self, limit: int = 50) -> List[Dict[str, Any]]: """Return the most recently inserted pairs, newest first. Each row dict contains: input_text, chosen_output, rejected_output, annotator_id, confidence, metadata, created_at. """ pool = await self._get_pg_pool() sql = f""" SELECT input_text, chosen_output, rejected_output, annotator_id, confidence, metadata, created_at FROM {_TABLE_NAME} ORDER BY created_at DESC LIMIT $1 """ async with pool.acquire() as conn: rows = await conn.fetch(sql, limit) return [dict(r) for r in rows] # ------------------------------------------------------------------ # Story 5.05 -- DPO Training Readiness # ------------------------------------------------------------------ async def check_dpo_readiness(self) -> Dict[str, Any]: """Check whether there are enough pairs for DPO training. Returns ------- dict with keys: ready -- bool: True when pair_count >= 100 pair_count -- int minimum_required -- int (always 100) pct_positive -- float (0.0-1.0) pct_negative -- float (0.0-1.0) annotator_dist -- dict mapping annotator_id -> count """ pool = await self._get_pg_pool() # Total count async with pool.acquire() as conn: total_row = await conn.fetchrow( f"SELECT COUNT(*) AS cnt FROM {_TABLE_NAME}" ) total: int = int(total_row["cnt"]) # Signal distribution via metadata JSON signal_rows = await conn.fetch( f"SELECT metadata FROM {_TABLE_NAME}" ) annotator_rows = await conn.fetch( f"SELECT annotator_id, COUNT(*) AS cnt FROM {_TABLE_NAME} " f"GROUP BY annotator_id" ) pos_count = 0 neg_count = 0 for row in signal_rows: meta_raw = row["metadata"] if isinstance(meta_raw, str): try: meta = json.loads(meta_raw) except (json.JSONDecodeError, TypeError): meta = {} else: meta = meta_raw or {} sig = meta.get("signal", "") if sig == "POSITIVE": pos_count += 1 elif sig == "NEGATIVE": neg_count += 1 annotator_dist: Dict[str, int] = { row["annotator_id"]: int(row["cnt"]) for row in annotator_rows } pct_positive = (pos_count / total) if total > 0 else 0.0 pct_negative = (neg_count / total) if total > 0 else 0.0 return { "ready": total >= _DPO_MINIMUM_PAIRS, "pair_count": total, "minimum_required": _DPO_MINIMUM_PAIRS, "pct_positive": pct_positive, "pct_negative": pct_negative, "annotator_dist": annotator_dist, } # --------------------------------------------------------------------------- # Re-export helper for external callers (Story 5.06) # --------------------------------------------------------------------------- __all__ = [ "FeedbackCollector", "_generate_fallback_response", "_TABLE_NAME", "_INTERACTION_KEY_PREFIX", "_INTERACTION_TTL_SECONDS", "_DPO_MINIMUM_PAIRS", "_DEFAULT_ANNOTATOR", ]