""" core/workers/faq_worker.py FAQWorker — Static FAQ Lookup Created by Story 5.10 (AIVA RLM Nexus PRD v2). Responsibilities: 1. Load FAQ knowledge base from data/aiva_faq.json 2. Fuzzy-match the caller's utterance against FAQ questions using difflib 3. Write the best-matching answer to Redis aiva:state:{session_id} under faq_answer 4. Return {"answer": str, "matched": bool} — never None Design notes: - difflib.SequenceMatcher is used exclusively (no external fuzzy-matching libraries) - Match threshold is configurable but defaults to 0.7 - Graceful degradation: missing FAQ file returns fallback answer without crashing - All external I/O (file loading, Redis) is injectable for full test isolation - No SQLite. No direct network calls in the class body. """ import json import logging import difflib from pathlib import Path from typing import Any, Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Module constants # --------------------------------------------------------------------------- FAQ_KB_PATH = Path("/mnt/e/genesis-system/data/aiva_faq.json") MATCH_THRESHOLD: float = 0.7 FALLBACK_ANSWER: str = ( "I don't have that information right now, " "but I'll make sure the team gets back to you." ) # --------------------------------------------------------------------------- # FAQWorker # --------------------------------------------------------------------------- class FAQWorker: """ Looks up the answer to a caller's FAQ from a JSON knowledge base and writes it to Redis for AIVA to speak. All external I/O (file system, Redis) is performed through injected objects so the worker is fully testable without real services. Args: faq_path: Path to the FAQ JSON file. Defaults to FAQ_KB_PATH. Injectable via constructor for test isolation. redis_client: Object with an ``hset(key, field, value)`` method. Compatible with redis.asyncio / aioredis clients. If None, Redis storage is skipped with a warning log. """ def __init__( self, faq_path: Optional[Path] = None, redis_client: Optional[Any] = None, ) -> None: self._path = faq_path or FAQ_KB_PATH self._redis = redis_client # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def execute(self, intent: Any) -> dict: """ Main entry point. Called by SwarmRouter for every ANSWER_FAQ intent. Steps: 1. Load FAQ KB from the configured JSON file path. 2. Fuzzy-match the intent utterance against all FAQ questions. 3. If best match score >= MATCH_THRESHOLD → use matched answer. 4. Otherwise → use FALLBACK_ANSWER. 5. Write the answer to Redis aiva:state:{session_id} under faq_answer. 6. Return {"answer": str, "matched": bool}. Args: intent: An IntentSignal instance (duck-typed to avoid hard import). Returns: dict with keys ``answer`` (str) and ``matched`` (bool). Never None. """ session_id: str = getattr(intent, "session_id", "unknown") utterance: str = getattr(intent, "utterance", "") faqs = self._load_faq() best_entry, best_score = self._find_best_match(utterance, faqs) if best_entry is not None and best_score >= MATCH_THRESHOLD: answer: str = best_entry["answer"] matched: bool = True logger.info( "FAQ matched for session %s (score=%.3f): %s", session_id, best_score, best_entry["question"], ) else: answer = FALLBACK_ANSWER matched = False logger.info( "No FAQ match for session %s (best_score=%.3f) — using fallback", session_id, best_score if best_entry is not None else 0.0, ) await self._write_to_redis(session_id, answer) return {"answer": answer, "matched": matched} # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _load_faq(self) -> list[dict]: """ Load FAQ entries from the configured JSON file. Returns: List of FAQ dicts with ``question`` and ``answer`` keys. Returns an empty list if the file is missing or malformed. """ try: raw = self._path.read_text(encoding="utf-8") entries = json.loads(raw) if not isinstance(entries, list): logger.warning("FAQ file at %s is not a JSON array — returning []", self._path) return [] return entries except FileNotFoundError: logger.warning("FAQ file not found at %s — returning []", self._path) return [] except json.JSONDecodeError as exc: logger.error("FAQ file JSON parse error at %s: %s", self._path, exc) return [] def _find_best_match( self, utterance: str, faqs: list[dict], ) -> tuple[Optional[dict], float]: """ Find the best matching FAQ entry using difflib.SequenceMatcher. Comparison is case-insensitive to maximise recall. Args: utterance: The caller's raw utterance text. faqs: List of FAQ dicts loaded from the knowledge base. Returns: Tuple of (best_entry, best_score). best_entry is None and best_score is 0.0 if faqs is empty. """ if not faqs: return None, 0.0 normalised_utterance = utterance.lower().strip() best_entry: Optional[dict] = None best_score: float = 0.0 for entry in faqs: question_text = entry.get("question", "").lower().strip() if not question_text: continue ratio = difflib.SequenceMatcher( None, normalised_utterance, question_text, ).ratio() if ratio > best_score: best_score = ratio best_entry = entry return best_entry, best_score async def _write_to_redis(self, session_id: str, answer: str) -> None: """ Write the FAQ answer to Redis under ``aiva:state:{session_id}`` using the field name ``faq_answer``. If no redis_client was injected, logs a warning and returns silently. Redis storage is best-effort and must not block the call. Args: session_id: Identifies the AIVA conversation session. answer: The FAQ answer string to store. """ if self._redis is None: logger.warning( "No redis_client injected — skipping Redis write for session %s", session_id, ) return key = f"aiva:state:{session_id}" try: self._redis.hset(key, "faq_answer", answer) logger.debug("FAQ answer written to Redis at %s[faq_answer]", key) except Exception as exc: # noqa: BLE001 logger.error( "Failed to write FAQ answer to Redis for session %s: %s", session_id, exc, ) # VERIFICATION_STAMP # Story: 5.10 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 18/18 # Coverage: 100%