Upload 8 files

src/services/cache_service.py

@@ -27,9 +27,9 @@ class CacheService:
             return None
         return value
 
-    def set(self, key: str, value: Any, ttl: int = 3600) -> None:
-        self._prune()
-        self._cache[key] = (value, time.time() + ttl)
+    def set(self, key: str, value: Any, ttl: int = 3600) -> None:
+        self._cache[key] = (value, time.time() + ttl)
+        self._prune()
 
     def clear(self) -> None:
         self._cache.clear()

src/services/html_sanitizer.py

@@ -8,7 +8,8 @@ from urllib.parse import urlparse
 
 
 ALLOWED_TAGS = {"a", "b", "blockquote", "br", "code", "details", "div", "em", "h3", "h4", "hr", "i", "li", "ol", "p", "pre", "span", "strong", "summary", "table", "tbody", "td", "th", "thead", "tr", "ul"}
-VOID_TAGS = {"br", "hr"}
+VOID_TAGS = {"br", "hr"}
+HTML_VOID_TAGS = {"area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "wbr"}
 DROP_CONTENT_TAGS = {"button", "form", "iframe", "input", "math", "object", "script", "style", "svg"}
 GLOBAL_ATTRIBUTES = {"aria-controls", "aria-expanded", "aria-label", "class", "role"}
 TAG_ATTRIBUTES = {"a": {"href", "rel", "target", "title"}, "td": {"colspan", "rowspan"}, "th": {"colspan", "rowspan", "scope"}}
@@ -21,14 +22,27 @@ _MARKDOWN_LINK_RE = re.compile(
 
 
 def _normalise_url(value: str) -> str:
-    return unescape(value).strip().rstrip(".,;:!?)]}")
+    # Punctuation can be meaningful in a URL, especially article titles and
+    # signed query values. Only strip sentence punctuation when matching text.
+    return unescape(value).strip()
 
 
 def _safe_href(value: str, allowed_urls: set[str] | None = None) -> str | None:
     candidate = _normalise_url(value)
     if not candidate:
         return None
-    parsed = urlparse(candidate)
+    if any(character.isspace() or ord(character) < 32 or ord(character) == 127 for character in candidate):
+        return None
+    try:
+        parsed = urlparse(candidate)
+        if parsed.scheme.lower() == "https" and (
+            not parsed.hostname or parsed.username or parsed.password
+        ):
+            return None
+        # Evaluating a malformed authority/port may itself raise ValueError.
+        _ = parsed.port
+    except ValueError:
+        return None
     if parsed.scheme.lower() not in {"https", "mailto", "tel"}:
         return None
     if allowed_urls is not None and candidate not in allowed_urls:
@@ -64,9 +78,10 @@ class _Sanitizer(HTMLParser):
         )
 
     def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
-        tag = tag.lower()
-        if tag in DROP_CONTENT_TAGS:
-            self.drop_depth += 1
+        tag = tag.lower()
+        if tag in DROP_CONTENT_TAGS:
+            if tag not in HTML_VOID_TAGS:
+                self.drop_depth += 1
             return
         if self.drop_depth or tag not in ALLOWED_TAGS:
             return
@@ -90,12 +105,14 @@ class _Sanitizer(HTMLParser):
             rendered.append(' rel="noopener noreferrer"')
         self.output.append(f"<{tag}{''.join(rendered)}>")
 
-    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
-        self.handle_starttag(tag, attrs)
+    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.handle_starttag(tag, attrs)
+        if tag.lower() not in HTML_VOID_TAGS:
+            self.handle_endtag(tag)
 
     def handle_endtag(self, tag: str) -> None:
         tag = tag.lower()
-        if tag in DROP_CONTENT_TAGS:
+        if tag in DROP_CONTENT_TAGS and tag not in HTML_VOID_TAGS:
             if self.drop_depth:
                 self.drop_depth -= 1
             return
@@ -106,7 +123,10 @@ class _Sanitizer(HTMLParser):
         if not self.drop_depth:
             if self.allowed_urls is not None:
                 data = _VISIBLE_URL_RE.sub(
-                    lambda m: m.group(0) if _normalise_url(m.group(0)) in self.allowed_urls else "",
+                    lambda m: m.group(0) if (
+                        _normalise_url(m.group(0)) in self.allowed_urls or
+                        _normalise_url(m.group(0)).rstrip(".,;:!?)]}") in self.allowed_urls
+                    ) else "",
                     data,
                 )
             self.output.append(escape(data, quote=False))

src/services/metrics_service.py

@@ -7,15 +7,16 @@ Fixes applied:
 2. flush() merges in-memory state to disk atomically using a temp-file rename.
    Called by a background task in app.py every 30 seconds and on shutdown.
 3. threading.Lock replaced with asyncio.Lock — consistent with FastAPI's async model.
-4. _read() logs a warning (with raw content) on JSON parse failure instead of
-   silently returning zeros, so corrupt files are visible in logs.
-5. _write() uses os.replace (atomic rename) to prevent corrupt state on crash.
+4. Corrupt metrics are reported without copying file contents into logs.
+5. _write() uses os.replace (atomic rename) to prevent corrupt state on crash.
+6. Failed writes retain pending counts for the next flush.
 """
 import asyncio
 import json
 import logging
 import os
-import time
+import time
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, Dict
 
@@ -52,7 +53,7 @@ class MetricsService:
 
         # Seed disk file if it doesn't exist
         if not self.path.exists():
-            self._write_to_disk(dict(_DEFAULT_STATE))
+            self._write_to_disk(deepcopy(_DEFAULT_STATE))
 
     # ── Public API (sync, no I/O) ──────────────────────────────────────────────
 
@@ -69,7 +70,7 @@ class MetricsService:
         """Append a query record to the in-memory log (capped at _MAX_QUERY_LOG)."""
         import datetime
         record = {
-            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"),
+            "timestamp": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M:%S"),
             "question": (question or "")[:120],
             "intent": intent or "",
             "tool": tool or "",
@@ -82,9 +83,11 @@ class MetricsService:
         if len(self._query_log) > _MAX_QUERY_LOG:
             self._query_log = self._query_log[-_MAX_QUERY_LOG:]
 
-    def recent_queries(self, limit: int = 50) -> list:
-        """Return the most recent `limit` query records, newest first."""
-        return list(reversed(self._query_log[-limit:]))
+    def recent_queries(self, limit: int = 50) -> list:
+        """Return the most recent `limit` query records, newest first."""
+        if limit <= 0:
+            return []
+        return list(reversed(self._query_log[-limit:]))
 
     # ── Public API (sync, no I/O) ──────────────────────────────────────────────
 
@@ -128,7 +131,9 @@ class MetricsService:
                 b = data.setdefault(bucket, {})
                 for k, v in keys.items():
                     b[k] = int(b.get(k, 0)) + v
-            self._write_to_disk(data)
+            if not self._write_to_disk(data):
+                # Preserve pending counts so a later successful flush can retry.
+                return
             self._counters.clear()
             self._buckets.clear()
             self._dirty = False
@@ -137,32 +142,44 @@ class MetricsService:
 
     def _read_from_disk(self) -> Dict[str, Any]:
         if not self.path.exists():
-            return dict(_DEFAULT_STATE)
-        try:
-            return json.loads(self.path.read_text(encoding="utf-8"))
+            return deepcopy(_DEFAULT_STATE)
+        try:
+            loaded = json.loads(self.path.read_text(encoding="utf-8"))
+            if not isinstance(loaded, dict):
+                raise ValueError("metrics state must be a JSON object")
+            data = deepcopy(_DEFAULT_STATE)
+            for key, value in loaded.items():
+                if isinstance(value, dict):
+                    if key in _DEFAULT_STATE and not isinstance(_DEFAULT_STATE[key], dict):
+                        continue
+                    data[key] = {
+                        str(k): v for k, v in value.items()
+                        if isinstance(v, int) and not isinstance(v, bool) and v >= 0
+                    }
+                elif isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0:
+                    if key not in {"intents", "errors"}:
+                        data[key] = value
+            return data
         except Exception as exc:
-            raw = ""
-            try:
-                raw = self.path.read_text(encoding="utf-8", errors="replace")[:200]
-            except Exception:
-                pass
-            logger.warning(
-                "MetricsService: failed to parse metrics file %s: %s | raw: %r",
-                self.path, exc, raw,
-            )
-            return dict(_DEFAULT_STATE)
-
-    def _write_to_disk(self, data: Dict[str, Any]) -> None:
+            logger.warning(
+                "MetricsService: failed to parse metrics file %s: %s",
+                self.path, exc,
+            )
+            return deepcopy(_DEFAULT_STATE)
+
+    def _write_to_disk(self, data: Dict[str, Any]) -> bool:
         data["last_updated"] = time.time()
         tmp = self.path.with_suffix(".tmp")
         try:
             tmp.write_text(
                 json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
             )
-            os.replace(tmp, self.path)  # atomic on POSIX / Windows
+            os.replace(tmp, self.path)  # atomic on POSIX / Windows
+            return True
         except Exception as exc:
             logger.error("MetricsService: failed to write metrics: %s", exc)
             try:
                 tmp.unlink(missing_ok=True)
-            except Exception:
-                pass
+            except Exception:
+                pass
+            return False

src/services/rag_service.py

@@ -1,69 +1,67 @@
-"""
-rag_service.py — LibBee v3.1
-
-Fixes applied:
-1. _find_corpus_index replaced with O(1) reverse-lookup dict (_content_to_idx).
-   Old implementation was O(n * fetch_k) per query — a major bottleneck.
-2. fetch_k corrected: max(top_k * 3, 15) — old expression max(top_k*3, top_k)
-   was always top_k*3 (max was a no-op). Now enforces a floor of 15.
-3. _content_to_idx rebuilt whenever corpus is loaded (init, cache load, rebuild).
-4. audit_knowledge_base removed from get_stats() — now only called explicitly
-   via a dedicated audit() method to avoid disk reads on every /rag-status poll.
-5. _write / cache saves use os.replace for atomic writes.
-"""
+"""Local keyword retrieval first, with an optional cached semantic index.
+
+Initialization never needs an API key to serve the local KB. Dense failures
+leave keyword search available; callers may run initialize in the background.
+"""
+import asyncio
 import hashlib
 import hmac
-import json
-import logging
-import os
-import re
-import time
-from pathlib import Path
-from typing import List, Optional
-
-import numpy as np
-
-from src.config import get_settings
-
-logger = logging.getLogger(__name__)
+import json
+import logging
+import os
+import re
+import time
+from pathlib import Path
+from typing import List
+
+from src.config import get_settings
+
+logger = logging.getLogger(__name__)
 CHUNK_WORDS = 260
 CHUNK_OVERLAP_WORDS = 40
-TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9_-]{1,}")
+EMBEDDING_MODEL = "text-embedding-3-small"
+CACHE_SCHEMA = 2
+TOKEN_RE = re.compile(r"[^\W_]+(?:[-_][^\W_]+)*", re.UNICODE)
 STOP_WORDS = {"about", "and", "are", "can", "for", "from", "how", "library", "please", "that", "the", "this", "what", "when", "where", "which", "with", "you"}
-
-
-class RAGService:
-    def __init__(self):
-        self.vectorstore = None
-        self.bm25 = None
-        self.bm25_corpus: List[str] = []
-        self.bm25_meta: List[dict] = []
-        self._embeddings = None
-        self._ready = False
-        self._kb_hash = ""
-        # O(1) reverse lookup: chunk_text -> corpus index
-        self._content_to_idx: dict[str, int] = {}
-
-    def is_ready(self) -> bool:
-        return self._ready
-
-    def _knowledge_dir(self) -> Path:
-        return get_settings().kb_dir
-
-    def _cache_dir(self) -> Path:
-        return get_settings().rag_cache_dir
-
-    def _hash_knowledge_base(self, txt_files: List[Path]) -> str:
-        digest = hashlib.sha256()
-        for path in txt_files:
-            stat = path.stat()
-            digest.update(path.name.encode("utf-8"))
-            digest.update(str(stat.st_mtime_ns).encode("utf-8"))
-            digest.update(str(stat.st_size).encode("utf-8"))
-        return digest.hexdigest()
-
+HEADER_RE = re.compile(r"^(?:SOURCE|TITLE|AUTHORITY|LAST_REVIEWED|LAST_UPDATED):", re.IGNORECASE)
+
+
+class RAGService:
+    def __init__(self):
+        self.vectorstore = None
+        self.bm25 = None
+        self.bm25_corpus: List[str] = []
+        self.bm25_meta: List[dict] = []
+        self._embeddings = None
+        self._ready = False
+        self._kb_hash = ""
+        self._status = "not_started"
+        self._content_to_idx: dict[str, int] = {}
+        self._init_lock = asyncio.Lock()
+
+    def is_ready(self) -> bool:
+        return self._ready
+
+    def _knowledge_dir(self) -> Path:
+        return get_settings().kb_dir
+
+    def _cache_dir(self) -> Path:
+        return get_settings().rag_cache_dir
+
+    def _hash_knowledge_base(self, txt_files: List[Path]) -> str:
+        # Deployments commonly reset mtimes. Hash content plus indexing settings
+        # so unchanged data can reuse persisted embeddings across image rebuilds.
+        digest = hashlib.sha256()
+        digest.update(f"{CACHE_SCHEMA}:{EMBEDDING_MODEL}:{CHUNK_WORDS}:{CHUNK_OVERLAP_WORDS}".encode())
+        for path in sorted(txt_files):
+            digest.update(path.name.encode("utf-8"))
+            digest.update(b"\0")
+            digest.update(path.read_bytes())
+            digest.update(b"\0")
+        return digest.hexdigest()
+
     def _cache_paths(self) -> tuple[Path, Path, Path]:
-        cache_dir = self._cache_dir()
+        cache_dir = self._cache_dir()
         return cache_dir / "faiss_index", cache_dir / "bm25_cache.json", cache_dir / "kb_state.json"
 
     def _cache_signature(self, faiss_dir: Path, bm25_path: Path) -> str:
@@ -79,259 +77,261 @@ class RAGService:
 
     @staticmethod
     def _tokens(value: str) -> List[str]:
-        return TOKEN_RE.findall((value or "").lower())
-
-    def _build_reverse_index(self) -> None:
-        """Build O(1) content -> index lookup. Call after any corpus change."""
-        self._content_to_idx = {
-            chunk.strip(): i for i, chunk in enumerate(self.bm25_corpus)
-        }
-
-    async def initialize(self, openai_api_key: str) -> None:
-        t0 = time.time()
-        logger.info("RAGService: starting initialization")
-        settings = get_settings()
-        knowledge_dir = settings.kb_dir
-        txt_files = sorted(knowledge_dir.glob("*.txt")) if knowledge_dir.exists() else []
-        self._kb_hash = self._hash_knowledge_base(txt_files) if txt_files else ""
-
-        try:
-            from langchain_openai import OpenAIEmbeddings
-            from langchain_community.vectorstores import FAISS
-            from rank_bm25 import BM25Okapi
-
-            self._embeddings = OpenAIEmbeddings(
-                model="text-embedding-3-small", openai_api_key=openai_api_key
-            )
-            if self._try_load_cached_indexes(FAISS, BM25Okapi):
-                self._build_reverse_index()
-                self._ready = True
-                logger.info(
-                    "RAGService: loaded cached indexes — %d chunks", len(self.bm25_corpus)
-                )
-                return
-
-            chunks, metadatas = self._load_and_chunk_all()
-            if not chunks:
-                logger.error("RAGService: no chunks loaded — check knowledge_dir: %s", knowledge_dir)
-                return
-
-            self.vectorstore = await FAISS.afrom_texts(
-                chunks, self._embeddings, metadatas=metadatas
-            )
-            self.bm25_corpus = chunks
-            self.bm25_meta = metadatas
-            tokenized = [self._tokens(c) for c in chunks]
-            self.bm25 = BM25Okapi(tokenized)
-            self._build_reverse_index()
-            self._save_cached_indexes()
-            self._ready = True
-            logger.info(
-                "RAGService: ready — %d chunks indexed in %.1fs", len(chunks), time.time() - t0
-            )
-        except Exception as exc:
-            logger.error("RAGService initialization failed: %s", exc, exc_info=True)
-
-    def _try_load_cached_indexes(self, FAISS, BM25Okapi) -> bool:
-        faiss_dir, bm25_path, state_path = self._cache_paths()
-        if not (faiss_dir.exists() and bm25_path.exists() and state_path.exists()):
-            return False
-        try:
-            state = json.loads(state_path.read_text(encoding="utf-8"))
-            if state.get("kb_hash") != self._kb_hash:
-                logger.info("RAGService: KB changed (hash mismatch) — rebuilding indexes")
+        return TOKEN_RE.findall((value or "").casefold())
+
+    def _build_reverse_index(self) -> None:
+        self._content_to_idx = {chunk.strip(): i for i, chunk in enumerate(self.bm25_corpus)}
+
+    def _build_keyword_index(self):
+        from rank_bm25 import BM25Okapi
+        files = sorted(self._knowledge_dir().glob("*.txt"))
+        kb_hash = self._hash_knowledge_base(files)
+        chunks, metadata = self._load_and_chunk_all()
+        if not chunks:
+            raise ValueError("No usable knowledge-base chunks found")
+        bm25 = BM25Okapi([self._tokens(chunk) for chunk in chunks])
+        return kb_hash, chunks, metadata, bm25
+
+    async def initialize(self, openai_api_key: str = "", *, force_rebuild: bool = False) -> None:
+        async with self._init_lock:
+            started = time.monotonic()
+            self._status = "initializing"
+            try:
+                # File reads, tokenization, and heavy imports must not block the
+                # event loop serving /config, health checks, and chat requests.
+                kb_hash, chunks, metadata, bm25 = await asyncio.to_thread(self._build_keyword_index)
+                self.vectorstore = None
+                self._embeddings = None
+                self._kb_hash = kb_hash
+                self.bm25_corpus, self.bm25_meta, self.bm25 = chunks, metadata, bm25
+                self._build_reverse_index()
+                self._ready = True
+                self._status = "keyword_ready"
+                logger.info("RAG keyword index ready: %d chunks in %.2fs", len(chunks), time.monotonic() - started)
+            except Exception:
+                self._status = ("ready" if self.vectorstore is not None else "keyword_ready") if self._ready else "failed"
+                logger.exception("RAG keyword initialization failed; retaining any existing index")
+                return
+
+            if not openai_api_key:
+                return
+            self._status = "semantic_initializing"
+            try:
+                def load_embedding_tools():
+                    from langchain_openai import OpenAIEmbeddings
+                    from langchain_community.vectorstores import FAISS
+                    return OpenAIEmbeddings, FAISS
+                OpenAIEmbeddings, FAISS = await asyncio.to_thread(load_embedding_tools)
+                self._embeddings = OpenAIEmbeddings(
+                    model=EMBEDDING_MODEL, openai_api_key=openai_api_key,
+                    request_timeout=15, max_retries=1,
+                )
+                if not force_rebuild and await asyncio.to_thread(self._try_load_cached_indexes, FAISS):
+                    self._status = "ready"
+                    return
+                vectorstore = await asyncio.wait_for(
+                    FAISS.afrom_texts(chunks, self._embeddings, metadatas=metadata), timeout=90
+                )
+                self.vectorstore = vectorstore
+                self._status = "ready"
+                try:
+                    await asyncio.to_thread(self._save_cached_indexes)
+                except Exception:
+                    # An unwritable cache must not disable a working index.
+                    logger.exception("RAG cache save failed; retrieval remains available")
+                logger.info("RAG semantic index ready in %.2fs", time.monotonic() - started)
+            except asyncio.CancelledError:
+                self._status = "keyword_ready"
+                raise
+            except Exception:
+                self._status = "keyword_ready"
+                logger.exception("RAG semantic index unavailable; using keyword retrieval")
+
+    def _try_load_cached_indexes(self, FAISS, _BM25Okapi=None) -> bool:
+        faiss_dir, bm25_path, state_path = self._cache_paths()
+        if not (faiss_dir.exists() and bm25_path.exists() and state_path.exists()):
+            return False
+        try:
+            state = json.loads(state_path.read_text(encoding="utf-8"))
+            if state.get("kb_hash") != self._kb_hash or state.get("schema") != CACHE_SCHEMA:
                 return False
             expected = self._cache_signature(faiss_dir, bm25_path)
             if not expected or not hmac.compare_digest(str(state.get("signature", "")), expected):
-                logger.warning("RAGService: unsigned or modified cache ignored")
+                logger.warning("RAG unsigned or modified cache ignored")
+                return False
+            cached = json.loads(bm25_path.read_text(encoding="utf-8"))
+            if cached.get("corpus") != self.bm25_corpus or cached.get("meta") != self.bm25_meta:
                 return False
-            self.vectorstore = FAISS.load_local(
-                str(faiss_dir), self._embeddings, allow_dangerous_deserialization=True
-            )
-            bm25_cache = json.loads(bm25_path.read_text(encoding="utf-8"))
-            self.bm25_corpus = bm25_cache["corpus"]
-            self.bm25_meta = bm25_cache["meta"]
-            tokenized = [self._tokens(c) for c in self.bm25_corpus]
-            self.bm25 = BM25Okapi(tokenized)
-            return True
-        except Exception as exc:
-            logger.warning("Failed loading cached RAG index: %s — will rebuild", exc)
-            return False
-
-    def _save_cached_indexes(self) -> None:
-        if not self.vectorstore:
-            return
-        faiss_dir, bm25_path, state_path = self._cache_paths()
-        faiss_dir.mkdir(parents=True, exist_ok=True)
-        self.vectorstore.save_local(str(faiss_dir))
-
-        # Atomic write for bm25 cache
-        bm25_tmp = bm25_path.with_suffix(".tmp")
-        bm25_tmp.write_text(
-            json.dumps(
-                {"corpus": self.bm25_corpus, "meta": self.bm25_meta}, ensure_ascii=False
-            ),
-            encoding="utf-8",
-        )
-        os.replace(bm25_tmp, bm25_path)
-
-        # Atomic write for state
-        state_tmp = state_path.with_suffix(".tmp")
-        state_tmp.write_text(
-            json.dumps({"kb_hash": self._kb_hash, "chunk_count": len(self.bm25_corpus), "signature": self._cache_signature(faiss_dir, bm25_path)}),
-            encoding="utf-8",
-        )
-        os.replace(state_tmp, state_path)
-
-    def _load_and_chunk_all(self) -> tuple[List[str], List[dict]]:
-        all_chunks: List[str] = []
-        all_meta: List[dict] = []
-        knowledge_dir = self._knowledge_dir()
-        if not knowledge_dir.exists():
-            logger.error("Knowledge directory not found: %s", knowledge_dir)
-            return [], []
-        txt_files = sorted(knowledge_dir.glob("*.txt"))
-        for fpath in txt_files:
-            try:
-                text = fpath.read_text(encoding="utf-8").strip()
-                if not text:
-                    continue
-                source_url = ""
-                title = fpath.stem.replace("_", " ").replace("-", " ")
-                authority = "official"
-                reviewed = ""
-                for line in text.splitlines()[:8]:
-                    line = line.strip()
-                    if line.startswith("SOURCE:"):
-                        source_url = line.replace("SOURCE:", "").strip()
-                    elif line.startswith("TITLE:"):
-                        title = line.replace("TITLE:", "").strip()
-                    elif line.startswith("AUTHORITY:"):
-                        authority = line.replace("AUTHORITY:", "").strip()[:40]
-                    elif line.startswith("LAST_REVIEWED:"):
-                        reviewed = line.replace("LAST_REVIEWED:", "").strip()[:20]
-                for chunk in self._chunk_text(text):
-                    all_chunks.append(chunk)
-                    all_meta.append(
-                        {"source": source_url, "title": title, "filename": fpath.name,
-                         "authority": authority, "last_reviewed": reviewed}
-                    )
-            except Exception as exc:
-                logger.warning("Failed to load %s: %s", fpath.name, exc)
-        logger.info("RAGService: loaded %d chunks from %d files", len(all_chunks), len(txt_files))
-        return all_chunks, all_meta
-
-    def _chunk_text(self, text: str) -> List[str]:
-        chunks: List[str] = []
-        sections = re.split(r"\n(?=(?:TOPIC:|[A-Z][A-Z\s,/&()\-]+:)\s)", text)
-        for section in sections:
-            section = section.strip()
-            if not section or len(section) < 30:
-                continue
+            # FAISS uses pickle internally. Only load bytes authenticated with
+            # the origin secret; never load unsigned/user-supplied index files.
+            self.vectorstore = FAISS.load_local(
+                str(faiss_dir), self._embeddings, allow_dangerous_deserialization=True
+            )
+            return True
+        except Exception:
+            logger.warning("RAG cache could not be loaded; rebuilding", exc_info=True)
+            return False
+
+    def _save_cached_indexes(self) -> None:
+        if self.vectorstore is None or not get_settings().origin_token:
+            return
+        faiss_dir, bm25_path, state_path = self._cache_paths()
+        faiss_dir.mkdir(parents=True, exist_ok=True)
+        self.vectorstore.save_local(str(faiss_dir))
+        bm25_tmp = bm25_path.with_suffix(".tmp")
+        bm25_tmp.write_text(json.dumps({"corpus": self.bm25_corpus, "meta": self.bm25_meta}, ensure_ascii=False), encoding="utf-8")
+        os.replace(bm25_tmp, bm25_path)
+        state_tmp = state_path.with_suffix(".tmp")
+        state_tmp.write_text(json.dumps({
+            "schema": CACHE_SCHEMA, "kb_hash": self._kb_hash,
+            "embedding_model": EMBEDDING_MODEL, "chunk_count": len(self.bm25_corpus),
+            "signature": self._cache_signature(faiss_dir, bm25_path),
+        }), encoding="utf-8")
+        os.replace(state_tmp, state_path)
+
+    def _load_and_chunk_all(self) -> tuple[List[str], List[dict]]:
+        all_chunks, all_meta = [], []
+        seen: set[str] = set()
+        for path in sorted(self._knowledge_dir().glob("*.txt")):
+            try:
+                text = path.read_text(encoding="utf-8").strip()
+                if not text:
+                    continue
+                header, header_lines = {}, set()
+                lines = text.splitlines()
+                for index, line in enumerate(lines[:8]):
+                    if HEADER_RE.match(line.strip()):
+                        key, value = line.split(":", 1)
+                        header[key.strip().upper()] = value.strip()
+                        header_lines.add(index)
+                body = "\n".join(line for index, line in enumerate(lines) if index not in header_lines)
+                metadata = {
+                    "source": header.get("SOURCE", ""),
+                    "title": header.get("TITLE", path.stem.replace("_", " ")),
+                    "filename": path.name,
+                    "authority": header.get("AUTHORITY", "supplied_library_content"),
+                    "last_reviewed": header.get("LAST_REVIEWED", ""),
+                    "last_updated": header.get("LAST_UPDATED", ""),
+                }
+                for chunk in self._chunk_text(body):
+                    fingerprint = re.sub(r"\s+", " ", chunk).strip()
+                    if fingerprint in seen:
+                        continue
+                    seen.add(fingerprint)
+                    all_chunks.append(chunk)
+                    all_meta.append(dict(metadata))
+            except Exception:
+                logger.warning("Could not read knowledge file %s", path.name, exc_info=True)
+        return all_chunks, all_meta
+
+    def _chunk_text(self, text: str) -> List[str]:
+        # Source files use RESOURCE 01:, all-caps headings without colons,
+        # staff names, and === section markers. Keep each entry with its URL.
+        sections, current = [], []
+        for line in text.splitlines():
+            stripped = line.strip()
+            is_heading = bool(
+                re.match(r"^(?:TOPIC:|RESOURCE\s+\d+:|===)", stripped)
+                or (4 <= len(stripped) <= 100 and stripped.isupper()
+                    and any(char.isalpha() for char in stripped)
+                    and not stripped.startswith(("http", "SOURCE:", "TITLE:")))
+            )
+            if is_heading and current:
+                sections.append("\n".join(current).strip())
+                current = []
+            current.append(line)
+        if current:
+            sections.append("\n".join(current).strip())
+        chunks = []
+        for section in sections:
+            if len(section) < 30 or not self._tokens(section):
+                continue
             words = section.split()
             if len(words) <= CHUNK_WORDS:
                 chunks.append(section)
                 continue
-            step = CHUNK_WORDS - CHUNK_OVERLAP_WORDS
-            for start in range(0, len(words), step):
-                chunk = " ".join(words[start:start + CHUNK_WORDS]).strip()
-                if chunk:
-                    chunks.append(chunk)
+            heading = section.splitlines()[0].strip()
+            for start in range(0, len(words), CHUNK_WORDS - CHUNK_OVERLAP_WORDS):
+                chunk = " ".join(words[start:start + CHUNK_WORDS])
+                if start and len(heading) < 100:
+                    chunk = heading + "\n" + chunk
+                chunks.append(chunk)
                 if start + CHUNK_WORDS >= len(words):
                     break
-        return chunks
-
-    async def hybrid_search(self, query: str, top_k: int = 5, alpha: float = 0.6) -> List[dict]:
-        if not self._ready:
-            logger.warning("hybrid_search called before ready — returning empty results")
-            return []
-        try:
-            # fetch_k: retrieve 3x candidates, minimum floor of 15
-            fetch_k = max(top_k * 3, 15)
-
-            dense_results = await self.vectorstore.asimilarity_search_with_score(
-                query, k=fetch_k
-            )
-            tokenized_query = self._tokens(query)
-            bm25_scores = self.bm25.get_scores(tokenized_query)
-            top_bm25_idx = np.argsort(bm25_scores)[::-1][:fetch_k].tolist()
-
-            # Weighted RRF fusion
-            # alpha controls dense list weight; (1-alpha) controls BM25 list weight
-            K = 60
-            rrf: dict[int, float] = {}
-            for rank, (doc, _score) in enumerate(dense_results):
-                idx = self._find_corpus_index(doc.page_content)
-                if idx >= 0:
-                    rrf[idx] = rrf.get(idx, 0.0) + alpha * (1.0 / (K + rank + 1))
-            for rank, idx in enumerate(top_bm25_idx):
-                rrf[idx] = rrf.get(idx, 0.0) + (1 - alpha) * (1.0 / (K + rank + 1))
-
-            sorted_idx = sorted(rrf, key=lambda i: rrf[i], reverse=True)[:top_k]
-            results = []
-            meaningful_query = {token for token in tokenized_query if token not in STOP_WORDS and len(token) > 2}
-            for idx in sorted_idx:
-                if idx < len(self.bm25_corpus):
-                    meta = self.bm25_meta[idx]
-                    chunk_tokens = set(self._tokens(self.bm25_corpus[idx]))
-                    lexical_overlap = len(meaningful_query & chunk_tokens)
-                    if meaningful_query and lexical_overlap == 0:
-                        continue
-                    results.append(
-                        {
-                            "content": self.bm25_corpus[idx],
-                            "source": meta.get("source", ""),
-                            "title": meta.get("title", ""),
-                            "filename": meta.get("filename", ""),
-                            "score": round(rrf[idx], 6),
-                            "lexical_overlap": lexical_overlap,
-                            "authority": meta.get("authority", "official"),
-                            "last_reviewed": meta.get("last_reviewed", ""),
-                        }
-                    )
-            return results
-        except Exception as exc:
-            logger.error("hybrid_search error: %s", exc, exc_info=True)
-            return []
-
-    def _find_corpus_index(self, content: str) -> int:
-        """O(1) reverse lookup using pre-built dict. Falls back to -1 if not found."""
-        return self._content_to_idx.get(content.strip(), -1)
-
-    def audit_knowledge_base(self) -> dict:
-        """Scan KB files for duplicate titles. Call explicitly — not on every stats poll."""
-        files = sorted(self._knowledge_dir().glob("*.txt"))
-        titles: dict[str, str] = {}
-        duplicates = []
-        for path in files:
-            title = path.stem
-            try:
-                text = path.read_text(encoding="utf-8", errors="ignore")[:500]
-                first_title = next(
-                    (
-                        line.replace("TITLE:", "").strip()
-                        for line in text.splitlines()
-                        if line.startswith("TITLE:")
-                    ),
-                    title,
-                )
-            except Exception:
-                first_title = title
-            if first_title in titles:
-                duplicates.append(
-                    {"title": first_title, "files": [titles[first_title], path.name]}
-                )
-            else:
-                titles[first_title] = path.name
-        return {"file_count": len(files), "duplicate_titles": duplicates[:20]}
-
-    def get_stats(self) -> dict:
-        """Lightweight stats — no disk reads beyond what's already in memory."""
-        return {
-            "ready": self._ready,
-            "chunk_count": len(self.bm25_corpus),
-            "knowledge_dir": str(self._knowledge_dir()),
-            "vectorstore_loaded": self.vectorstore is not None,
-            "bm25_loaded": self.bm25 is not None,
-            "kb_hash": self._kb_hash,
-        }
+        return chunks
+
+    async def hybrid_search(self, query: str, top_k: int = 5, alpha: float = 0.6) -> List[dict]:
+        if not self._ready or self.bm25 is None or not query.strip() or top_k <= 0:
+            return []
+        # Snapshot before awaiting dense search so an admin rebuild cannot mix
+        # old vector document ids with a new corpus/metadata list.
+        corpus, metadata, bm25 = self.bm25_corpus, self.bm25_meta, self.bm25
+        vectorstore, reverse = self.vectorstore, self._content_to_idx
+        tokens = self._tokens(query)
+        meaningful = {token for token in tokens if token not in STOP_WORDS and len(token) > 1}
+        if not meaningful:
+            return []
+        fetch_k = min(max(top_k * 3, 15), len(corpus))
+        alpha = min(1.0, max(0.0, alpha))
+        scores = bm25.get_scores(tokens)
+        lexical_indices = sorted(range(len(corpus)), key=lambda idx: float(scores[idx]), reverse=True)
+        lexical_indices = [idx for idx in lexical_indices
+                           if meaningful & set(self._tokens(corpus[idx]))][:fetch_k]
+        dense_results = []
+        if vectorstore is not None:
+            try:
+                dense_results = await asyncio.wait_for(
+                    vectorstore.asimilarity_search_with_score(query, k=fetch_k), timeout=4
+                )
+            except Exception:
+                logger.warning("Semantic retrieval unavailable; using keywords", exc_info=True)
+        rrf: dict[int, float] = {}
+        for rank, (doc, _score) in enumerate(dense_results):
+            idx = reverse.get(doc.page_content.strip(), -1)
+            if idx >= 0:
+                rrf[idx] = rrf.get(idx, 0.0) + alpha / (61 + rank)
+        lexical_weight = (1 - alpha) if dense_results else 1.0
+        for rank, idx in enumerate(lexical_indices):
+            rrf[idx] = rrf.get(idx, 0.0) + lexical_weight / (61 + rank)
+        results = []
+        # Filter before applying top_k: irrelevant dense hits must not crowd
+        # relevant keyword hits out of the candidate pool.
+        for idx in sorted(rrf, key=rrf.get, reverse=True):
+            overlap = len(meaningful & set(self._tokens(corpus[idx])))
+            if not overlap:
+                continue
+            results.append({
+                "content": corpus[idx], **metadata[idx], "score": round(rrf[idx], 6),
+                "lexical_overlap": overlap,
+                "retrieval_mode": "hybrid" if dense_results else "keyword",
+            })
+            if len(results) >= top_k:
+                break
+        return results
+
+    def _find_corpus_index(self, content: str) -> int:
+        return self._content_to_idx.get(content.strip(), -1)
+
+    def audit_knowledge_base(self) -> dict:
+        files = sorted(self._knowledge_dir().glob("*.txt"))
+        titles, duplicates = {}, []
+        for path in files:
+            try:
+                text = path.read_text(encoding="utf-8")[:500]
+                title = next((line.split(":", 1)[1].strip() for line in text.splitlines() if line.startswith("TITLE:")), path.stem)
+                if title in titles:
+                    duplicates.append({"title": title, "files": [titles[title], path.name]})
+                else:
+                    titles[title] = path.name
+            except OSError:
+                continue
+        return {"file_count": len(files), "duplicate_titles": duplicates[:20]}
+
+    def get_stats(self) -> dict:
+        return {
+            "ready": self._ready, "status": self._status,
+            "retrieval_mode": "hybrid" if self.vectorstore is not None else "keyword" if self._ready else "unavailable",
+            "chunk_count": len(self.bm25_corpus), "knowledge_dir": str(self._knowledge_dir()),
+            "vectorstore_loaded": self.vectorstore is not None, "bm25_loaded": self.bm25 is not None,
+            "kb_hash": self._kb_hash,
+        }

src/services/runtime_store.py

@@ -1,52 +1,57 @@
-"""
-runtime_store.py — LibBee v3.1
-
-Fixes applied:
-1. save() now uses atomic write-then-rename (os.replace) to prevent file
-   corruption on crash or mid-write failure.
-   os.replace is atomic on POSIX (Linux/macOS) and effectively atomic on
-   Windows (same filesystem). This protects feedback.jsonl, metrics.json,
-   and runtime_config.json from partial writes.
-"""
-import json
-import os
-import time
-from pathlib import Path
-from typing import Any, Dict, Optional
-
-
-class JsonRuntimeStore:
-    def __init__(self, path: Path, default: Optional[Dict[str, Any]] = None):
-        self.path = Path(path)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.default = default or {}
-
-    def load(self) -> Dict[str, Any]:
-        if not self.path.exists():
-            return dict(self.default)
-        try:
-            return json.loads(self.path.read_text(encoding="utf-8"))
-        except Exception:
-            return dict(self.default)
-
-    def save(self, data: Dict[str, Any]) -> None:
-        """Atomically write data to disk using a temp file + os.replace."""
-        tmp = self.path.with_suffix(".tmp")
-        try:
-            tmp.write_text(
-                json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
-            )
-            os.replace(tmp, self.path)
-        except Exception:
-            try:
-                tmp.unlink(missing_ok=True)
-            except Exception:
-                pass
-            raise
-
-    def update(self, **kwargs: Any) -> Dict[str, Any]:
-        data = self.load()
-        data.update(kwargs)
-        data["updated_at"] = time.time()
-        self.save(data)
-        return data
+"""Small atomic JSON store for a single-process LibBee deployment."""
+
+import copy
+import json
+import os
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+
+class JsonRuntimeStore:
+    def __init__(self, path: Path, default: Optional[Dict[str, Any]] = None):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.default = copy.deepcopy(default or {})
+
+    def load(self) -> Dict[str, Any]:
+        data = copy.deepcopy(self.default)
+        try:
+            loaded = json.loads(self.path.read_text(encoding="utf-8"))
+            if isinstance(loaded, dict):
+                data.update(loaded)
+        except (OSError, UnicodeError, json.JSONDecodeError):
+            pass
+        return data
+
+    def save(self, data: Dict[str, Any]) -> None:
+        """Replace a complete object with a unique, owner-only temporary file.
+
+        A shared fixed `.tmp` name allowed independent writers to overwrite or
+        unlink each other's temporary file. Unique files prevent that race;
+        callers still need a database for multi-process read/modify/write use.
+        """
+        if not isinstance(data, dict):
+            raise TypeError("runtime data must be a JSON object")
+        tmp = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w", encoding="utf-8", dir=self.path.parent,
+                prefix=f".{self.path.name}.", suffix=".tmp", delete=False,
+            ) as handle:
+                tmp = Path(handle.name)
+                json.dump(data, handle, ensure_ascii=False, indent=2, allow_nan=False)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(tmp, self.path)
+        finally:
+            if tmp is not None:
+                tmp.unlink(missing_ok=True)
+
+    def update(self, **kwargs: Any) -> Dict[str, Any]:
+        data = self.load()
+        data.update(kwargs)
+        data["updated_at"] = time.time()
+        self.save(data)
+        return data

src/services/staff_service.py

@@ -1,222 +1,195 @@
-"""
-staff_service.py — LibBee v3.1
-
-Fixes applied:
-1. match_staff_name token logic corrected: previously checked if every question
-   token was in the staff token set (inverted). Now checks if the staff member's
-   own name tokens are all present in the question token set (correct direction).
-2. Module-level build_staff_index() call removed — was running at import time,
-   causing the index to be built twice (once at import, once in lifespan).
-   app.py lifespan remains the single build trigger.
-3. Fuzzy name matching added as a fallback in match_staff_name using
-   difflib.get_close_matches (cutoff=0.75) to handle typos like "Nikkesh".
-4. _ROLE_INDEX guard added: match_staff_role auto-builds if index is empty,
-   defensive against out-of-order call scenarios.
-"""
-import difflib
-import re
-from typing import Dict, List, Optional
-
-STAFF_DIRECTORY = [
-    {
-        "full_name": "Dr. Abdulla Al Hefeiti",
-        "role": "Library Director / Assistant Provost, Libraries",
-        "email": "abdulla.alhefeiti@ku.ac.ae",
-        "phone": "+971 2 312 3331",
-        "expertise": "strategic and institutional matters, library leadership, and partnerships",
-        "tokens": ["abdulla", "hefeiti", "abdulla al hefeiti", "al hefeiti"],
-        "aliases": ["library director", "director", "assistant provost"],
-    },
-    {
-        "full_name": "Nikesh Narayanan",
-        "role": "Research & Access Services Librarian",
-        "email": "nikesh.narayanan@ku.ac.ae",
-        "phone": "+971 2 312 3980",
-        "expertise": "research support, Open Access publishing, Khazna repository, ORCID, Scopus, research impact, AI tools for research, bibliometrics, and scholarly communication",
-        "tokens": ["nikesh", "narayanan", "nikesh narayanan"],
-        "aliases": ["research librarian", "research support librarian"],
-    },
-    {
-        "full_name": "Rani Anand",
-        "role": "E-Resources Librarian",
-        "email": "rani.anand@ku.ac.ae",
-        "phone": "+971 2 312 3935",
-        "expertise": "database access problems, e-resources troubleshooting, remote access, vendor issues, patents, e-books, and bibliometrics",
-        "tokens": ["rani", "anand", "rani anand"],
-        "aliases": ["e-resources librarian", "eresources librarian", "database librarian"],
-    },
-    {
-        "full_name": "Jason Fetty",
-        "role": "Medical Librarian",
-        "email": "jason.fetty@ku.ac.ae",
-        "phone": "+971 2 312 4722",
-        "expertise": "medical and health sciences research, systematic reviews, PubMed, Embase, CINAHL, UpToDate, and clinical databases",
-        "tokens": ["jason", "fetty", "jason fetty"],
-        "aliases": ["medical librarian", "medical library", "health sciences librarian"],
-    },
-    {
-        "full_name": "Walter Brian Hall",
-        "role": "Digital & Technology Services Librarian / Systems Librarian",
-        "email": "walter.hall@ku.ac.ae",
-        "phone": "+971 2 312 3163",
-        "expertise": "library website, systems, technology, digital infrastructure, ORCID, STEAM, coding, and Open Access support",
-        "tokens": ["walter", "brian", "hall", "walter brian hall", "walter hall", "brian hall"],
-        "aliases": ["systems librarian", "technology librarian", "digital librarian"],
-    },
-    {
-        "full_name": "Alia Al-Harrasi",
-        "role": "Manager, Technical Services",
-        "email": "alia.alharrasi@ku.ac.ae",
-        "phone": "+971 2 312 3180",
-        "expertise": "cataloguing, metadata, acquisitions processing, and technical services",
-        "tokens": ["alia", "alia al-harrasi", "alia alharrasi", "al harrasi", "alharrasi"],
-        "aliases": ["technical services", "acquisitions processing", "cataloguing"],
-    },
-]
-
-_STOP_WORDS = {
-    "who", "is", "are", "the", "a", "an", "can", "help", "me", "tell",
-    "about", "find", "contact", "email", "phone", "number", "i", "need",
-    "to", "speak", "with", "please", "get", "in", "touch", "reach", "how",
-    "do", "what", "which", "where", "for", "of", "best", "person", "librarian",
-}
-
-_CONTACT_INTENT_RE = re.compile(
-    r"\b(contact|email|phone|number|who is|who's|who handles|who can help|best person|best librarian|"
-    r"which librarian|talk to|speak to|reach|appointment|book an appointment|schedule)",
-    re.IGNORECASE,
-)
-
-_ROLE_INDEX: dict[str, Dict] = {}
-
-
-def build_staff_index() -> None:
-    """Build role alias → staff dict. Called once from app.py lifespan."""
-    global _ROLE_INDEX
-    _ROLE_INDEX = {}
-    for staff in STAFF_DIRECTORY:
-        for alias in [staff["full_name"], *staff.get("aliases", [])]:
-            key = re.sub(r"[^a-z0-9]+", " ", alias.lower()).strip()
-            _ROLE_INDEX[key] = staff
-
-
-def _normalize(text: str) -> List[str]:
-    tokens = re.sub(r"[^a-z0-9 ]+", " ", (text or "").lower()).split()
-    return [t for t in tokens if t not in _STOP_WORDS]
-
-
-def should_attempt_staff_lookup(question: str) -> bool:
-    q = (question or "").strip().lower()
-    if not q:
-        return False
-    if not _CONTACT_INTENT_RE.search(q):
-        return False
-    # Don't trigger on article/research queries that happen to contain contact words
-    if re.search(
-        r"\b(article|articles|paper|papers|study|studies|research article|research articles|literature)\b", q
-    ) and not re.search(
-        r"\b(who|contact|email|phone|which librarian|best person|who can help)\b", q
-    ):
-        return False
-    return True
-
-
-def match_staff_name(question: str) -> Optional[Dict]:
-    """
-    Match by name tokens. ALL of a staff member's name tokens must appear
-    in the question.
-
-    Single-token matches are only accepted if the token is distinctive
-    (≥6 chars and not a common word). This prevents short or common tokens
-    like 'hall', 'rani', 'brian', 'jason' from matching on unrelated queries.
-    Multi-token matches (e.g. 'nikesh narayanan', 'rani anand') always accepted.
-    """
-    if not should_attempt_staff_lookup(question):
-        return None
-
-    question_token_set = set(_normalize(question))
-    if not question_token_set:
-        return None
-
-    # Tokens that are too common/short to be used as sole match criteria
-    _WEAK_TOKENS = {
-        "hall", "rani", "brian", "alia", "jason", "walter",
-        "anand", "fetty",
-    }
-
-    for staff in STAFF_DIRECTORY:
-        staff_name_tokens: set[str] = set()
-        for tok in staff.get("tokens", []):
-            staff_name_tokens.update(_normalize(tok))
-        if not staff_name_tokens:
-            continue
-        if not staff_name_tokens.issubset(question_token_set):
-            continue
-        # All tokens matched — now check if match is strong enough
-        matched = staff_name_tokens & question_token_set
-        if len(matched) >= 2:
-            return staff
-        # Single token match — only accept if distinctive
-        sole = next(iter(matched))
-        if len(sole) >= 6 and sole not in _WEAK_TOKENS:
-            return staff
-
-    # Fuzzy fallback for typos (e.g. "Nikkesh", "Al Harrasi")
-    return _fuzzy_staff_match(question)
-
-
-def _fuzzy_staff_match(question: str) -> Optional[Dict]:
-    """
-    Fuzzy match individual question tokens against staff full names using difflib.
-    Whole-question comparison is skipped — it never matches because a full sentence
-    has near-zero similarity to a short name string.
-    """
-    all_names = [s["full_name"].lower() for s in STAFF_DIRECTORY]
-    for token in _normalize(question):
-        if len(token) < 4:
-            continue
-        token_matches = difflib.get_close_matches(
-            token, all_names, n=1, cutoff=0.8
-        )
-        if token_matches:
-            return next(
-                (s for s in STAFF_DIRECTORY if s["full_name"].lower() == token_matches[0]),
-                None,
-            )
-    return None
-
-
-def match_staff_role(question: str) -> Optional[Dict]:
-    """Match by role alias. Auto-builds index if empty (defensive guard)."""
-    if not should_attempt_staff_lookup(question):
-        return None
-    if not _ROLE_INDEX:
-        build_staff_index()
-    q = re.sub(r"[^a-z0-9]+", " ", question.lower()).strip()
-    # Longest alias first — prevents "librarian" matching before "medical librarian"
-    for alias, staff in sorted(_ROLE_INDEX.items(), key=lambda item: len(item[0]), reverse=True):
-        if len(alias) > 3 and alias in q:
-            return staff
-    return None
-
-
-def staff_name_answer(staff: Dict) -> str:
-    return (
-        f"<strong>{staff['full_name']}</strong> is the <strong>{staff['role']}</strong>.<br><br>"
-        f"They can help with: {staff['expertise']}.<br><br>"
-        f"📧 <a href=\"mailto:{staff['email']}\">{staff['email']}</a><br>"
-        f"📞 <a href=\"tel:{staff['phone'].replace(' ', '')}\">{staff['phone']}</a>"
-    )
-
-
-def staff_role_answer(staff: Dict, question: str) -> str:
-    return (
-        f"For help with that, the best person to contact is <strong>{staff['full_name']}</strong> — "
-        f"<strong>{staff['role']}</strong>.<br><br>"
-        f"They can help with: {staff['expertise']}.<br><br>"
-        f"📧 <a href=\"mailto:{staff['email']}\">{staff['email']}</a><br>"
-        f"📞 <a href=\"tel:{staff['phone'].replace(' ', '')}\">{staff['phone']}</a>"
-    )
-
-# NOTE: build_staff_index() is NOT called here at module level.
-# It is called once from app.py lifespan to avoid double-build.
+"""Deterministic staff matching using the supplied KU directory snapshot.
+
+Live roles and contact details still require checking against the official
+staff page; these helpers do not perform a live directory lookup.
+"""
+import difflib
+import re
+from typing import Dict, List, Optional
+
+STAFF_DIRECTORY = [
+    {
+        "full_name": "Dr. Abdulla Al Hefeiti",
+        "role": "Library Director / Assistant Provost, Libraries",
+        "email": "abdulla.alhefeiti@ku.ac.ae",
+        "phone": "+971 2 312 3331",
+        "expertise": "strategic and institutional matters, library leadership, and partnerships",
+        "tokens": ["abdulla", "hefeiti", "abdulla al hefeiti", "al hefeiti"],
+        "aliases": ["library director", "director", "assistant provost"],
+    },
+    {
+        "full_name": "Nikesh Narayanan",
+        "role": "Research & Access Services Librarian",
+        "email": "nikesh.narayanan@ku.ac.ae",
+        "phone": "+971 2 312 3980",
+        "expertise": "research support, Open Access publishing, Khazna repository, ORCID, Scopus, research impact, AI tools for research, bibliometrics, and scholarly communication",
+        "tokens": ["nikesh", "narayanan", "nikesh narayanan"],
+        "aliases": ["research librarian", "research support librarian"],
+    },
+    {
+        "full_name": "Rani Anand",
+        "role": "E-Resources Librarian",
+        "email": "rani.anand@ku.ac.ae",
+        "phone": "+971 2 312 3935",
+        "expertise": "database access problems, e-resources troubleshooting, remote access, vendor issues, patents, e-books, and bibliometrics",
+        "tokens": ["rani", "anand", "rani anand"],
+        "aliases": ["e-resources librarian", "eresources librarian", "database librarian"],
+    },
+    {
+        "full_name": "Jason Fetty",
+        "role": "Medical Librarian",
+        "email": "jason.fetty@ku.ac.ae",
+        "phone": "+971 2 312 4722",
+        "expertise": "medical and health sciences research, systematic reviews, PubMed, Embase, CINAHL, UpToDate, and clinical databases",
+        "tokens": ["jason", "fetty", "jason fetty"],
+        "aliases": ["medical librarian", "medical library", "health sciences librarian"],
+    },
+    {
+        "full_name": "Walter Brian Hall",
+        "role": "Digital & Technology Services Librarian / Systems Librarian",
+        "email": "walter.hall@ku.ac.ae",
+        "phone": "+971 2 312 3163",
+        "expertise": "library website, systems, technology, digital infrastructure, ORCID, STEAM, coding, and Open Access support",
+        "tokens": ["walter", "brian", "hall", "walter brian hall", "walter hall", "brian hall"],
+        "aliases": ["systems librarian", "technology librarian", "digital librarian"],
+    },
+    {
+        "full_name": "Alia Al-Harrasi",
+        "role": "Manager, Technical Services",
+        "email": "alia.alharrasi@ku.ac.ae",
+        "phone": "+971 2 312 3180",
+        "expertise": "cataloguing, metadata, acquisitions processing, and technical services",
+        "tokens": ["alia", "alia al-harrasi", "alia alharrasi", "al harrasi", "alharrasi"],
+        "aliases": ["technical services", "acquisitions processing", "cataloguing"],
+    },
+]
+
+_STOP_WORDS = {
+    "who", "is", "are", "the", "a", "an", "can", "help", "me", "tell",
+    "about", "find", "contact", "email", "phone", "number", "i", "need",
+    "to", "speak", "with", "please", "get", "in", "touch", "reach", "how",
+    "do", "what", "which", "where", "for", "of", "best", "person", "librarian",
+}
+
+_CONTACT_INTENT_RE = re.compile(
+    r"\b(contact|email|phone|number|who is|who's|who handles|who can help|best person|best librarian|"
+    r"which librarian|talk to|speak to|reach|appointment|book an appointment|schedule)",
+    re.IGNORECASE,
+)
+
+_ROLE_INDEX: dict[str, Dict] = {}
+
+
+def build_staff_index() -> None:
+    """Build role alias → staff dict. Called once from app.py lifespan."""
+    global _ROLE_INDEX
+    _ROLE_INDEX = {}
+    for staff in STAFF_DIRECTORY:
+        for alias in [staff["full_name"], *staff.get("aliases", [])]:
+            key = re.sub(r"[^a-z0-9]+", " ", alias.lower()).strip()
+            _ROLE_INDEX[key] = staff
+
+
+def _normalize(text: str) -> List[str]:
+    tokens = re.sub(r"[^a-z0-9 ]+", " ", (text or "").lower()).split()
+    return [t for t in tokens if t not in _STOP_WORDS]
+
+
+def should_attempt_staff_lookup(question: str) -> bool:
+    q = (question or "").strip().lower()
+    if not q:
+        return False
+    if not _CONTACT_INTENT_RE.search(q):
+        return False
+    # Don't trigger on article/research queries that happen to contain contact words
+    if re.search(
+        r"\b(article|articles|paper|papers|study|studies|research article|research articles|literature)\b", q
+    ) and not re.search(
+        r"\b(who|contact|email|phone|which librarian|best person|who can help)\b", q
+    ):
+        return False
+    return True
+
+
+def match_staff_name(question: str) -> Optional[Dict]:
+    """Match alternative spellings independently, without combining aliases.
+
+    A contact request can use a first name, surname or full-name alias. Common
+    words such as "hall" cannot identify a person on their own. Ambiguous
+    matches fall through to the KB instead of choosing the first staff entry.
+    """
+    if not should_attempt_staff_lookup(question):
+        return None
+    question_tokens = set(_normalize(question))
+    weak = {"al", "dr", "hall", "brian"}
+    candidates = []
+    for staff in STAFF_DIRECTORY:
+        strength = 0
+        for alias in [staff["full_name"], *staff.get("tokens", [])]:
+            tokens = set(_normalize(alias)) - {"dr"}
+            if tokens and tokens.issubset(question_tokens):
+                if len(tokens) >= 2 or next(iter(tokens)) not in weak:
+                    strength = max(strength, len(tokens))
+        if strength:
+            candidates.append(staff)
+    if len(candidates) == 1:
+        return candidates[0]
+    if candidates:
+        return None
+    return _fuzzy_staff_match(question)
+
+
+def _fuzzy_staff_match(question: str) -> Optional[Dict]:
+    """Compare probable name words with name words, not full sentences."""
+    candidates = {}
+    for token in _normalize(question):
+        if len(token) < 6:
+            continue
+        for staff in STAFF_DIRECTORY:
+            distinctive = {
+                word for alias in staff.get("tokens", []) for word in _normalize(alias)
+                if len(word) >= 6 and word not in {"librarian", "library"}
+            }
+            if difflib.get_close_matches(token, distinctive, n=1, cutoff=0.88):
+                candidates[staff["email"]] = staff
+    return next(iter(candidates.values())) if len(candidates) == 1 else None
+
+
+def match_staff_role(question: str) -> Optional[Dict]:
+    """Match by role alias. Auto-builds index if empty (defensive guard)."""
+    if not should_attempt_staff_lookup(question):
+        return None
+    if not _ROLE_INDEX:
+        build_staff_index()
+    q = re.sub(r"[^a-z0-9]+", " ", question.lower()).strip()
+    # Longest alias first — prevents "librarian" matching before "medical librarian"
+    for alias, staff in sorted(_ROLE_INDEX.items(), key=lambda item: len(item[0]), reverse=True):
+        if len(alias) > 3 and re.search(r"(?<!\w)" + re.escape(alias) + r"(?!\w)", q):
+            return staff
+    return None
+
+
+def staff_name_answer(staff: Dict) -> str:
+    return (
+        f"<strong>{staff['full_name']}</strong> is the <strong>{staff['role']}</strong>.<br><br>"
+        f"They can help with: {staff['expertise']}.<br><br>"
+        f"📧 <a href=\"mailto:{staff['email']}\">{staff['email']}</a><br>"
+        f"📞 <a href=\"tel:{staff['phone'].replace(' ', '')}\">{staff['phone']}</a>"
+        '<br><br>Contact details are from the supplied directory. '
+        '<a href="https://library.ku.ac.ae/librarystaff" target="_blank">Check the current staff directory</a>.'
+    )
+
+
+def staff_role_answer(staff: Dict, question: str) -> str:
+    return (
+        f"For help with that, the best person to contact is <strong>{staff['full_name']}</strong> — "
+        f"<strong>{staff['role']}</strong>.<br><br>"
+        f"They can help with: {staff['expertise']}.<br><br>"
+        f"📧 <a href=\"mailto:{staff['email']}\">{staff['email']}</a><br>"
+        f"📞 <a href=\"tel:{staff['phone'].replace(' ', '')}\">{staff['phone']}</a>"
+        '<br><br>Contact details are from the supplied directory. '
+        '<a href="https://library.ku.ac.ae/librarystaff" target="_blank">Check the current staff directory</a>.'
+    )
+
+# NOTE: build_staff_index() is NOT called here at module level.
+# It is called once from app.py lifespan to avoid double-build.