diff --git a/README.md b/README.md index cd390d0..80f39b6 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Two workspaces, never mixed. **Live research** uses actual EIA RSS, Federal Regi 1. Start the local app (setup below). Source checks run on opening the live workspace, then while the tab is visible and auto-check is enabled. The polling floor is 60 seconds, with longer provider Cache-Control/Retry-After respected. These are publication feeds, not tick streams. 2. Inspect source status, capture times and publication dates. First-seen documents are labelled new to this installation, not newly published. Errors preserve old evidence with a stale-source warning. -3. Filter the inbox or select documents. Ask a focused question using **Evidence retrieval only** or **AI brief with citations**. Retrieval uses SQLite FTS5/BM25 over the latest captured document versions. +3. Filter the inbox or select documents. Ask a focused question using **Evidence retrieval only** or **AI brief with citations**. Retrieval runs a Haystack 2.x BM25 pipeline over the latest captured document versions. 4. Inspect the returned passages, exact citations and version diffs. AI output is a draft interpretation, not verified truth. No model call occurs on background source refresh. 5. Export the complete job/evidence JSON. Jobs persist their stages, result and errors; cancellation suppresses acceptance of in-flight results. Restarted unfinished jobs become `interrupted`, never falsely `completed`. @@ -67,6 +67,43 @@ python -m infodesk.app Open **http://127.0.0.1:8000/**. The local database defaults to `data/desk.sqlite3` and survives restarts. Override its path with `INFODESK_DB`. There is no automatic `.env` loader; set variables in your shell. +### Connect in Cursor + +The same local tools are available over MCP stdio. GitHub Pages has no MCP server and does not pretend to. + +```json +{ + "mcpServers": { + "info-desk": { + "command": "python", + "args": ["-m", "infodesk.mcp_server"], + "cwd": "/absolute/path/to/info-desk", + "env": { + "INFODESK_DB": "/absolute/path/to/info-desk/data/desk.sqlite3" + } + } + } +} +``` + +Point `command` at the virtualenv interpreter that has `pip install -e ".[dev]"` installed. After the local app has captured live documents, `search_sources` reuses that Haystack BM25 index. `get_claim` reads the recorded claim table. `propose_write` inserts a pending draft only. + +List the same tools from the running local API: + +```powershell +curl -s http://127.0.0.1:8000/api/mcp/tools +``` + +#### MCP checklist + +| Check | What is actually true | +|---|---| +| Auth | Loopback FastAPI + stdio. No API key, login, or hosted MCP. Trusted-Host still rejects unexpected Host headers. | +| Tool schema | Extra fields (`approve`, unknown names) and missing required fields return a visible 422/tool error. | +| Writes need `approve()` | `propose_write` cannot approve itself. `POST /api/drafts/{id}/decide` or `store.approve()` writes the note. | +| 429 / timeout | `search_sources` surfaces existing live-check 429/timeout details. It does not invent a successful fetch. | +| Eval / harness | Recorded harness, Pages replay JSON and retrieval eval stay on their existing paths. MCP is local/live only. | + To preview only the public replay: ```powershell @@ -101,13 +138,17 @@ The default model request timeout is an uncalibrated resource budget, not a late |---|---|---| | `sources.py`, `tools.py` | Allowlisted stored excerpts | Capture metadata and controlled reads; editorial additions excluded from evidence | | `extract.py`, `claims.py` | Curated claim rules + analysed source text | Claim cells, named quantities and attribution; absence is not denial | +| `haystack_retrieval.py` | Latest captured `live_chunks` rows | Haystack `Document` + `InMemoryBM25Retriever` Pipeline; lexical ranking only | +| `mcp_tools.py`, `mcp_server.py` | Live store + recorded claims | stdio/HTTP MCP: `search_sources`, `get_claim`, `propose_write`; no self-approve | | `interpret.py` | Evidence-backed findings; optional Ollama | Deterministic memo and bounded finding ordering | | `validate.py`, `policy.py` | Quantities, licenses, configured instruction-pattern checks | Release guard; no model-controlled database write | | `store.py` | SQLite transactions and conditional status updates | Persistent run snapshots, deduplication by complete draft identity, atomic approval and note insertion | | `workbench.py` | The same analysis service | Shared payload for local API and generated replay | | `public/` | Replay JSON or local API | Responsive source selector, claim inspector, review queue, exports and visible evaluations | -**Why this scope:** the live workspace demonstrates retrieval-augmented generation with lexical search, versioned evidence and bounded model output. The recorded workspace remains a fixed-rule comparison, not RAG or a general extraction benchmark. Neither independently verifies the publisher's claims. +Live ask retrieval is: **captures (SQLite ledger) → Haystack `Document` → BM25 retriever → existing citation layer**. Haystack does not write the ledger, approve drafts, or call the model. Citations, claims, `approve()`, and allowlisted connectors stay in Python. This is local lexical BM25, not embeddings, semantic search, or a production service. GitHub Pages still serves the recorded case; Haystack runs only on the live localhost path. Captures are flat paragraph windows, so AutoMergingRetriever is unused. + +**Why this scope:** the live workspace demonstrates retrieval-augmented generation with lexical Haystack BM25, versioned evidence and bounded model output. The recorded workspace remains a fixed-rule comparison, not RAG or a general extraction benchmark. Neither independently verifies the publisher's claims. The local API binds to loopback, rejects unexpected Host headers and cross-origin writes, and accepts only registered source IDs. It is **single-user**. Do not expose it publicly without authentication, per-user authorization, request limits and a production storage design. @@ -122,6 +163,8 @@ The local API binds to loopback, rejects unexpected Host headers and cross-origi | `POST /api/drafts/{id}/decide` | `{"decision":"approve"}` or `{"decision":"reject"}` | | `GET /api/db` | Current local ledger state | | `GET /api/harness` | Isolated regression evaluation, not mutations to your ledger | +| `GET /api/mcp/tools` | List local MCP tools and input schemas | +| `POST /api/mcp/tools/{name}` | Invoke one listed tool; `propose_write` still needs human `approve()` | All supplied public-source cases currently produce `hold` or `verify_first`; the UI correctly prevents approval. Positive approval, rejection, replay and concurrency paths are exercised with clearly labelled SYNTHETIC tests. Exporting a memo never approves a note. diff --git a/evals/README.md b/evals/README.md index a81689b..59181b2 100644 --- a/evals/README.md +++ b/evals/README.md @@ -6,7 +6,7 @@ Run from the repository root: python -m infodesk.evaluate_retrieval --output test-results/retrieval-evaluation.json ``` -This exercises the actual SQLite FTS5/BM25 retriever using an isolated in-memory database. It makes no network requests or model calls and never modifies the live database. CI saves the full result as an artifact. +This exercises the live Haystack BM25 retriever using captured versions loaded into an isolated in-memory store. It makes no network requests or model calls and never modifies the live database. CI saves the full result as an artifact. The committed baseline JSON below was measured against SQLite FTS5/BM25 on 10 September 2026; CI reruns the same annotations through Haystack and does not treat that file as a pass/fail gate. ## Dataset and scoring diff --git a/public/index.html b/public/index.html index 42aeeaf..ab0ae38 100644 --- a/public/index.html +++ b/public/index.html @@ -51,7 +51,7 @@ -

Under the hood

From a source update to a reviewable answer.

The live pipeline collects official records, keeps document versions and retrieves passages with SQLite FTS5 / BM25. A local model writes a short answer with exact source quotations. Failed checks stay failed; no recording is substituted for a live result.

ComponentDepends onProduces / enforces
Live connectorsEIA RSS, Federal Register API, OFAC HTMLConditional HTTP requests, provider cooldowns, response limits and visible source errors
Version storeValidated source documentsAppend-only captures, fingerprints and diffs; publication and fetch dates stay separate
RetrievalLatest stored versions + user-selected scopeBM25-ranked passages with document/version IDs; no-match abstention
AI briefQuestion + retrieved passages + structured-output modelQuote and numeric validation, visible failures, reviewable interpretation — no automatic publication
Job workerLocal API + SQLite checkpointsResponsive progress, one-job backpressure, cancellation and restart recovery to an interrupted state

The separate recorded case remains a reproducible regression lab. Live coverage is publisher summaries, abstracts and license titles, not every full document. Lexical retrieval can miss relevant passages, and citation matching cannot prove semantic correctness. Public multi-user hosting requires authentication and a deployed backend.

+

Under the hood

From a source update to a reviewable answer.

The live pipeline collects official records, keeps document versions and retrieves passages with a Haystack BM25 pipeline. A local model writes a short answer with exact source quotations. Failed checks stay failed; no recording is substituted for a live result.

ComponentDepends onProduces / enforces
Live connectorsEIA RSS, Federal Register API, OFAC HTMLConditional HTTP requests, provider cooldowns, response limits and visible source errors
Version storeValidated source documentsAppend-only captures, fingerprints and diffs; publication and fetch dates stay separate
RetrievalLatest stored versions + Haystack BM25 PipelineLexical BM25-ranked passages with document/version IDs; no-match abstention. Haystack does not write the ledger or call the model
AI briefQuestion + retrieved passages + structured-output modelQuote and numeric validation, visible failures, reviewable interpretation — no automatic publication
Job workerLocal API + SQLite checkpointsResponsive progress, one-job backpressure, cancellation and restart recovery to an interrupted state

The separate recorded case remains a reproducible regression lab. Live coverage is publisher summaries, abstracts and license titles, not every full document. Lexical retrieval can miss relevant passages, and citation matching cannot prove semantic correctness. Public multi-user hosting requires authentication and a deployed backend.

diff --git a/pyproject.toml b/pyproject.toml index 9d39f58..055b1e7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,8 @@ dependencies = [ "httpx>=0.27", "beautifulsoup4>=4.12", "defusedxml>=0.7", + "haystack-ai==2.31.0", + "mcp==2.2.0", ] [project.optional-dependencies] @@ -20,6 +22,7 @@ dev = ["pytest>=8.0", "httpx>=0.27"] [project.scripts] info-desk = "infodesk.app:main" +info-desk-mcp = "infodesk.mcp_server:main" [tool.setuptools.packages.find] where = ["src"] diff --git a/src/infodesk/app.py b/src/infodesk/app.py index bdeb5d8..8fa873b 100644 --- a/src/infodesk/app.py +++ b/src/infodesk/app.py @@ -19,6 +19,7 @@ from .workbench import analyse from .live_store import LiveStore from .live_service import LiveRequest, LiveService +from .mcp_tools import ToolError, bind_store, call_tool, list_tools ROOT = Path(__file__).resolve().parents[2] PUBLIC = ROOT / "public" @@ -39,6 +40,7 @@ def create_app(db_path: str | Path | None = None) -> FastAPI: async def lifespan(app): app.state.store = LiveStore(db_path or os.environ.get("INFODESK_DB", str(ROOT / "data" / "desk.sqlite3"))) app.state.live = LiveService(app.state.store) + bind_store(app.state.store) try: yield finally: @@ -175,6 +177,23 @@ def harness(): # Each evaluation uses an isolated in-memory database, never the user's ledger. return run_harness() + @app.get("/api/mcp/tools") + def mcp_tools(): + return { + "tools": list_tools(), + "transport": ["stdio", "http"], + "approval": "propose_write cannot approve. Human approve() writes the note.", + "pages": "Recorded GitHub Pages has no MCP server.", + } + + @app.post("/api/mcp/tools/{name}") + def mcp_call(name: str, body: dict): + try: + return call_tool(store(), name, body) + except ToolError as exc: + status = 404 if str(exc).startswith("Unknown tool") else 422 + raise HTTPException(status, str(exc)) from exc + @app.get("/") def index(): return FileResponse(PUBLIC / "index.html") diff --git a/src/infodesk/evaluate_retrieval.py b/src/infodesk/evaluate_retrieval.py index 62d2893..ad2edc2 100644 --- a/src/infodesk/evaluate_retrieval.py +++ b/src/infodesk/evaluate_retrieval.py @@ -8,6 +8,8 @@ from pathlib import Path from time import perf_counter +import haystack + from .live_store import LiveStore from .sources import ROOT, load_sources @@ -76,7 +78,7 @@ def evaluate(dataset_path, repeats): 'dataset_sha256': hashlib.sha256(raw).hexdigest(), 'corpus_sha256': {sid: hashlib.sha256(s['body'].encode()).hexdigest() for sid, s in sources.items()}, 'runtime': {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, - 'platform': platform.platform()}, + 'haystack': haystack.__version__, 'platform': platform.platform()}, 'summary': {'cases': len(rows), 'answerable': len(answerable), 'unanswerable': len(negative), 'mean_evidence_recall_at_returned_limit': mean([r['evidence_recall'] for r in answerable]), 'top1_hit_rate': mean([r['top1_relevant'] for r in answerable]), @@ -88,7 +90,8 @@ def evaluate(dataset_path, repeats): 'Corpus is the supplied dated excerpts, not current live connector coverage or independently verified facts.', 'No answer generation or semantic correctness is scored. Context returned for an unanswerable question is not necessarily a wrong answer.', 'No retrieval confidence or safe abstention can be inferred from non-empty search results.', - 'Local in-memory retrieval timing excludes network, persistence I/O and model inference.'], + 'Local in-memory retrieval timing excludes network, persistence I/O and model inference.', + 'Live ranking is Haystack lexical BM25 over captured versions, not embeddings or a semantic index.'], 'cases': rows, } diff --git a/src/infodesk/haystack_retrieval.py b/src/infodesk/haystack_retrieval.py new file mode 100644 index 0000000..4be70cb --- /dev/null +++ b/src/infodesk/haystack_retrieval.py @@ -0,0 +1,78 @@ +"""Lexical BM25 retrieval over already-captured live document versions. + +Haystack ranks passages only. It does not write the SQLite ledger, approve +drafts, or call a model. Captures are flat paragraph windows, so this module +does not invent hierarchy metadata or use AutoMergingRetriever. +""" +import os + +# Keep CI and local tests offline; BM25 does not need provider credentials. +os.environ.setdefault("HAYSTACK_TELEMETRY_ENABLED", "False") + +from haystack import Document, Pipeline +from haystack.components.retrievers.in_memory import InMemoryBM25Retriever +from haystack.document_stores.in_memory import InMemoryDocumentStore + +# GUESS: six passages bound CPU-model context. Relevance is not calibrated. +PASSAGE_LIMIT = 6 + + +def documents_from_captures(rows): + """Map ledger chunks to Haystack Documents. Title is searchable; body stays exact.""" + documents = [] + for row in rows: + title = row["title"] or "" + body = row["body"] + content = f"{title}\n{body}".strip() if title and title != body else body + documents.append(Document( + id=str(row["chunk_id"]), + content=content, + meta={ + "chunk_id": row["chunk_id"], + "body": body, + "version_id": row["version_id"], + "document_id": row["document_id"], + "source_id": row["source_id"], + "url": row["url"], + "title": row["title"], + "received_at": row["received_at"], + "published_at": row["published_at"], + "text_hash": row["text_hash"], + "coverage": row["coverage"], + }, + )) + return documents + + +def retrieve_passages(question, rows, top_k=PASSAGE_LIMIT): + """Run a Haystack 2.x BM25 Pipeline and return the existing passage dicts.""" + if not rows: + return [] + store = InMemoryDocumentStore() + store.write_documents(documents_from_captures(rows)) + pipeline = Pipeline() + pipeline.add_component( + "retriever", + InMemoryBM25Retriever(document_store=store, top_k=top_k), + ) + result = pipeline.run({"retriever": {"query": question}}) + passages = [] + for document in result["retriever"]["documents"]: + if document.score is None or document.score <= 0: + continue + meta = document.meta + passages.append({ + "chunk_id": meta["chunk_id"], + "body": meta["body"], + "version_id": meta["version_id"], + "document_id": meta["document_id"], + "source_id": meta["source_id"], + "url": meta["url"], + "title": meta["title"], + "received_at": meta["received_at"], + "published_at": meta["published_at"], + "text_hash": meta["text_hash"], + "coverage": meta["coverage"], + "rank": document.score, + }) + return passages diff --git a/src/infodesk/live_service.py b/src/infodesk/live_service.py index 9aac8aa..991bc05 100644 --- a/src/infodesk/live_service.py +++ b/src/infodesk/live_service.py @@ -229,7 +229,7 @@ def _run(self, job_id, kind, request): if kind == "ask": started = time.perf_counter() result["passages"] = self.store.retrieve(request.question, request.source_ids, request.document_ids) - self.store.job_step(job_id, f"Retrieved {len(result['passages'])} passages with SQLite FTS5 / BM25", + self.store.job_step(job_id, f"Retrieved {len(result['passages'])} passages with Haystack BM25", (time.perf_counter()-started)*1000) if not result["passages"]: result["answer"] = {"statements": [], "insufficient_evidence": True, diff --git a/src/infodesk/live_store.py b/src/infodesk/live_store.py index 7d473ca..23c12f6 100644 --- a/src/infodesk/live_store.py +++ b/src/infodesk/live_store.py @@ -5,6 +5,7 @@ from datetime import datetime, timezone from uuid import uuid4 +from .haystack_retrieval import retrieve_passages from .store import Store, serialized @@ -119,21 +120,21 @@ def retrieve(self, question, source_ids, document_ids=()): tokens = list(dict.fromkeys(word.lower() for word in re.findall(r"[^\W_]+", question, re.UNICODE) if len(word)>2 and word.lower() not in stop)) if not tokens: return [] - match = " OR ".join('"' + token.replace('"', '""') + '"' for token in tokens) - params = [match, *source_ids] + params = list(source_ids) where = "" if document_ids: where = " AND d.id IN (" + ",".join("?" for _ in document_ids) + ")" params.extend(document_ids) - # GUESS: six passages bound CPU-model context. Relevance is not calibrated. - return [dict(row) for row in self.conn.execute("""SELECT c.id AS chunk_id, c.body, + # Latest captured versions only. Haystack ranks; SQLite does not MATCH here. + rows = [dict(row) for row in self.conn.execute("""SELECT c.id AS chunk_id, c.body, c.version_id, d.id AS document_id, d.source_id, d.url, d.title, - v.received_at, v.published_at, v.text_hash, v.coverage, bm25(live_fts) AS rank - FROM live_fts JOIN live_chunks c ON c.id=live_fts.chunk_id + v.received_at, v.published_at, v.text_hash, v.coverage + FROM live_chunks c JOIN live_documents d ON d.id=c.document_id AND d.latest_version=c.version_id JOIN live_versions v ON v.id=c.version_id - WHERE live_fts MATCH ? AND d.source_id IN (""" + ",".join("?" for _ in source_ids) + ")" - + where + " ORDER BY rank, c.id LIMIT 6", params).fetchall()] + WHERE d.source_id IN (""" + ",".join("?" for _ in source_ids) + ")" + + where + " ORDER BY c.id", params).fetchall()] + return retrieve_passages(question, rows) @serialized def new_job(self, kind, request): diff --git a/src/infodesk/mcp_server.py b/src/infodesk/mcp_server.py new file mode 100644 index 0000000..3387fa3 --- /dev/null +++ b/src/infodesk/mcp_server.py @@ -0,0 +1,64 @@ +"""stdio MCP server for the local Info Desk tools. + +Haystack retrieves; this process does not approve drafts. Run: +`python -m infodesk.mcp_server` after `pip install -e .`. +""" +from __future__ import annotations + +import os +from pathlib import Path + +from mcp.server import MCPServer +from mcp.types import ToolAnnotations + +from .live_store import LiveStore +from .mcp_tools import bind_store, call_tool, current_store + +ROOT = Path(__file__).resolve().parents[2] + +desk_mcp = MCPServer("Info Desk") + + +@desk_mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, openWorldHint=False)) +def search_sources( + question: str, + source_ids: list[str], + document_ids: list[str] | None = None, +) -> dict: + """Retrieve live captured passages with the Haystack BM25 pipeline. + + Read-only. Does not fetch sources, write the ledger, or approve drafts. + Empty results stay empty. 429/timeout source checks stay visible. + """ + return call_tool(current_store(), "search_sources", { + "question": question, + "source_ids": source_ids, + "document_ids": document_ids or [], + }) + + +@desk_mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, openWorldHint=False)) +def get_claim(claim_id: str, source_id: str) -> dict: + """Return one recorded claim cell and its exact supporting sentence.""" + return call_tool(current_store(), "get_claim", {"claim_id": claim_id, "source_id": source_id}) + + +@desk_mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=False)) +def propose_write(headline: str, body: str, case_id: str = "mcp") -> dict: + """Create a pending draft only. This tool cannot approve or write a note.""" + return call_tool(current_store(), "propose_write", { + "headline": headline, "body": body, "case_id": case_id, + }) + + +def main() -> None: + store = LiveStore(os.environ.get("INFODESK_DB", str(ROOT / "data" / "desk.sqlite3"))) + bind_store(store) + try: + desk_mcp.run("stdio") + finally: + store.close() + + +if __name__ == "__main__": + main() diff --git a/src/infodesk/mcp_tools.py b/src/infodesk/mcp_tools.py new file mode 100644 index 0000000..f3180cf --- /dev/null +++ b/src/infodesk/mcp_tools.py @@ -0,0 +1,157 @@ +"""Local MCP tools over the existing desk. Haystack ranks; approve() still writes notes.""" +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from .claims import CLAIMS, classify_sentence +from .connectors import FEEDS +from .sources import SOURCES + +LIVE_SOURCE_IDS = tuple(FEEDS) + +_bound_store = None + + +class ToolError(ValueError): + """Visible tool-schema or tool-dispatch failure. Not a silent fallback.""" + + +def bind_store(store) -> None: + global _bound_store + _bound_store = store + + +def current_store(): + if _bound_store is None: + raise ToolError("MCP store is not bound. Start the local app or stdio server.") + return _bound_store + + +class SearchSourcesInput(BaseModel): + model_config = ConfigDict(extra="forbid") + question: str = Field(min_length=1, max_length=1000) + source_ids: list[Literal["eia", "federal-register", "ofac-live"]] = Field( + min_length=1, max_length=len(LIVE_SOURCE_IDS) + ) + document_ids: list[str] = Field(default_factory=list, max_length=10) + + +class GetClaimInput(BaseModel): + model_config = ConfigDict(extra="forbid") + claim_id: str = Field(min_length=1, max_length=40) + source_id: Literal["ofac", "white-house", "ap"] + + +class ProposeWriteInput(BaseModel): + model_config = ConfigDict(extra="forbid") + headline: str = Field(min_length=1, max_length=200) + body: str = Field(min_length=1, max_length=4000) + case_id: str = Field(default="mcp", min_length=1, max_length=80) + + +def search_sources(store, payload: SearchSourcesInput) -> dict: + """Retrieve latest captured passages via the live Haystack BM25 path.""" + if len(set(payload.source_ids)) != len(payload.source_ids): + raise ToolError("Choose each live source at most once") + checks = store.latest_checks() + source_errors = [] + for source_id in payload.source_ids: + check = checks.get(source_id) + if not check: + continue + detail = check.get("detail") or "" + if check.get("status") in {"error", "cooldown"} or "429" in detail or "timeout" in detail.lower(): + source_errors.append({ + "source_id": source_id, + "status": check.get("status"), + "detail": detail, + }) + passages = store.retrieve(payload.question, payload.source_ids, payload.document_ids) + return { + "passages": passages, + "source_errors": source_errors, + "engine": "haystack-bm25", + "ledger_written": False, + "approved": False, + } + + +def get_claim(store, payload: GetClaimInput) -> dict: + """Return one recorded claim cell and its exact supporting sentence.""" + spec = next((item for item in CLAIMS if item[0] == payload.claim_id), None) + if spec is None: + raise ToolError(f"Unknown claim_id: {payload.claim_id}") + source = SOURCES[payload.source_id] + cell = classify_sentence(source["body"], spec[2]) + return { + "claim_id": payload.claim_id, + "label": spec[1], + "source_id": payload.source_id, + "status": cell["status"], + "quote": cell["quote"], + "url": source["url"], + "ledger_written": False, + "approved": False, + } + + +def propose_write(store, payload: ProposeWriteInput) -> dict: + """Insert a pending draft only. This tool never calls approve().""" + draft_id = store.insert_draft( + payload.case_id, "publish_draft", payload.headline, payload.body, "mcp_propose_write" + ) + return { + "draft_id": draft_id, + "status": "pending", + "note_written": False, + "approved": False, + "notes": store.note_count(), + "requires": "Human store.approve(draft_id) or POST /api/drafts/{id}/decide. This tool cannot approve.", + } + + +TOOLS = { + "search_sources": { + "description": "Retrieve evidence passages from latest live captures using Haystack BM25. Does not fetch, ingest, or approve.", + "input": SearchSourcesInput, + "run": search_sources, + "read_only": True, + }, + "get_claim": { + "description": "Return one recorded claim cell and the exact supporting sentence from the stored excerpt.", + "input": GetClaimInput, + "run": get_claim, + "read_only": True, + }, + "propose_write": { + "description": "Create a pending draft note. A separate human approve() is required before a note is written.", + "input": ProposeWriteInput, + "run": propose_write, + "read_only": False, + }, +} + + +def list_tools() -> list[dict]: + return [ + { + "name": name, + "description": spec["description"], + "inputSchema": spec["input"].model_json_schema(), + "readOnly": spec["read_only"], + } + for name, spec in TOOLS.items() + ] + + +def call_tool(store, name: str, arguments: dict | None) -> dict: + spec = TOOLS.get(name) + if spec is None: + raise ToolError(f"Unknown tool: {name}") + try: + payload = spec["input"].model_validate(arguments or {}) + except ValidationError as exc: + raise ToolError(str(exc)) from exc + return spec["run"](store, payload) diff --git a/tests/test_haystack_retrieval.py b/tests/test_haystack_retrieval.py new file mode 100644 index 0000000..880b8df --- /dev/null +++ b/tests/test_haystack_retrieval.py @@ -0,0 +1,94 @@ +"""SYNTHETIC checks that live retrieval actually runs Haystack, not a stub.""" +import pytest +from haystack import Document, Pipeline +from haystack.components.retrievers.in_memory import InMemoryBM25Retriever + +from infodesk import haystack_retrieval as hr +from infodesk.live_store import LiveStore +from infodesk.haystack_retrieval import documents_from_captures, retrieve_passages + + +def capture(chunk_id, body, title="SYNTHETIC gas report", source_id="eia"): + return { + "chunk_id": chunk_id, + "body": body, + "version_id": int(str(chunk_id).split(":")[0]), + "document_id": f"doc-{chunk_id}", + "source_id": source_id, + "url": "https://www.eia.gov/todayinenergy/detail.php?id=SYNTHETIC", + "title": title, + "received_at": "2026-09-09T12:00:00+00:00", + "published_at": "2026-09-09", + "text_hash": "SYNTHETIC", + "coverage": "SYNTHETIC test document", + } + + +def test_production_module_imports_haystack_pipeline(): + assert hr.Document is Document + assert hr.Pipeline is Pipeline + assert hr.InMemoryBM25Retriever is InMemoryBM25Retriever + docs = documents_from_captures([capture("1:0", "SYNTHETIC gas output reached 12 units.")]) + assert len(docs) == 1 and isinstance(docs[0], Document) + assert docs[0].meta["body"] == "SYNTHETIC gas output reached 12 units." + + +def test_SYNTHETIC_haystack_pipeline_ranks_matching_passage(monkeypatch): + ran = {} + original = Pipeline.run + + def wrapped(self, *args, **kwargs): + ran["used"] = True + return original(self, *args, **kwargs) + + monkeypatch.setattr(hr.Pipeline, "run", wrapped) + rows = [ + capture("1:0", "SYNTHETIC gas output reached 12 units.", title="gas report"), + capture("2:0", "Unrelated license titles only.", title="licenses", source_id="ofac-live"), + ] + hits = retrieve_passages("gas output", rows) + assert ran.get("used") is True + assert hits and hits[0]["chunk_id"] == "1:0" + assert "gas output" in hits[0]["body"] + assert retrieve_passages("unicorns", rows) == [] + + +def test_SYNTHETIC_retrieve_uses_haystack_not_sqlite_fts(monkeypatch): + store = LiveStore() + try: + store.ingest("eia", { + "title": "SYNTHETIC gas report", + "body": "SYNTHETIC gas output reached 12 units.", + "published_at": "2026-09-09", + "url": "https://www.eia.gov/todayinenergy/detail.php?id=SYNTHETIC", + "coverage": "SYNTHETIC test document", + }) + # FTS5 still has the capture. If retrieve() were still MATCH-based, + # stubbing Haystack would still return the gas passage. + monkeypatch.setattr("infodesk.live_store.retrieve_passages", + lambda *args, **kwargs: [{ + "chunk_id": "haystack-sentinel", + "body": "Haystack sentinel passage", + "version_id": 1, + "document_id": "sentinel", + "source_id": "eia", + "url": "https://www.eia.gov/todayinenergy/detail.php?id=SYNTHETIC", + "title": "sentinel", + "received_at": "t", + "published_at": "p", + "text_hash": "h", + "coverage": "c", + "rank": 1.0, + }]) + hits = store.retrieve("gas output", ["eia"]) + assert hits == [{"chunk_id": "haystack-sentinel", "body": "Haystack sentinel passage", + "version_id": 1, "document_id": "sentinel", "source_id": "eia", + "url": "https://www.eia.gov/todayinenergy/detail.php?id=SYNTHETIC", + "title": "sentinel", "received_at": "t", "published_at": "p", + "text_hash": "h", "coverage": "c", "rank": 1.0}] + monkeypatch.setattr("infodesk.live_store.retrieve_passages", + lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("Haystack step removed"))) + with pytest.raises(RuntimeError, match="Haystack step removed"): + store.retrieve("gas output", ["eia"]) + finally: + store.close() diff --git a/tests/test_mcp.py b/tests/test_mcp.py new file mode 100644 index 0000000..5afd766 --- /dev/null +++ b/tests/test_mcp.py @@ -0,0 +1,177 @@ +"""SYNTHETIC MCP checks. No external HTTP or model calls in CI.""" +import asyncio +import json +import os +import sys + +import pytest +from fastapi.testclient import TestClient +from mcp import Client, StdioServerParameters + +from infodesk.app import create_app +from infodesk.live_store import LiveStore +from infodesk.mcp_server import desk_mcp +from infodesk.mcp_tools import ToolError, bind_store, call_tool, list_tools +from infodesk.sources import ROOT + + +def live_document(body="SYNTHETIC gas output reached 12 units."): + return {"title": "SYNTHETIC gas report", "body": body, "published_at": "2026-09-09", + "url": "https://www.eia.gov/todayinenergy/detail.php?id=SYNTHETIC", + "coverage": "SYNTHETIC test document"} + + +def seed_live(store): + store.ingest("eia", live_document()) + store.add_check("eia", "error", "429 Too Many Requests", 12, "2026-09-09T12:02:00+00:00") + store.add_check("federal-register", "error", "ReadTimeout: SYNTHETIC connector timeout", 15, + "2026-09-09T12:02:00+00:00") + + +def test_SYNTHETIC_each_mcp_tool_is_invoked_and_propose_write_needs_approve(): + store = LiveStore() + try: + seed_live(store) + names = {item["name"] for item in list_tools()} + assert names == {"search_sources", "get_claim", "propose_write"} + + found = call_tool(store, "search_sources", {"question": "gas output", "source_ids": ["eia"]}) + assert found["engine"] == "haystack-bm25" + assert found["passages"] and "12 units" in found["passages"][0]["body"] + assert found["ledger_written"] is False + assert any(item["detail"].startswith("429") for item in found["source_errors"]) + + timed = call_tool(store, "search_sources", + {"question": "gas output", "source_ids": ["federal-register"]}) + assert any("timeout" in item["detail"].lower() for item in timed["source_errors"]) + + claim = call_tool(store, "get_claim", {"claim_id": "royalty", "source_id": "white-house"}) + assert claim["status"] == "stated" + assert claim["quote"] + assert "$200 billion" in claim["quote"] + + draft = call_tool(store, "propose_write", + {"headline": "SYNTHETIC draft", "body": "Pending human review."}) + assert draft["draft_id"] and draft["approved"] is False and draft["note_written"] is False + assert store.note_count() == 0 + assert store.approve(draft["draft_id"]) + assert store.note_count() == 1 + finally: + store.close() + + +def test_SYNTHETIC_schema_invalid_and_unknown_tool_are_visible(): + store = LiveStore() + try: + with pytest.raises(ToolError, match="question"): + call_tool(store, "search_sources", {"source_ids": ["eia"]}) + with pytest.raises(ToolError, match="extra"): + call_tool(store, "propose_write", { + "headline": "x", "body": "y", "approve": True, + }) + with pytest.raises(ToolError, match="Unknown claim_id"): + call_tool(store, "get_claim", {"claim_id": "invented", "source_id": "ofac"}) + with pytest.raises(ToolError, match="Unknown tool"): + call_tool(store, "apply_packet", {"headline": "no"}) + finally: + store.close() + + +def test_SYNTHETIC_http_lists_and_invokes_mcp_tools(tmp_path): + with TestClient(create_app(tmp_path / "SYNTHETIC-mcp.sqlite3")) as client: + listed = client.get("/api/mcp/tools") + assert listed.status_code == 200 + assert {item["name"] for item in listed.json()["tools"]} == { + "search_sources", "get_claim", "propose_write", + } + assert listed.json()["pages"].startswith("Recorded") + client.app.state.store.ingest("eia", live_document()) + search = client.post("/api/mcp/tools/search_sources", + json={"question": "gas output", "source_ids": ["eia"]}) + assert search.status_code == 200 and search.json()["passages"] + claim = client.post("/api/mcp/tools/get_claim", + json={"claim_id": "gl_46d", "source_id": "ofac"}) + assert claim.status_code == 200 and claim.json()["status"] == "stated" + draft = client.post("/api/mcp/tools/propose_write", + json={"headline": "SYNTHETIC", "body": "Do not persist yet."}) + assert draft.status_code == 200 and client.app.state.store.note_count() == 0 + invalid = client.post("/api/mcp/tools/search_sources", json={"source_ids": ["eia"]}) + assert invalid.status_code == 422 + unknown = client.post("/api/mcp/tools/apply_packet", json={}) + assert unknown.status_code == 404 + + +def test_SYNTHETIC_official_mcp_client_invokes_each_tool(): + store = LiveStore() + seed_live(store) + bind_store(store) + + async def exercise(): + async with Client(desk_mcp, raise_exceptions=True) as client: + tools = await client.list_tools() + assert {item.name for item in tools.tools} == { + "search_sources", "get_claim", "propose_write", + } + search = await client.call_tool("search_sources", + {"question": "gas output", "source_ids": ["eia"]}) + assert not search.is_error + payload = json.loads(search.content[0].text) + assert payload["passages"] and payload["engine"] == "haystack-bm25" + claim = await client.call_tool("get_claim", + {"claim_id": "royalty", "source_id": "white-house"}) + assert not claim.is_error + assert "$200 billion" in json.loads(claim.content[0].text)["quote"] + draft = await client.call_tool("propose_write", + {"headline": "SYNTHETIC", "body": "Pending."}) + assert not draft.is_error + assert json.loads(draft.content[0].text)["note_written"] is False + invalid = await client.call_tool("search_sources", {"source_ids": ["eia"]}) + assert invalid.is_error + assert "question" in invalid.content[0].text + + try: + asyncio.run(exercise()) + assert store.note_count() == 0 + finally: + store.close() + + +def test_SYNTHETIC_stdio_mcp_server_invokes_tools(tmp_path): + db = tmp_path / "SYNTHETIC-stdio.sqlite3" + store = LiveStore(db) + try: + seed_live(store) + finally: + store.close() + + async def exercise(): + params = StdioServerParameters( + command=sys.executable, + args=["-m", "infodesk.mcp_server"], + env={**os.environ, "INFODESK_DB": str(db), "HAYSTACK_TELEMETRY_ENABLED": "False"}, + cwd=str(ROOT), + ) + async with Client(params, raise_exceptions=True) as client: + tools = await client.list_tools() + assert {item.name for item in tools.tools} == { + "search_sources", "get_claim", "propose_write", + } + search = await client.call_tool("search_sources", + {"question": "gas output", "source_ids": ["eia"]}) + assert not search.is_error + claim = await client.call_tool("get_claim", + {"claim_id": "nabep", "source_id": "white-house"}) + assert not claim.is_error + draft = await client.call_tool("propose_write", + {"headline": "SYNTHETIC", "body": "stdio draft"}) + assert not draft.is_error + assert json.loads(draft.content[0].text)["note_written"] is False + + asyncio.run(exercise()) + check = LiveStore(db) + try: + assert check.note_count() == 0 + drafts = check.snapshot()["drafts"] + assert drafts and drafts[0]["status"] == "pending" + finally: + check.close()