Upload 40 files

.dockerignore

@@ -0,0 +1,18 @@
+.git
+.gitignore
+.env
+.env.*
+__pycache__/
+*.py[cod]
+*.log
+.DS_Store
+metadata_search_data/
+paper_summaries/
+PDFs/
+Figures/
+Summary/
+CSS/
+JS/
+Assets/
+Data/
+*.html

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+metadata_search/data/papers.csv filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,9 @@
+.env
+.env.*
+__pycache__/
+*.py[cod]
+*.log
+.DS_Store
+metadata_search_data/
+paper_summaries/
+runtime_data/

Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PORT=7860 \
+    HOST=0.0.0.0 \
+    WATERSCOPE_CORS=1 \
+    WATERSCOPE_CORS_ORIGIN=* \
+    METADATA_AUTO_REBUILD=0 \
+    METADATA_SEMANTIC_SEARCH=1 \
+    USE_MCP=0 \
+    WATERSCOPE_SPACE_ID=EnvironmentalAI/WaterScopeAI
+
+WORKDIR /app
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r /app/requirements.txt
+
+COPY . /app
+
+EXPOSE 7860
+
+CMD ["python", "-m", "metadata_search.app"]

README.md

@@ -1,11 +1,23 @@
 ---
 title: WaterScopeAI Search
-emoji: 🦀
-colorFrom: red
-colorTo: gray
+emoji: 🔎
+colorFrom: blue
+colorTo: green
 sdk: docker
+app_port: 7860
 pinned: false
-license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# WaterScopeAI Search System
+
+CPU-based hybrid publication search, retrieval, and MCP backend for WaterScope-AI.
+
+This Space is intentionally separated from the GPU model Space (`EnvironmentalAI/WaterScopeAI`). Large corpus assets and the canonical prebuilt Qdrant index will be connected from a Hugging Face Dataset repository in the next deployment phase.
+
+## Initial deployment test
+
+- `/` — service information
+- `/api/health` — backend/resource health
+- `/api/search?q=activated+sludge&top_k=10` — publication search
+
+The first package includes `papers.csv` and the paper manifest so BM25 search can be tested before the prebuilt Qdrant index and document assets are connected. `METADATA_AUTO_REBUILD=0` prevents the CPU Space from rebuilding the vector index automatically; if no prebuilt Qdrant index is present, search falls back to BM25.

backend/__init__.py

@@ -0,0 +1 @@
+

backend/chat_settings.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import os
+import re
+
+MAX_CHAT_PAPERS = int(os.getenv("MAX_CHAT_PAPERS", "4"))
+CHAT_SUMMARY_MAX_CHARS = int(os.getenv("CHAT_SUMMARY_MAX_CHARS", "14000"))
+
+_SECTION_HEADING = re.compile(r"^##\s+\d+\.", re.MULTILINE)
+_QUESTION_STOPWORDS = frozenset(
+    {
+        "about",
+        "also",
+        "based",
+        "compare",
+        "could",
+        "does",
+        "first",
+        "from",
+        "have",
+        "paper",
+        "papers",
+        "please",
+        "provide",
+        "range",
+        "second",
+        "source",
+        "summarize",
+        "summary",
+        "that",
+        "the",
+        "their",
+        "these",
+        "third",
+        "this",
+        "those",
+        "was",
+        "were",
+        "what",
+        "when",
+        "which",
+        "would",
+        "your",
+    }
+)
+_ORDINAL_PAPER = re.compile(
+    r"\b(?P<ordinal>first|1st|second|2nd|third|3rd|fourth|4th)\s+paper\b",
+    re.I,
+)
+_ORDINAL_INDEX = {
+    "first": 0,
+    "1st": 0,
+    "second": 1,
+    "2nd": 1,
+    "third": 2,
+    "3rd": 2,
+    "fourth": 3,
+    "4th": 3,
+}
+
+
+def focus_document_ids_for_question(message: str, document_ids: list[str]) -> list[str]:
+    """When the user names a specific paper, answer from that paper only."""
+    if len(document_ids) <= 1:
+        return document_ids
+    match = _ORDINAL_PAPER.search(message or "")
+    if not match:
+        return document_ids
+    index = _ORDINAL_INDEX.get(match.group("ordinal").lower())
+    if index is None or index >= len(document_ids):
+        return document_ids
+    return [document_ids[index]]
+
+
+def enhance_ordinal_question(message: str, sources: list[dict]) -> str:
+    """When the user says 'second paper', inject that paper's title into the prompt."""
+    if not _ORDINAL_PAPER.search(message or "") or len(sources) != 1:
+        return message
+    title = (sources[0].get("title") or "").strip()
+    if not title:
+        return message
+    return f"{message}\n\nAnswer specifically about the paper titled: {title}"
+
+
+def _question_terms(question: str) -> list[str]:
+    terms: list[str] = []
+    lowered = (question or "").lower()
+    if "ph" in lowered:
+        terms.append("ph")
+    for token in re.findall(r"[a-z0-9]+", lowered):
+        if token in _QUESTION_STOPWORDS or len(token) < 3:
+            continue
+        terms.append(token)
+    return list(dict.fromkeys(terms))
+
+
+def _extract_section(text: str, section_number: int, section_label: str) -> str:
+    pattern = re.compile(
+        rf"^##\s*{section_number}\.\s*{re.escape(section_label)}.*?(?=^##\s+\d+\.|\Z)",
+        re.MULTILINE | re.DOTALL | re.IGNORECASE,
+    )
+    match = pattern.search(text)
+    return match.group(0).strip() if match else ""
+
+
+def _matching_lines(text: str, terms: list[str]) -> str:
+    if not terms:
+        return ""
+    lines: list[str] = []
+    for line in text.splitlines():
+        lowered = line.lower()
+        if any(term in lowered for term in terms):
+            lines.append(line.rstrip())
+    return "\n".join(lines).strip()
+
+
+def _fit_blocks(blocks: list[str], budget: int) -> str:
+    selected: list[str] = []
+    used = 0
+    for block in blocks:
+        block = block.strip()
+        if not block:
+            continue
+        extra = len(block) + (2 if selected else 0)
+        if used + extra > budget:
+            remaining = budget - used - (2 if selected else 0)
+            if remaining > 120:
+                selected.append(block[:remaining].rstrip() + "…")
+            break
+        selected.append(block)
+        used += extra
+    return "\n\n".join(selected)
+
+
+def excerpt_summary(text: str, *, question: str = "", n_papers: int = 1) -> str:
+    """Prefer identity + key results (+ question matches) instead of blind head truncation."""
+    per_paper = max(2500, CHAT_SUMMARY_MAX_CHARS // max(n_papers, 1))
+    if len(text) <= per_paper:
+        return text
+
+    blocks: list[str] = []
+    identity = _extract_section(text, 1, "Paper Identity")
+    key_results = _extract_section(text, 5, "Key Results")
+    takeaway = _extract_section(text, 10, "Concise Takeaway")
+    terms = _question_terms(question)
+    matched = _matching_lines(text, terms)
+
+    if identity:
+        blocks.append(identity)
+    if key_results:
+        blocks.append(key_results)
+    elif matched:
+        blocks.append(matched)
+    if takeaway and takeaway not in blocks:
+        blocks.append(takeaway)
+    if matched and matched not in blocks:
+        blocks.append(matched)
+
+    excerpt = _fit_blocks(blocks, per_paper)
+    if excerpt:
+        return excerpt
+
+    return text[:per_paper] + "\n\n[Additional sections omitted for length.]"
+
+
+def truncate_summary(text: str, *, n_papers: int = 1, question: str = "") -> str:
+    return excerpt_summary(text, question=question, n_papers=n_papers)

backend/mcp_client.py

@@ -0,0 +1,68 @@
+"""Persistent MCP client for the local paper retrieval server."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+class PaperMCPClient:
+    """Keeps one MCP session open for retrieval tool calls."""
+
+    def __init__(self, project_root: Path | None = None):
+        self.project_root = project_root or PROJECT_ROOT
+        self._stdio_ctx = None
+        self._session_ctx = None
+        self._session: ClientSession | None = None
+
+    async def connect(self) -> None:
+        if self._session is not None:
+            return
+
+        server_script = os.getenv("MCP_SERVER_SCRIPT", "waterscope_retrieval_server.py")
+        server_params = StdioServerParameters(
+            command=sys.executable,
+            args=[str(self.project_root / server_script)],
+            cwd=str(self.project_root),
+            env={**os.environ},
+        )
+
+        self._stdio_ctx = stdio_client(server_params)
+        read, write = await self._stdio_ctx.__aenter__()
+        self._session_ctx = ClientSession(read, write)
+        self._session = await self._session_ctx.__aenter__()
+        await self._session.initialize()
+
+    async def close(self) -> None:
+        if self._session_ctx is not None:
+            await self._session_ctx.__aexit__(None, None, None)
+            self._session_ctx = None
+            self._session = None
+        if self._stdio_ctx is not None:
+            await self._stdio_ctx.__aexit__(None, None, None)
+            self._stdio_ctx = None
+
+    async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
+        if self._session is None:
+            raise RuntimeError("MCP client is not connected.")
+
+        payload = {key: value for key, value in (arguments or {}).items() if value is not None}
+        result = await self._session.call_tool(name, payload)
+
+        if result.isError:
+            message = result.content[0].text if result.content else "Unknown MCP tool error"
+            raise RuntimeError(message)
+
+        if not result.content:
+            return {}
+
+        text = result.content[0].text
+        return json.loads(text)

backend/mcp_sync.py

@@ -0,0 +1,85 @@
+"""Synchronous MCP client for Flask (stdio subprocess)."""
+
+from __future__ import annotations
+
+import asyncio
+import atexit
+import os
+import threading
+from typing import Any
+
+from backend.mcp_client import PaperMCPClient
+
+_client: PaperMCPClient | None = None
+_connect_lock = threading.Lock()
+_shutdown_registered = False
+
+_loop: asyncio.AbstractEventLoop | None = None
+_loop_thread: threading.Thread | None = None
+_loop_ready = threading.Event()
+
+
+def use_mcp() -> bool:
+    return os.getenv("USE_MCP", "1").lower() in {"1", "true", "yes"}
+
+
+def _start_loop_thread() -> asyncio.AbstractEventLoop:
+    global _loop, _loop_thread
+    if _loop is not None and _loop.is_running():
+        return _loop
+
+    def _runner() -> None:
+        global _loop
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        _loop = loop
+        _loop_ready.set()
+        loop.run_forever()
+
+    _loop_ready.clear()
+    _loop_thread = threading.Thread(target=_runner, daemon=True, name="mcp-async-loop")
+    _loop_thread.start()
+    _loop_ready.wait()
+    if _loop is None:
+        raise RuntimeError("MCP event loop failed to start.")
+    return _loop
+
+
+def _run_async(coro, *, timeout: float = 600):
+    loop = _start_loop_thread()
+    future = asyncio.run_coroutine_threadsafe(coro, loop)
+    return future.result(timeout=timeout)
+
+
+def ensure_mcp_connected() -> PaperMCPClient:
+    global _client, _shutdown_registered
+    if _client is not None and _client._session is not None:
+        return _client
+    with _connect_lock:
+        if _client is not None and _client._session is not None:
+            return _client
+        client = PaperMCPClient()
+        _run_async(client.connect())
+        _client = client
+        if not _shutdown_registered:
+            atexit.register(shutdown_mcp)
+            _shutdown_registered = True
+        return _client
+
+
+def shutdown_mcp() -> None:
+    global _client, _loop
+    if _client is None:
+        return
+    try:
+        _run_async(_client.close(), timeout=30)
+    except Exception:
+        pass
+    _client = None
+    if _loop is not None and _loop.is_running():
+        _loop.call_soon_threadsafe(_loop.stop)
+
+
+def call_mcp_tool(name: str, arguments: dict[str, Any] | None = None) -> Any:
+    client = ensure_mcp_connected()
+    return _run_async(client.call_tool(name, arguments))

backend/model.py

@@ -0,0 +1,13 @@
+"""WaterScope HF Space backend helpers."""
+
+from __future__ import annotations
+
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+
+def llm_backend_name() -> str:
+    return "waterscope"

backend/paper_linking.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+import hashlib
+import re
+
+from local_rag.summaries import load_manifest, summary_path
+
+_RP_SEGMENT_RE = re.compile(r";\s*")
+_FILENAME_AUTHORS_RE = re.compile(r"^(.+?)\s*-\s*\d{4}\s*-\s*")
+_ET_AL_RE = re.compile(r"\bet\s+al\b\.?", re.I)
+
+_NON_AUTHOR_PREFIXES = (
+    "not specified",
+    "not explicitly",
+    "not listed",
+    "not available",
+    "not provided",
+    "unknown",
+    "co-author",
+)
+
+
+def _is_non_author_text(value: str) -> bool:
+    text = (value or "").strip().lower()
+
+    if not text:
+        return True
+
+    if text in {"n/a", "na", "none"}:
+        return True
+
+    return text.startswith(_NON_AUTHOR_PREFIXES)
+
+
+def _surname_key(name: str) -> str:
+    """Normalize an author label for surname-level deduplication."""
+    text = (name or "").strip()
+    if not text:
+        return ""
+
+    # For "... Surname et al.", always use the word immediately before "et al."
+    et_al_match = re.search(r"\bet\s+al\b\.?", text, flags=re.I)
+    if et_al_match:
+        before_et_al = text[:et_al_match.start()].strip()
+        parts = before_et_al.split()
+        if parts:
+            return parts[-1].strip(".,;:").lower()
+
+    # "Surname, Initials" format
+    if "," in text:
+        return text.split(",", 1)[0].strip().lower()
+
+    # Fallback
+    parts = text.split()
+    return parts[0].strip(".,;:").lower() if parts else ""
+
+
+def parse_authors_from_filename(file_name: str) -> list[str]:
+    """Extract author names from 'Authors - YYYY - Title.ext' filenames."""
+    name = (file_name or "").strip()
+    if not name:
+        return []
+
+    stem = name.rsplit(".", 1)[0]
+    match = _FILENAME_AUTHORS_RE.match(stem)
+    if not match:
+        return []
+
+    author_part = match.group(1).strip()
+    if _ET_AL_RE.search(author_part):
+        return [author_part]
+    return [part.strip() for part in re.split(r"\s+and\s+", author_part, flags=re.I) if part.strip()]
+
+
+def parse_authors_from_rp(rp: str) -> str:
+    """Extract usable author names from RP and keep the more specific duplicate."""
+    text = (rp or "").strip()
+    if not text or _is_non_author_text(text):
+        return ""
+
+    names: list[str] = []
+    by_surname: dict[str, int] = {}
+
+    def _prefer(current: str, candidate: str) -> str:
+        # Prefer the more specific/longer representation
+        if len(candidate) > len(current):
+            return candidate
+        return current
+
+    for segment in _RP_SEGMENT_RE.split(text):
+        segment = segment.strip()
+
+        if not segment or _is_non_author_text(segment):
+            continue
+
+        name = segment.split("(", 1)[0].strip().rstrip(",")
+
+        if not name or _is_non_author_text(name):
+            continue
+
+        key = _surname_key(name)
+        if not key:
+            continue
+
+        if key in by_surname:
+            idx = by_surname[key]
+            names[idx] = _prefer(names[idx], name)
+        else:
+            by_surname[key] = len(names)
+            names.append(name)
+
+    return "; ".join(names)
+
+
+def resolve_authors(row: dict) -> str:
+    """
+    Resolve authors using the best available metadata source.
+
+    Priority:
+    1. AU  - full author list from publication metadata
+    2. RP  - author information parsed from full text
+    3. PDF/TXT filename - fallback only
+    """
+
+    # 1. Prefer the full author list from metadata
+    au = str(row.get("AU") or "").strip()
+    if au:
+        return au
+
+    # 2. Fall back to RP
+    rp_authors = parse_authors_from_rp(row.get("RP") or "").strip()
+    if rp_authors:
+        return rp_authors
+
+    # 3. Final fallback: filename
+    filename_authors = parse_authors_from_filename(
+        row.get("PDF_file") or row.get("TXT_file") or ""
+    )
+
+    return "; ".join(filename_authors)
+
+
+def format_authors(authors: str, *, max_names: int | None = None) -> str:
+    """Display string for UI cards. Default shows every parsed author."""
+    if not authors:
+        return ""
+    parts = [part.strip() for part in authors.split(";") if part.strip()]
+    if max_names is None or len(parts) <= max_names:
+        return "; ".join(parts)
+    return "; ".join(parts[:max_names]) + "; et al."
+
+
+def document_id_for_txt_file(txt_file: str) -> str | None:
+    name = (txt_file or "").strip()
+    if not name:
+        return None
+
+    manifest = load_manifest()
+    files = manifest.get("files", [])
+
+    for entry in files:
+        if entry.get("file_name") == name:
+            return entry.get("document_id")
+
+    lowered = name.lower()
+    for entry in files:
+        file_name = entry.get("file_name", "")
+        if file_name.lower() == lowered:
+            return entry.get("document_id")
+
+    if not name.endswith(".txt"):
+        with_txt = f"{name}.txt"
+        for entry in files:
+            if entry.get("file_name") == with_txt:
+                return entry.get("document_id")
+
+    if summary_path(name) is not None:
+        stem = name[:-4] if name.lower().endswith(".txt") else name
+        return hashlib.sha1(stem.encode("utf-8")).hexdigest()[:16]
+
+    return None
+
+
+def enrich_metadata_row(row: dict, row_index: int) -> dict:
+    txt_file = (row.get("TXT_file") or row.get("txt_file") or "").strip()
+    document_id = document_id_for_txt_file(txt_file)
+    authors = resolve_authors(row)
+    return {
+        "metadata_row_index": row_index,
+        "title": (row.get("TI") or "").strip(),
+        "abstract": (row.get("AB") or "").strip(),
+        "keywords": (row.get("DE") or "").strip(),
+        "authors": authors,
+        "doi": (row.get("DI") or "").strip(),
+        "year": (row.get("PY") or "").strip(),
+        "txt_file": txt_file,
+        "document_id": document_id,
+        "has_chat": document_id is not None,
+    }

backend/paper_voice.py

@@ -0,0 +1,65 @@
+"""Shared instructions so chat answers read as paper-grounded research, not summary meta-talk."""
+
+from __future__ import annotations
+
+import re
+
+PAPER_VOICE_RULES = """\
+Answer as a research assistant who has read the selected papers.
+- Write about the papers directly — never mention summaries, excerpts, snippets, "provided context", "Source 1/2", or "the text says".
+- Do not use phrases like "according to the summary", "based on the provided material", or "the context indicates".
+- Cite papers by **title** (and year when helpful), not by source number.
+- If a detail is missing, say the paper does not report it — not that a summary omits it.
+- Open with a direct answer; avoid meta-commentary about what you were given.\
+"""
+
+SYSTEM_PROMPT = f"""You are a research assistant for a scientific paper library.
+
+{PAPER_VOICE_RULES}
+
+Use ONLY the paper material in the user message. Do not use outside knowledge or invent facts.
+
+Formatting:
+- Write in clear markdown.
+- Use **bold** for paper titles and key terms on first mention.
+- Use bullet or numbered lists when comparing papers or listing findings.
+- Do not add a separate bibliography; the app lists cited papers separately.
+"""
+
+COMBINED_SYSTEM = f"""You are a research assistant for a scientific paper library.
+
+{PAPER_VOICE_RULES}
+
+Use ONLY the paper material in the user message. Do not use outside knowledge or invent facts.
+- For comparisons, use a ### heading per paper.
+- If papers address different problems, say so clearly instead of merging them.
+- Write in clear markdown with a direct answer first when possible.
+"""
+
+_META_LEAK_RE = re.compile(
+    r"\b("
+    r"according to (?:the )?(?:paper )?summar(?:y|ies)|"
+    r"based on (?:the )?(?:paper )?summar(?:y|ies)|"
+    r"(?:the )?provided (?:context|summar(?:y|ies))|"
+    r"in (?:the )?summar(?:y|ies)|"
+    r"from (?:the )?(?:paper )?summar(?:y|ies)|"
+    r"the (?:context|excerpt|snippet) (?:indicates|suggests|states|mentions)|"
+    r"as (?:mentioned|stated|noted) in (?:the )?(?:summary|context|provided material)"
+    r")\b",
+    re.I,
+)
+
+
+def prefix_context(context: str) -> str:
+    if not context.strip():
+        return PAPER_VOICE_RULES
+    return f"{PAPER_VOICE_RULES}\n\n---\n\n{context}"
+
+
+def sanitize_answer(text: str) -> str:
+    """Light cleanup if the model still leaks summary/meta phrasing."""
+    cleaned = _META_LEAK_RE.sub("", text)
+    cleaned = re.sub(r"\bIn\s*,", "In this paper,", cleaned)
+    cleaned = re.sub(r" {2,}", " ", cleaned)
+    cleaned = re.sub(r" +\n", "\n", cleaned)
+    return cleaned.strip()

backend/waterscope_agent.py

@@ -0,0 +1,114 @@
+"""Client for the WaterScopeAI Hugging Face Space /agent_run endpoint."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import os
+import re
+from functools import lru_cache
+from typing import Any
+
+from gradio_client import Client
+
+SPACE_ID = os.getenv("WATERSCOPE_SPACE_ID", "EnvironmentalAI/WaterScopeAI")
+TIMEOUT_SEC = int(os.getenv("WATERSCOPE_TIMEOUT_SEC", "600"))
+
+_SPACE_EXECUTOR = concurrent.futures.ThreadPoolExecutor(
+    max_workers=2,
+    thread_name_prefix="waterscope-space",
+)
+
+MODELS = ("base", "dpo", "da_dpo", "tulucore")
+APPLICATIONS = (
+    "general_qa",
+    "document_qa",
+    "paper_synthesis",
+    "method_recommendation",
+    "research_gaps",
+    "claim_check",
+)
+
+_SYNTHESIS = re.compile(r"\b(summarize|summary|overview|synthesize|main finding)\b", re.I)
+_METHODS = re.compile(r"\b(method|approach|model|algorithm|recommend)\b", re.I)
+_GAPS = re.compile(r"\b(gap|limitation|future work|open question|unclear)\b", re.I)
+_CLAIM = re.compile(r"\b(does the paper|is it true|claim|support|evidence that)\b", re.I)
+
+
+def infer_application(question: str, *, explicit: str | None = None) -> str:
+    if explicit and explicit != "auto":
+        if explicit not in APPLICATIONS:
+            raise ValueError(f"Unknown application {explicit!r}")
+        return explicit
+    text = question.lower()
+    if _SYNTHESIS.search(text):
+        return "paper_synthesis"
+    if _GAPS.search(text):
+        return "research_gaps"
+    if _CLAIM.search(text):
+        return "claim_check"
+    if _METHODS.search(text):
+        return "method_recommendation"
+    return "document_qa"
+
+
+@lru_cache(maxsize=1)
+def _client() -> Client:
+    token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN")
+    kwargs: dict[str, Any] = {"httpx_kwargs": {"timeout": TIMEOUT_SEC}}
+    if token:
+        kwargs["hf_token"] = token
+    return Client(SPACE_ID, **kwargs)
+
+
+def run_agent(
+    *,
+    context: str,
+    question: str,
+    model: str | None = None,
+    application: str = "document_qa",
+    max_new_tokens: int | None = None,
+) -> tuple[str, str]:
+    """Call /agent_run. Returns (answer, diagnostics)."""
+    selected_model = model or os.getenv("WATERSCOPE_MODEL", "dpo")
+    if selected_model not in MODELS:
+        raise ValueError(f"Unknown model {selected_model!r}. Choose from: {MODELS}")
+
+    app_type = infer_application(question, explicit=application if application != "auto" else None)
+    tokens = max_new_tokens or int(os.getenv("WATERSCOPE_MAX_TOKENS", "1024"))
+
+    def _predict() -> tuple[str, str]:
+        response = _client().predict(
+            selected_model,
+            app_type,
+            context,
+            question,
+            tokens,
+            api_name="/agent_run",
+        )
+        answer = response[0] if response else ""
+        diagnostics = response[1] if len(response) > 1 else ""
+        return answer, diagnostics
+
+    # Run in a worker thread so Gradio asyncio does not clash with MCP's loop.
+    answer, diagnostics = _SPACE_EXECUTOR.submit(_predict).result(timeout=TIMEOUT_SEC)
+    return answer, diagnostics
+
+
+def agent_answer(
+    context: str,
+    question: str,
+    *,
+    application: str = "auto",
+    model: str | None = None,
+    max_new_tokens: int | None = None,
+) -> tuple[str, str, str]:
+    """Returns (answer, application_used, diagnostics)."""
+    app_type = infer_application(question, explicit=application if application != "auto" else None)
+    answer, diagnostics = run_agent(
+        context=context,
+        question=question,
+        model=model,
+        application=app_type,
+        max_new_tokens=max_new_tokens,
+    )
+    return answer, app_type, diagnostics

local_rag/__init__.py

@@ -0,0 +1,5 @@
+"""Summary catalog helpers for paper-grounded chat."""
+
+from local_rag.summaries import list_papers, summaries_for_documents
+
+__all__ = ["list_papers", "summaries_for_documents"]

local_rag/config.py

@@ -0,0 +1,23 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+@dataclass(frozen=True)
+class LocalRagConfig:
+    data_dir: Path
+
+    @classmethod
+    def from_env(cls) -> "LocalRagConfig":
+        data_dir = Path(os.getenv("LOCAL_RAG_DATA_DIR", PROJECT_ROOT / "local_rag_data"))
+        return cls(data_dir=data_dir)
+
+    @property
+    def manifest_path(self) -> Path:
+        return self.data_dir / "manifest.json"

local_rag/summaries.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import json
+import os
+import re
+from pathlib import Path
+
+from local_rag.config import LocalRagConfig
+
+TITLE_PATTERN = re.compile(r"\*\*Title:\*\*\s*(.+)", re.MULTILINE)
+_FILENAME_TITLE = re.compile(r"^.+?\s-\s\d{4}\s-\s(.+)$")
+
+
+def summary_search_dirs() -> list[Path]:
+    """Directories to search for summary .txt files (first match wins)."""
+    dirs: list[Path] = []
+    for raw in (os.getenv("PAPER_SUMMARIES_DIRS") or "").split(":"):
+        raw = raw.strip()
+        if raw:
+            dirs.append(Path(raw).expanduser().resolve())
+
+    override = os.getenv("PAPER_SUMMARIES_DIR")
+    if override:
+        dirs.append(Path(override).expanduser().resolve())
+
+    project_root = Path(__file__).resolve().parents[1]
+    dirs.append(project_root / "paper_summaries")
+
+    unique: list[Path] = []
+    seen: set[str] = set()
+    for path in dirs:
+        key = str(path)
+        if key in seen:
+            continue
+        seen.add(key)
+        unique.append(path)
+    return unique
+
+
+def summaries_dir() -> Path:
+    for path in summary_search_dirs():
+        if path.exists():
+            return path
+    return Path(__file__).resolve().parents[1] / "paper_summaries"
+
+
+def _manifest_path(config: LocalRagConfig | None = None) -> Path:
+    cfg = config or LocalRagConfig.from_env()
+    return cfg.manifest_path
+
+
+def load_manifest(config: LocalRagConfig | None = None) -> dict:
+    path = _manifest_path(config)
+    if not path.exists():
+        raise FileNotFoundError(f"Manifest not found: {path}. Rebuild local_rag_data/manifest.json first.")
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def title_from_filename(file_name: str) -> str:
+    stem = Path(file_name).stem
+    match = _FILENAME_TITLE.match(stem)
+    if match:
+        return match.group(1).strip()
+    return stem
+
+
+def extract_title(file_name: str, summary: str | None = None) -> str:
+    if summary:
+        match = TITLE_PATTERN.search(summary)
+        if match:
+            return match.group(1).strip()
+    return title_from_filename(file_name)
+
+
+def summary_path(file_name: str) -> Path | None:
+    for root in summary_search_dirs():
+        candidate = root / file_name
+        if candidate.exists():
+            return candidate
+    return None
+
+
+def load_summary(file_name: str) -> str | None:
+    path = summary_path(file_name)
+    if path is None:
+        return None
+    return path.read_text(encoding="utf-8")
+
+
+def list_papers(config: LocalRagConfig | None = None) -> list[dict]:
+    """List indexed papers without reading every summary file into memory."""
+    manifest = load_manifest(config)
+    papers: list[dict] = []
+    for entry in manifest.get("files", []):
+        file_name = entry["file_name"]
+        papers.append(
+            {
+                "document_id": entry["document_id"],
+                "file_name": file_name,
+                "title": title_from_filename(file_name),
+                "chunk_count": entry.get("chunk_count", 0),
+                "has_summary": summary_path(file_name) is not None,
+            }
+        )
+    papers.sort(key=lambda item: item["title"].lower())
+    return papers
+
+
+def summaries_for_documents(document_ids: list[str], config: LocalRagConfig | None = None) -> list[dict]:
+    manifest = load_manifest(config)
+    by_id = {entry["document_id"]: entry for entry in manifest.get("files", [])}
+    selected: list[dict] = []
+    for document_id in document_ids:
+        entry = by_id.get(document_id)
+        if not entry:
+            continue
+        file_name = entry["file_name"]
+        summary = load_summary(file_name)
+        if not summary:
+            continue
+        selected.append(
+            {
+                "document_id": document_id,
+                "title": extract_title(file_name, summary),
+                "file_name": file_name,
+                "text": summary,
+            }
+        )
+    return selected

local_rag_data/saved_papers.json

@@ -0,0 +1 @@
+[]

mcp_tools/__init__.py

@@ -0,0 +1 @@
+"""Shared tool implementations for WaterScope MCP servers."""

mcp_tools/local_rag_tools.py

@@ -0,0 +1,37 @@
+"""Summary MCP tools for paper-grounded chat."""
+
+from __future__ import annotations
+
+import json
+
+from backend.chat_settings import truncate_summary
+from local_rag.summaries import summaries_for_documents
+
+
+def get_paper_summaries_tool(document_ids: list[str]) -> str:
+    return json.dumps({"sources": summaries_for_documents(document_ids)}, indent=2)
+
+
+def build_paper_context_tool(document_ids: list[str], question: str = "") -> str:
+    sources = summaries_for_documents(document_ids)
+    if not sources:
+        return json.dumps({"error": "No summaries found for the selected papers."}, indent=2)
+
+    n_papers = len(sources)
+    parts: list[str] = []
+    for index, source in enumerate(sources, start=1):
+        parts.append(
+            f"Paper {index}: {source['title']}\n"
+            f"File: {source['file_name']}\n\n"
+            f"{truncate_summary(source['text'], n_papers=n_papers, question=question)}"
+        )
+    context = "\n\n---\n\n".join(parts)
+    return json.dumps(
+        {
+            "context": context,
+            "sources": [{"document_id": s["document_id"], "title": s["title"]} for s in sources],
+            "n_papers": n_papers,
+            "question": question,
+        },
+        indent=2,
+    )

mcp_tools/metadata_tools.py

@@ -0,0 +1,25 @@
+"""Metadata search MCP tool implementations."""
+
+from __future__ import annotations
+
+import json
+
+from metadata_search.search_service import document_ids_from_ranks, search_metadata, search_score_mode
+
+
+def search_metadata_tool(query: str, top_k: int = 10) -> str:
+    results = search_metadata(query, top_k=top_k)
+    return json.dumps(
+        {"query": query, "results": results, "score_mode": search_score_mode()},
+        indent=2,
+    )
+
+
+def select_papers_from_search_tool(query: str, ranks: list[int], top_k: int = 10) -> str:
+    results = search_metadata(query, top_k=top_k)
+    document_ids = document_ids_from_ranks(results, ranks)
+    selected = [item for item in results if int(item["rank"]) in ranks]
+    return json.dumps(
+        {"query": query, "document_ids": document_ids, "papers": selected},
+        indent=2,
+    )

metadata_search/__init__.py

@@ -0,0 +1,12 @@
+"""CSV metadata search: field-weighted BM25 + Qdrant semantic + hybrid RRF."""
+
+from metadata_search.bm25_search import BM25Index
+from metadata_search.hybrid_search import weighted_rrf_scores
+from metadata_search.qdrant_metadata import MetadataSearchHit, MetadataVectorStore
+
+__all__ = [
+    "BM25Index",
+    "MetadataSearchHit",
+    "MetadataVectorStore",
+    "weighted_rrf_scores",
+]

metadata_search/agent_orchestrator.py

@@ -0,0 +1,339 @@
+"""Agent mode: fixed MCP tool pipeline → HF Space answer."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from typing import Any
+
+from backend.mcp_sync import call_mcp_tool, use_mcp
+from backend.model import llm_backend_name
+from backend.waterscope_agent import infer_application
+from metadata_search.mcp_retrieval import (
+    build_context_for_documents,
+    list_papers_catalog,
+    search_papers,
+)
+from metadata_search.boolean_query import focus_tokens
+from metadata_search.search_query_planner import plan_search_query
+from metadata_search.search_service import document_ids_from_ranks
+
+MAX_AUTO_PAPERS = int(os.getenv("AGENT_MAX_PAPERS", "4"))
+
+_DISCOVERY = re.compile(
+    r"\b(find|search|show|list|recommend|retrieve|get|papers?\s+(?:on|about|for|related))\b",
+    re.I,
+)
+
+
+def _is_paper_discovery(message: str) -> bool:
+    return bool(_DISCOVERY.search(message))
+
+
+def _enhance_agent_question(message: str, n_papers: int) -> str:
+    if not _is_paper_discovery(message):
+        return message
+    return (
+        f"{message}\n\n"
+        f"You are given exactly {n_papers} papers (Paper 1 through Paper {n_papers}). "
+        "Write one short paragraph per paper in numeric order. "
+        "Every paper must appear — do not skip any. "
+        "Do not mention summaries or provided context; write as if discussing the papers directly."
+    )
+
+
+def _paper_line(row: dict[str, Any]) -> tuple[str, str, str]:
+    title = (row.get("title") or "Untitled").strip()
+    year = (row.get("year") or "").strip() or "n.d."
+    authors = (row.get("authors_display") or row.get("authors") or "").strip() or "Authors not listed"
+    return title, year, authors
+
+
+def _abstract_snippet(row: dict[str, Any], *, limit: int = 220) -> str:
+    abstract = (row.get("full_abstract") or row.get("abstract") or "").strip()
+    if not abstract:
+        return ""
+    if len(abstract) <= limit:
+        return abstract
+    return f"{abstract[:limit].rstrip()}…"
+
+
+def _build_discovery_answer(picked: list[dict[str, Any]], search_query: str) -> str:
+    lines = [f"Found **{len(picked)} papers** for “{search_query}” (search rank order):", ""]
+    for index, row in enumerate(picked, start=1):
+        title, year, authors = _paper_line(row)
+        lines.append(f"{index}. **{title}** ({year}) — {authors}")
+        snippet = _abstract_snippet(row)
+        if snippet:
+            lines.append(f"   {snippet}")
+        lines.append("")
+    lines.append(
+        "All of the above are selected in the Publications panel — check boxes to chat about specific papers."
+    )
+    return "\n".join(lines)
+
+
+def _selected_paper_payload(picked: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    payload: list[dict[str, Any]] = []
+    for index, row in enumerate(picked, start=1):
+        title, year, authors = _paper_line(row)
+        payload.append(
+            {
+                "index": index,
+                "rank": row.get("rank"),
+                "title": title,
+                "year": year,
+                "authors": authors,
+                "document_id": row.get("document_id"),
+            }
+        )
+    return payload
+
+
+def _papers_with_summaries() -> dict[str, dict[str, Any]]:
+    return {
+        str(paper["document_id"]): paper
+        for paper in list_papers_catalog()
+        if paper.get("document_id") and paper.get("has_summary")
+    }
+
+
+def _rank_candidates_by_message(
+    results: list[dict[str, Any]],
+    selection_hint: str,
+    *,
+    limit: int,
+) -> list[int]:
+    index = _papers_with_summaries()
+    tokens = focus_tokens(selection_hint)
+    scored: list[tuple[float, int]] = []
+    for row in results:
+        rank = int(row["rank"])
+        doc_id = row.get("document_id")
+        if not doc_id or str(doc_id) not in index:
+            continue
+        haystack = " ".join(
+            [
+                str(row.get("title") or ""),
+                str(row.get("abstract") or row.get("full_abstract") or ""),
+                str(row.get("keywords") or ""),
+            ]
+        ).lower()
+        overlap = sum(1 for token in tokens if token in haystack)
+        score = overlap + float(row.get("final_score") or row.get("score") or 0.0) * 0.01
+        scored.append((score, rank))
+    scored.sort(key=lambda item: (-item[0], item[1]))
+    if not scored:
+        return [int(row["rank"]) for row in results if row.get("document_id") in index][:limit]
+    return [rank for _, rank in scored[:limit]]
+
+
+def _parse_tool_payload(raw: Any) -> dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str):
+        return json.loads(raw)
+    raise TypeError(f"Unexpected tool payload type: {type(raw)!r}")
+
+
+def _mcp_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Run a retrieval MCP tool (stdio subprocess)."""
+    return _parse_tool_payload(call_mcp_tool(name, arguments))
+
+
+def _invoke_retrieval_tool(name: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Agent retrieval — MCP when USE_MCP=1, else direct Python."""
+    args = arguments or {}
+
+    if name == "search_metadata":
+        query = str(args["query"])
+        top_k = int(args.get("top_k", 10))
+        # Direct in-process search — MCP subprocess cannot share local Qdrant with Flask.
+        results, score_mode = search_papers(query, top_k=top_k)
+        return {"query": query, "results": results, "score_mode": score_mode}
+
+    if name == "select_papers_from_search":
+        query = str(args["query"])
+        ranks = [int(rank) for rank in args["ranks"]]
+        top_k = int(args.get("top_k", 10))
+        results, score_mode = search_papers(query, top_k=top_k)
+        document_ids = document_ids_from_ranks(results, ranks)
+        selected = [item for item in results if int(item["rank"]) in ranks]
+        return {
+            "query": query,
+            "document_ids": document_ids,
+            "papers": selected,
+            "score_mode": score_mode,
+        }
+
+    if name == "build_paper_context":
+        document_ids = [str(item) for item in args["document_ids"]]
+        question = str(args.get("question") or "")
+        context, sources = build_context_for_documents(document_ids, question=question)
+        return {"context": context, "sources": sources, "n_papers": len(sources), "question": question}
+
+    raise ValueError(f"Unknown retrieval tool: {name}")
+
+
+def _step(step: int, tool: str, arguments: dict[str, Any], output: dict[str, Any]) -> dict[str, Any]:
+    return {"step": step, "tool": tool, "input": arguments, "output": output}
+
+
+def agent_chat_payload(body: dict[str, Any]) -> dict[str, Any]:
+    """Fixed MCP pipeline: search → select → context → HF Space answer."""
+    from metadata_search.chat_api import _generate_answer
+
+    message = (body.get("message") or "").strip()
+    if not message:
+        raise ValueError("message is required")
+
+    search_plan = plan_search_query(message, explicit_query=(body.get("search_query") or "").strip())
+    search_query = search_plan.search_query
+    top_k = min(int(body.get("search_top_k") or body.get("top_k") or 10), 50)
+    discovery = _is_paper_discovery(message)
+    steps: list[dict[str, Any]] = []
+
+    steps.append(
+        _step(
+            0,
+            "plan_search_query",
+            {"message": message},
+            {
+                "search_query": search_query,
+                "selection_hint": search_plan.selection_hint,
+                "planner": search_plan.planner,
+                "parsed": search_plan.parsed,
+            },
+        )
+    )
+
+    search_payload = _invoke_retrieval_tool(
+        "search_metadata",
+        {"query": search_query, "top_k": top_k},
+    )
+    search_results = list(search_payload.get("results") or [])
+    search_score_mode = str(search_payload.get("score_mode") or "rrf")
+    steps.append(
+        _step(
+            1,
+            "search_metadata",
+            {"query": search_query, "top_k": top_k},
+            {
+                "n_results": len(search_results),
+                "top_titles": [row.get("title") for row in search_results[:5]],
+            },
+        )
+    )
+
+    ranks = _rank_candidates_by_message(
+        search_results,
+        search_plan.selection_hint,
+        limit=MAX_AUTO_PAPERS,
+    )
+    if not ranks:
+        raise ValueError(
+            "Agent could not find papers with linked summaries. "
+            "Try a different query or select papers manually."
+        )
+
+    select_payload = _invoke_retrieval_tool(
+        "select_papers_from_search",
+        {"query": search_query, "ranks": ranks, "top_k": top_k},
+    )
+    document_ids = [str(item) for item in select_payload.get("document_ids") or []]
+    picked_rows = list(select_payload.get("papers") or [])
+    steps.append(
+        _step(
+            2,
+            "select_papers_from_search",
+            {"ranks": ranks},
+            {
+                "document_ids": document_ids,
+                "papers": [{"rank": row.get("rank"), "title": row.get("title")} for row in picked_rows],
+            },
+        )
+    )
+
+    if discovery:
+        answer = _build_discovery_answer(picked_rows, search_query)
+        strategy = "discovery_template"
+        steps.append(
+            _step(
+                3,
+                "format_discovery_results",
+                {"n_papers": len(picked_rows), "query": search_query},
+                {"strategy": strategy},
+            )
+        )
+        sources: list[dict[str, Any]] = []
+        context = ""
+        diagnostics = None
+    else:
+        context_payload = _invoke_retrieval_tool(
+            "build_paper_context",
+            {"document_ids": document_ids, "question": message},
+        )
+        context = str(context_payload.get("context") or "")
+        sources = list(context_payload.get("sources") or [])
+        if not context or not sources:
+            raise ValueError("Agent could not assemble paper context. Try a different query.")
+        steps.append(
+            _step(
+                3,
+                "build_paper_context",
+                {"document_ids": document_ids, "question": message},
+                {"n_papers": len(sources), "context_chars": len(context)},
+            )
+        )
+
+        application = body.get("application")
+        if not application or application == "auto":
+            application = infer_application(message, explicit=None)
+
+        answer, strategy, diagnostics = _generate_answer(
+            _enhance_agent_question(message, len(sources)),
+            context,
+            application=application,
+            model=body.get("model"),
+        )
+        steps.append(
+            _step(
+                4,
+                "waterscope_agent_run",
+                {"model": body.get("model"), "application": application},
+                {"strategy": strategy},
+            )
+        )
+
+    payload: dict[str, Any] = {
+        "answer": answer,
+        "strategy": strategy,
+        "agent_mode": True,
+        "discovery_mode": discovery,
+        "agent_steps": steps,
+        "search_query": search_query,
+        "search_plan": {
+            "search_query": search_plan.search_query,
+            "selection_hint": search_plan.selection_hint,
+            "planner": search_plan.planner,
+            "parsed": search_plan.parsed,
+        },
+        "document_ids": document_ids,
+        "sources": [
+            {
+                "document_id": source["document_id"],
+                "title": source["title"],
+                "file_name": source["file_name"],
+            }
+            for source in sources
+        ],
+        "search_results": picked_rows,
+        "selected_papers": _selected_paper_payload(picked_rows),
+        "score_mode": search_score_mode,
+        "retrieval": "mcp" if use_mcp() else "direct",
+    }
+    if diagnostics is not None:
+        payload["diagnostics"] = diagnostics
+    payload["llm_backend"] = llm_backend_name()
+    return payload

metadata_search/app.py

@@ -0,0 +1,319 @@
+#!/usr/bin/env python3
+"""Flask + MCP retrieval + WaterScope agent UI."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+from flask import Flask, jsonify, request, send_from_directory
+
+from metadata_search.chat_api import chat_payload, get_papers_payload
+from metadata_search.config import METADATA_SEARCH_PORT, PAPER_CSV, PAPER_FIGURES_DIR, PROJECT_ROOT
+from metadata_search.paper_figures import (
+    figure_paper_count,
+    figures_payload_for_paper,
+    figures_dir,
+)
+from metadata_search.mcp_retrieval import (
+    list_saved_papers,
+    remove_saved_paper,
+    save_paper_record,
+    search_papers,
+)
+from metadata_search.search_service import search_score_mode
+from backend.mcp_sync import use_mcp
+from backend.model import llm_backend_name
+
+DEFAULT_TOP_K = 10
+ALLOWED_TOP_K = {10, 50, 100, 200}
+
+if not Path(PAPER_CSV).is_file():
+    raise SystemExit(f"Paper dataset not found: {PAPER_CSV}. Set PAPER_CSV in .env.")
+
+app = Flask(__name__)
+UI_DIR = PROJECT_ROOT / "github_pages"
+
+
+def _page(name: str):
+    return send_from_directory(UI_DIR, name)
+
+
+def _parse_top_k(raw: str | None) -> int:
+    try:
+        value = int(raw or DEFAULT_TOP_K)
+    except (TypeError, ValueError):
+        return DEFAULT_TOP_K
+    return value if value in ALLOWED_TOP_K else DEFAULT_TOP_K
+
+
+@app.get("/")
+def home_page():
+    if UI_DIR.is_dir():
+        return _page("index.html")
+    return jsonify(
+        {
+            "service": "WaterScopeAI Search System",
+            "status": "running",
+            "health": "/api/health",
+            "search": "/api/search?q=activated+sludge&top_k=10",
+        }
+    )
+
+
+@app.get("/demo")
+def demo_page():
+    return _page("demo.html")
+
+
+@app.get("/chat")
+def chat_page():
+    return _page("chat.html")
+
+
+@app.get("/instructions")
+def instructions_page():
+    return _page("instructions.html")
+
+
+@app.get("/agent")
+def agent_page():
+    return _page("agent.html")
+
+
+@app.get("/Data/<path:filename>")
+def mcqa_data(filename: str):
+    return send_from_directory(UI_DIR / "Data", filename)
+
+
+@app.get("/assets/<path:filename>")
+def assets(filename: str):
+    response = send_from_directory(UI_DIR, filename)
+    if filename.endswith((".js", ".css")):
+        response.headers["Cache-Control"] = "no-cache"
+    return response
+
+
+@app.before_request
+def handle_api_preflight():
+    if request.method == "OPTIONS" and request.path.startswith("/api/"):
+        return ("", 204)
+
+
+def _cors_allow_origin() -> str | None:
+    configured = os.getenv("WATERSCOPE_CORS_ORIGIN", "*").strip()
+    origin = request.headers.get("Origin")
+    if configured in {"*", ""}:
+        return origin or "*"
+    allowed = {item.strip() for item in configured.split(",") if item.strip()}
+    if origin and (origin in allowed or "localhost" in origin or "127.0.0.1" in origin):
+        return origin
+    return configured if configured in allowed else None
+
+
+@app.after_request
+def cors_headers(response):
+    if os.getenv("WATERSCOPE_CORS", "0").lower() in {"1", "true", "yes"}:
+        allow_origin = _cors_allow_origin()
+        if allow_origin:
+            response.headers["Access-Control-Allow-Origin"] = allow_origin
+        response.headers["Access-Control-Allow-Headers"] = "Content-Type"
+        response.headers["Access-Control-Allow-Methods"] = "GET, POST, DELETE, OPTIONS"
+    return response
+
+
+@app.get("/pdfs/<path:filename>")
+def paper_pdf_file(filename: str):
+    root_raw = os.getenv("PAPER_PDF_DIR", "").strip()
+    if not root_raw:
+        return jsonify({"error": "PDF assets are not configured."}), 404
+
+    root = Path(root_raw).resolve()
+    if not root.is_dir():
+        return jsonify({"error": "PDF directory not found."}), 404
+
+    safe_name = Path(filename).name
+    target = (root / safe_name).resolve()
+    if target.parent != root or not target.is_file():
+        return jsonify({"error": "PDF file not found."}), 404
+
+    return send_from_directory(root, safe_name, as_attachment=True)
+
+
+@app.get("/figures/<path:paper_key>/<path:filename>")
+def paper_figure_file(paper_key: str, filename: str):
+    root = figures_dir()
+    if root is None:
+        return jsonify({"error": "Figure assets are not configured."}), 404
+    paper_dir = root / paper_key
+    if not paper_dir.is_dir():
+        return jsonify({"error": "Paper figures not found."}), 404
+    target = (paper_dir / filename).resolve()
+    if not str(target).startswith(str(paper_dir.resolve())):
+        return jsonify({"error": "Invalid figure path."}), 400
+    if not target.is_file():
+        return jsonify({"error": "Figure file not found."}), 404
+    return send_from_directory(paper_dir, filename)
+
+
+@app.get("/api/papers/figures")
+def api_paper_figures():
+    txt_file = request.args.get("txt_file", "").strip()
+    pdf_file = request.args.get("pdf_file", "").strip()
+    paper_key = request.args.get("paper_key", "").strip()
+    metadata_row_index_raw = request.args.get("metadata_row_index", "").strip()
+
+    metadata_row_index: int | None = None
+    if metadata_row_index_raw:
+        try:
+            metadata_row_index = int(metadata_row_index_raw)
+        except ValueError:
+            return jsonify({"error": "metadata_row_index must be an integer."}), 400
+
+    if not any([txt_file, pdf_file, paper_key, metadata_row_index is not None]):
+        return jsonify({"error": "Provide txt_file, pdf_file, paper_key, or metadata_row_index."}), 400
+
+    payload = figures_payload_for_paper(
+        txt_file=txt_file,
+        pdf_file=pdf_file,
+        paper_key=paper_key,
+        metadata_row_index=metadata_row_index,
+    )
+    if payload is None:
+        return jsonify({"error": "No figures found for this paper."}), 404
+    return jsonify(payload)
+
+
+@app.get("/api/search")
+def api_search():
+    query = request.args.get("q", "").strip()
+    if not query:
+        return jsonify({"error": "q is required"}), 400
+    top_k = _parse_top_k(request.args.get("top_k"))
+    try:
+        from metadata_search.boolean_query import parse_boolean_query, plan_summary
+
+        results, score_mode = search_papers(query, top_k=top_k)
+        parsed = parse_boolean_query(query)
+        return jsonify(
+            {
+                "query": query,
+                "search_query": parsed.raw if parsed.raw != query else query,
+                "results": results,
+                "score_mode": score_mode,
+                "query_parsed": plan_summary(parsed),
+            }
+        )
+    except Exception as exc:
+        message = str(exc) or repr(exc) or type(exc).__name__
+        return jsonify({"error": message}), 500
+
+
+@app.get("/api/saved-papers")
+def api_saved_papers():
+    return jsonify({"papers": list_saved_papers()})
+
+
+@app.post("/api/saved-papers")
+def api_create_saved_paper():
+    body = request.get_json(force=True, silent=True) or {}
+    try:
+        return jsonify({"paper": save_paper_record(body)})
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+
+
+@app.delete("/api/saved-papers/<paper_id>")
+def api_delete_saved_paper(paper_id: str):
+    if not remove_saved_paper(paper_id):
+        return jsonify({"error": "Saved paper not found."}), 404
+    return jsonify({"ok": True})
+
+
+@app.get("/api/health")
+def api_health():
+    summaries_dir = os.getenv("PAPER_SUMMARIES_DIR", "")
+    from metadata_search.search_service import get_rows, get_vector_store
+
+    try:
+        vector_points = get_vector_store().points_count()
+        paper_rows = len(get_rows())
+    except Exception:
+        vector_points = 0
+        paper_rows = 0
+    return jsonify(
+        {
+            "status": "ok",
+            "checks": {
+                "paper_csv": Path(PAPER_CSV).is_file(),
+                "summaries_dir": bool(summaries_dir and Path(summaries_dir).is_dir()),
+                "metadata_vector_index": vector_points > 0,
+                "paper_figures_dir": PAPER_FIGURES_DIR.is_dir(),
+            },
+            "paper_count": paper_rows,
+            "metadata_vector_points": vector_points,
+            "figure_paper_count": figure_paper_count(),
+            "llm_backend": llm_backend_name(),
+            "retrieval": "mcp" if use_mcp() else "direct",
+            "search_score_mode": search_score_mode(),
+        }
+    )
+
+
+@app.get("/api/papers")
+def api_papers():
+    try:
+        return jsonify(get_papers_payload())
+    except FileNotFoundError as exc:
+        return jsonify({"error": str(exc)}), 503
+    except Exception as exc:
+        return jsonify({"error": f"Could not load papers: {exc}"}), 500
+
+
+@app.post("/api/chat")
+def api_chat():
+    body = request.get_json(force=True, silent=True) or {}
+    try:
+        return jsonify(chat_payload(body))
+    except ValueError as exc:
+        message = str(exc)
+        if "RUNTIME_ERROR" in message or "invalid state" in message.lower():
+            return jsonify(
+                {
+                    "error": (
+                        "WaterScope HF Space is unavailable right now "
+                        f"(EnvironmentalAI/WaterScopeAI: {message}). "
+                        "Retrieval succeeded, but the answer model could not run."
+                    )
+                }
+            ), 503
+        return jsonify({"error": message}), 400
+    except Exception as exc:
+        return jsonify(
+            {
+                "error": (
+                    "Could not reach the WaterScope HF Space model. "
+                    f"Check network and EnvironmentalAI/WaterScopeAI ({type(exc).__name__}: {exc})"
+                )
+            }
+        ), 503
+
+
+def main() -> None:
+    from metadata_search.search_service import auto_rebuild_on_startup
+
+    auto_rebuild_on_startup()
+    if use_mcp():
+        from backend.mcp_sync import ensure_mcp_connected
+
+        ensure_mcp_connected()
+    debug = os.getenv("FLASK_DEBUG", "0").lower() in {"1", "true", "yes"}
+    app.run(host=os.getenv("HOST", "0.0.0.0"), debug=debug, use_reloader=False, port=METADATA_SEARCH_PORT)
+
+
+if __name__ == "__main__":
+    main()

metadata_search/bm25_search.py

@@ -0,0 +1,64 @@
+"""Field-weighted BM25 over paper title, abstract, and keywords."""
+
+from __future__ import annotations
+
+from rank_bm25 import BM25Okapi
+
+from metadata_search.config import ABSTRACT_COL, KEYWORD_COL, TITLE_COL
+from metadata_search.corpus import tokenize
+
+
+class BM25Index:
+    """Fielded BM25 index for a static list of paper records."""
+
+    def __init__(
+        self,
+        rows: list[dict[str, str]],
+        *,
+        title_col: str = TITLE_COL,
+        abstract_col: str = ABSTRACT_COL,
+        keyword_col: str = KEYWORD_COL,
+        title_weight: float = 2.0,
+        keyword_weight: float = 1.5,
+    ) -> None:
+        self.rows = rows
+        self.title_col = title_col
+        self.abstract_col = abstract_col
+        self.keyword_col = keyword_col
+        self.title_weight = title_weight
+        self.keyword_weight = keyword_weight
+
+        self.title_documents = [tokenize(str(row.get(title_col) or "")) for row in rows]
+        self.abstract_documents = [tokenize(str(row.get(abstract_col) or "")) for row in rows]
+        self.keyword_documents = [tokenize(str(row.get(keyword_col) or "")) for row in rows]
+
+        self.title_bm25 = self._build_bm25(self.title_documents)
+        self.abstract_bm25 = self._build_bm25(self.abstract_documents)
+        self.keyword_bm25 = self._build_bm25(self.keyword_documents)
+
+    def _build_bm25(self, documents: list[list[str]]) -> BM25Okapi | None:
+        if not documents or not any(documents):
+            return None
+        return BM25Okapi(documents)
+
+    def score(self, query: str) -> list[float]:
+        if not query.strip() or not self.rows:
+            return [0.0] * len(self.rows)
+
+        query_terms = tokenize(query)
+        title_scores = self.title_bm25.get_scores(query_terms) if self.title_bm25 else [0.0] * len(self.rows)
+        abstract_scores = (
+            self.abstract_bm25.get_scores(query_terms) if self.abstract_bm25 else [0.0] * len(self.rows)
+        )
+        keyword_scores = (
+            self.keyword_bm25.get_scores(query_terms) if self.keyword_bm25 else [0.0] * len(self.rows)
+        )
+
+        return [
+            float(
+                self.title_weight * title_scores[index]
+                + abstract_scores[index]
+                + self.keyword_weight * keyword_scores[index]
+            )
+            for index in range(len(self.rows))
+        ]

metadata_search/boolean_query.py

@@ -0,0 +1,345 @@
+"""Google Scholar-style boolean query parsing (+required, -exclude, \"phrases\")."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from metadata_search.corpus import tokenize
+
+_QUOTED = re.compile(r'"([^"]+)"')
+_DISCOVERY_PREFIX = re.compile(
+    r"^(?:please\s+)?(?:find|search|show|list|recommend|retrieve|get)\s+"
+    r"(?:me\s+)?(?:some\s+)?(?:papers?\s+)?(?:on|about|for|related to)\s+",
+    re.I,
+)
+_BUT_NOT = re.compile(r"\s+but\s+not\s+", re.I)
+_BUT_EXCLUDE = re.compile(r"\s+but\s+(?=-)", re.I)
+_AND_NOT = re.compile(r"\s+and\s+not\s+", re.I)
+_TRAILING_BUT = re.compile(r"\s+but\s*$", re.I)
+_NL_NEGATION = re.compile(
+    r"(?:,\s*|\s+)(?:but\s+not|not|excluding|without|except)\s+(.+)$",
+    re.I,
+)
+_EXCLUDED_FILLER_PREFIX = re.compile(
+    r"^(?:anything|something|any|studies|papers|research|work|articles?)\s+"
+    r"(?:on|about|for|regarding|related to)\s+",
+    re.I,
+)
+_EXCLUDED_LEADING_PREP = re.compile(r"^(?:on|about|for|regarding|related to)\s+", re.I)
+_TRAILING_FILLER = re.compile(
+    r"(?:,\s*)?\b(?:though|tho|however|please|thanks|thank you|just|really|actually|anyway|right|ok|okay|hmm|um|uh|lol)\b[\s,.!?]*$",
+    re.I,
+)
+_LEADING_FILLER = re.compile(
+    r"^[\s,.!?]*\b(?:please|just|really|actually|maybe|kind of|sort of)\b\s+",
+    re.I,
+)
+_FILLER_WORDS = frozenset(
+    {
+        "though",
+        "tho",
+        "however",
+        "please",
+        "thanks",
+        "just",
+        "really",
+        "actually",
+        "anyway",
+        "right",
+        "ok",
+        "okay",
+        "hmm",
+        "um",
+        "uh",
+    }
+)
+
+
+def _strip_conversational_fillers(text: str) -> str:
+    """Remove trailing/leading chat filler ('… though', 'please …')."""
+    cleaned = text.strip()
+    if not cleaned:
+        return ""
+    for _ in range(5):
+        updated = _TRAILING_FILLER.sub("", cleaned).strip(" ,.")
+        updated = _LEADING_FILLER.sub("", updated).strip(" ,.")
+        if updated == cleaned:
+            break
+        cleaned = updated
+    return cleaned
+
+
+def _clean_excluded_phrase(phrase: str) -> str:
+    """Drop chat filler from NL exclusions: 'anything on drinking water though' → 'drinking water'."""
+    text = _strip_conversational_fillers(phrase.strip())
+    if not text:
+        return ""
+    for _ in range(3):
+        updated = _EXCLUDED_FILLER_PREFIX.sub("", text).strip()
+        updated = _EXCLUDED_LEADING_PREP.sub("", updated).strip()
+        updated = _strip_conversational_fillers(updated)
+        if updated == text:
+            break
+        text = updated
+    return text
+
+
+def normalize_search_query(query: str) -> str:
+    """Strip chat phrasing (e.g. 'Find papers on …') before boolean parsing."""
+    text = (query or "").strip()
+    if not text:
+        return ""
+    text = _DISCOVERY_PREFIX.sub("", text).strip(" .")
+    text = _BUT_NOT.sub(" not ", text)
+    text = _AND_NOT.sub(" not ", text)
+    text = _BUT_EXCLUDE.sub(" ", text)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+
+def has_natural_language_negation(query: str) -> bool:
+    text = (query or "").strip()
+    return bool(text and _NL_NEGATION.search(text))
+
+
+def split_natural_language_negation(query: str) -> tuple[str, tuple[str, ...]]:
+    """Split trailing ', not X' / ' not X' style constraints from the positive query."""
+    text = (query or "").strip()
+    if not text:
+        return "", ()
+    match = _NL_NEGATION.search(text)
+    if not match:
+        return text, ()
+    positive = text[: match.start()].strip(" ,.")
+    positive = _TRAILING_BUT.sub("", positive).strip(" ,.")
+    excluded = _clean_excluded_phrase(match.group(1).strip(" ,."))
+    if not positive:
+        positive = text
+        excluded = ""
+    return positive, (excluded,) if excluded else ()
+
+
+def has_boolean_syntax(query: str) -> bool:
+    text = (query or "").strip()
+    if not text:
+        return False
+    if has_natural_language_negation(text):
+        return True
+    if _QUOTED.search(text):
+        return True
+    return bool(re.search(r"(?:^|\s)[+-]", text))
+
+
+@dataclass(frozen=True)
+class BooleanQuery:
+    raw: str
+    ranking_text: str
+    required_phrases: tuple[str, ...] = ()
+    required_terms: tuple[str, ...] = ()
+    excluded_phrases: tuple[str, ...] = ()
+    excluded_terms: tuple[str, ...] = ()
+    optional_terms: tuple[str, ...] = ()
+
+
+def _format_exclusion(phrase: str) -> str:
+    """Multi-word exclusions must be quoted or only the first token is excluded."""
+    text = phrase.strip()
+    if not text:
+        return ""
+    if " " in text:
+        return f'-"{text}"'
+    return f"-{text}"
+
+
+def _format_required(term: str) -> str:
+    text = term.strip()
+    if not text:
+        return ""
+    if " " in text:
+        return f'"{text}"'
+    return f"+{text}"
+
+
+def _build_effective_raw(
+    *,
+    required_phrases: list[str],
+    required_terms: list[str],
+    optional_terms: list[str],
+    excluded_phrases: list[str],
+    excluded_terms: list[str],
+    fallback: str,
+) -> str:
+    parts: list[str] = []
+    for phrase in required_phrases:
+        parts.append(f'"{phrase}"')
+    for term in required_terms:
+        parts.append(_format_required(term))
+    for term in optional_terms:
+        parts.append(term)
+    for phrase in excluded_phrases:
+        parts.append(_format_exclusion(phrase))
+    for term in excluded_terms:
+        parts.append(_format_exclusion(term))
+    return " ".join(part for part in parts if part).strip() or fallback
+
+
+def parse_boolean_query(query: str) -> BooleanQuery:
+    """Parse +term, -term, \"phrase\", and natural-language \"not X\" constraints."""
+    raw = normalize_search_query(query)
+    if not raw:
+        return BooleanQuery(raw="", ranking_text="")
+
+    positive_text, nl_excluded_phrases = split_natural_language_negation(raw)
+
+    required_phrases: list[str] = []
+    excluded_phrases: list[str] = list(nl_excluded_phrases)
+    required_terms: list[str] = []
+    excluded_terms: list[str] = []
+    optional_terms: list[str] = []
+    ranking_parts: list[str] = []
+
+    def _add_term(bucket: list[str], token: str) -> None:
+        normalized = token.strip().lower()
+        if normalized:
+            bucket.append(normalized)
+
+    remaining = positive_text
+    for match in _QUOTED.finditer(positive_text):
+        phrase = match.group(1).strip()
+        if not phrase:
+            continue
+        start = match.start()
+        prefix = positive_text[max(0, start - 1) : start]
+        if prefix == "-":
+            excluded_phrases.append(_clean_excluded_phrase(phrase))
+        else:
+            required_phrases.append(phrase)
+            ranking_parts.append(phrase)
+        remaining = remaining.replace(match.group(0), " ", 1)
+
+    tokens = remaining.split()
+    index = 0
+    while index < len(tokens):
+        token = tokens[index].strip()
+        if not token:
+            index += 1
+            continue
+        if token.startswith("+"):
+            term = token[1:].strip("+-")
+            _add_term(required_terms, term)
+            if term:
+                ranking_parts.append(term)
+            index += 1
+            continue
+        if token.startswith("-"):
+            term = token[1:].strip("+-")
+            phrase_parts: list[str] = [term] if term else []
+            next_index = index + 1
+            while next_index < len(tokens):
+                follower = tokens[next_index].strip()
+                if not follower or follower.startswith(("+", "-")):
+                    break
+                phrase_parts.append(follower)
+                next_index += 1
+            phrase = " ".join(part for part in phrase_parts if part).strip()
+            phrase = _clean_excluded_phrase(phrase)
+            if " " in phrase:
+                excluded_phrases.append(phrase)
+            elif phrase:
+                _add_term(excluded_terms, phrase)
+            index = next_index
+            continue
+        _add_term(optional_terms, token)
+        if token.lower() not in _FILLER_WORDS:
+            ranking_parts.append(token)
+        index += 1
+
+    excluded_phrases = [_clean_excluded_phrase(phrase) for phrase in excluded_phrases]
+    excluded_phrases = [phrase for phrase in excluded_phrases if phrase]
+    optional_terms = [term for term in optional_terms if term.lower() not in _FILLER_WORDS]
+    ranking_parts = [part for part in ranking_parts if part.lower() not in _FILLER_WORDS]
+
+    ranking_text = " ".join(part for part in ranking_parts if part).strip() or positive_text or raw
+    effective_raw = _build_effective_raw(
+        required_phrases=required_phrases,
+        required_terms=required_terms,
+        optional_terms=optional_terms,
+        excluded_phrases=excluded_phrases,
+        excluded_terms=excluded_terms,
+        fallback=raw,
+    )
+    return BooleanQuery(
+        raw=effective_raw,
+        ranking_text=ranking_text,
+        required_phrases=tuple(required_phrases),
+        required_terms=tuple(required_terms),
+        excluded_phrases=tuple(excluded_phrases),
+        excluded_terms=tuple(excluded_terms),
+        optional_terms=tuple(optional_terms),
+    )
+
+
+def _haystack(row: dict[str, object]) -> str:
+    return " ".join(
+        [
+            str(row.get("title") or row.get("TI") or ""),
+            str(row.get("abstract") or row.get("full_abstract") or row.get("AB") or ""),
+            str(row.get("keywords") or row.get("DE") or ""),
+        ]
+    ).lower()
+
+
+def _contains_term(haystack: str, term: str) -> bool:
+    normalized = term.lower().strip()
+    if not normalized:
+        return True
+    if " " in normalized:
+        return normalized in haystack
+    return bool(re.search(rf"\b{re.escape(normalized)}\b", haystack)) or normalized in haystack
+
+
+def matches_boolean_query(row: dict[str, object], parsed: BooleanQuery) -> bool:
+    haystack = _haystack(row)
+    for phrase in parsed.required_phrases:
+        if phrase.lower() not in haystack:
+            return False
+    for term in parsed.required_terms:
+        if not _contains_term(haystack, term):
+            return False
+    for phrase in parsed.excluded_phrases:
+        if phrase.lower() in haystack:
+            return False
+    for term in parsed.excluded_terms:
+        if _contains_term(haystack, term):
+            return False
+    return True
+
+
+def focus_tokens(text: str, *, limit: int = 12) -> list[str]:
+    """Tokens for re-ranking agent paper picks (no hardcoded stopword list)."""
+    tokens = tokenize(text or "")
+    seen: set[str] = set()
+    focused: list[str] = []
+    for token in tokens:
+        if token.startswith(("lemma:", "phrase:", "concept:")):
+            continue
+        if token in seen or len(token) < 3:
+            continue
+        seen.add(token)
+        focused.append(token)
+        if len(focused) >= limit:
+            break
+    return focused
+
+
+def plan_summary(parsed: BooleanQuery) -> dict[str, object]:
+    return {
+        "raw": parsed.raw,
+        "ranking_text": parsed.ranking_text,
+        "required_phrases": list(parsed.required_phrases),
+        "required_terms": list(parsed.required_terms),
+        "excluded_terms": list(parsed.excluded_terms),
+        "excluded_phrases": list(parsed.excluded_phrases),
+        "optional_terms": list(parsed.optional_terms),
+    }

metadata_search/chat_api.py

@@ -0,0 +1,85 @@
+"""Chat API for the WaterScope research agent (HF Space answers)."""
+
+from __future__ import annotations
+
+import os
+
+from backend.chat_settings import MAX_CHAT_PAPERS, enhance_ordinal_question, focus_document_ids_for_question
+from backend.mcp_sync import use_mcp
+from backend.model import llm_backend_name
+from backend.paper_voice import prefix_context, sanitize_answer
+from backend.waterscope_agent import agent_answer
+from metadata_search.mcp_retrieval import build_context_for_documents, list_papers_catalog, list_saved_papers
+
+
+def _generate_answer(
+    message: str,
+    context: str,
+    *,
+    application: str | None = None,
+    model: str | None = None,
+) -> tuple[str, str, str]:
+    answer, app_used, diagnostics = agent_answer(
+        prefix_context(context),
+        message,
+        application=application or os.getenv("WATERSCOPE_APPLICATION", "auto"),
+        model=model or os.getenv("WATERSCOPE_MODEL"),
+    )
+    return sanitize_answer(answer), f"waterscope:{app_used}", diagnostics
+
+
+def get_papers_payload() -> dict:
+    papers = list_papers_catalog()
+    saved_ids = {paper.get("document_id") for paper in list_saved_papers() if paper.get("document_id")}
+    for paper in papers:
+        paper["is_saved"] = paper.get("document_id") in saved_ids
+    return {"papers": papers, "max_select": MAX_CHAT_PAPERS}
+
+
+def chat_payload(body: dict) -> dict:
+    message = (body.get("message") or "").strip()
+    document_ids = body.get("document_ids") or []
+    application = body.get("application")
+    agent_mode = bool(body.get("agent_mode"))
+
+    if not message:
+        raise ValueError("message is required")
+
+    if agent_mode and not document_ids:
+        from metadata_search.agent_orchestrator import agent_chat_payload
+
+        return agent_chat_payload(body)
+
+    if not document_ids:
+        raise ValueError("Select at least one paper, or enable agent mode.")
+    if len(document_ids) > MAX_CHAT_PAPERS:
+        raise ValueError(f"Select at most {MAX_CHAT_PAPERS} papers.")
+
+    document_ids = focus_document_ids_for_question(message, document_ids)
+    context, sources = build_context_for_documents(document_ids, question=message)
+    message = enhance_ordinal_question(message, sources)
+
+    answer, strategy, diagnostics = _generate_answer(
+        message,
+        context,
+        application=application,
+        model=body.get("model"),
+    )
+
+    return {
+        "answer": answer,
+        "strategy": strategy,
+        "include_excerpts": False,
+        "sources": [
+            {
+                "document_id": source["document_id"],
+                "title": source["title"],
+                "file_name": source["file_name"],
+            }
+            for source in sources
+        ],
+        "chunks": [],
+        "retrieval": "mcp" if use_mcp() else "direct",
+        "diagnostics": diagnostics,
+        "llm_backend": llm_backend_name(),
+    }

metadata_search/config.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+# .env wins over stale shell exports
+load_dotenv(override=True)
+
+PACKAGE_DIR = Path(__file__).resolve().parent
+PROJECT_ROOT = PACKAGE_DIR.parent
+DEFAULT_CSV = PACKAGE_DIR / "data" / "papers.csv"
+
+PAPER_CSV = Path(os.getenv("PAPER_CSV", str(DEFAULT_CSV)))
+EMBEDDING_MODEL = os.getenv("METADATA_EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
+METADATA_QDRANT_PATH = Path(
+    os.getenv("METADATA_QDRANT_PATH", str(PROJECT_ROOT / "metadata_search_data" / "qdrant"))
+)
+METADATA_QDRANT_COLLECTION = os.getenv("METADATA_QDRANT_COLLECTION", "papers_metadata")
+METADATA_QDRANT_MANIFEST = Path(
+    os.getenv(
+        "METADATA_QDRANT_MANIFEST",
+        str(PROJECT_ROOT / "metadata_search_data" / "qdrant_manifest.json"),
+    )
+)
+METADATA_SEARCH_PORT = int(os.getenv("PORT", os.getenv("METADATA_SEARCH_PORT", "7860")))
+PAPER_FIGURES_DIR = Path(
+    os.getenv("PAPER_FIGURES_DIR", str(PROJECT_ROOT / "results_metadata (2025-2026)"))
+)
+
+TITLE_COL = os.getenv("METADATA_TITLE_COL", "TI")
+ABSTRACT_COL = os.getenv("METADATA_ABSTRACT_COL", "AB")
+KEYWORD_COL = os.getenv("METADATA_KEYWORD_COL", "DE")

metadata_search/corpus.py

@@ -0,0 +1,101 @@
+"""Paper corpus helpers: CSV loading, embedding text, boolean tokenization."""
+
+from __future__ import annotations
+
+import csv
+import re
+from pathlib import Path
+
+from metadata_search.config import ABSTRACT_COL, KEYWORD_COL, TITLE_COL
+
+TOKEN_RE = re.compile(r"[a-z0-9]+(?:[-'][a-z0-9]+)?")
+
+STOPWORDS = frozenset(
+    "a an and are as at be been being by can did do does for from had has have "
+    "how in into is it its may of on or our that the their these this those to "
+    "using was were what when where which with within without".split()
+)
+
+IRREGULAR_LEMMAS = {
+    "analyses": "analysis",
+    "bacteria": "bacterium",
+    "children": "child",
+    "data": "datum",
+    "indices": "index",
+    "matrices": "matrix",
+    "media": "medium",
+    "mice": "mouse",
+    "species": "species",
+}
+
+ALIASES = {
+    "mbr": "membrane_bioreactor",
+    "membrane bioreactor": "membrane_bioreactor",
+    "wwtp": "wastewater_treatment_plant",
+    "wastewater treatment plant": "wastewater_treatment_plant",
+    "pfas": "per_polyfluoroalkyl_substances",
+    "per- and polyfluoroalkyl substances": "per_polyfluoroalkyl_substances",
+    "lca": "life_cycle_assessment",
+    "life cycle assessment": "life_cycle_assessment",
+    "cfd": "computational_fluid_dynamics",
+    "computational fluid dynamics": "computational_fluid_dynamics",
+    "gis": "geographic_information_system",
+    "geographic information system": "geographic_information_system",
+}
+
+
+def load_rows(csv_path: str | Path) -> list[dict[str, str]]:
+    with Path(csv_path).open(newline="", encoding="utf-8-sig") as handle:
+        return list(csv.DictReader(handle))
+
+
+def paper_text(
+    row: dict[str, str],
+    *,
+    title_col: str = TITLE_COL,
+    abstract_col: str = ABSTRACT_COL,
+    keyword_col: str = KEYWORD_COL,
+) -> str:
+    title = (row.get(title_col) or "").strip()
+    abstract = (row.get(abstract_col) or "").strip()
+    keywords = (row.get(keyword_col) or "").strip()
+    return f"{title}\n\n{abstract}\n\nKeywords: {keywords}".strip()
+
+
+def lemmatize(token: str) -> str:
+    if token in IRREGULAR_LEMMAS:
+        return IRREGULAR_LEMMAS[token]
+    if len(token) > 4 and token.endswith("ies"):
+        return f"{token[:-3]}y"
+    if len(token) > 5 and token.endswith(("ches", "shes", "sses", "xes", "zes")):
+        return token[:-2]
+    if len(token) > 3 and token.endswith("s") and not token.endswith(
+        ("ss", "us", "is", "ics")
+    ):
+        return token[:-1]
+    return token
+
+
+def _surface_tokens(text: str) -> list[str]:
+    return [token for token in TOKEN_RE.findall((text or "").lower()) if token not in STOPWORDS]
+
+
+def tokenize(text: str, *, include_phrases: bool = True) -> list[str]:
+    """Lexical features for boolean query matching."""
+    surface = _surface_tokens(text)
+    lemmas = [lemmatize(token) for token in surface]
+    features = list(surface)
+    features.extend(f"lemma:{lemma}" for lemma in lemmas)
+
+    if include_phrases:
+        features.extend(f"phrase:{left}_{right}" for left, right in zip(lemmas, lemmas[1:]))
+
+    lowered = " ".join(TOKEN_RE.findall((text or "").lower()))
+    surface_set = set(surface)
+    for alias, concept in ALIASES.items():
+        normalized_alias = " ".join(TOKEN_RE.findall(alias.lower()))
+        if (" " not in normalized_alias and normalized_alias in surface_set) or (
+            " " in normalized_alias and f" {normalized_alias} " in f" {lowered} "
+        ):
+            features.append(f"concept:{concept}")
+    return features

metadata_search/data/papers.csv

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:58382d4d472d662ba676c2bb666af4c34eab054f155479c5aa23839cf7fb3de7
+size 24137195

metadata_search/embedder.py

@@ -0,0 +1,48 @@
+"""Sentence-transformer embeddings for metadata vector search."""
+
+from __future__ import annotations
+
+from functools import lru_cache
+
+from metadata_search.config import EMBEDDING_MODEL
+
+
+@lru_cache(maxsize=1)
+def _load_model(model_name: str):
+    try:
+        from sentence_transformers import SentenceTransformer
+    except ImportError as exc:
+        raise SystemExit(
+            "Missing dependency: sentence-transformers. Install with `pip install -r requirements.txt`."
+        ) from exc
+
+    try:
+        return SentenceTransformer(model_name, local_files_only=True)
+    except Exception:
+        return SentenceTransformer(model_name)
+
+
+class MetadataEmbedder:
+    def __init__(self, model_name: str = EMBEDDING_MODEL, *, batch_size: int = 64):
+        self.model_name = model_name
+        self.batch_size = batch_size
+        self.model = _load_model(model_name)
+
+    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        if not texts:
+            return []
+        vectors = self.model.encode(
+            texts,
+            batch_size=self.batch_size,
+            normalize_embeddings=True,
+            show_progress_bar=len(texts) > 128,
+        )
+        return vectors.tolist()
+
+    def embed_query(self, query: str) -> list[float]:
+        vector = self.model.encode(
+            query,
+            normalize_embeddings=True,
+            show_progress_bar=False,
+        )
+        return vector.tolist()

metadata_search/hybrid_search.py

@@ -0,0 +1,46 @@
+"""RRF fusion for metadata BM25 + semantic (Qdrant) rankings."""
+
+from __future__ import annotations
+
+
+def rank_scores(scores: list[float], *, minimum_score: float = 0.0) -> list[int]:
+    """Return 1-based rank per row index; 0 means below minimum_score."""
+    ranked = sorted(
+        (index for index, score in enumerate(scores) if score > minimum_score),
+        key=lambda index: scores[index],
+        reverse=True,
+    )
+    ranks = [0] * len(scores)
+    for rank, index in enumerate(ranked, start=1):
+        ranks[index] = rank
+    return ranks
+
+
+def reciprocal_rank_scores(ranks: list[int], *, rrf_k: int = 60) -> list[float]:
+    return [1.0 / (rrf_k + rank) if rank > 0 else 0.0 for rank in ranks]
+
+
+def weighted_rrf_scores(
+    bm25_scores: list[float],
+    semantic_scores: list[float],
+    *,
+    bm25_weight: float = 0.5,
+    semantic_weight: float = 0.5,
+    rrf_k: int = 60,
+) -> tuple[list[float], list[int], list[float], list[int], list[float]]:
+    total_weight = bm25_weight + semantic_weight
+    if total_weight <= 0:
+        raise ValueError("bm25_weight + semantic_weight must be > 0")
+
+    bm25_weight = bm25_weight / total_weight
+    semantic_weight = semantic_weight / total_weight
+
+    bm25_ranks = rank_scores(bm25_scores, minimum_score=0.0)
+    semantic_ranks = rank_scores(semantic_scores)
+    bm25_rrf = reciprocal_rank_scores(bm25_ranks, rrf_k=rrf_k)
+    semantic_rrf = reciprocal_rank_scores(semantic_ranks, rrf_k=rrf_k)
+    final_scores = [
+        bm25_weight * bm25_score + semantic_weight * semantic_score
+        for bm25_score, semantic_score in zip(bm25_rrf, semantic_rrf, strict=True)
+    ]
+    return final_scores, bm25_ranks, bm25_rrf, semantic_ranks, semantic_rrf

metadata_search/mcp_retrieval.py

@@ -0,0 +1,76 @@
+"""Retrieval layer for Flask — MCP tools or direct Python (USE_MCP=0)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from backend.chat_settings import truncate_summary
+from backend.mcp_sync import call_mcp_tool, use_mcp
+from local_rag.summaries import list_papers, summaries_for_documents
+
+from metadata_search.saved_papers import delete_saved, list_saved, save_paper
+from metadata_search.search_service import search_metadata
+
+
+def search_papers(query: str, *, top_k: int = 10) -> tuple[list[dict[str, Any]], str]:
+    from metadata_search.search_service import score_mode_for_results
+
+    results = search_metadata(query, top_k=top_k)
+    return results, score_mode_for_results(results)
+
+
+def list_saved_papers() -> list[dict[str, Any]]:
+    return list_saved()
+
+
+def save_paper_record(payload: dict[str, Any]) -> dict[str, Any]:
+    return save_paper(payload)
+
+
+def remove_saved_paper(paper_id: str) -> bool:
+    return delete_saved(paper_id)
+
+
+def list_papers_catalog() -> list[dict[str, Any]]:
+    return list_papers()
+
+
+def build_context_for_documents(
+    document_ids: list[str],
+    *,
+    question: str = "",
+) -> tuple[str, list[dict[str, Any]]]:
+    if not use_mcp():
+        sources = summaries_for_documents(document_ids)
+        return _build_summary_context(sources, question=question)
+
+    if question:
+        payload = call_mcp_tool(
+            "build_paper_context",
+            {"document_ids": document_ids, "question": question},
+        )
+    else:
+        payload = call_mcp_tool("build_paper_context", {"document_ids": document_ids})
+
+    if payload.get("error"):
+        raise ValueError(str(payload["error"]))
+
+    summary_payload = call_mcp_tool("get_paper_summaries", {"document_ids": document_ids})
+    sources = summary_payload.get("sources", [])
+    if not sources:
+        raise ValueError("No summaries found for the selected papers.")
+    return str(payload.get("context", "")), sources
+
+
+def _build_summary_context(sources: list[dict], *, question: str = "") -> tuple[str, list[dict]]:
+    if not sources:
+        raise ValueError("No summaries found for the selected papers.")
+    n_papers = len(sources)
+    parts: list[str] = []
+    for index, source in enumerate(sources, start=1):
+        parts.append(
+            f"Paper {index}: {source['title']}\n"
+            f"File: {source['file_name']}\n\n"
+            f"{truncate_summary(source['text'], n_papers=n_papers, question=question)}"
+        )
+    return "\n\n---\n\n".join(parts), sources

metadata_search/paper_figures.py

@@ -0,0 +1,224 @@
+"""Index and serve extracted paper figures from results_metadata folders."""
+
+from __future__ import annotations
+
+import json
+import logging
+import unicodedata
+from pathlib import Path
+from urllib.parse import quote
+
+from metadata_search.config import PAPER_FIGURES_DIR
+
+logger = logging.getLogger(__name__)
+
+STEM_PREFIX_MIN = 40
+
+_index: dict[str, str] | None = None
+_manifests: dict[str, dict] | None = None
+
+
+def figures_dir() -> Path | None:
+    root = PAPER_FIGURES_DIR
+    if not root.is_dir():
+        return None
+    return root
+
+
+def normalize_stem(name: str) -> str:
+    text = unicodedata.normalize("NFKC", (name or "").strip())
+    lower = text.lower()
+    for suffix in (".txt", ".pdf"):
+        if lower.endswith(suffix):
+            text = text[: -len(suffix)]
+            break
+    return text.strip().lower()
+
+
+def stems_match(left: str, right: str, *, min_prefix: int = STEM_PREFIX_MIN) -> bool:
+    a = normalize_stem(left)
+    b = normalize_stem(right)
+    if not a or not b:
+        return False
+    if a == b:
+        return True
+    shorter, longer = (a, b) if len(a) <= len(b) else (b, a)
+    if len(shorter) >= min_prefix and longer.startswith(shorter):
+        return True
+    return False
+
+
+def _register_stem(stem_index: dict[str, str], key: str, folder: str) -> None:
+    stem = normalize_stem(key)
+    if not stem:
+        return
+    existing = stem_index.get(stem)
+    if existing and existing != folder:
+        logger.debug("Figure stem collision: %r -> %s (kept %s)", key, folder, existing)
+        return
+    stem_index[stem] = folder
+
+
+def _load_index() -> tuple[dict[str, str], dict[str, dict]]:
+    global _index, _manifests
+    if _index is not None and _manifests is not None:
+        return _index, _manifests
+
+    stem_index: dict[str, str] = {}
+    manifests: dict[str, dict] = {}
+    root = figures_dir()
+    if root is None:
+        _index = {}
+        _manifests = {}
+        return _index, _manifests
+
+    for entry in sorted(root.iterdir()):
+        if not entry.is_dir():
+            continue
+        manifest_path = entry / "manifest.json"
+        if not manifest_path.is_file():
+            continue
+        try:
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            logger.warning("Skipping figure manifest %s: %s", manifest_path, exc)
+            continue
+
+        folder = entry.name
+        manifests[folder] = manifest
+        _register_stem(stem_index, folder, folder)
+
+        source_pdf = (manifest.get("source_pdf") or "").strip()
+        if source_pdf:
+            _register_stem(stem_index, Path(source_pdf).name, folder)
+
+    _index = stem_index
+    _manifests = manifests
+    logger.info("Indexed figures for %s papers under %s", len(manifests), root)
+    return _index, _manifests
+
+
+def figure_paper_count() -> int:
+    _, manifests = _load_index()
+    return len(manifests)
+
+
+def _lookup_folder(*, txt_file: str = "", pdf_file: str = "", paper_key: str = "") -> str | None:
+    stem_index, manifests = _load_index()
+    if not manifests:
+        return None
+
+    if paper_key:
+        if paper_key in manifests:
+            return paper_key
+        target = normalize_stem(paper_key)
+        for folder in manifests:
+            if normalize_stem(folder) == target:
+                return folder
+
+    for candidate in (txt_file, pdf_file):
+        stem = normalize_stem(candidate)
+        if not stem:
+            continue
+        if stem in stem_index:
+            return stem_index[stem]
+        for key, folder in stem_index.items():
+            if stems_match(stem, key):
+                return folder
+    return None
+
+
+def resolve_figure_folder(
+    *,
+    txt_file: str = "",
+    pdf_file: str = "",
+    paper_key: str = "",
+    metadata_row_index: int | None = None,
+) -> str | None:
+    folder = _lookup_folder(txt_file=txt_file, pdf_file=pdf_file, paper_key=paper_key)
+    if folder or metadata_row_index is None:
+        return folder
+
+    try:
+        from metadata_search.search_service import get_rows
+
+        rows = get_rows()
+        if metadata_row_index < 0 or metadata_row_index >= len(rows):
+            return None
+        row = rows[metadata_row_index]
+        return _lookup_folder(
+            txt_file=(row.get("TXT_file") or row.get("txt_file") or "").strip(),
+            pdf_file=(row.get("PDF_file") or row.get("pdf_file") or "").strip(),
+        )
+    except Exception as exc:
+        logger.debug("Figure lookup via metadata_row_index failed: %s", exc)
+        return None
+
+
+def figure_image_url(folder: str, filename: str) -> str:
+    return f"/figures/{quote(folder, safe='')}/{quote(filename, safe='')}"
+
+
+def figures_payload_for_folder(folder: str) -> dict | None:
+    _, manifests = _load_index()
+    manifest = manifests.get(folder)
+    if not manifest:
+        return None
+
+    items: list[dict] = []
+    for entry in manifest.get("items") or []:
+        image = (entry.get("image") or "").strip()
+        if not image:
+            continue
+        items.append(
+            {
+                "kind": entry.get("kind") or "figure",
+                "label": entry.get("label") or "",
+                "page": entry.get("page"),
+                "caption": (entry.get("caption") or "").strip(),
+                "image": image,
+                "image_url": figure_image_url(folder, image),
+                "caption_file": entry.get("caption_file"),
+                "row_count": entry.get("row_count"),
+                "column_count": entry.get("column_count"),
+            }
+        )
+
+    source_pdf = (manifest.get("source_pdf") or "").strip()
+    return {
+        "paper_key": folder,
+        "source_pdf": Path(source_pdf).name if source_pdf else folder,
+        "item_count": len(items),
+        "items": items,
+    }
+
+
+def figures_payload_for_paper(
+    *,
+    txt_file: str = "",
+    pdf_file: str = "",
+    paper_key: str = "",
+    metadata_row_index: int | None = None,
+) -> dict | None:
+    folder = resolve_figure_folder(
+        txt_file=txt_file,
+        pdf_file=pdf_file,
+        paper_key=paper_key,
+        metadata_row_index=metadata_row_index,
+    )
+    if not folder:
+        return None
+    return figures_payload_for_folder(folder)
+
+
+def paper_has_figures(
+    *,
+    txt_file: str = "",
+    pdf_file: str = "",
+    metadata_row_index: int | None = None,
+) -> bool:
+    return resolve_figure_folder(
+        txt_file=txt_file,
+        pdf_file=pdf_file,
+        metadata_row_index=metadata_row_index,
+    ) is not None

metadata_search/qdrant_metadata.py

@@ -0,0 +1,237 @@
+"""Qdrant vector index for paper metadata (title / abstract / keywords)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+from qdrant_client import QdrantClient
+from qdrant_client.http import models as qmodels
+
+from metadata_search.config import (
+    ABSTRACT_COL,
+    EMBEDDING_MODEL,
+    KEYWORD_COL,
+    METADATA_QDRANT_COLLECTION,
+    METADATA_QDRANT_MANIFEST,
+    METADATA_QDRANT_PATH,
+    PAPER_CSV,
+    TITLE_COL,
+)
+from metadata_search.corpus import load_rows, paper_text
+from metadata_search.embedder import MetadataEmbedder
+
+
+@dataclass(frozen=True)
+class MetadataSearchHit:
+    row_index: int
+    score: float
+
+
+class MetadataVectorStore:
+    """One Qdrant point per papers.csv row."""
+
+    def __init__(
+        self,
+        *,
+        qdrant_path: Path = METADATA_QDRANT_PATH,
+        collection_name: str = METADATA_QDRANT_COLLECTION,
+        model_name: str = EMBEDDING_MODEL,
+        title_weight: float = 2.0,
+        abstract_weight: float = 1.0,
+        keyword_weight: float = 1.5,
+        batch_size: int = 64,
+    ) -> None:
+        self.qdrant_path = qdrant_path
+        self.collection_name = collection_name
+        self.model_name = model_name
+        self.title_weight = title_weight
+        self.abstract_weight = abstract_weight
+        self.keyword_weight = keyword_weight
+        self.batch_size = batch_size
+        self.qdrant_path.mkdir(parents=True, exist_ok=True)
+        self.client = QdrantClient(path=str(self.qdrant_path))
+        self.manifest_path = METADATA_QDRANT_MANIFEST
+
+    def points_count(self) -> int:
+        if not self.client.collection_exists(self.collection_name):
+            return 0
+        info = self.client.get_collection(self.collection_name)
+        return int(info.points_count or 0)
+
+    def _content_hash(self, rows: list[dict[str, str]]) -> str:
+        digest = hashlib.sha256()
+        for row in rows:
+            digest.update(paper_text(row).encode("utf-8"))
+            digest.update(b"\0")
+        digest.update(self.model_name.encode("utf-8"))
+        digest.update(
+            f"{self.title_weight}:{self.abstract_weight}:{self.keyword_weight}".encode("utf-8")
+        )
+        return digest.hexdigest()
+
+    def _load_manifest(self) -> dict[str, Any]:
+        if not self.manifest_path.exists():
+            return {}
+        return json.loads(self.manifest_path.read_text(encoding="utf-8"))
+
+    def _write_manifest(self, payload: dict[str, Any]) -> None:
+        self.manifest_path.parent.mkdir(parents=True, exist_ok=True)
+        self.manifest_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+    def is_current(self, rows: list[dict[str, str]]) -> bool:
+        manifest = self._load_manifest()
+        if not manifest:
+            return False
+        if manifest.get("row_count") != len(rows):
+            return False
+        if manifest.get("content_hash") != self._content_hash(rows):
+            return False
+        if manifest.get("model_name") != self.model_name:
+            return False
+        return self.points_count() == len(rows)
+
+    def build_weighted_vectors(self, rows: list[dict[str, str]]) -> list[list[float]]:
+        embedder = MetadataEmbedder(self.model_name, batch_size=self.batch_size)
+        title_texts = [str(row.get(TITLE_COL) or "") for row in rows]
+        abstract_texts = [str(row.get(ABSTRACT_COL) or "") for row in rows]
+        keyword_texts = [str(row.get(KEYWORD_COL) or "") for row in rows]
+        field_texts = title_texts + abstract_texts + keyword_texts
+        field_embeddings = np.asarray(
+            embedder.embed_documents(field_texts),
+            dtype=float,
+        )
+        count = len(rows)
+        title_embeddings = field_embeddings[:count]
+        abstract_embeddings = field_embeddings[count : 2 * count]
+        keyword_embeddings = field_embeddings[2 * count :]
+
+        title_mask = np.asarray([bool(text.strip()) for text in title_texts], dtype=float)[:, None]
+        abstract_mask = np.asarray([bool(text.strip()) for text in abstract_texts], dtype=float)[:, None]
+        keyword_mask = np.asarray([bool(text.strip()) for text in keyword_texts], dtype=float)[:, None]
+
+        combined = (
+            self.title_weight * title_embeddings * title_mask
+            + self.abstract_weight * abstract_embeddings * abstract_mask
+            + self.keyword_weight * keyword_embeddings * keyword_mask
+        )
+        norms = np.linalg.norm(combined, axis=1, keepdims=True)
+        combined = np.divide(combined, norms, out=np.zeros_like(combined), where=norms != 0)
+        return combined.tolist()
+
+    def reset_collection(self, vector_size: int) -> None:
+        if self.client.collection_exists(self.collection_name):
+            self.client.delete_collection(self.collection_name)
+        self.client.create_collection(
+            collection_name=self.collection_name,
+            vectors_config=qmodels.VectorParams(
+                size=vector_size,
+                distance=qmodels.Distance.COSINE,
+            ),
+        )
+
+    def upsert_rows(self, rows: list[dict[str, str]], vectors: list[list[float]]) -> None:
+        if len(rows) != len(vectors):
+            raise ValueError("rows and vectors length mismatch")
+        if not rows:
+            raise ValueError("No rows to index")
+
+        vector_size = len(vectors[0])
+        self.reset_collection(vector_size)
+
+        points: list[qmodels.PointStruct] = []
+        for row_index, (row, vector) in enumerate(zip(rows, vectors, strict=True)):
+            points.append(
+                qmodels.PointStruct(
+                    id=row_index,
+                    vector=vector,
+                    payload={
+                        "row_index": row_index,
+                        "title": (row.get(TITLE_COL) or "").strip(),
+                        "year": (row.get("PY") or "").strip(),
+                        "txt_file": (row.get("TXT_file") or row.get("txt_file") or "").strip(),
+                    },
+                )
+            )
+
+        batch = 1000
+        for start in range(0, len(points), batch):
+            end = min(start + batch, len(points))
+        
+            self.client.upsert(
+                collection_name=self.collection_name,
+                points=points[start:end],
+            )
+        
+            print(f"Qdrant indexing: {end}/{len(points)} points")
+
+        self._write_manifest(
+            {
+                "csv_path": str(Path(PAPER_CSV).resolve()),
+                "model_name": self.model_name,
+                "row_count": len(rows),
+                "content_hash": self._content_hash(rows),
+                "title_weight": self.title_weight,
+                "abstract_weight": self.abstract_weight,
+                "keyword_weight": self.keyword_weight,
+                "collection_name": self.collection_name,
+                "qdrant_path": str(self.qdrant_path.resolve()),
+            }
+        )
+
+    def rebuild(self, rows: list[dict[str, str]] | None = None, *, force: bool = False) -> dict[str, Any]:
+        rows = rows if rows is not None else load_rows(PAPER_CSV)
+        if not force and self.is_current(rows):
+            return {
+                "rebuilt": False,
+                "row_count": len(rows),
+                "points_count": self.points_count(),
+                "message": "Metadata vector index is already up to date.",
+            }
+
+        vectors = self.build_weighted_vectors(rows)
+        self.upsert_rows(rows, vectors)
+        return {
+            "rebuilt": True,
+            "row_count": len(rows),
+            "points_count": self.points_count(),
+            "model_name": self.model_name,
+            "qdrant_path": str(self.qdrant_path.resolve()),
+        }
+
+    def search(self, query_vector: list[float], *, limit: int) -> list[MetadataSearchHit]:
+        if not self.client.collection_exists(self.collection_name):
+            return []
+
+        response = self.client.query_points(
+            collection_name=self.collection_name,
+            query=query_vector,
+            limit=limit,
+            with_payload=True,
+        )
+
+        hits: list[MetadataSearchHit] = []
+        for point in response.points:
+            payload = point.payload or {}
+            row_index = int(payload.get("row_index", point.id))
+            hits.append(
+                MetadataSearchHit(
+                    row_index=row_index,
+                    score=float(point.score or 0.0),
+                )
+            )
+        return hits
+
+    def semantic_scores(self, query_vector: list[float], row_count: int) -> list[float]:
+        """Cosine similarity for every row (via Qdrant top-N over full collection)."""
+        scores = [0.0] * row_count
+        if row_count <= 0:
+            return scores
+        for hit in self.search(query_vector, limit=row_count):
+            if 0 <= hit.row_index < row_count:
+                scores[hit.row_index] = hit.score
+        return scores

metadata_search/saved_papers.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import json
+import os
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from backend.paper_linking import enrich_metadata_row, format_authors
+from metadata_search.corpus import load_rows
+from metadata_search.config import PAPER_CSV, PROJECT_ROOT
+
+_metadata_rows: list[dict[str, str]] | None = None
+
+
+def _get_metadata_rows() -> list[dict[str, str]]:
+    global _metadata_rows
+    if _metadata_rows is None:
+        _metadata_rows = load_rows(PAPER_CSV)
+    return _metadata_rows
+
+
+def enrich_saved_record(paper: dict[str, Any]) -> dict[str, Any]:
+    """Fill authors (and display helpers) from bibliographic CSV when missing."""
+    enriched = dict(paper)
+    row_index = paper.get("metadata_row_index")
+    if row_index is not None and not enriched.get("authors"):
+        rows = _get_metadata_rows()
+        if 0 <= row_index < len(rows):
+            meta = enrich_metadata_row(rows[row_index], row_index)
+            enriched["authors"] = meta["authors"]
+    authors = (enriched.get("authors") or "").strip()
+    enriched["authors_display"] = format_authors(authors)
+    return enriched
+
+
+def _store_path() -> Path:
+    override = os.getenv("SAVED_PAPERS_PATH")
+    if override:
+        return Path(override).expanduser().resolve()
+    return PROJECT_ROOT / "local_rag_data" / "saved_papers.json"
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+
+def _load_all() -> list[dict[str, Any]]:
+    path = _store_path()
+    if not path.exists():
+        return []
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _save_all(papers: list[dict[str, Any]]) -> None:
+    path = _store_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(papers, indent=2), encoding="utf-8")
+
+
+def _dedupe_key(paper: dict[str, Any]) -> str | None:
+    document_id = (paper.get("document_id") or "").strip()
+    if document_id:
+        return f"doc:{document_id}"
+    row_index = paper.get("metadata_row_index")
+    if row_index is not None:
+        return f"row:{row_index}"
+    doi = (paper.get("doi") or "").strip().lower()
+    if doi:
+        return f"doi:{doi}"
+    title = (paper.get("title") or "").strip().lower()
+    if title:
+        return f"title:{title}"
+    return None
+
+
+def list_saved() -> list[dict[str, Any]]:
+    papers = [enrich_saved_record(paper) for paper in _load_all()]
+    papers.sort(key=lambda item: item.get("saved_at", ""), reverse=True)
+    return papers
+
+
+def is_saved(*, document_id: str | None = None, metadata_row_index: int | None = None) -> bool:
+    for paper in _load_all():
+        if document_id and paper.get("document_id") == document_id:
+            return True
+        if metadata_row_index is not None and paper.get("metadata_row_index") == metadata_row_index:
+            return True
+    return False
+
+
+def save_paper(payload: dict[str, Any]) -> dict[str, Any]:
+    papers = _load_all()
+    candidate = {
+        "id": payload.get("id") or str(uuid.uuid4()),
+        "title": (payload.get("title") or "").strip(),
+        "authors": (payload.get("authors") or "").strip(),
+        "abstract": (payload.get("abstract") or "").strip(),
+        "keywords": (payload.get("keywords") or "").strip(),
+        "doi": (payload.get("doi") or "").strip(),
+        "year": (payload.get("year") or "").strip(),
+        "document_id": (payload.get("document_id") or "").strip() or None,
+        "metadata_row_index": payload.get("metadata_row_index"),
+        "txt_file": (payload.get("txt_file") or "").strip(),
+        "source": payload.get("source") or "metadata",
+        "score_label": payload.get("score_label"),
+        "rrf_score": payload.get("rrf_score"),
+        "final_score": payload.get("final_score"),
+        "bm25_score": payload.get("bm25_score"),
+        "semantic_score": payload.get("semantic_score"),
+        "bm25_rank": payload.get("bm25_rank"),
+        "semantic_rank": payload.get("semantic_rank"),
+        "bm25_rrf_score": payload.get("bm25_rrf_score"),
+        "semantic_rrf_score": payload.get("semantic_rrf_score"),
+        "saved_at": payload.get("saved_at") or _now_iso(),
+    }
+    if not candidate["title"]:
+        raise ValueError("title is required")
+
+    key = _dedupe_key(candidate)
+    if key:
+        for index, existing in enumerate(papers):
+            if _dedupe_key(existing) == key:
+                merged = enrich_saved_record(
+                    {**existing, **candidate, "id": existing["id"], "saved_at": _now_iso()}
+                )
+                papers[index] = merged
+                _save_all(papers)
+                return merged
+
+    saved = enrich_saved_record(candidate)
+    papers.append(saved)
+    _save_all(papers)
+    return saved
+
+
+def delete_saved(paper_id: str) -> bool:
+    papers = _load_all()
+    remaining = [paper for paper in papers if paper.get("id") != paper_id]
+    if len(remaining) == len(papers):
+        return False
+    _save_all(remaining)
+    return True

metadata_search/search_query_planner.py

@@ -0,0 +1,113 @@
+"""LLM-backed search query planning for agent retrieval (Scholar-style operators)."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass
+
+from metadata_search.boolean_query import parse_boolean_query, plan_summary
+
+_JSON_BLOCK = re.compile(r"\{[\s\S]*\}")
+
+
+@dataclass(frozen=True)
+class SearchPlan:
+    search_query: str
+    selection_hint: str
+    planner: str
+    parsed: dict[str, object]
+
+
+def _extract_json_object(text: str) -> dict[str, object]:
+    match = _JSON_BLOCK.search(text or "")
+    if not match:
+        raise ValueError("Search planner did not return JSON.")
+    payload = json.loads(match.group(0))
+    if not isinstance(payload, dict):
+        raise ValueError("Search planner JSON must be an object.")
+    return payload
+
+
+def _normalize_plan(raw: dict[str, object], *, planner: str, fallback_message: str) -> SearchPlan:
+    search_query = str(raw.get("search_query") or raw.get("query") or "").strip()
+    selection_hint = str(raw.get("selection_hint") or raw.get("focus") or fallback_message).strip()
+    if not search_query:
+        raise ValueError("Search planner JSON missing search_query.")
+    parsed = parse_boolean_query(search_query)
+    return SearchPlan(
+        search_query=parsed.raw,
+        selection_hint=selection_hint or fallback_message,
+        planner=planner,
+        parsed=plan_summary(parsed),
+    )
+
+
+def _plan_with_rules(message: str) -> SearchPlan:
+    text = message.strip()
+    parsed = parse_boolean_query(text)
+    return SearchPlan(
+        search_query=parsed.raw,
+        selection_hint=text,
+        planner="rules",
+        parsed=plan_summary(parsed),
+    )
+
+
+def _plan_with_llm(message: str) -> SearchPlan:
+    from backend.waterscope_agent import run_agent
+
+    instructions = (
+        "You convert research chat messages into bibliographic search queries.\n"
+        "Do NOT include conversational prefixes like 'find papers on'.\n"
+        "Do NOT include filler words (though, however, please, just, really, etc.) in search_query.\n"
+        "Use Google Scholar-style operators when they help:\n"
+        '- "exact phrase" for multi-word concepts\n'
+        "- +term for required concepts\n"
+        "- -term to exclude a single word\n"
+        '- -"exact phrase" to exclude multi-word topics (always quote multi-word exclusions)\n'
+        "- plain terms for optional relevance\n\n"
+        'Example: "microplastics not drinking water though" → search_query: microplastics -"drinking water"\n\n'
+        'Return ONLY one JSON object, no markdown:\n'
+        '{"search_query":"...", "selection_hint":"what kinds of papers best answer the user"}\n'
+    )
+    raw, _ = run_agent(
+        context=instructions,
+        question=message,
+        model=os.getenv("AGENT_SEARCH_PLANNER_MODEL") or os.getenv("WATERSCOPE_MODEL", "dpo"),
+        application="general_qa",
+        max_new_tokens=int(os.getenv("AGENT_SEARCH_PLANNER_MAX_TOKENS", "256")),
+    )
+    payload = _extract_json_object(raw)
+    return _normalize_plan(payload, planner="llm", fallback_message=message)
+
+
+def planner_backend() -> str:
+    explicit = os.getenv("AGENT_SEARCH_PLANNER", "auto").strip().lower()
+    if explicit in {"llm", "rules"}:
+        return explicit
+    return "auto"
+
+
+def plan_search_query(message: str, *, explicit_query: str = "") -> SearchPlan:
+    """Build the metadata search string used by agent MCP tools."""
+    if explicit_query.strip():
+        parsed = parse_boolean_query(explicit_query.strip())
+        return SearchPlan(
+            search_query=parsed.raw,
+            selection_hint=message.strip() or explicit_query.strip(),
+            planner="explicit",
+            parsed=plan_summary(parsed),
+        )
+
+    backend = planner_backend()
+    if backend == "rules":
+        return _plan_with_rules(message)
+    if backend == "llm":
+        return _plan_with_llm(message)
+
+    try:
+        return _plan_with_llm(message)
+    except Exception:
+        return _plan_with_rules(message)

metadata_search/search_service.py

@@ -0,0 +1,316 @@
+"""Hybrid metadata search: BM25 + Qdrant semantic vectors fused with weighted RRF."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from backend.paper_linking import enrich_metadata_row, format_authors
+from metadata_search.boolean_query import matches_boolean_query, parse_boolean_query, plan_summary
+from metadata_search.bm25_search import BM25Index
+from metadata_search.config import PAPER_CSV
+from metadata_search.corpus import load_rows
+from metadata_search.embedder import MetadataEmbedder
+from metadata_search.hybrid_search import weighted_rrf_scores
+from metadata_search.qdrant_metadata import MetadataVectorStore
+from metadata_search.saved_papers import is_saved, list_saved
+
+try:
+    from metadata_search.paper_figures import paper_has_figures as _paper_has_figures
+except Exception:
+    def _paper_has_figures(**_kwargs: object) -> bool:
+        return False
+
+DEFAULT_TOP_K = 10
+BM25_WEIGHT = 0.5
+SEMANTIC_WEIGHT = 0.5
+RRF_K = 60
+TITLE_WEIGHT = 2.0
+KEYWORD_WEIGHT = 1.5
+
+logger = logging.getLogger(__name__)
+
+_rows: list[dict[str, str]] | None = None
+_bm25_index: BM25Index | None = None
+_vector_store: MetadataVectorStore | None = None
+_embedder: MetadataEmbedder | None = None
+
+
+def use_semantic_search() -> bool:
+    """Hybrid BM25 + Qdrant semantic RRF when true; BM25-only fallback when false."""
+    return os.getenv("METADATA_SEMANTIC_SEARCH", "1").lower() in {"1", "true", "yes"}
+
+
+def metadata_auto_rebuild_enabled() -> bool:
+    """Whether this runtime is allowed to create/rebuild the metadata vector index."""
+    return os.getenv("METADATA_AUTO_REBUILD", "0").lower() in {"1", "true", "yes"}
+
+
+def search_score_mode() -> str:
+    return "rrf" if use_semantic_search() else "bm25"
+
+
+def score_mode_for_results(results: list[dict[str, object]]) -> str:
+    if not results:
+        return search_score_mode()
+    return "rrf" if results[0].get("score_label") == "RRF" else "bm25"
+
+
+def get_rows() -> list[dict[str, str]]:
+    global _rows
+    if _rows is None:
+        _rows = load_rows(PAPER_CSV)
+    return _rows
+
+
+def get_bm25_index() -> BM25Index:
+    global _bm25_index
+    if _bm25_index is None:
+        _bm25_index = BM25Index(
+            get_rows(),
+            title_weight=TITLE_WEIGHT,
+            keyword_weight=KEYWORD_WEIGHT,
+        )
+    return _bm25_index
+
+
+def get_vector_store() -> MetadataVectorStore:
+    global _vector_store
+    if _vector_store is None:
+        _vector_store = MetadataVectorStore()
+    return _vector_store
+
+
+def get_embedder() -> MetadataEmbedder:
+    global _embedder
+    if _embedder is None:
+        _embedder = MetadataEmbedder()
+    return _embedder
+
+
+def ensure_metadata_index(*, force: bool = False) -> dict:
+    """Build or refresh the metadata Qdrant index when papers.csv changes."""
+    rows = get_rows()
+    store = get_vector_store()
+    if force or not store.is_current(rows):
+        logger.info("Rebuilding metadata vector index for %s papers …", len(rows))
+        return store.rebuild(rows, force=True)
+    return {
+        "rebuilt": False,
+        "row_count": len(rows),
+        "points_count": store.points_count(),
+    }
+
+
+def _score_fields(
+    final_score: float,
+    bm25_score: float,
+    *,
+    use_semantic: bool,
+) -> dict[str, object]:
+    return {
+        "final_score": final_score,
+        "rrf_score": final_score if use_semantic else None,
+        "score": final_score,
+        "score_label": "RRF" if use_semantic else "BM25",
+        "bm25_score": bm25_score,
+    }
+
+
+def hybrid_search(query: str, *, top_k: int = DEFAULT_TOP_K, semantic: bool | None = None) -> list[dict[str, object]]:
+    rows = get_rows()
+    parsed = parse_boolean_query(query)
+    ranking_query = parsed.ranking_text or query
+    bm25_scores = get_bm25_index().score(ranking_query)
+    use_semantic = use_semantic_search() if semantic is None else semantic
+
+    semantic_scores: list[float]
+    if use_semantic:
+        try:
+            store = get_vector_store()
+            if store.points_count() == 0:
+                if metadata_auto_rebuild_enabled():
+                    ensure_metadata_index(force=True)
+                else:
+                    raise RuntimeError(
+                        "Prebuilt Qdrant index is unavailable and METADATA_AUTO_REBUILD=0; "
+                        "falling back to BM25 search."
+                    )
+            query_vector = get_embedder().embed_query(ranking_query)
+            semantic_scores = store.semantic_scores(query_vector, len(rows))
+        except Exception as exc:
+            logger.warning("Semantic search unavailable, using BM25 only: %s", exc)
+            use_semantic = False
+            semantic_scores = [0.0] * len(bm25_scores)
+    else:
+        semantic_scores = [0.0] * len(bm25_scores)
+
+    if use_semantic:
+        final_scores, bm25_ranks, bm25_rrf, semantic_ranks, semantic_rrf = weighted_rrf_scores(
+            bm25_scores,
+            semantic_scores,
+            bm25_weight=BM25_WEIGHT,
+            semantic_weight=SEMANTIC_WEIGHT,
+            rrf_k=RRF_K,
+        )
+    else:
+        ranked = sorted(
+            range(len(bm25_scores)),
+            key=lambda index: bm25_scores[index],
+            reverse=True,
+        )
+        bm25_ranks = [0] * len(bm25_scores)
+        for rank, index in enumerate(ranked, start=1):
+            bm25_ranks[index] = rank
+        semantic_ranks = [0] * len(bm25_scores)
+        final_scores = list(bm25_scores)
+        bm25_rrf = list(bm25_scores)
+        semantic_rrf = [0.0] * len(bm25_scores)
+
+    ranked_indices = sorted(
+        range(len(final_scores)),
+        key=lambda index: final_scores[index],
+        reverse=True,
+    )
+
+    results: list[dict[str, object]] = []
+    for row_index in ranked_indices:
+        row = rows[row_index]
+        enriched = enrich_metadata_row(row, row_index)
+        match_row = {
+            "title": enriched["title"],
+            "abstract": enriched["abstract"],
+            "full_abstract": enriched["abstract"],
+            "keywords": enriched["keywords"],
+        }
+        if not matches_boolean_query(match_row, parsed):
+            continue
+
+        rank = len(results) + 1
+        abstract = enriched["abstract"]
+        display_abstract = abstract
+        if len(display_abstract) > 450:
+            display_abstract = f"{display_abstract[:450].rstrip()}..."
+        saved = is_saved(
+            document_id=enriched.get("document_id"),
+            metadata_row_index=enriched.get("metadata_row_index"),
+        )
+        txt_file = enriched.get("txt_file", "")
+        pdf_file = (row.get("PDF_file") or row.get("pdf_file") or "").strip()
+        has_figures = _paper_has_figures(
+            txt_file=txt_file,
+            pdf_file=pdf_file,
+            metadata_row_index=enriched.get("metadata_row_index"),
+        )
+        results.append(
+            {
+                "rank": rank,
+                **_score_fields(
+                    final_scores[row_index],
+                    bm25_scores[row_index],
+                    use_semantic=use_semantic,
+                ),
+                "semantic_score": semantic_scores[row_index],
+                "bm25_rank": bm25_ranks[row_index],
+                "semantic_rank": semantic_ranks[row_index],
+                "bm25_rrf_score": bm25_rrf[row_index] if use_semantic else bm25_scores[row_index],
+                "semantic_rrf_score": semantic_rrf[row_index] if use_semantic else 0.0,
+                "title": enriched["title"],
+                "authors": enriched["authors"],
+                "authors_display": format_authors(enriched["authors"]),
+                "abstract": display_abstract,
+                "full_abstract": abstract,
+                "keywords": enriched["keywords"],
+                "year": enriched["year"],
+                "doi": enriched["doi"],
+                "document_id": enriched.get("document_id"),
+                "metadata_row_index": enriched.get("metadata_row_index"),
+                "txt_file": txt_file,
+                "pdf_file": pdf_file,
+                "has_figures": has_figures,
+                "is_saved": saved,
+                "search_query_parsed": plan_summary(parsed),
+            }
+        )
+        if len(results) >= top_k:
+            break
+    return results
+
+
+def document_ids_from_ranks(results: list[dict[str, object]], ranks: list[int]) -> list[str]:
+    by_rank = {int(item["rank"]): item for item in results}
+    ids: list[str] = []
+    for rank in ranks:
+        item = by_rank.get(rank)
+        if not item:
+            raise ValueError(f"No search result at rank {rank}")
+        doc_id = item.get("document_id")
+        if not doc_id:
+            raise ValueError(f"Result #{rank} has no linked chat document_id")
+        ids.append(str(doc_id))
+    return ids
+
+
+def search_metadata(query: str, *, top_k: int = DEFAULT_TOP_K) -> list[dict[str, object]]:
+    """Metadata search with save_payload for MCP and API consumers."""
+    base_results = hybrid_search(query, top_k=top_k)
+    results: list[dict[str, object]] = []
+    saved_list = list_saved()
+    for item in base_results:
+        saved = bool(item.get("is_saved"))
+        saved_match = next(
+            (
+                saved_item
+                for saved_item in saved_list
+                if saved_item.get("metadata_row_index") == item.get("metadata_row_index")
+                or (
+                    item.get("document_id")
+                    and saved_item.get("document_id") == item.get("document_id")
+                )
+            ),
+            None,
+        )
+        save_payload = {
+            "title": item["title"],
+            "authors": item.get("authors", ""),
+            "abstract": item.get("full_abstract") or item["abstract"],
+            "keywords": item["keywords"],
+            "doi": item["doi"],
+            "year": item["year"],
+            "document_id": item.get("document_id"),
+            "metadata_row_index": item.get("metadata_row_index"),
+            "txt_file": item.get("txt_file", ""),
+            "pdf_file": item.get("pdf_file", ""),
+            "source": "metadata",
+            "score_label": item.get("score_label"),
+            "rrf_score": item.get("rrf_score"),
+            "final_score": item.get("final_score"),
+            "bm25_score": item.get("bm25_score"),
+            "semantic_score": item.get("semantic_score"),
+            "bm25_rank": item.get("bm25_rank"),
+            "semantic_rank": item.get("semantic_rank"),
+            "bm25_rrf_score": item.get("bm25_rrf_score"),
+            "semantic_rrf_score": item.get("semantic_rrf_score"),
+        }
+        if saved_match:
+            save_payload["saved_id"] = saved_match["id"]
+        results.append(
+            {
+                **item,
+                "abstract": item["abstract"],
+                "is_saved": saved,
+                "save_payload": save_payload,
+            }
+        )
+    return results
+
+
+def auto_rebuild_on_startup() -> None:
+    if not use_semantic_search():
+        return
+    if not metadata_auto_rebuild_enabled():
+        return
+    try:
+        ensure_metadata_index(force=False)
+    except Exception as exc:
+        logger.warning("Metadata index auto-rebuild skipped: %s", exc)

requirements.txt

@@ -0,0 +1,16 @@
+# WaterScopeAI Search System - CPU Space
+Flask>=3.0,<4
+python-dotenv>=1.0,<2
+httpx>=0.27,<1
+gradio_client>=1.4,<2
+huggingface_hub>=0.27,<2
+
+# MCP 1.x is required by the current FastMCP/client implementation.
+mcp>=1.0,<2
+
+# Metadata retrieval
+qdrant-client>=1.11,<2
+sentence-transformers>=3.0,<4
+rank-bm25>=0.2.2,<1
+numpy>=1.26,<3
+pandas>=2.0,<3

retrieval_tools.py

@@ -0,0 +1,53 @@
+"""Bibliographic search and summary context for the WaterScope HF Space."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parent
+
+
+def _configure_paths() -> None:
+    os.environ.setdefault("PAPER_CSV", str(_ROOT / "metadata_search" / "data" / "papers.csv"))
+    os.environ.setdefault("METADATA_QDRANT_PATH", str(_ROOT / "metadata_search_data" / "qdrant"))
+    os.environ.setdefault("LOCAL_RAG_DATA_DIR", str(_ROOT / "local_rag_data"))
+    os.environ.setdefault("PAPER_SUMMARIES_DIR", str(_ROOT / "paper_summaries"))
+    os.environ.setdefault("METADATA_SEMANTIC_SEARCH", os.getenv("METADATA_SEMANTIC_SEARCH", "1"))
+    os.environ.setdefault("METADATA_AUTO_REBUILD", "0")
+
+
+_configure_paths()
+
+from mcp_tools.local_rag_tools import build_paper_context_tool
+from mcp_tools.metadata_tools import search_metadata_tool, select_papers_from_search_tool
+
+
+def search_papers(query: str, top_k: int = 10) -> dict:
+    return json.loads(search_metadata_tool(query, top_k=top_k))
+
+
+def select_papers(query: str, ranks: list[int], top_k: int = 10) -> dict:
+    return json.loads(select_papers_from_search_tool(query, ranks, top_k=top_k))
+
+
+def build_context(document_ids: list[str], question: str = "") -> dict:
+    return json.loads(build_paper_context_tool(document_ids, question=question))
+
+
+def format_search_results(payload: dict) -> str:
+    lines: list[str] = []
+    for row in payload.get("results", []):
+        authors = (row.get("authors_display") or row.get("authors") or "").strip()
+        lines.append(f"{row.get('rank')}. {row.get('title')} ({row.get('year', 'n.d.')})")
+        if authors:
+            lines.append(f"   {authors}")
+        abstract = (row.get("abstract") or "").strip()
+        if abstract:
+            lines.append(f"   {abstract[:220]}{'…' if len(abstract) > 220 else ''}")
+        lines.append("")
+    if not lines:
+        return "No papers matched that query."
+    mode = payload.get("score_mode", "rrf")
+    return f"Score mode: {mode}\n\n" + "\n".join(lines).strip()

waterscope_retrieval_server.py

@@ -0,0 +1,52 @@
+"""MCP retrieval tools used by the agent loop (summaries)."""
+
+from __future__ import annotations
+
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+from mcp.server.fastmcp import FastMCP
+
+from mcp_tools.local_rag_tools import build_paper_context_tool, get_paper_summaries_tool
+from mcp_tools.metadata_tools import search_metadata_tool, select_papers_from_search_tool
+
+mcp = FastMCP("waterscope-retrieval")
+
+
+@mcp.tool()
+def search_metadata(query: str, top_k: int = 10) -> str:
+    """Search title, abstract, and keywords in the bibliographic CSV."""
+    return search_metadata_tool(query, top_k=top_k)
+
+
+@mcp.tool()
+def select_papers_from_search(query: str, ranks: list[int], top_k: int = 10) -> str:
+    """Return document_ids for 1-based search result ranks."""
+    return select_papers_from_search_tool(query, ranks, top_k=top_k)
+
+
+@mcp.tool()
+def get_paper_summaries(document_ids: list[str]) -> str:
+    """Load pre-written summaries for chat context."""
+    return get_paper_summaries_tool(document_ids)
+
+
+@mcp.tool()
+def build_paper_context(document_ids: list[str], question: str = "") -> str:
+    """Assemble truncated multi-paper context for WaterScope /agent_run."""
+    return build_paper_context_tool(document_ids, question=question)
+
+
+if __name__ == "__main__":
+    transport = os.getenv("MCP_TRANSPORT", "stdio")
+    if transport == "streamable-http":
+        mcp.run(
+            transport="streamable-http",
+            host=os.getenv("MCP_HOST", "127.0.0.1"),
+            port=int(os.getenv("MCP_PORT", "8001")),
+        )
+    else:
+        mcp.run(transport="stdio")