From 0b36d1221bf512144131b0a8d7e4e22855515125 Mon Sep 17 00:00:00 2001 From: Oded Falik Date: Fri, 28 Aug 2026 11:22:30 -0700 Subject: [PATCH] Make paper search bounded and observable --- CHANGELOG.md | 8 + README.md | 125 +++--------- paper_intelligence/server.py | 38 ++-- paper_intelligence/tools/__init__.py | 32 ++- paper_intelligence/tools/search.py | 282 +++++++++++++++++++++------ paper_intelligence/utils/__init__.py | 22 ++- tests/test_integration.py | 18 +- tests/test_reliability.py | 138 +++++++++++++ 8 files changed, 463 insertions(+), 200 deletions(-) create mode 100644 tests/test_reliability.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4915f0e..b767eb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Changed +- First-use PDF conversion and embedding now run as an observable background job; `search` returns promptly with polling instructions instead of exceeding MCP client deadlines. +- `get_paper_info` reports queued, processing, ready, incomplete, or failed state without initializing the embedding model. +- Search no longer blocks on an unconditional whole-library R2 pull. + +### Fixed +- Direct PDF searches inspect only the requested PDF and its output directory, never unrelated sibling temp content. + ## [0.3.0] - 2026-01-02 ### Added diff --git a/README.md b/README.md index d31b338..892fed9 100644 --- a/README.md +++ b/README.md @@ -104,119 +104,50 @@ Any MCP-compatible client can use paper-intelligence: ## 📖 MCP Tools -### `process_paper` - -
-Full pipeline: Convert PDF → Index headers → Create embeddings - -**Parameters:** -- `pdf_path` (string): Path to PDF file -- `use_llm` (boolean, optional): Enhanced accuracy mode (default: false) -- `chunk_size` (integer, optional): Text chunk size for embedding (default: 512) -- `chunk_overlap` (integer, optional): Overlap between chunks (default: 50) - -**Example:** -``` -Process the paper at ~/Downloads/attention-is-all-you-need.pdf -``` - -**Output Structure:** -``` -attention-is-all-you-need/ -├── attention-is-all-you-need.md # Converted markdown -├── metadata.json # Processing version info -├── index.json # Header hierarchy -├── chroma/ # Embeddings database -└── images/ # Extracted figures -``` - -
- ### `search` -
-Unified search with grep and/or semantic RAG +Search one or more explicitly requested PDFs, processed paper directories, or library +directories with `grep`, `rag`, or `hybrid` mode. Direct PDF searches never inspect +sibling files or directories. **Parameters:** -- `query` (string): Search query (text, regex, or semantic) -- `paper_dirs` (array): List of paper directories to search +- `query` (string): Text, regex, or semantic query +- `sources` (array): PDF paths, paper directories, or an explicitly selected library directory - `mode` (string, optional): `"grep"`, `"rag"`, or `"hybrid"` (default: hybrid) - `top_k` (integer, optional): Number of results (default: 5) -- `regex` (boolean, optional): Treat query as regex (default: false) - -**Example Queries:** -``` -# Semantic search across papers -Search for "attention mechanism implementation" in my processed papers - -# Exact text search -Search for "transformer" using grep mode - -# Regex search -Search for "BERT|GPT|T5" with regex enabled -``` - -**Returns:** Results with line numbers, surrounding context, and header location. - -
+- `regex` (boolean, optional): Treat the grep query as a regex (default: false) -### `convert_pdf` +A new or incomplete PDF is converted, indexed, and embedded in the background because +this normally takes **1–3 minutes**, longer than the roughly 30-second deadline used by +many MCP clients. The first call returns promptly: -
-Convert PDF to Markdown (without embeddings) - -**Parameters:** -- `pdf_path` (string): Path to PDF file -- `output_dir` (string, optional): Custom output directory -- `use_llm` (boolean, optional): Enhanced accuracy mode - -**Returns:** `markdown_path`, `images_dir`, `image_count` - -
- -### `get_paper_info` - -
-Check processing status of a paper - -**Parameters:** -- `paper_dir` (string): Path to paper directory - -**Example Response:** ```json { - "has_markdown": true, - "has_index": true, - "has_embeddings": true, - "has_images": true, - "image_count": 12, - "version": "0.2.0", - "processed_at": "2025-01-15T10:30:00Z" + "success": true, + "status": "processing", + "message": "First-use processing ... is continuing in the background.", + "processing": [{ + "paper_dir": "/path/to/paper", + "retry_after_seconds": 30, + "next_step": "Call get_paper_info ..." + }] } ``` -
- -### `index_markdown` / `embed_document` - -
-Individual pipeline steps (for advanced use) +Processing continues after that response. Poll `get_paper_info` using `paper_dir`; when +it reports `status: "ready"`, retry the original search. Already-processed grep searches +do not initialize the semantic model. RAG and hybrid searches initialize it when needed. +Search no longer performs an unconditional remote-library sync, which previously allowed +a local query to block for up to five minutes. -**`index_markdown`** — Extract header hierarchy into searchable JSON -```python -index_markdown(markdown_path="~/Downloads/paper/paper.md") -``` +### `get_paper_info` -**`embed_document`** — Create embeddings for semantic search -```python -embed_document( - markdown_path="~/Downloads/paper/paper.md", - chunk_size=512, - chunk_overlap=50 -) -``` +Check a paper's processing state without loading the embedding model. Pass either the +paper directory returned by `search` or the original PDF path. -
+Statuses are `queued`, `processing`, `ready`, `incomplete`, or `failed`. Failed responses +include the background job's error message. A ready response includes artifact presence, +metadata, and a lightweight local chunk count when available. ## 📊 Example Output diff --git a/paper_intelligence/server.py b/paper_intelligence/server.py index 2dbb188..dbe1a76 100644 --- a/paper_intelligence/server.py +++ b/paper_intelligence/server.py @@ -1,10 +1,8 @@ """Paper Intelligence MCP Server. Provides AI agents with efficient, searchable access to PDF documents. -PDFs are automatically processed on first search (1-3 minutes), then -all subsequent searches are instant. - -Papers are synced to/from Cloudflare R2 via rclone for shared access. +First-use PDF processing runs in the background (normally 1-3 minutes), so +search returns promptly with observable status instead of exceeding MCP deadlines. """ from typing import Literal @@ -16,9 +14,10 @@ "paper-intelligence", instructions=( "Search PDF documents efficiently. " - "Pass PDF paths directly to search - they're auto-processed on first use " - "(takes 1-3 minutes, then instant). " - "Use get_paper_info to check processing status." + "Pass PDF paths directly to search. First use queues 1-3 minute background " + "processing and returns status='processing'; call get_paper_info on the returned " + "paper_dir until status='ready', then retry search. Searches inspect only the " + "sources explicitly supplied." ), ) @@ -36,8 +35,9 @@ def search( ) -> dict: """Search PDF documents and paper directories. - PDFs are automatically processed on first use (1-3 minutes per PDF). - Subsequent searches are instant. + First use queues background processing (normally 1-3 minutes) and returns + status="processing" promptly. Processing continues after the tool call returns. + Call get_paper_info with the returned paper_dir and retry when status="ready". Args: query: Search query (text, regex pattern, or semantic query) @@ -52,13 +52,12 @@ def search( Returns: Search results with content, location, and relevance scores """ - from .sync import pull from .tools.search import search as _search - # Pull latest from R2 before searching - pull() - - result = _search( + # Search is deliberately local and bounded. The previous unconditional whole- + # library R2 pull could block every query for five minutes, even when all requested + # sources were already indexed. + return _search( query=query, sources=sources, mode=mode, @@ -69,20 +68,13 @@ def search( use_llm=use_llm, ) - # If any papers were auto-processed, push to R2 - if result.get("processing_notes"): - from .sync import push - push() - - return result - @mcp.tool() def get_paper_info(paper_dir: str) -> dict: - """Check processing status of a paper directory. + """Check processing status without loading the embedding model. Args: - paper_dir: Path to paper directory + paper_dir: Paper directory, or the original PDF path Returns: Processing status and metadata diff --git a/paper_intelligence/tools/__init__.py b/paper_intelligence/tools/__init__.py index fa893dd..edc9814 100644 --- a/paper_intelligence/tools/__init__.py +++ b/paper_intelligence/tools/__init__.py @@ -1,17 +1,31 @@ -from .convert import convert_pdf -from .embed import embed_document, query_paper -from .index import get_header_context, index_markdown, search_headers +"""Paper tool exports with ML-heavy modules loaded only when requested.""" + +from importlib import import_module + from .search import get_paper_info, grep_search, rag_search, search +_LAZY_EXPORTS = { + "convert_pdf": ".convert", + "index_markdown": ".index", + "get_header_context": ".index", + "search_headers": ".index", + "embed_document": ".embed", + "query_paper": ".embed", +} + __all__ = [ - "convert_pdf", - "index_markdown", - "get_header_context", - "search_headers", - "embed_document", - "query_paper", + *_LAZY_EXPORTS, "search", "grep_search", "rag_search", "get_paper_info", ] + + +def __getattr__(name: str): + module_name = _LAZY_EXPORTS.get(name) + if module_name is None: + raise AttributeError(name) + value = getattr(import_module(module_name, __name__), name) + globals()[name] = value + return value diff --git a/paper_intelligence/tools/search.py b/paper_intelligence/tools/search.py index 793e320..05365ee 100644 --- a/paper_intelligence/tools/search.py +++ b/paper_intelligence/tools/search.py @@ -5,14 +5,23 @@ """ import json +import os import re +import sqlite3 +import threading +from concurrent.futures import Future, ThreadPoolExecutor +from datetime import datetime, timezone from pathlib import Path from typing import Literal, Optional from ..metadata import check_version_compatibility -from ..utils.chromadb_client import RAGClient from ..utils.markdown_parser import MarkdownParser +PROCESSING_STATE_FILENAME = ".paper-intelligence-processing.json" +_PROCESSING_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="paper-processing") +_PROCESSING_JOBS: dict[str, Future] = {} +_PROCESSING_LOCK = threading.Lock() + def _process_paper_if_needed( path: Path, @@ -27,17 +36,15 @@ def _process_paper_if_needed( Returns: Tuple of (paper_dir, error_message). If error, paper_dir is None. """ - from .convert import convert_pdf, find_duplicate, get_output_dir + from .convert import convert_pdf, get_output_dir from .embed import embed_document from .index import index_markdown # Handle PDF files if path.is_file() and path.suffix.lower() == ".pdf": - # Check for duplicate by content hash - dup = find_duplicate(path, papers_dir=path.parent) - if dup: - return path.parent / dup, None - + # A direct PDF source is intentionally isolated from sibling content. Looking + # for duplicates by scanning its parent made /tmp PDFs inspect unrelated temp + # directories (and could hit permission errors). paper_dir = get_output_dir(path) # Check if already processed and compatible @@ -112,11 +119,110 @@ def _ensure_fully_processed(paper_dir: Path) -> tuple[Optional[Path], Optional[s return paper_dir, None +def _processing_state_path(paper_dir: Path) -> Path: + return paper_dir / PROCESSING_STATE_FILENAME + + +def _read_processing_state(paper_dir: Path) -> Optional[dict]: + state_path = _processing_state_path(paper_dir) + try: + return json.loads(state_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def _write_processing_state(paper_dir: Path, state: dict) -> None: + paper_dir.mkdir(parents=True, exist_ok=True) + state_path = _processing_state_path(paper_dir) + temporary_path = state_path.with_suffix(f"{state_path.suffix}.tmp") + temporary_path.write_text(json.dumps(state, indent=2), encoding="utf-8") + temporary_path.replace(state_path) + + +def _is_fully_processed(paper_dir: Path) -> bool: + artifacts_exist = all( + (paper_dir / name).exists() for name in ("paper.md", "index.json", "chroma") + ) + return artifacts_exist and check_version_compatibility(paper_dir)["is_compatible"] + + +def _run_processing_job(source: Path, paper_dir: Path, use_llm: bool) -> None: + started_at = datetime.now(timezone.utc).isoformat() + _write_processing_state(paper_dir, { + "status": "processing", + "source": str(source), + "paper_dir": str(paper_dir), + "started_at": started_at, + "worker_pid": os.getpid(), + "message": "PDF processing is running in the background.", + }) + + try: + if source.is_file() and source.name == "paper.md": + processed_dir, error = _process_paper_if_needed(source.parent, use_llm) + else: + processed_dir, error = _process_paper_if_needed(source, use_llm) + except Exception as exc: + processed_dir, error = None, f"Background processing failed: {exc}" + + completed_at = datetime.now(timezone.utc).isoformat() + if processed_dir: + _write_processing_state(paper_dir, { + "status": "completed", + "source": str(source), + "paper_dir": str(processed_dir), + "started_at": started_at, + "completed_at": completed_at, + "message": "Processing completed. Retry search with the same source.", + }) + else: + _write_processing_state(paper_dir, { + "status": "failed", + "source": str(source), + "paper_dir": str(paper_dir), + "started_at": started_at, + "completed_at": completed_at, + "message": error or "Paper processing failed.", + }) + + +def _schedule_processing(source: Path, paper_dir: Path, use_llm: bool) -> dict: + """Start or resume one background processing job and return observable status.""" + key = str(paper_dir) + with _PROCESSING_LOCK: + future = _PROCESSING_JOBS.get(key) + if future is None or future.done(): + queued_at = datetime.now(timezone.utc).isoformat() + _write_processing_state(paper_dir, { + "status": "queued", + "source": str(source), + "paper_dir": key, + "queued_at": queued_at, + "message": "PDF processing is queued and will continue after this call returns.", + }) + _PROCESSING_JOBS[key] = _PROCESSING_EXECUTOR.submit( + _run_processing_job, source, paper_dir, use_llm + ) + + state = _read_processing_state(paper_dir) or {} + return { + **state, + "status": state.get("status", "queued"), + "source": str(source), + "paper_dir": key, + "retry_after_seconds": 30, + "next_step": ( + f"Call get_paper_info with paper_dir={key!r}; when status is 'ready', " + "retry search with the original source." + ), + } + + def _find_paper_dirs( search_paths: list[str], auto_process: bool = True, use_llm: bool = False, -) -> tuple[list[Path], list[str]]: +) -> tuple[list[Path], list[dict]]: """Find all paper directories, optionally processing PDFs. Args: @@ -125,62 +231,51 @@ def _find_paper_dirs( use_llm: Use LLM for PDF conversion Returns: - Tuple of (paper_dirs, processing_messages) + Tuple of (ready paper_dirs, background processing jobs) """ paper_dirs = [] - messages = [] + processing_jobs = [] for path_str in search_paths: path = Path(path_str).expanduser().resolve() - # Handle PDF files + # A direct PDF never discovers or reads sibling directories. First-use work + # runs asynchronously because conversion and embedding normally exceed MCP + # client deadlines. if path.is_file() and path.suffix.lower() == ".pdf": - if auto_process: - paper_dir, error = _process_paper_if_needed(path, use_llm) - if paper_dir: - paper_dirs.append(paper_dir) - messages.append(f"Processed PDF: {path.name}") - elif error: - messages.append(error) + from .convert import get_output_dir + + paper_dir = get_output_dir(path) + if _is_fully_processed(paper_dir): + paper_dirs.append(paper_dir) + elif auto_process: + processing_jobs.append(_schedule_processing(path, paper_dir, use_llm)) continue - # Handle paper.md files directly + # Handle paper.md files directly. if path.is_file() and path.name == "paper.md": - if auto_process: - paper_dir, error = _ensure_fully_processed(path.parent) - if paper_dir: - paper_dirs.append(paper_dir) - elif error: - messages.append(error) - else: + if _is_fully_processed(path.parent) or not auto_process: paper_dirs.append(path.parent) + else: + processing_jobs.append(_schedule_processing(path, path.parent, use_llm)) continue - # Handle directories + # Directory discovery is limited to the directory explicitly supplied by the + # caller and its immediate paper children. if path.is_dir(): - # Check if this is a paper directory - if (path / "paper.md").exists(): - if auto_process: - paper_dir, error = _ensure_fully_processed(path) - if paper_dir: - paper_dirs.append(paper_dir) - elif error: - messages.append(error) + candidates = [path] if (path / "paper.md").exists() else [ + subdir for subdir in path.iterdir() + if subdir.is_dir() and (subdir / "paper.md").exists() + ] + for candidate in candidates: + if _is_fully_processed(candidate) or not auto_process: + paper_dirs.append(candidate) else: - paper_dirs.append(path) - # Also check subdirectories - for subdir in path.iterdir(): - if subdir.is_dir() and (subdir / "paper.md").exists(): - if auto_process: - paper_dir, error = _ensure_fully_processed(subdir) - if paper_dir: - paper_dirs.append(paper_dir) - elif error: - messages.append(error) - else: - paper_dirs.append(subdir) + processing_jobs.append( + _schedule_processing(candidate / "paper.md", candidate, use_llm) + ) - return paper_dirs, messages + return paper_dirs, processing_jobs def grep_search( @@ -286,6 +381,8 @@ def rag_search( Returns: List of matches with content, score, and metadata """ + from ..utils.chromadb_client import RAGClient + results = [] for paper_dir in paper_dirs: @@ -336,8 +433,8 @@ def search( ) -> dict: """Search PDF documents and paper directories. - Automatically processes PDFs on first use (may take 1-3 minutes per PDF). - Subsequent searches on the same PDF are instant. + Queues first-use PDF processing in the background (normally 1-3 minutes) + and returns actionable status immediately. Retry once get_paper_info is ready. Args: query: Search query (text, regex pattern, or semantic query) @@ -352,17 +449,33 @@ def search( Returns: Result with results list, num_results, and success status """ - # Find all paper directories, auto-processing PDFs as needed - dirs, processing_messages = _find_paper_dirs(sources, auto_process=True, use_llm=use_llm) + # Discover only explicitly requested sources. Slow first-use processing is queued + # so the MCP response stays within normal client deadlines. + dirs, processing_jobs = _find_paper_dirs(sources, auto_process=True, use_llm=use_llm) if not dirs: + if processing_jobs: + return { + "results": [], + "num_results": 0, + "query": query, + "mode": mode, + "success": True, + "status": "processing", + "message": ( + "First-use processing takes about 1-3 minutes and is continuing " + "in the background. Check get_paper_info, then retry this search." + ), + "processing": processing_jobs, + } return { "results": [], "num_results": 0, "query": query, "mode": mode, - "success": False if processing_messages else True, - "message": "; ".join(processing_messages) if processing_messages else "No paper directories found", + "success": True, + "status": "no_sources", + "message": "No paper directories found for the requested sources.", } results = [] @@ -417,8 +530,15 @@ def search( "mode": mode, "success": True, } - if processing_messages: - result["processing_notes"] = processing_messages + if processing_jobs: + result["status"] = "partial" + result["processing"] = processing_jobs + result["message"] = ( + "Searched ready sources; remaining first-use processing continues in " + "the background." + ) + else: + result["status"] = "ready" return result except Exception as e: @@ -446,6 +566,26 @@ def _deduplicate_results(results: list[dict]) -> list[dict]: return unique_results +def _get_chroma_collection_count(chroma_dir: Path) -> Optional[int]: + """Read Chroma's count without loading the embedding model. + + Status calls must remain cheap and must not download or initialize the semantic + model merely to report an already-built collection's size. + """ + database = chroma_dir / "chroma" / "chroma.sqlite3" + if not database.exists(): + return None + try: + connection = sqlite3.connect(f"file:{database}?mode=ro", uri=True, timeout=1) + try: + row = connection.execute("SELECT COUNT(*) FROM embeddings").fetchone() + return int(row[0]) if row else 0 + finally: + connection.close() + except (OSError, sqlite3.Error, ValueError): + return None + + def get_paper_info(paper_dir: str) -> dict: """Get information about a paper directory. @@ -462,6 +602,9 @@ def get_paper_info(paper_dir: str) -> dict: from ..metadata import check_version_compatibility, read_metadata path = Path(paper_dir).expanduser().resolve() + if path.is_file() and path.suffix.lower() == ".pdf": + from .convert import get_output_dir + path = get_output_dir(path) if not path.is_dir(): return { @@ -479,6 +622,21 @@ def get_paper_info(paper_dir: str) -> dict: "success": True, } + version_info = check_version_compatibility(path) + processing = _read_processing_state(path) + if processing and processing.get("status") in {"queued", "processing", "failed"}: + info["processing"] = processing + info["status"] = processing["status"] + elif ( + info["has_markdown"] + and info["has_index"] + and info["has_embeddings"] + and version_info["is_compatible"] + ): + info["status"] = "ready" + else: + info["status"] = "incomplete" + # Add paths for easy access if info["has_markdown"]: info["markdown_path"] = str(path / "paper.md") @@ -502,16 +660,12 @@ def get_paper_info(paper_dir: str) -> dict: except Exception: info["header_count"] = 0 - # Get chunk count from ChromaDB + # Count chunks directly from the local Chroma database. Constructing RAGClient + # here initializes a HuggingFace model and made a status check take minutes. if info["has_embeddings"]: - try: - rag_client = RAGClient(persist_directory=path / "chroma") - info["chunk_count"] = rag_client.get_collection_count("paper") - except Exception: - info["chunk_count"] = 0 + info["chunk_count"] = _get_chroma_collection_count(path / "chroma") - # Check version compatibility - version_info = check_version_compatibility(path) + # Report version compatibility. info["version"] = { "processed_version": version_info["processed_version"], "current_version": version_info["current_version"], diff --git a/paper_intelligence/utils/__init__.py b/paper_intelligence/utils/__init__.py index 9f67495..6e4b372 100644 --- a/paper_intelligence/utils/__init__.py +++ b/paper_intelligence/utils/__init__.py @@ -1,4 +1,20 @@ -from .chromadb_client import RAGClient, create_documents_from_markdown -from .markdown_parser import MarkdownParser +"""Utility exports, loaded lazily so lightweight tools avoid ML imports.""" -__all__ = ["RAGClient", "create_documents_from_markdown", "MarkdownParser"] +from importlib import import_module + +_EXPORTS = { + "RAGClient": ".chromadb_client", + "create_documents_from_markdown": ".chromadb_client", + "MarkdownParser": ".markdown_parser", +} + +__all__ = list(_EXPORTS) + + +def __getattr__(name: str): + module_name = _EXPORTS.get(name) + if module_name is None: + raise AttributeError(name) + value = getattr(import_module(module_name, __name__), name) + globals()[name] = value + return value diff --git a/tests/test_integration.py b/tests/test_integration.py index 0891ce3..5817d86 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -399,12 +399,22 @@ def test_search_auto_processes_pdf(self, temp_output_dir): ) assert result["success"] - # Should have processing notes about the PDF - assert "processing_notes" in result or result["num_results"] > 0 + assert result["status"] == "processing" + assert result["processing"][0]["retry_after_seconds"] == 30 - # Paper directory should now exist + # Processing continues independently of the first MCP response. paper_dir = temp_output_dir / "test_auto" - assert paper_dir.exists() + from paper_intelligence.tools.search import _PROCESSING_JOBS + _PROCESSING_JOBS[str(paper_dir)].result(timeout=300) + + completed = search( + query="the", + sources=[str(pdf_copy)], + mode="grep", + top_k=5, + ) + assert completed["status"] == "ready" + assert completed["num_results"] > 0 assert (paper_dir / "paper.md").exists() assert (paper_dir / "index.json").exists() assert (paper_dir / "chroma").exists() diff --git a/tests/test_reliability.py b/tests/test_reliability.py new file mode 100644 index 0000000..b0b6282 --- /dev/null +++ b/tests/test_reliability.py @@ -0,0 +1,138 @@ +"""Regression tests for bounded, source-scoped MCP behavior.""" + +import importlib +import json +import sqlite3 +import subprocess +import sys +from pathlib import Path + + +def _ready_paper(path: Path) -> Path: + path.mkdir() + (path / "paper.md").write_text("# Manual\n\nA bounded local search.\n", encoding="utf-8") + (path / "index.json").write_text('{"flat_headers": []}', encoding="utf-8") + (path / "chroma").mkdir() + return path + + +def test_direct_pdf_does_not_discover_sibling_temp_content(tmp_path, monkeypatch): + """A /tmp PDF must not inspect unrelated sibling metadata or directories.""" + search_module = importlib.import_module("paper_intelligence.tools.search") + source = tmp_path / "requested.pdf" + source.write_bytes(b"%PDF-1.4\n") + unrelated = tmp_path / ".vnc-0" + unrelated.mkdir() + (unrelated / "metadata.json").write_text("unrelated", encoding="utf-8") + + scheduled = [] + + def fake_schedule(job_source, paper_dir, use_llm): + scheduled.append((job_source, paper_dir)) + return {"status": "queued", "paper_dir": str(paper_dir)} + + monkeypatch.setattr(search_module, "_schedule_processing", fake_schedule) + + ready, jobs = search_module._find_paper_dirs([str(source)]) + + assert ready == [] + assert scheduled == [(source.resolve(), tmp_path / "requested")] + assert jobs == [{"status": "queued", "paper_dir": str(tmp_path / "requested")}] + + +def test_first_use_search_returns_actionable_background_status(tmp_path, monkeypatch): + search_module = importlib.import_module("paper_intelligence.tools.search") + + source = tmp_path / "slow.pdf" + source.write_bytes(b"%PDF-1.4\n") + paper_dir = tmp_path / "slow" + + monkeypatch.setattr( + search_module, + "_schedule_processing", + lambda *_: { + "status": "processing", + "paper_dir": str(paper_dir), + "retry_after_seconds": 30, + "next_step": "Call get_paper_info, then retry search.", + }, + ) + + result = search_module.search("query", [str(source)], mode="grep") + + assert result["success"] is True + assert result["status"] == "processing" + assert result["results"] == [] + assert "continuing in the background" in result["message"] + assert result["processing"][0]["paper_dir"] == str(paper_dir) + assert "get_paper_info" in result["processing"][0]["next_step"] + + +def test_processing_job_persists_completion_status(tmp_path, monkeypatch): + search_module = importlib.import_module("paper_intelligence.tools.search") + + source = tmp_path / "paper.pdf" + source.write_bytes(b"%PDF-1.4\n") + paper_dir = tmp_path / "paper" + + monkeypatch.setattr( + search_module, + "_process_paper_if_needed", + lambda *_: (paper_dir, None), + ) + + search_module._run_processing_job(source, paper_dir, use_llm=False) + state = json.loads( + (paper_dir / search_module.PROCESSING_STATE_FILENAME).read_text(encoding="utf-8") + ) + + assert state["status"] == "completed" + assert state["paper_dir"] == str(paper_dir) + assert "Retry search" in state["message"] + + +def test_status_module_import_does_not_load_embedding_stack(): + code = ( + "import sys; from paper_intelligence.tools import get_paper_info, search; " + "assert callable(get_paper_info) and callable(search); " + "assert 'paper_intelligence.utils.chromadb_client' not in sys.modules" + ) + subprocess.run([sys.executable, "-c", code], check=True, timeout=10) + + +def test_get_paper_info_counts_chunks_without_embedding_model(tmp_path): + from paper_intelligence.tools.search import get_paper_info + + source = tmp_path / "manual.pdf" + source.write_bytes(b"%PDF-1.4\n") + paper_dir = _ready_paper(tmp_path / "manual") + database_dir = paper_dir / "chroma" / "chroma" + database_dir.mkdir() + connection = sqlite3.connect(database_dir / "chroma.sqlite3") + connection.execute("CREATE TABLE embeddings (id TEXT)") + connection.executemany("INSERT INTO embeddings VALUES (?)", [("a",), ("b",)]) + connection.commit() + connection.close() + + result = get_paper_info(str(source)) + + assert result["status"] == "ready" + assert result["chunk_count"] == 2 + + +def test_mcp_search_does_not_sync_whole_library(tmp_path, monkeypatch): + """Already-indexed local searches must not enter the five-minute rclone path.""" + import paper_intelligence.server as server + import paper_intelligence.sync as sync + + paper_dir = _ready_paper(tmp_path / "manual") + + def sync_must_not_run(): + raise AssertionError("search must not run an unconditional library sync") + + monkeypatch.setattr(sync, "pull", sync_must_not_run) + result = server.search("bounded", [str(paper_dir)], mode="grep") + + assert result["success"] is True + assert result["status"] == "ready" + assert result["num_results"] == 1