diff --git a/.ai-team/TASK.md b/.ai-team/TASK.md index 8982ec2..5ce958b 100644 --- a/.ai-team/TASK.md +++ b/.ai-team/TASK.md @@ -1,61 +1,120 @@ # Current Task -- ID: `AR-003` -- Title: `Simplify the assignment plan to module level` -- Status: `done` -- Owner: `zzg` -- Next owner: `user/team` +- ID: `AR-005` +- Title: `Finish practical M05 search-to-reader pipeline` +- Status: `active` +- Owner: `wdwd200` +- Next owner: `wdwd200` ## Goal -根据用户反馈，把任务拆解入口从 122 个子任务收敛为模块级分配。保留此前详细版作为附录，但当前团队先按模块认领负责人、边界、主要产出、依赖和验收目标。 +Improve practical paper-search precision and later materialize legal OA full text locally for Reader. + +## Stage 1 scope + +Implement lightweight semantic relevance selection after multi-source search and deduplication: + +1. Send every deduplicated connector candidate to the optional LLM in batches of 8 using the original user research intent and only bounded title/abstract context. +2. Strictly validate every batch response and retain valid scores 2/3; reject scores 0/1. +3. If LLM screening is unavailable or a batch fails, retain candidates fail-open and use lexical relevance only for deterministic fallback ordering. +4. Build a seed-prioritized Reader workset capped at 20 papers. Persist relevant overflow candidates as deferred bibliographic records without OA/materialization in the current run. +5. Persist, resolve OA, materialize legal full text, and hand off only the current Reader workset. + +Stage 1 does not change query rewriting, Reader/Writer, OA providers, agents, embeddings, BM25, vector stores, rerankers, MCP, or the `PaperRecord` contract. + +## Stage 2 scope + +Materialize only legally available OA PDF full text after persistence and OA resolution: + +1. Preserve explicitly supplied seed papers after normal deduplication, bypass semantic rejection, and prioritize them in the current Reader workset. +2. Materialize only `open_access` records with an HTTP(S) `fulltext_url` into `paper-projects//02_literature/full_text/`. +3. Stream bounded downloads to a temporary file, validate them as PDFs readable by the existing Reader dependency, atomically rename on success, and upsert the same `PaperRecord.full_text_path`. +4. Preserve paper and OA records on every materialization failure, record a diagnostic, and continue processing remaining papers. + +Stage 2 does not change Reader, bypass access controls, perform browser automation, add OCR, introduce PDF architecture, or run live Unpaywall testing. + +## Query rewrite scope + +Use the existing optional LLM to generate exactly three validated, complementary scholarly retrieval queries: core topic, academic terminology/synonym, and a faithful complementary angle. The original `ResearchState.idea` remains the authoritative research intent and is passed separately to LLM relevance screening. + +On unavailable, failed, invalid, duplicate, or near-identical LLM output, use three conservative deterministic query variants derived only from the original idea. Do not restore unconditional review or benchmark query suffixes. ## Acceptance scenarios -- [x] `docs/AutoResearch_任务拆解书.md` 只作为模块级分配入口，覆盖 M00–M17。 -- [x] 每个模块说明目标、边界、主要交付物、前置模块、负责人建议和当前状态。 -- [x] 文档提供模块负责人验收模板和第一批模块分配建议。 -- [x] 原 122 子任务版本保留为 `docs/AutoResearch_任务拆解书_详细版.md`，不丢失后续拆解素材。 -- [x] README、Project-to-Act 和 VibeCollab 任务状态保持一致，不改变五业务 Agent 和证据门禁边界。 +- [x] Offline tests cover all-candidate batch semantic screening, strict validation, 0/1 rejection, 2/3 retention, failed-batch fail-open behavior, original-intent context, and dedupe-before-screening. +- [x] Offline tests cover lexical-only fallback ordering, deterministic fallback retention, score priority/tie-breaking, bounded Reader worksets, deferred persistence, and deferred OA/materialization exclusion. +- [x] Existing OA tests continue passing. +- [x] Full requested checks pass. +- [x] Multiple explicit seed papers bypass semantic screening while connector duplicates are removed. +- [x] Offline materialization tests cover successful PDF storage and persistence, Reader path consumption, failures, size limits, unsupported responses, ineligible OA records, repeat execution, and search integration. +- [x] Materializer diagnostics retain the exception type and safe bounded reasons; streamed downloads without `Content-Length` still enforce the byte limit. +- [x] Offline query-rewrite tests cover exactly-three valid LLM output, distinctness, failure/invalid/duplicate fallbacks, original-intent preservation, rewritten lexical ranking, and semantic screening context. ## Invariants -- 当前入口不要求团队一次性拆到函数、接口或工单；模块负责人后续再自行细化。 -- 详细版是参考材料，不代表所有子任务都已实现或已验收。 -- 真实论文试点、生产数据库、Web UI、灾备和性能仍保持原有“外部阻塞/待开发”状态。 -- 不提交 token、密钥、私有来源、完整论文、原始提示或虚构外部结果。 +- Preserve exactly five business agents and existing service architecture. +- LLM use remains optional and environment-configured; offline tests require no API key. +- OA enrichment remains after final relevance retention and continues to use existing providers and persistence. +- Full-text materialization only follows an OA record with an HTTP(S) full-text URL, and only persists a Reader-readable PDF after atomic local completion. +- Search-query rewriting improves retrieval only; it never overwrites the original research idea used to decide semantic relevance. +- Lexical relevance is fallback ordering only and never gates candidates when semantic screening is unavailable or a batch fails. +- Do not claim scientific truth, evidence validity, or paper conclusions from relevance screening. ## Decisions -- 使用 M00–M17 作为稳定模块 ID，先按模块分配负责人。 -- 保留详细拆解版，避免未来重新整理时丢失依赖和验收素材。 -- `.project-to-act` 继续作为长期项目事实源；`.ai-team/TASK.md` 只记录本次短周期交付。 +- Keep relevance selection inside `PaperSearchService` as focused private methods, with the existing `LLMService` injected by `AutoResearchApplication`. +- Batch semantic screening uses a constant batch size of 8 and validates exact requested-ID coverage with integer scores 0 through 3. +- Rank successful score-3 candidates before score-2 candidates; tie-break by DOI or normalized title. Place failed-batch candidates afterward using lexical score and stable identity. +- Treat seed papers as explicit selections: deduplicate them first, exclude connector duplicates, bypass semantic screening, and prioritize them in the 20-paper workset. +- Persist all deduplicated candidates as bibliographic records, but invoke evidence/OA/materialization only for the current Reader workset; relevant overflow remains deferred. +- The existing `papers.search_completed` event retains `paper_ids` and also records `reader_workset_paper_ids` and `deferred_paper_ids` so deferred papers are distinguishable from semantically rejected records. +- Use `FullTextMaterializer` after OA resolution. It stores deterministic PDF filenames under the existing `02_literature` project folder and upserts the existing paper record only after validation and atomic rename. +- Full-text materialization includes short `ValueError` reasons in diagnostics and remains bounded to 160 characters; other exception diagnostics remain type-only to avoid leaking transport details. +- `OrchestratorAgent` validates exactly three normalized-distinct, non-near-identical LLM queries. `PaperSearchAgent` passes `ResearchState.idea` as the semantic relevance context while lexical selection uses `search_queries`. +- The opt-in live smoke script generates a unique UTC timestamp-plus-random project ID unless `--project-id` is supplied; explicit project reuse is refused when its normal full-text directory already contains PDFs. +- Live E2E verification is stricter than production fail-open behavior: it requires real query rewrite, no semantic-screening fallback or failed batches, and at least one local-PDF Reader success. ## Completed -- 将原 `docs/AutoResearch_任务拆解书.md` 移为 `docs/AutoResearch_任务拆解书_详细版.md`。 -- 新建模块级 `docs/AutoResearch_任务拆解书.md`，包含 18 个模块、执行阶段、负责人模板和首批分配建议。 -- 保留 README 任务书入口，并将 AR-003 的状态、范围和下一步写入协作账本。 +- AR-004 OA discovery is present and must remain unchanged in behavior for retained papers. +- Existing search connectors, dedupe, persistence, evidence, knowledge, and Reader handoff are available. +- Stage 1 seed handling now preserves all unique user-supplied papers and prevents a matching connector record from duplicating them. +- Stage 2 now materializes qualifying OA PDFs locally and exposes the resulting path through the existing `PaperRecord.full_text_path` contract. +- Stage-2 cleanup now reports safe download failure reasons and validates the no-content-length streaming byte cap. +- Scholarly query rewriting now replaces the former fixed review/benchmark expansion without changing `ResearchState` or contracts. +- All deduplicated connector candidates now receive batched semantic screening when available; lexical ordering is fail-open fallback only. +- Relevant overflow candidates are retained as bibliographic records and deferred from OA/materialization when the Reader workset is full. +- Search-completed events now explicitly identify both the current Reader workset and deferred paper IDs. +- Live smoke tooling now prevents stale local PDFs from counting as a new run and treats any semantic batch failure as an incomplete E2E verification. +- Manual live OA-only verification passed for DOI `10.3389/frai.2024.1293084`: Unpaywall returned `open_access` with `cc-by`, a 733733-byte PDF was materialized locally, and `PaperReaderService` consumed it with 10 locators. +- Manual live E2E verification passed for "How can retrieval augmented generation reduce hallucinations in large language models?": a configured OpenAI-compatible LLM generated exactly three queries, real scholarly connectors returned 17 observed/deduplicated candidates, and real semantic screening completed with zero batch failures. +- The live E2E run placed 10 papers in the Reader workset with zero deferred; 8 resolved `open_access`, 2 had no open full text found, 4 PDFs were materialized locally, Reader consumed all 4 with zero failures, and the full search-to-Reader smoke result succeeded. + +## Live Verification Limitation + +Some legitimate OA records return HTML landing pages rather than direct PDF responses. `FullTextMaterializer` intentionally rejects HTML and does not scrape landing pages, automate browsers, or bypass access controls. Therefore `open_access` does not guarantee local PDF materialization. ## Pending -- 用户/团队需要为模块负责人模板填写实际 owner、分支、目标日期和 evidence ID。 -- 模块负责人确认后，再选择需要展开的模块，不默认展开全部模块。 -- 真实论文试点仍等待 idea、领域、目标 venue、授权全文和实验资源边界。 +- Opt-in live M05 smoke tooling was manually executed successfully by the user; do not rerun it during offline verification. +- Real email/API credentials are intentionally not stored in source, task files, fixtures, records, events, or reports. ## Next step -先认领 M01–M04、M05–M07、M08、M09–M10、M11–M12、M13–M15/M17、M16 这 7 个模块包；完成模块边界确认后，再只对当前瓶颈模块做子任务拆解。 +Prepare the combined AR-004 + AR-005 M05 pull request; do not execute live network modes or commit/push as part of this request. ## Verification -- [x] Project-to-Act `--check` — managed 配置有效，唯一账本保持不变。 -- [x] 模块文档结构检查 — M00–M17 全部存在，详细版文件可访问。 -- [x] `node .ai-team/check.mjs --base main` — AR-003 acceptance 5/5，任务范围和变更同步有效。 -- [x] Project-to-Act `--validate` — managed 配置 valid，issues 为空。 +- [x] `.venv\Scripts\python.exe -m pytest tests/test_relevance_selection.py -q` - 14 passed +- [x] `.venv\Scripts\python.exe -m pytest tests/test_query_rewrite.py tests/test_relevance_selection.py tests/test_open_access_service.py tests/test_full_text_service.py -q` - 27 passed +- [x] `.venv\Scripts\python.exe -m pytest tests/test_live_m05_smoke.py -q` - 8 passed; no network calls +- [x] `.venv\Scripts\python.exe -m pytest -q` - 53 passed +- [x] `.venv\Scripts\python.exe -m pytest -W error -q -p no:cacheprovider` - 53 passed +- [x] `.venv\Scripts\python.exe -m ruff check src tests scripts` - all checks passed +- [x] `.venv\Scripts\autoresearch.exe doctor` - healthy, five agents +- [x] `node .ai-team/check.mjs --base upstream/main` - valid; 8/8 functional progress +- [x] `git diff --check` - no whitespace errors; Windows line-ending warnings only ## Handoff note -- From: `zzg` -- To: `user/team` -- Summary: 当前分配入口已经降为模块级；详细拆解保留在附录。先分模块负责人，后续只对真正需要的模块继续细化。 +AR-005 stages 1, 2, query rewriting, bounded worksets, OA PDF materialization, and opt-in live smoke verification are complete. Live credentials remain local and unrecorded. diff --git a/.gitignore b/.gitignore index 2639a56..1b991bf 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,9 @@ dist/ var/ .langgraph_api/ +# Opt-in live M05 smoke artifacts remain local for manual inspection. +/paper-projects/live-m05-smoke-*/ + # Local reference repository; not part of the AutoResearch source tree DeepReason-Agents-Framework/ diff --git a/scripts/live_m05_smoke.py b/scripts/live_m05_smoke.py new file mode 100644 index 0000000..5a528b2 --- /dev/null +++ b/scripts/live_m05_smoke.py @@ -0,0 +1,384 @@ +"""Opt-in live M05 smoke checks. + +This script never runs as part of pytest and never prompts for credentials. +PowerShell setup uses the existing environment names: + + $env:UNPAYWALL_EMAIL="" + $env:LLM_API_KEY="" + $env:LLM_BASE_URL="" + $env:LLM_MODEL="" + $env:LLM_PROVIDER="" + + python scripts/live_m05_smoke.py --oa-only --doi "" + python scripts/live_m05_smoke.py --e2e --idea "" +""" + +from __future__ import annotations + +import argparse +import secrets +import sys +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import urlsplit + +from autoresearch.application import AutoResearchApplication +from autoresearch.config import get_settings +from autoresearch.contracts import ( + OpenAccessStatus, + PaperRecord, + ProjectCreate, + ResearchState, + RunStatus, +) +from autoresearch.open_access_service import OPEN_ACCESS_RECORD_KIND + +DEFAULT_DOI = "10.1371/journal.pone.0000308" +DEFAULT_IDEA = "evidence provenance in reproducible research workflows" +LIVE_PROJECT_PREFIX = "live-m05-smoke" + + +def configured_label(value: object) -> str: + """Return a non-sensitive configuration status label.""" + + return "yes" if value else "no" + + +def _generated_project_id() -> str: + timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") + return f"{LIVE_PROJECT_PREFIX}-{timestamp}-{secrets.token_hex(3)}" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Opt-in live M05 OA or E2E smoke test") + modes = parser.add_mutually_exclusive_group(required=True) + modes.add_argument( + "--oa-only", + action="store_true", + help="verify one real DOI OA-to-Reader chain", + ) + modes.add_argument("--e2e", action="store_true", help="verify the bounded live M05 pipeline") + parser.add_argument( + "--doi", + default=DEFAULT_DOI, + help=f"DOI for --oa-only (default: known PLOS OA DOI {DEFAULT_DOI})", + ) + parser.add_argument("--idea", default=DEFAULT_IDEA, help="research idea for --e2e") + parser.add_argument( + "--project-id", + help="reuse this isolated project only when its full_text directory has no existing PDFs", + ) + parser.add_argument( + "--per-connector-limit", + type=int, + default=3, + help="small live retrieval limit per connector and query (default: 3)", + ) + return parser + + +def _ensure_project(application: AutoResearchApplication, project_id: str) -> None: + if application.projects.get(project_id) is None: + application.create_project( + ProjectCreate( + project_id=project_id, + title="Live M05 smoke test", + idea="Temporary isolated project for an opt-in live M05 check.", + ), + owner="live-smoke", + ) + + +def _pdf_directory(application: AutoResearchApplication, project_id: str) -> Path: + return _pdf_directory_for_root(application.settings.projects_dir, project_id) + + +def _pdf_directory_for_root(projects_dir: Path, project_id: str) -> Path: + root = Path(projects_dir).resolve() + project_dir = (root / project_id).resolve() + if not project_dir.is_relative_to(root): + raise ValueError("project path escaped projects directory") + return project_dir / "02_literature" / "full_text" + + +def _project_id_for_run(args: argparse.Namespace, settings: object) -> str | None: + if not args.project_id: + project_id = _generated_project_id() + print(f"Generated isolated project ID: {project_id}") + return project_id + + directory = _pdf_directory_for_root(settings.projects_dir, args.project_id) + existing_pdfs = sorted(directory.glob("*.pdf")) if directory.is_dir() else [] + if existing_pdfs: + print("Refusing to reuse a project ID with existing PDFs; choose a new --project-id.") + print(f"Existing PDF directory: {directory}") + return None + return args.project_id + + +def _e2e_success( + *, + query_rewrite_used: bool, + semantic_unavailable: bool, + semantic_failure_count: int, + reader_success_count: int, +) -> bool: + return ( + query_rewrite_used + and not semantic_unavailable + and semantic_failure_count == 0 + and reader_success_count > 0 + ) + + +def _print_pdf_directory(directory: Path) -> None: + print(f"PDF directory: {directory}") + files = sorted(directory.glob("*.pdf")) if directory.is_dir() else [] + if not files: + print("Downloaded PDF files: none") + return + print("Downloaded PDF files:") + for path in files: + print(f"- {path.name} ({path.stat().st_size} bytes)") + + +def _oa_only(args: argparse.Namespace) -> int: + settings = get_settings() + if not settings.unpaywall_email: + print("UNPAYWALL_EMAIL configured: no") + print("Set UNPAYWALL_EMAIL in the environment before running --oa-only.") + return 2 + print("UNPAYWALL_EMAIL configured: yes") + project_id = _project_id_for_run(args, settings) + if project_id is None: + return 2 + + application = AutoResearchApplication(settings.model_copy(update={"network_enabled": True})) + try: + _ensure_project(application, project_id) + paper = PaperRecord( + project_id=project_id, + title="Live M05 OA smoke paper", + doi=args.doi, + source="live_smoke", + ) + persisted, _ = application.search._persist(paper) + oa = application.open_access.resolve(persisted) + materialized = application.full_text.materialize(persisted, oa) + paper = materialized.paper + + local_path = Path(paper.full_text_path) if paper.full_text_path else None + pdf_valid = bool(local_path and application.full_text._valid_pdf(local_path)) + reader_succeeded = False + locator_count = 0 + example_locator = "none" + if pdf_valid: + try: + card, _ = application.reader.read(paper) + reader_succeeded = True + locator_count = len(card.locators) + example_locator = card.locators[0] if card.locators else "none" + except Exception as exc: + print(f"Reader error: {type(exc).__name__}") + + print("\nOA SMOKE") + print(f"DOI: {args.doi}") + print(f"OA status: {oa.status.value}") + print(f"OA provider/source: {oa.source or 'unknown'}") + print(f"Checked providers: {', '.join(oa.checked_sources) or 'none'}") + print(f"Full-text URL found: {configured_label(oa.fulltext_url)}") + print(f"License: {oa.license or 'unknown'}") + print( + "PDF downloaded: " + f"{configured_label(local_path and local_path.is_file() and pdf_valid)}" + ) + print(f"Local PDF path: {local_path or 'none'}") + print( + "Local PDF bytes: " + f"{local_path.stat().st_size if local_path and local_path.is_file() else 'none'}" + ) + print(f"Reader succeeded: {configured_label(reader_succeeded)}") + print(f"Reader locator count: {locator_count}") + print(f"Example locator: {example_locator}") + if oa.diagnostics or materialized.diagnostics: + print("Diagnostics:") + for diagnostic in [*oa.diagnostics, *materialized.diagnostics][:8]: + print(f"- {diagnostic[:240]}") + _print_pdf_directory(_pdf_directory(application, project_id)) + + url_is_http = bool( + oa.fulltext_url + and urlsplit(oa.fulltext_url).scheme in {"http", "https"} + ) + success = ( + oa.status is OpenAccessStatus.OPEN_ACCESS + and url_is_http + and pdf_valid + and local_path is not None + and local_path.is_file() + and local_path.stat().st_size > 0 + and reader_succeeded + ) + return 0 if success else 1 + finally: + application.close() + + +def _e2e(args: argparse.Namespace) -> int: + settings = get_settings() + if not settings.llm_configured: + print("LLM configured: no") + print("Configure LLM_PROVIDER, LLM_API_KEY, LLM_BASE_URL, and LLM_MODEL before --e2e.") + return 2 + if args.per_connector_limit < 1: + print("--per-connector-limit must be at least 1.") + return 2 + project_id = _project_id_for_run(args, settings) + if project_id is None: + return 2 + + print("LLM configured: yes") + application = AutoResearchApplication(settings.model_copy(update={"network_enabled": True})) + try: + _ensure_project(application, project_id) + before_ids = { + paper["paper_id"] + for paper in application.store.list("paper", project_id=project_id) + } + state = ResearchState( + project_id=project_id, + idea=args.idea, + run_status=RunStatus.RUNNING, + ) + scoped = application.orchestrator_agent.scope(state.model_dump(mode="json")) + queries = scoped["search_queries"] + rewrite_fallback = any( + "query rewriting unavailable" in diagnostic + or "query rewriting returned invalid" in diagnostic + for diagnostic in scoped.get("diagnostics", []) + ) + if len(queries) != 3 or rewrite_fallback: + print("LLM query rewrite did not complete with three valid live queries.") + return 1 + print("LLM query rewrite used: yes") + outcome = application.search.search( + project_id, + queries, + per_connector_limit=args.per_connector_limit, + relevance_context=args.idea, + ) + after_ids = { + paper["paper_id"] + for paper in application.store.list("paper", project_id=project_id) + } + observed_count = len(after_ids - before_ids) + events = application.store.events(project_id) + event = next( + item["payload"] + for item in reversed(events) + if item["event_type"] == "papers.search_completed" + ) + workset_ids = event.get("reader_workset_paper_ids", event.get("paper_ids", [])) + deferred_ids = event.get("deferred_paper_ids", []) + semantic_failures = [ + diagnostic + for diagnostic in outcome.diagnostics + if "LLM semantic screening failed for batch" in diagnostic + ] + semantic_unavailable = any( + "semantic relevance screening unavailable" in diagnostic + for diagnostic in outcome.diagnostics + ) + + reader_successes: list[tuple[PaperRecord, int, str]] = [] + reader_failures = 0 + for paper in outcome.papers: + try: + card, _ = application.reader.read(paper) + if paper.full_text_path and Path(paper.full_text_path).is_file(): + reader_successes.append( + ( + paper, + len(card.locators), + card.locators[0] if card.locators else "none", + ) + ) + except Exception: + reader_failures += 1 + + oa_records = [ + application.store.get(OPEN_ACCESS_RECORD_KIND, paper_id) + for paper_id in workset_ids + ] + oa_records = [record for record in oa_records if record] + oa_open = sum( + record["status"] == OpenAccessStatus.OPEN_ACCESS.value for record in oa_records + ) + oa_no_fulltext = sum(record["status"] == "no_open_fulltext_found" for record in oa_records) + oa_unknown = sum( + record["status"] == OpenAccessStatus.UNKNOWN.value for record in oa_records + ) + downloaded = [ + paper + for paper in outcome.papers + if paper.full_text_path and Path(paper.full_text_path).is_file() + ] + print("\nM05 LIVE E2E") + print(f"Original idea:\n{args.idea}") + print("Rewritten queries:") + for index, query in enumerate(queries, start=1): + print(f"{index}. {query}") + print("Retrieval:") + print(f"Raw/observed candidates: {observed_count}") + print(f"Reader workset: {len(workset_ids)}") + print(f"Deferred candidates: {len(deferred_ids)}") + print(f"LLM semantic screening used: {configured_label(not semantic_unavailable)}") + print(f"LLM semantic screening batch failures: {len(semantic_failures)}") + print("Reader workset papers:") + for paper in outcome.papers: + print(f"[selected] {paper.title}") + print( + "Semantic score labels: unavailable from the current production outcome; " + "no scores were fabricated." + ) + print("OA:") + print(f"OA open: {oa_open}") + print(f"No OA full text found: {oa_no_fulltext}") + print(f"OA unknown: {oa_unknown}") + print("PDF:") + print(f"Downloaded PDFs: {len(downloaded)}") + download_failures = sum( + "materialization failed" in diagnostic + for diagnostic in outcome.diagnostics + ) + print(f"Download failures: {download_failures}") + print(f"Reader locally readable papers: {len(reader_successes)}") + print(f"Reader failures: {reader_failures}") + print("Downloaded files:") + for paper in downloaded: + print(f"- {paper.full_text_path}") + _print_pdf_directory(_pdf_directory(application, project_id)) + + success = _e2e_success( + query_rewrite_used=True, + semantic_unavailable=semantic_unavailable, + semantic_failure_count=len(semantic_failures), + reader_success_count=len(reader_successes), + ) + print(f"\nFull search-to-Reader chain completed: {configured_label(success)}") + return 0 if success else 1 + finally: + application.close() + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return _oa_only(args) if args.oa_only else _e2e(args) + except Exception as exc: + print(f"Smoke test failed: {type(exc).__name__}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/autoresearch/agents/orchestrator.py b/src/autoresearch/agents/orchestrator.py index 3e21e5f..ea4dfe8 100644 --- a/src/autoresearch/agents/orchestrator.py +++ b/src/autoresearch/agents/orchestrator.py @@ -42,41 +42,73 @@ def __init__( self.llm = llm @staticmethod - def _queries(idea: str) -> list[str]: - cleaned = re.sub(r"[^\w\u4e00-\u9fff -]+", " ", idea) - tokens = [token for token in cleaned.split() if len(token) > 1] - base = " ".join(tokens[:12]) or idea[:120] - return list( - dict.fromkeys( - [ - base, - f"{base} systematic review", - f"{base} benchmark evaluation", - ] + def _query_tokens(value: str) -> list[str]: + cleaned = re.sub(r"[^\w\u4e00-\u9fff -]+", " ", value) + return [token for token in cleaned.split() if len(token) > 1] + + @classmethod + def _fallback_queries(cls, idea: str) -> list[str]: + tokens = cls._query_tokens(idea) + base = " ".join(tokens[:10]) or idea.strip()[:120] + if len(tokens) >= 4: + first = " ".join(tokens[: len(tokens) // 2]) + second = " ".join(tokens[len(tokens) // 2 : 10]) + candidates = [base, first, second] + if len({cls._normalized_query(query) for query in candidates}) == 3: + return candidates + return [base, f"{base} research", f"{base} literature"] + + @staticmethod + def _normalized_query(value: str) -> str: + return " ".join(re.findall(r"\w+", value.lower(), flags=re.UNICODE)) + + @classmethod + def _valid_queries(cls, value: object) -> list[str] | None: + if not isinstance(value, list) or len(value) != 3: + return None + queries = [item.strip() for item in value if isinstance(item, str)] + if len(queries) != 3 or any(not query or len(query) > 160 for query in queries): + return None + normalized = [cls._normalized_query(query) for query in queries] + if any(not query for query in normalized) or len(set(normalized)) != 3: + return None + token_sets = [set(query.split()) for query in normalized] + if any( + len(first & second) / len(first | second) >= 0.8 + for index, first in enumerate(token_sets) + for second in token_sets[index + 1 :] + ): + return None + return queries + + def _queries(self, idea: str) -> tuple[list[str], str | None]: + fallback = self._fallback_queries(idea) + if not self.llm.available: + return fallback, "LLM query rewriting unavailable; deterministic fallback used." + try: + result = self.llm.complete_json( + system=( + "Return JSON with exactly three strings in search_queries. They must be " + "concise, distinct scholarly keyword queries for the user's research intent: " + "core topic, academic terminology or synonym, and a faithful complementary " + "angle. Do not add unsupported methods, datasets, populations, diseases, " + "interventions, benchmarks, " + "or review terminology unless the user asks for them." + ), + user=idea, + ) + queries = self._valid_queries(result.get("search_queries") if result else None) + if queries is not None: + return queries, "Managed LLM generated three scholarly search queries." + return fallback, "LLM query rewriting returned invalid output; fallback used." + except Exception as exc: + return fallback, ( + f"LLM query rewriting unavailable: {type(exc).__name__}; fallback used." ) - ) def scope(self, raw_state: WorkflowState) -> dict: state = ResearchState.model_validate(raw_state) - queries = self._queries(state.idea) - llm_note = None - if self.llm.available: - try: - result = self.llm.complete_json( - system=( - "You decompose a research idea. Return JSON with a 'search_queries' " - "array only. Do not claim novelty or results." - ), - user=state.idea, - ) - candidates = result.get("search_queries", []) if result else [] - if candidates: - queries = list( - dict.fromkeys([str(item)[:300] for item in candidates if str(item).strip()]) - )[:8] - llm_note = "Managed LLM refined the search query plan." - except Exception as exc: - llm_note = f"LLM query refinement unavailable: {type(exc).__name__}; fallback used." + queries, llm_note = self._queries(state.idea) charter_id = new_id("charter") charter = { diff --git a/src/autoresearch/agents/paper_search.py b/src/autoresearch/agents/paper_search.py index 7183db3..b6de5f0 100644 --- a/src/autoresearch/agents/paper_search.py +++ b/src/autoresearch/agents/paper_search.py @@ -45,6 +45,7 @@ def run(self, raw_state: WorkflowState) -> dict: state.project_id, state.search_queries, seed_papers=seeds, + relevance_context=state.idea, ) diagnostics = [*state.diagnostics, *outcome.diagnostics] if not outcome.papers: diff --git a/src/autoresearch/application.py b/src/autoresearch/application.py index 1da0cfc..cfacf4e 100644 --- a/src/autoresearch/application.py +++ b/src/autoresearch/application.py @@ -36,11 +36,17 @@ from autoresearch.evidence import EvidenceService from autoresearch.evolution_service import EvolutionService, ExperienceService from autoresearch.execution_service import ExecutionService +from autoresearch.full_text_service import FullTextMaterializer from autoresearch.gates import GateService from autoresearch.graph import build_research_graph from autoresearch.handoffs import HandoffService from autoresearch.knowledge import KnowledgeService from autoresearch.llm import LLMService +from autoresearch.open_access_service import ( + OpenAccessResolver, + OpenAlexOAProvider, + UnpaywallProvider, +) from autoresearch.profile_service import UserProfileService from autoresearch.project_service import ProjectService from autoresearch.reader_service import PaperReaderService @@ -66,11 +72,33 @@ def __init__(self, settings: Settings | None = None): self.settings.template_dir, self.store, ) + oa_providers = [] + + if self.settings.unpaywall_email: + oa_providers.append( + UnpaywallProvider(self.settings.unpaywall_email) + ) + + oa_providers.append(OpenAlexOAProvider()) + + self.open_access = OpenAccessResolver( + self.store, + oa_providers, + ) + self.full_text = FullTextMaterializer(self.store, self.settings.projects_dir) + self.search = PaperSearchService( self.store, self.evidence, self.knowledge, network_enabled=self.settings.network_enabled, + open_access=( + self.open_access + if self.settings.network_enabled + else None + ), + llm=self.llm, + materializer=self.full_text, ) self.reader = PaperReaderService(self.store, self.evidence, self.knowledge) self.writing = WritingService( diff --git a/src/autoresearch/config.py b/src/autoresearch/config.py index a1590b0..0f0a18f 100644 --- a/src/autoresearch/config.py +++ b/src/autoresearch/config.py @@ -29,6 +29,10 @@ class Settings(BaseSettings): template_dir: Path = Field( default=Path("paper-projects/_template"), alias="AUTORESEARCH_TEMPLATE_DIR" ) + unpaywall_email: str | None = Field( + default=None, + alias="UNPAYWALL_EMAIL", + ) network_enabled: bool = Field(default=False, alias="AUTORESEARCH_NETWORK_ENABLED") @property @@ -55,6 +59,7 @@ def safe_summary(self) -> dict[str, object]: "network_enabled": self.network_enabled, "db_path": str(self.db_path), "checkpoint_path": str(self.checkpoint_path), + "unpaywall_email_configured": bool(self.unpaywall_email), } diff --git a/src/autoresearch/contracts.py b/src/autoresearch/contracts.py index 34158de..40f8b71 100644 --- a/src/autoresearch/contracts.py +++ b/src/autoresearch/contracts.py @@ -90,6 +90,12 @@ class KnowledgePartition(StrEnum): PROJECTS = "projects" +class OpenAccessStatus(StrEnum): + OPEN_ACCESS = "open_access" + NO_OPEN_FULLTEXT_FOUND = "no_open_fulltext_found" + UNKNOWN = "unknown" + + class ClaimStatus(StrEnum): UNKNOWN = "unknown" HYPOTHESIS = "hypothesis" @@ -203,6 +209,18 @@ class PaperRecord(BaseModel): retrieved_at: datetime = Field(default_factory=utc_now) +class OpenAccessRecord(BaseModel): + paper_id: str + project_id: str + status: OpenAccessStatus + fulltext_url: str | None = None + license: str | None = None + source: str | None = None + checked_sources: list[str] = Field(default_factory=list) + diagnostics: list[str] = Field(default_factory=list) + checked_at: datetime = Field(default_factory=utc_now) + + class ReadingCard(BaseModel): card_id: str = Field(default_factory=lambda: new_id("card")) project_id: str diff --git a/src/autoresearch/full_text_service.py b/src/autoresearch/full_text_service.py new file mode 100644 index 0000000..46bfc25 --- /dev/null +++ b/src/autoresearch/full_text_service.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import hashlib +import os +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from urllib.parse import urlsplit + +import httpx +from pypdf import PdfReader + +from autoresearch.contracts import OpenAccessRecord, OpenAccessStatus, PaperRecord +from autoresearch.storage import RecordStore + + +@dataclass +class MaterializationResult: + paper: PaperRecord + diagnostics: list[str] = field(default_factory=list) + + +class FullTextMaterializer: + """Materialize legal OA PDFs into the existing project literature tree.""" + + def __init__( + self, + store: RecordStore, + projects_dir: Path, + *, + timeout: float = 20.0, + max_bytes: int = 25 * 1024 * 1024, + client: httpx.Client | None = None, + ): + self.store = store + self.projects_dir = Path(projects_dir).resolve() + self.timeout = timeout + self.max_bytes = max_bytes + self.client = client + + @staticmethod + def _filename(paper: PaperRecord) -> str: + identity = (paper.doi or paper.title).strip().lower() + digest = hashlib.sha256(identity.encode()).hexdigest()[:24] + return f"paper-{digest}.pdf" + + @staticmethod + def _valid_pdf(path: Path) -> bool: + try: + return path.is_file() and path.stat().st_size > 0 and bool(PdfReader(path).pages) + except Exception: + return False + + def _target_path(self, paper: PaperRecord) -> Path: + project_dir = (self.projects_dir / paper.project_id).resolve() + if not project_dir.is_relative_to(self.projects_dir): + raise ValueError("project path escaped projects directory") + target_dir = project_dir / "02_literature" / "full_text" + target_dir.mkdir(parents=True, exist_ok=True) + return target_dir / self._filename(paper) + + def _persist_path(self, paper: PaperRecord, target: Path) -> PaperRecord: + updated = paper.model_copy(update={"full_text_path": str(target)}) + self.store.put( + "paper", + updated.paper_id, + updated, + project_id=updated.project_id, + partition="papers", + ) + return updated + + def materialize( + self, paper: PaperRecord, oa_record: OpenAccessRecord + ) -> MaterializationResult: + diagnostics: list[str] = [] + if oa_record.status is not OpenAccessStatus.OPEN_ACCESS: + return MaterializationResult(paper) + url = (oa_record.fulltext_url or "").strip() + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"}: + diagnostics.append( + f"Full-text materialization skipped for {paper.paper_id}: URL is not HTTP(S)." + ) + return MaterializationResult(paper, diagnostics) + + try: + target = self._target_path(paper) + if self._valid_pdf(target): + return MaterializationResult(self._persist_path(paper, target)) + + temporary_name: str | None = None + try: + stream = ( + self.client.stream("GET", url) + if self.client is not None + else httpx.stream("GET", url, timeout=self.timeout, follow_redirects=True) + ) + with stream as response: + response.raise_for_status() + content_length = response.headers.get("content-length") + if content_length and int(content_length) > self.max_bytes: + raise ValueError("response exceeds maximum download size") + with tempfile.NamedTemporaryFile( + mode="wb", + dir=target.parent, + prefix=f".{target.name}.", + suffix=".part", + delete=False, + ) as temporary: + temporary_name = temporary.name + total = 0 + for chunk in response.iter_bytes(chunk_size=64 * 1024): + total += len(chunk) + if total > self.max_bytes: + raise ValueError("response exceeds maximum download size") + temporary.write(chunk) + temporary_path = Path(temporary_name) + content_type = response.headers.get("content-type", "").split(";", 1)[0].lower() + if content_type and content_type != "application/pdf": + raise ValueError(f"unsupported content type: {content_type}") + if not self._valid_pdf(temporary_path): + raise ValueError("downloaded content is not a readable PDF") + os.replace(temporary_path, target) + temporary_name = None + finally: + if temporary_name: + Path(temporary_name).unlink(missing_ok=True) + return MaterializationResult(self._persist_path(paper, target)) + except Exception as exc: + reason = str(exc).strip() if isinstance(exc, ValueError) else "" + detail = f": {reason[:160]}" if reason else "" + diagnostics.append( + f"Full-text materialization failed for {paper.paper_id}: " + f"{type(exc).__name__}{detail}; paper retained." + ) + return MaterializationResult(paper, diagnostics) diff --git a/src/autoresearch/open_access_service.py b/src/autoresearch/open_access_service.py new file mode 100644 index 0000000..0c8e6e5 --- /dev/null +++ b/src/autoresearch/open_access_service.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass, field +from typing import Protocol + +import httpx + +from autoresearch.contracts import ( + OpenAccessRecord, + OpenAccessStatus, + PaperRecord, +) +from autoresearch.storage import RecordStore + +OPEN_ACCESS_RECORD_KIND = "paper_open_access" + + +@dataclass(frozen=True) +class OpenAccessLookupResult: + """A provider's report about a paper's legally available full text.""" + + fulltext_url: str | None = None + license: str | None = None + diagnostics: list[str] = field(default_factory=list) + + +class OpenAccessProvider(Protocol): + """Read-only OA lookup provider used by :class:`OpenAccessResolver`.""" + + name: str + + def lookup(self, paper: PaperRecord) -> OpenAccessLookupResult: ... + + +class UnpaywallProvider: + name = "unpaywall" + + def __init__(self, email: str, timeout: float = 30.0) -> None: + self.email = email + self.timeout = timeout + + def lookup(self, paper: PaperRecord) -> OpenAccessLookupResult: + if not paper.doi: + return OpenAccessLookupResult() + + response = httpx.get( + f"https://api.unpaywall.org/v2/{paper.doi}", + params={"email": self.email}, + timeout=self.timeout, + ) + + if response.status_code == 404: + return OpenAccessLookupResult( + diagnostics=["unpaywall: DOI not found"] + ) + + response.raise_for_status() + + payload = response.json() + + if not payload.get("is_oa"): + return OpenAccessLookupResult( + diagnostics=["unpaywall: no OA location found"] + ) + + location = payload.get("best_oa_location") or {} + + fulltext_url = ( + location.get("url_for_pdf") + or location.get("url_for_landing_page") + or location.get("url") + ) + + return OpenAccessLookupResult( + fulltext_url=fulltext_url, + license=location.get("license"), + ) + + +class OpenAlexOAProvider: + name = "openalex" + + def __init__(self, timeout: float = 30.0) -> None: + self.timeout = timeout + + def lookup(self, paper: PaperRecord) -> OpenAccessLookupResult: + if not paper.doi: + return OpenAccessLookupResult() + + response = httpx.get( + f"https://api.openalex.org/works/https://doi.org/{paper.doi}", + timeout=self.timeout, + ) + + if response.status_code == 404: + return OpenAccessLookupResult( + diagnostics=["openalex: DOI not found"] + ) + + response.raise_for_status() + + payload = response.json() + location = payload.get("best_oa_location") or {} + + fulltext_url = ( + location.get("pdf_url") + or location.get("landing_page_url") + or (payload.get("open_access") or {}).get("oa_url") + ) + + if not fulltext_url: + return OpenAccessLookupResult( + diagnostics=["openalex: no OA location found"] + ) + + return OpenAccessLookupResult( + fulltext_url=fulltext_url, + license=location.get("license"), + ) + + +class OpenAccessResolver: + """Resolve and persist the latest legal OA entry point for a paper.""" + + def __init__( + self, + store: RecordStore, + providers: Sequence[OpenAccessProvider], + ) -> None: + self.store = store + self.providers = tuple(providers) + + def resolve(self, paper: PaperRecord) -> OpenAccessRecord: + if not paper.doi: + record = OpenAccessRecord( + paper_id=paper.paper_id, + project_id=paper.project_id, + status=OpenAccessStatus.UNKNOWN, + diagnostics=["OA lookup skipped because paper has no DOI."], + ) + self._persist(record) + return record + + checked_sources: list[str] = [] + diagnostics: list[str] = [] + provider_failed = False + + for provider in self.providers: + provider_name = provider.name + checked_sources.append(provider_name) + + try: + result = provider.lookup(paper) + except Exception as exc: + provider_failed = True + diagnostics.append( + f"{provider_name} failed: {type(exc).__name__}" + ) + continue + + diagnostics.extend(result.diagnostics) + + if result.fulltext_url and result.fulltext_url.strip(): + record = OpenAccessRecord( + paper_id=paper.paper_id, + project_id=paper.project_id, + status=OpenAccessStatus.OPEN_ACCESS, + fulltext_url=result.fulltext_url, + license=result.license, + source=provider_name, + checked_sources=checked_sources, + diagnostics=diagnostics, + ) + self._persist(record) + return record + + status = ( + OpenAccessStatus.UNKNOWN + if provider_failed + else OpenAccessStatus.NO_OPEN_FULLTEXT_FOUND + ) + + record = OpenAccessRecord( + paper_id=paper.paper_id, + project_id=paper.project_id, + status=status, + checked_sources=checked_sources, + diagnostics=diagnostics, + ) + self._persist(record) + return record + + def _persist(self, record: OpenAccessRecord) -> None: + self.store.put( + OPEN_ACCESS_RECORD_KIND, + record.paper_id, + record, + project_id=record.project_id, + ) diff --git a/src/autoresearch/search_service.py b/src/autoresearch/search_service.py index 0af79c6..aaa2964 100644 --- a/src/autoresearch/search_service.py +++ b/src/autoresearch/search_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import re from dataclasses import dataclass, field from typing import Any, Protocol @@ -15,7 +16,10 @@ WikiPage, ) from autoresearch.evidence import EvidenceService +from autoresearch.full_text_service import FullTextMaterializer from autoresearch.knowledge import KnowledgeService +from autoresearch.llm import LLMService +from autoresearch.open_access_service import OpenAccessResolver from autoresearch.storage import RecordStore @@ -25,6 +29,12 @@ class SearchOutcome: diagnostics: list[str] = field(default_factory=list) +@dataclass(frozen=True) +class _CandidateDecision: + paper: PaperRecord + relevance_score: int | None + + class ScholarlyConnector(Protocol): name: str @@ -139,6 +149,9 @@ def __init__( *, network_enabled: bool, connectors: list[ScholarlyConnector] | None = None, + open_access: OpenAccessResolver | None = None, + llm: LLMService | None = None, + materializer: FullTextMaterializer | None = None, ): self.store = store self.evidence = evidence @@ -149,6 +162,9 @@ def __init__( CrossrefConnector(), SemanticScholarConnector(), ] + self.open_access = open_access + self.llm = llm + self.materializer = materializer @staticmethod def _dedupe_key(paper: PaperRecord) -> str: @@ -195,6 +211,176 @@ def _persist(self, paper: PaperRecord) -> tuple[PaperRecord, EvidenceItem]: ) return paper, evidence + def _persist_bibliographic(self, paper: PaperRecord) -> None: + self.store.put( + "paper", + paper.paper_id, + paper, + project_id=paper.project_id, + partition=KnowledgePartition.PAPERS.value, + ) + + @staticmethod + def _normalized_tokens(value: str) -> list[str]: + return re.findall(r"\w+", value.lower(), flags=re.UNICODE) + + @classmethod + def _lexical_score(cls, paper: PaperRecord, queries: list[str]) -> float: + title = cls._normalized_tokens(paper.title) + abstract = cls._normalized_tokens(paper.abstract) + title_set = set(title) + abstract_set = set(abstract) + best = 0.0 + for query in queries: + query_tokens = set(cls._normalized_tokens(query)) + if not query_tokens: + continue + score = 2 * len(query_tokens & title_set) + len(query_tokens & abstract_set) + query_text = " ".join(cls._normalized_tokens(query)) + document_text = " ".join(title + abstract) + if query_text and query_text in document_text: + score += 0.5 + best = max(best, float(score)) + return best + + @classmethod + def _identity(cls, paper: PaperRecord) -> str: + return (paper.doi or "").strip().lower() or " ".join(cls._normalized_tokens(paper.title)) + + def _semantic_screen_batches( + self, + queries: list[str], + candidates: list[PaperRecord], + diagnostics: list[str], + *, + relevance_context: str | None = None, + ) -> dict[str, int | None]: + decisions: dict[str, int | None] = {} + if not candidates: + return decisions + if self.llm is None or not getattr(self.llm, "available", True): + diagnostics.append( + "LLM semantic relevance screening unavailable; all connector candidates " + "retained for lexical fallback ordering." + ) + return {paper.paper_id: None for paper in candidates} + + context = relevance_context or "\n".join(f"- {query}" for query in queries if query.strip()) + batch_size = 8 + abstract_limit = 1200 + for batch_number, start in enumerate(range(0, len(candidates), batch_size), start=1): + batch = candidates[start : start + batch_size] + requested_ids = [paper.paper_id for paper in batch] + payload = [ + { + "paper_id": paper.paper_id, + "title": paper.title, + "abstract": (paper.abstract or "")[:abstract_limit], + } + for paper in batch + ] + user = ( + f"Original research intent:\n{context}\n\n" + "Judge each paper only for whether it is worth sending to Reader. " + "Papers:\n" + f"{json.dumps(payload, ensure_ascii=False)}" + ) + try: + result = self.llm.complete_json( + system=( + "Return JSON exactly as {\"scores\":[{\"paper_id\":\"...\"," + "\"relevance_score\":0}]}. Include every requested paper exactly once. " + "Scores: 0 clearly unrelated, 1 weakly related, 2 relevant, " + "3 highly relevant. Judge relevance to the original research intent only; " + "do not judge scientific truth, evidence validity, or conclusions." + ), + user=user, + ) + batch_scores = self._validate_batch_scores(result, requested_ids) + except Exception as exc: + diagnostics.append( + f"LLM semantic screening failed for batch {batch_number} " + f"({len(batch)} papers): {type(exc).__name__}; batch retained unscored." + ) + continue + decisions.update(batch_scores) + return decisions + + @staticmethod + def _validate_batch_scores( + result: object, requested_ids: list[str] + ) -> dict[str, int]: + if not isinstance(result, dict) or not isinstance(result.get("scores"), list): + raise ValueError("invalid batch response") + scores = result["scores"] + if len(scores) != len(requested_ids): + raise ValueError("batch response count mismatch") + requested = set(requested_ids) + validated: dict[str, int] = {} + for item in scores: + if not isinstance(item, dict): + raise ValueError("invalid batch score item") + paper_id = item.get("paper_id") + score = item.get("relevance_score") + if paper_id not in requested or paper_id in validated: + raise ValueError("unknown or duplicate paper_id") + if not isinstance(score, int) or isinstance(score, bool) or score not in {0, 1, 2, 3}: + raise ValueError("invalid relevance score") + validated[paper_id] = score + if set(validated) != requested: + raise ValueError("missing paper_id") + return validated + + @classmethod + def _fallback_order( + cls, candidates: list[PaperRecord], queries: list[str] + ) -> list[_CandidateDecision]: + return [ + _CandidateDecision(paper, None) + for paper in sorted( + candidates, + key=lambda paper: (-cls._lexical_score(paper, queries), cls._identity(paper)), + ) + ] + + @classmethod + def _workset( + cls, + seeds: list[PaperRecord], + connector_decisions: list[_CandidateDecision], + queries: list[str], + diagnostics: list[str], + ) -> tuple[list[PaperRecord], list[PaperRecord]]: + relevant = [ + decision + for decision in connector_decisions + if decision.relevance_score is None or decision.relevance_score >= 2 + ] + ordered = sorted( + relevant, + key=lambda decision: ( + 2 if decision.relevance_score is None else 3 - decision.relevance_score, + -cls._lexical_score(decision.paper, queries) + if decision.relevance_score is None + else 0, + cls._identity(decision.paper), + ), + ) + if len(seeds) > 20: + diagnostics.append( + f"{len(seeds) - 20} explicit seed papers deferred because the Reader " + "workset is capped at 20." + ) + capacity = max(0, 20 - len(seeds)) + current = [*seeds[:20], *(decision.paper for decision in ordered[:capacity])] + deferred = [*seeds[20:], *(decision.paper for decision in ordered[capacity:])] + if deferred: + diagnostics.append( + f"{len(deferred)} relevant or fail-open connector/seed candidates deferred by " + "the Reader workset limit." + ) + return current, deferred + def search( self, project_id: str, @@ -202,9 +388,11 @@ def search( *, seed_papers: list[PaperRecord] | None = None, per_connector_limit: int = 5, + relevance_context: str | None = None, ) -> SearchOutcome: outcome = SearchOutcome() - candidates = list(seed_papers or []) + seeds = list(seed_papers or []) + discovered: list[PaperRecord] = [] if not self.network_enabled: outcome.diagnostics.append( "Network search is disabled; only user-supplied seed papers were processed." @@ -214,7 +402,7 @@ def search( for connector in self.connectors: try: for raw in connector.search(query, per_connector_limit): - candidates.append( + discovered.append( PaperRecord(project_id=project_id, source=connector.name, **raw) ) except Exception as exc: @@ -223,14 +411,53 @@ def search( ) seen: set[str] = set() - for paper in candidates: - if paper.project_id != project_id: - paper = paper.model_copy(update={"project_id": project_id}) - key = self._dedupe_key(paper) - if key in seen: - continue - seen.add(key) + unique_seeds: list[PaperRecord] = [] + unique_discovered: list[PaperRecord] = [] + for candidates, destination in ( + (seeds, unique_seeds), + (discovered, unique_discovered), + ): + for paper in candidates: + if paper.project_id != project_id: + paper = paper.model_copy(update={"project_id": project_id}) + key = self._dedupe_key(paper) + if key in seen: + continue + seen.add(key) + destination.append(paper) + + all_candidates = [*unique_seeds, *unique_discovered] + for paper in all_candidates: + self._persist_bibliographic(paper) + + scores = self._semantic_screen_batches( + queries, + unique_discovered, + outcome.diagnostics, + relevance_context=relevance_context, + ) + if self.llm is None or not getattr(self.llm, "available", True): + connector_decisions = self._fallback_order(unique_discovered, queries) + else: + connector_decisions = [ + _CandidateDecision(paper, scores.get(paper.paper_id)) + for paper in unique_discovered + ] + current, deferred = self._workset( + unique_seeds, + connector_decisions, + queries, + outcome.diagnostics, + ) + for paper in current: persisted, _ = self._persist(paper) + oa_record = None + if self.open_access is not None: + oa_record = self.open_access.resolve(persisted) + if self.materializer is not None and oa_record is not None: + materialized = self.materializer.materialize(persisted, oa_record) + persisted = materialized.paper + outcome.diagnostics.extend(materialized.diagnostics) outcome.papers.append(persisted) if not outcome.papers: @@ -242,6 +469,8 @@ def search( { "queries": queries, "paper_ids": [paper.paper_id for paper in outcome.papers], + "reader_workset_paper_ids": [paper.paper_id for paper in outcome.papers], + "deferred_paper_ids": [paper.paper_id for paper in deferred], "diagnostics": outcome.diagnostics, }, project_id=project_id, diff --git a/tests/test_full_text_service.py b/tests/test_full_text_service.py new file mode 100644 index 0000000..82ba474 --- /dev/null +++ b/tests/test_full_text_service.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +from io import BytesIO +from pathlib import Path + +import httpx +from pypdf import PdfWriter + +from autoresearch.contracts import ( + KnowledgePartition, + OpenAccessRecord, + OpenAccessStatus, + PaperRecord, + WikiPage, +) +from autoresearch.evidence import EvidenceService +from autoresearch.full_text_service import FullTextMaterializer +from autoresearch.knowledge import KnowledgeService +from autoresearch.reader_service import PaperReaderService +from autoresearch.search_service import PaperSearchService +from autoresearch.storage import RecordStore + + +def _pdf_bytes() -> bytes: + output = BytesIO() + writer = PdfWriter() + writer.add_blank_page(width=72, height=72) + writer.write(output) + return output.getvalue() + + +def _paper() -> PaperRecord: + return PaperRecord( + paper_id="paper_1", + project_id="project_1", + title="Legal OA PDF", + doi="10.1234/legal-oa", + source="test", + ) + + +def _oa(url: str | None = "https://example.test/paper.pdf") -> OpenAccessRecord: + return OpenAccessRecord( + paper_id="paper_1", + project_id="project_1", + status=OpenAccessStatus.OPEN_ACCESS, + fulltext_url=url, + ) + + +def _materializer( + tmp_path: Path, handler, *, max_bytes: int = 25 * 1024 * 1024 +) -> tuple[FullTextMaterializer, RecordStore]: + store = RecordStore(tmp_path / "records.sqlite3") + projects_dir = tmp_path / "paper-projects" + (projects_dir / "project_1" / "02_literature").mkdir(parents=True) + client = httpx.Client(transport=httpx.MockTransport(handler)) + return FullTextMaterializer(store, projects_dir, max_bytes=max_bytes, client=client), store + + +class StaticConnector: + name = "offline" + + def search(self, query: str, limit: int) -> list[dict[str, str]]: + return [{"title": "Integrated PDF", "doi": "10.1234/integrated"}] + + +class StaticResolver: + def resolve(self, paper: PaperRecord) -> OpenAccessRecord: + return OpenAccessRecord( + paper_id=paper.paper_id, + project_id=paper.project_id, + status=OpenAccessStatus.OPEN_ACCESS, + fulltext_url="https://example.test/paper.pdf", + ) + + +def test_materializes_pdf_persists_path_and_reader_consumes_it(tmp_path: Path): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response( + 200, + headers={"content-type": "application/pdf"}, + content=_pdf_bytes(), + ) + + materializer, store = _materializer(tmp_path, handler) + result = materializer.materialize(_paper(), _oa()) + + assert result.diagnostics == [] + assert result.paper.full_text_path is not None + assert Path(result.paper.full_text_path).is_file() + assert "02_literature" in result.paper.full_text_path + assert store.get("paper", "paper_1")["full_text_path"] == result.paper.full_text_path + knowledge = KnowledgeService(store) + knowledge.add_page( + WikiPage( + page_id=result.paper.paper_id, + project_id=result.paper.project_id, + partition=KnowledgePartition.PAPERS, + title=result.paper.title, + body="test paper", + ) + ) + reader = PaperReaderService(store, EvidenceService(store), knowledge) + card, _ = reader.read(result.paper) + assert card.locators == ["p.1"] + + repeated = materializer.materialize(result.paper, _oa()) + assert repeated.paper.full_text_path == result.paper.full_text_path + assert calls == 1 + + +def test_download_failures_and_oversized_content_leave_path_unset(tmp_path: Path): + def failing_handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectTimeout("offline", request=request) + + failing, _ = _materializer(tmp_path / "failure", failing_handler) + failed = failing.materialize(_paper(), _oa()) + assert failed.paper.full_text_path is None + assert failed.diagnostics and "failed" in failed.diagnostics[0] + + def oversized_handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "application/pdf", "content-length": "100"}, + content=b"x", + ) + + oversized, _ = _materializer(tmp_path / "oversized", oversized_handler, max_bytes=10) + result = oversized.materialize(_paper(), _oa()) + assert result.paper.full_text_path is None + assert result.diagnostics == [ + "Full-text materialization failed for paper_1: " + "ValueError: response exceeds maximum download size; paper retained." + ] + + +def test_streaming_size_limit_applies_without_content_length(tmp_path: Path): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "application/pdf"}, + stream=httpx.ByteStream(b"x" * 11), + ) + + materializer, _ = _materializer(tmp_path, handler, max_bytes=10) + result = materializer.materialize(_paper(), _oa()) + + assert result.paper.full_text_path is None + assert result.diagnostics == [ + "Full-text materialization failed for paper_1: " + "ValueError: response exceeds maximum download size; paper retained." + ] + + +def test_rejects_non_document_and_skips_ineligible_records(tmp_path: Path): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response( + 200, + headers={"content-type": "text/html"}, + content=b"paywall", + ) + + materializer, _ = _materializer(tmp_path, handler) + unsupported = materializer.materialize(_paper(), _oa()) + assert unsupported.paper.full_text_path is None + assert unsupported.diagnostics == [ + "Full-text materialization failed for paper_1: " + "ValueError: unsupported content type: text/html; paper retained." + ] + + non_oa = _oa() + non_oa.status = OpenAccessStatus.NO_OPEN_FULLTEXT_FOUND + assert materializer.materialize(_paper(), non_oa).paper.full_text_path is None + assert materializer.materialize(_paper(), _oa(None)).paper.full_text_path is None + assert calls == 1 + + +def test_search_materializes_resolved_oa_pdf(tmp_path: Path): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "application/pdf"}, + content=_pdf_bytes(), + ) + + materializer, store = _materializer(tmp_path, handler) + service = PaperSearchService( + store, + EvidenceService(store), + KnowledgeService(store), + network_enabled=True, + connectors=[StaticConnector()], + open_access=StaticResolver(), + materializer=materializer, + ) + outcome = service.search("project_1", ["integrated"]) + + assert len(outcome.papers) == 1 + path = outcome.papers[0].full_text_path + assert path is not None and Path(path).is_file() + assert store.get("paper", outcome.papers[0].paper_id)["full_text_path"] == path diff --git a/tests/test_live_m05_smoke.py b/tests/test_live_m05_smoke.py new file mode 100644 index 0000000..e7181da --- /dev/null +++ b/tests/test_live_m05_smoke.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import importlib.util +from argparse import Namespace +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +@pytest.fixture +def smoke_module(): + path = Path(__file__).resolve().parents[1] / "scripts" / "live_m05_smoke.py" + spec = importlib.util.spec_from_file_location("live_m05_smoke", path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_parser_requires_exactly_one_live_mode(smoke_module): + with pytest.raises(SystemExit): + smoke_module.build_parser().parse_args([]) + with pytest.raises(SystemExit): + smoke_module.build_parser().parse_args(["--oa-only", "--e2e"]) + + +def test_configuration_status_does_not_expose_secret(smoke_module): + secret = "person@example.test;super-secret-api-key" + assert smoke_module.configured_label(secret) == "yes" + assert smoke_module.configured_label("") == "no" + + +def test_oa_only_missing_email_short_circuits_before_application(monkeypatch, smoke_module, capsys): + class Settings: + unpaywall_email = None + + monkeypatch.setattr(smoke_module, "get_settings", lambda: Settings()) + monkeypatch.setattr( + smoke_module, + "AutoResearchApplication", + lambda *_args, **_kwargs: pytest.fail("application must not be created"), + ) + + result = smoke_module._oa_only(Namespace(doi="10.0000/example", project_id="live")) + + assert result == 2 + output = capsys.readouterr().out + assert "UNPAYWALL_EMAIL configured: no" in output + assert "10.0000/example" not in output + + +def test_e2e_missing_llm_short_circuits_before_application(monkeypatch, smoke_module, capsys): + class Settings: + llm_configured = False + + monkeypatch.setattr(smoke_module, "get_settings", lambda: Settings()) + monkeypatch.setattr( + smoke_module, + "AutoResearchApplication", + lambda *_args, **_kwargs: pytest.fail("application must not be created"), + ) + + result = smoke_module._e2e( + Namespace(idea="private research idea", project_id="live", per_connector_limit=3) + ) + + assert result == 2 + output = capsys.readouterr().out + assert "LLM configured: no" in output + assert "private research idea" not in output + + +def test_generated_live_project_ids_are_isolated(monkeypatch, smoke_module): + values = iter(["first", "second"]) + monkeypatch.setattr(smoke_module.secrets, "token_hex", lambda _size: next(values)) + + first = smoke_module._generated_project_id() + second = smoke_module._generated_project_id() + + assert first.startswith("live-m05-smoke-") + assert second.startswith("live-m05-smoke-") + assert first != second + + +def test_existing_pdf_refuses_explicit_project_id(tmp_path, monkeypatch, smoke_module, capsys): + full_text = tmp_path / "existing" / "02_literature" / "full_text" + full_text.mkdir(parents=True) + (full_text / "old.pdf").write_bytes(b"old") + + class Settings: + unpaywall_email = "configured" + projects_dir = tmp_path + + monkeypatch.setattr(smoke_module, "get_settings", lambda: Settings()) + monkeypatch.setattr( + smoke_module, + "AutoResearchApplication", + lambda *_args, **_kwargs: pytest.fail("stale state must short-circuit before application"), + ) + + result = smoke_module._oa_only(Namespace(doi="10.0000/example", project_id="existing")) + + assert result == 2 + assert "Refusing to reuse a project ID with existing PDFs" in capsys.readouterr().out + + +def _mock_e2e_application(tmp_path: Path, diagnostics: list[str]): + pdf_path = tmp_path / "fresh.pdf" + pdf_path.write_bytes(b"fresh") + paper = SimpleNamespace( + paper_id="paper-live", + title="Live paper", + full_text_path=str(pdf_path), + ) + + class Store: + def list(self, *_args, **_kwargs): + return [{"paper_id": paper.paper_id}] + + def events(self, _project_id): + return [ + { + "event_type": "papers.search_completed", + "payload": {"paper_ids": [paper.paper_id], "deferred_paper_ids": []}, + } + ] + + def get(self, _kind, _record_id): + return {"status": "open_access"} + + class Application: + settings = SimpleNamespace(projects_dir=tmp_path) + projects = SimpleNamespace(get=lambda _project_id: {"project_id": "run"}) + store = Store() + orchestrator_agent = SimpleNamespace( + scope=lambda _state: {"search_queries": ["one", "two", "three"], "diagnostics": []} + ) + search = SimpleNamespace( + search=lambda *_args, **_kwargs: SimpleNamespace( + papers=[paper], diagnostics=diagnostics + ) + ) + reader = SimpleNamespace(read=lambda _paper: (SimpleNamespace(locators=["p.1"]), None)) + + def close(self): + pass + + return Application() + + +def _configured_settings(tmp_path: Path, llm_api_key: str = ""): + class Settings: + llm_configured = True + projects_dir = tmp_path + + def __init__(self): + self.llm_api_key = llm_api_key + + def model_copy(self, **_kwargs): + return self + + return Settings() + + +def test_e2e_batch_failure_returns_nonzero_despite_readable_pdf( + tmp_path, monkeypatch, smoke_module, capsys +): + settings = _configured_settings(tmp_path) + application = _mock_e2e_application( + tmp_path, + [ + "LLM semantic screening failed for batch 1 (1 papers): " + "RuntimeError; batch retained unscored." + ], + ) + monkeypatch.setattr(smoke_module, "get_settings", lambda: settings) + monkeypatch.setattr(smoke_module, "AutoResearchApplication", lambda _settings: application) + + result = smoke_module._e2e(Namespace(idea="idea", project_id="run", per_connector_limit=3)) + + assert result == 1 + assert "LLM semantic screening batch failures: 1" in capsys.readouterr().out + + +def test_e2e_clean_mocked_flow_succeeds_without_exposing_secret( + tmp_path, monkeypatch, smoke_module, capsys +): + secret = "super-secret-relay-key" + settings = _configured_settings(tmp_path, secret) + application = _mock_e2e_application(tmp_path, []) + monkeypatch.setattr(smoke_module, "get_settings", lambda: settings) + monkeypatch.setattr(smoke_module, "AutoResearchApplication", lambda _settings: application) + + result = smoke_module._e2e(Namespace(idea="idea", project_id="run", per_connector_limit=3)) + + assert result == 0 + output = capsys.readouterr().out + assert "Full search-to-Reader chain completed: yes" in output + assert secret not in output diff --git a/tests/test_open_access_service.py b/tests/test_open_access_service.py new file mode 100644 index 0000000..fcaa1b7 --- /dev/null +++ b/tests/test_open_access_service.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from autoresearch.contracts import ( + OpenAccessStatus, + PaperRecord, +) +from autoresearch.evidence import EvidenceService +from autoresearch.knowledge import KnowledgeService +from autoresearch.open_access_service import ( + OPEN_ACCESS_RECORD_KIND, + OpenAccessLookupResult, + OpenAccessResolver, +) +from autoresearch.search_service import PaperSearchService +from autoresearch.storage import RecordStore + + +@dataclass +class FakeProvider: + name: str + result: OpenAccessLookupResult | None = None + error: Exception | None = None + calls: int = 0 + received_paper_ids: list[str] = field(default_factory=list) + + def lookup(self, paper: PaperRecord) -> OpenAccessLookupResult: + self.calls += 1 + self.received_paper_ids.append(paper.paper_id) + if self.error is not None: + raise self.error + return self.result or OpenAccessLookupResult() + + +@dataclass +class FakeSearchConnector: + name: str + records: list[dict[str, Any]] + + def search(self, query: str, limit: int) -> list[dict[str, Any]]: + return self.records[:limit] + + +@dataclass +class RecordingResolver: + store: RecordStore + paper_ids: list[str] = field(default_factory=list) + persisted_at_call: list[bool] = field(default_factory=list) + + def resolve(self, paper: PaperRecord) -> None: + self.paper_ids.append(paper.paper_id) + self.persisted_at_call.append( + self.store.get("paper", paper.paper_id) is not None + ) + + +def _paper() -> PaperRecord: + return PaperRecord( + paper_id="paper_1", + project_id="project_1", + title="Offline OA test paper", + doi="10.1234/offline-oa-test", + source="test", + ) + + +def _resolver(tmp_path: Path, providers: list[FakeProvider]) -> OpenAccessResolver: + return OpenAccessResolver(RecordStore(tmp_path / "records.sqlite3"), providers) + + +def test_first_provider_finds_oa(tmp_path: Path): + first = FakeProvider( + "unpaywall", + OpenAccessLookupResult( + fulltext_url="https://repository.example/paper.pdf", + license="cc-by", + diagnostics=["matched repository full text"], + ), + ) + second = FakeProvider("openalex", OpenAccessLookupResult()) + resolver = _resolver(tmp_path, [first, second]) + + record = resolver.resolve(_paper()) + + assert record.status is OpenAccessStatus.OPEN_ACCESS + assert record.fulltext_url == "https://repository.example/paper.pdf" + assert record.license == "cc-by" + assert record.source == "unpaywall" + assert record.checked_sources == ["unpaywall"] + assert record.diagnostics == ["matched repository full text"] + assert first.calls == 1 + assert second.calls == 0 + + +def test_first_provider_misses_and_second_provider_finds_oa(tmp_path: Path): + first = FakeProvider("unpaywall", OpenAccessLookupResult()) + second = FakeProvider( + "openalex", + OpenAccessLookupResult(fulltext_url="https://openalex.example/paper.pdf"), + ) + resolver = _resolver(tmp_path, [first, second]) + + record = resolver.resolve(_paper()) + + assert record.status is OpenAccessStatus.OPEN_ACCESS + assert record.fulltext_url == "https://openalex.example/paper.pdf" + assert record.source == "openalex" + assert record.checked_sources == ["unpaywall", "openalex"] + + +def test_all_providers_successfully_miss(tmp_path: Path): + first = FakeProvider("unpaywall", OpenAccessLookupResult()) + second = FakeProvider("openalex", OpenAccessLookupResult()) + resolver = _resolver(tmp_path, [first, second]) + + record = resolver.resolve(_paper()) + + assert record.status is OpenAccessStatus.NO_OPEN_FULLTEXT_FOUND + assert record.fulltext_url is None + assert record.checked_sources == ["unpaywall", "openalex"] + assert record.diagnostics == [] + + +def test_provider_failure_returns_unknown(tmp_path: Path): + first = FakeProvider("unpaywall", error=TimeoutError()) + second = FakeProvider("openalex", error=ConnectionError()) + resolver = _resolver(tmp_path, [first, second]) + + record = resolver.resolve(_paper()) + + assert record.status is OpenAccessStatus.UNKNOWN + assert record.checked_sources == ["unpaywall", "openalex"] + assert record.diagnostics == [ + "unpaywall failed: TimeoutError", + "openalex failed: ConnectionError", + ] + + +def test_repeated_resolution_updates_one_current_record(tmp_path: Path): + provider = FakeProvider( + "unpaywall", + OpenAccessLookupResult(fulltext_url="https://repository.example/first.pdf"), + ) + resolver = _resolver(tmp_path, [provider]) + paper = _paper() + + first = resolver.resolve(paper) + provider.result = OpenAccessLookupResult( + fulltext_url="https://repository.example/second.pdf", + license="cc-by-nc", + ) + second = resolver.resolve(paper) + + stored = resolver.store.get(OPEN_ACCESS_RECORD_KIND, paper.paper_id) + current_records = resolver.store.list( + OPEN_ACCESS_RECORD_KIND, + project_id=paper.project_id, + ) + assert first.fulltext_url == "https://repository.example/first.pdf" + assert second.fulltext_url == "https://repository.example/second.pdf" + assert stored is not None + assert stored["fulltext_url"] == "https://repository.example/second.pdf" + assert len(current_records) == 1 + + +def test_paper_without_doi_returns_unknown_without_calling_provider(tmp_path: Path): + provider = FakeProvider( + "unpaywall", + OpenAccessLookupResult(fulltext_url="https://repository.example/paper.pdf"), + ) + resolver = _resolver(tmp_path, [provider]) + paper = _paper().model_copy(update={"doi": None}) + + record = resolver.resolve(paper) + + assert record.status is OpenAccessStatus.UNKNOWN + assert record.checked_sources == [] + assert any("OA lookup skipped" in diagnostic and "no DOI" in diagnostic + for diagnostic in record.diagnostics) + assert provider.calls == 0 + + +def test_search_resolves_deduplicated_persisted_paper_only_when_configured( + tmp_path: Path, +): + records = [ + { + "title": "Deduplicated OA paper", + "abstract": "Offline abstract", + "doi": "10.1234/integration-oa", + "url": "https://example.invalid/paper", + "source_record_id": "first", + }, + { + "title": "Duplicate OA paper", + "abstract": "Duplicate abstract", + "doi": "10.1234/integration-oa", + "url": "https://example.invalid/duplicate", + "source_record_id": "second", + }, + ] + store = RecordStore(tmp_path / "configured.sqlite3") + resolver = RecordingResolver(store) + service = PaperSearchService( + store, + EvidenceService(store), + KnowledgeService(store), + network_enabled=True, + connectors=[FakeSearchConnector("offline", records)], + open_access=resolver, + ) + + outcome = service.search("project_1", ["offline"]) + + assert len(outcome.papers) == 1 + paper = outcome.papers[0] + assert resolver.paper_ids == [paper.paper_id] + assert resolver.persisted_at_call == [True] + assert store.get("paper", paper.paper_id) is not None + + disabled_store = RecordStore(tmp_path / "disabled.sqlite3") + disabled_resolver = RecordingResolver(disabled_store) + disabled_service = PaperSearchService( + disabled_store, + EvidenceService(disabled_store), + KnowledgeService(disabled_store), + network_enabled=True, + connectors=[FakeSearchConnector("offline", records)], + open_access=None, + ) + + disabled_outcome = disabled_service.search("project_1", ["offline"]) + + assert len(disabled_outcome.papers) == 1 + assert disabled_resolver.paper_ids == [] + assert disabled_store.list(OPEN_ACCESS_RECORD_KIND) == [] diff --git a/tests/test_query_rewrite.py b/tests/test_query_rewrite.py new file mode 100644 index 0000000..747973d --- /dev/null +++ b/tests/test_query_rewrite.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from autoresearch.agents.orchestrator import OrchestratorAgent +from autoresearch.contracts import PaperRecord, ResearchState +from autoresearch.evidence import EvidenceService +from autoresearch.handoffs import HandoffService +from autoresearch.knowledge import KnowledgeService +from autoresearch.search_service import PaperSearchService +from autoresearch.state_machine import StateMachine +from autoresearch.storage import RecordStore + + +@dataclass +class FakeLLM: + response: dict[str, Any] | Exception | None + available: bool = True + calls: list[str] = field(default_factory=list) + + def complete_json(self, *, system: str, user: str) -> dict[str, Any] | None: + self.calls.append(user) + if isinstance(self.response, Exception): + raise self.response + return self.response + + +def _agent(tmp_path: Path, llm: FakeLLM) -> OrchestratorAgent: + store = RecordStore(tmp_path / "records.sqlite3") + evidence = EvidenceService(store) + return OrchestratorAgent(store, evidence, HandoffService(store), StateMachine(), llm) + + +def test_valid_llm_query_rewrite_returns_exactly_three_and_preserves_idea(tmp_path: Path): + idea = "Evaluate reliable clinical decision support for emergency triage" + llm = FakeLLM( + { + "search_queries": [ + "clinical decision support emergency triage", + "healthcare decision support diagnostic assistance", + "emergency triage reliability clinical AI", + ] + } + ) + result = _agent(tmp_path, llm).scope(ResearchState(project_id="demo", idea=idea).model_dump()) + + assert result["idea"] == idea + assert result["search_queries"] == llm.response["search_queries"] + assert len(result["search_queries"]) == 3 + assert len({query.lower() for query in result["search_queries"]}) == 3 + + +def test_query_rewrite_failures_invalid_and_duplicates_use_fallback(tmp_path: Path): + idea = "Assess climate risk communication for coastal communities" + invalid_llm = FakeLLM({"search_queries": ["only one"]}) + duplicate_llm = FakeLLM( + { + "search_queries": [ + "climate risk coastal communities", + "coastal communities climate risk", + "climate communication coastal communities", + ] + } + ) + failing_llm = FakeLLM(RuntimeError("offline")) + + fallback = _agent(tmp_path / "invalid", invalid_llm)._fallback_queries(idea) + cases = (("invalid", invalid_llm), ("duplicate", duplicate_llm), ("failure", failing_llm)) + for directory, llm in cases: + result = _agent(tmp_path / directory, llm).scope( + ResearchState(project_id="demo", idea=idea).model_dump() + ) + assert result["search_queries"] == fallback + assert len(result["search_queries"]) == 3 + assert len({query.lower() for query in result["search_queries"]}) == 3 + assert all("systematic review" not in query for query in result["search_queries"]) + assert all("benchmark evaluation" not in query for query in result["search_queries"]) + + +def test_rewritten_queries_still_drive_lexical_ranking(tmp_path: Path): + queries = [ + "clinical decision support emergency triage", + "healthcare diagnostic assistance reliability", + "emergency triage clinical AI", + ] + store = RecordStore(tmp_path / "records.sqlite3") + service = PaperSearchService( + store, + EvidenceService(store), + KnowledgeService(store), + network_enabled=False, + ) + paper = PaperRecord( + project_id="demo", + title="Reliable clinical decision support for emergency triage", + source="test", + ) + + assert service._lexical_score(paper, queries) > 0 diff --git a/tests/test_relevance_selection.py b/tests/test_relevance_selection.py new file mode 100644 index 0000000..b2ca66c --- /dev/null +++ b/tests/test_relevance_selection.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from autoresearch.contracts import OpenAccessRecord, OpenAccessStatus, PaperRecord +from autoresearch.evidence import EvidenceService +from autoresearch.full_text_service import MaterializationResult +from autoresearch.knowledge import KnowledgeService +from autoresearch.search_service import PaperSearchService +from autoresearch.storage import RecordStore + + +@dataclass +class FakeConnector: + records: list[dict[str, Any]] + name: str = "offline" + + def search(self, query: str, limit: int) -> list[dict[str, Any]]: + return self.records[:limit] + + +@dataclass +class BatchLLM: + responses: list[dict[str, Any] | Exception] + available: bool = True + calls: list[str] = field(default_factory=list) + + def complete_json(self, *, system: str, user: str) -> dict[str, Any]: + self.calls.append(user) + response = self.responses[len(self.calls) - 1] + if isinstance(response, Exception): + raise response + return response + + +@dataclass +class RecordingResolver: + paper_ids: list[str] = field(default_factory=list) + + def resolve(self, paper: PaperRecord) -> OpenAccessRecord: + self.paper_ids.append(paper.paper_id) + return OpenAccessRecord( + paper_id=paper.paper_id, + project_id=paper.project_id, + status=OpenAccessStatus.OPEN_ACCESS, + fulltext_url="https://example.test/paper.pdf", + ) + + +@dataclass +class RecordingMaterializer: + paper_ids: list[str] = field(default_factory=list) + + def materialize( + self, paper: PaperRecord, oa_record: OpenAccessRecord + ) -> MaterializationResult: + self.paper_ids.append(paper.paper_id) + return MaterializationResult(paper) + + +def _service( + tmp_path: Path, + records: list[dict[str, Any]], + llm=None, + open_access=None, + materializer=None, +): + store = RecordStore(tmp_path / "records.sqlite3") + prepared = [] + for index, record in enumerate(records): + item = dict(record) + item.setdefault("paper_id", item.get("doi") or f"paper-{index}") + prepared.append(item) + return PaperSearchService( + store, + EvidenceService(store), + KnowledgeService(store), + network_enabled=True, + connectors=[FakeConnector(prepared)], + llm=llm, + open_access=open_access, + materializer=materializer, + ) + + +def _paper(title: str, abstract: str = "", doi: str | None = None) -> PaperRecord: + return PaperRecord(project_id="p1", title=title, abstract=abstract, doi=doi, source="test") + + +def _batch_scores(titles: list[str], score: int = 3) -> dict[str, Any]: + return {"scores": [{"paper_id": title, "relevance_score": score} for title in titles]} + + +def test_all_deduplicated_connector_candidates_reach_batch_screening(tmp_path: Path): + records = [ + {"title": f"Paper {index}", "doi": f"10/paper-{index}"} + for index in range(10) + ] + llm = BatchLLM( + [ + _batch_scores([record["doi"] for record in records[:8]], 3), + _batch_scores([record["doi"] for record in records[8:]], 3), + ] + ) + outcome = _service(tmp_path, records, llm).search("p1", ["topic"], per_connector_limit=20) + + assert len(llm.calls) == 2 + assert all('"paper_id"' in call for call in llm.calls) + assert len(outcome.papers) == 10 + + +def test_scores_three_and_two_are_retained_but_zero_and_one_rejected(tmp_path: Path): + records = [{"title": f"Score {score}", "doi": f"10/score-{score}"} for score in range(4)] + llm = BatchLLM([_batch_scores([record["doi"] for record in records], 2)]) + llm.responses[0] = { + "scores": [ + {"paper_id": "10/score-0", "relevance_score": 0}, + {"paper_id": "10/score-1", "relevance_score": 1}, + {"paper_id": "10/score-2", "relevance_score": 2}, + {"paper_id": "10/score-3", "relevance_score": 3}, + ] + } + outcome = _service(tmp_path, records, llm).search("p1", ["topic"], per_connector_limit=20) + + assert {paper.title for paper in outcome.papers} == {"Score 2", "Score 3"} + + +def test_original_relevance_context_is_used_not_rewritten_queries(tmp_path: Path): + records = [{"title": "Candidate", "abstract": "topic", "doi": "10/candidate"}] + llm = BatchLLM([_batch_scores(["10/candidate"], 3)]) + outcome = _service(tmp_path, records, llm).search( + "p1", + ["rewritten retrieval terms"], + relevance_context="Original user research question", + ) + + assert len(outcome.papers) == 1 + assert "Original user research question" in llm.calls[0] + assert "rewritten retrieval terms" not in llm.calls[0] + + +def test_dedupe_happens_before_batch_screening(tmp_path: Path): + records = [ + {"title": "First", "doi": "10/duplicate"}, + {"title": "Second copy", "doi": "10/duplicate"}, + {"title": "Unique", "doi": "10/unique"}, + ] + llm = BatchLLM([_batch_scores(["10/duplicate", "10/unique"], 3)]) + outcome = _service(tmp_path, records, llm).search("p1", ["topic"], per_connector_limit=20) + + assert len(llm.calls) == 1 + assert '"paper_id": "10/duplicate"' in llm.calls[0] + assert "Second copy" not in {paper.title for paper in outcome.papers} + + +def test_batch_validation_failures_retain_that_batch_and_successful_batches_apply(tmp_path: Path): + records = [{"title": f"Paper {index}", "doi": f"10/paper-{index}"} for index in range(10)] + invalid = {"scores": [{"paper_id": "10/paper-0", "relevance_score": 3}]} + successful = _batch_scores([record["doi"] for record in records[8:]], 3) + llm = BatchLLM([invalid, successful]) + outcome = _service(tmp_path, records, llm).search("p1", ["topic"], per_connector_limit=20) + + titles = {paper.title for paper in outcome.papers} + assert {"Paper 8", "Paper 9"} <= titles + assert len(titles) == 10 # failed batch is fail-open, not dropped + assert [paper.title for paper in outcome.papers[:2]] == ["Paper 8", "Paper 9"] + assert any("batch 1" in diagnostic for diagnostic in outcome.diagnostics) + + +def test_batch_validation_rejects_duplicate_unknown_and_invalid_scores(): + requested = ["a", "b"] + cases = [ + { + "scores": [ + {"paper_id": "a", "relevance_score": 3}, + {"paper_id": "a", "relevance_score": 2}, + ] + }, + { + "scores": [ + {"paper_id": "a", "relevance_score": 3}, + {"paper_id": "c", "relevance_score": 2}, + ] + }, + { + "scores": [ + {"paper_id": "a", "relevance_score": "3"}, + {"paper_id": "b", "relevance_score": 2}, + ] + }, + ] + for case in cases: + try: + PaperSearchService._validate_batch_scores(case, requested) + except ValueError: + pass + else: + raise AssertionError("invalid batch was accepted") + + +def test_unavailable_llm_uses_deterministic_lexical_order_without_rejection(tmp_path: Path): + records = [ + {"title": "Weak", "abstract": "", "doi": "10/weak"}, + {"title": "Topic strong", "abstract": "", "doi": "10/strong"}, + ] + llm = BatchLLM([], available=False) + first_service = _service(tmp_path / "one", records, llm) + first = first_service.search("p1", ["topic"], per_connector_limit=20) + second = _service(tmp_path / "two", records, llm).search( + "p1", ["topic"], per_connector_limit=20 + ) + + assert [paper.title for paper in first.papers] == ["Topic strong", "Weak"] + assert [paper.title for paper in first.papers] == [paper.title for paper in second.papers] + assert any("unavailable" in diagnostic for diagnostic in first.diagnostics) + + +def test_unavailable_llm_fallback_caps_and_records_deferred_ids(tmp_path: Path): + records = [ + { + "title": f"Paper {index}", + "abstract": "topic" if index % 2 == 0 else "", + "doi": f"10/paper-{index}", + } + for index in range(25) + ] + llm = BatchLLM([], available=False) + service = _service(tmp_path, records, llm) + outcome = service.search("p1", ["topic"], per_connector_limit=30) + + event = next( + event["payload"] + for event in service.store.events("p1") + if event["event_type"] == "papers.search_completed" + ) + assert len(outcome.papers) == 20 + assert len(service.store.list("paper", project_id="p1")) == 25 + assert event["reader_workset_paper_ids"] == [paper.paper_id for paper in outcome.papers] + assert len(event["deferred_paper_ids"]) == 5 + assert set(event["deferred_paper_ids"]).isdisjoint(event["reader_workset_paper_ids"]) + assert llm.calls == [] + + +def test_reader_workset_caps_at_twenty_and_defers_relevant_records(tmp_path: Path): + records = [{"title": f"Paper {index}", "doi": f"10/paper-{index}"} for index in range(25)] + llm = BatchLLM( + [ + _batch_scores([record["doi"] for record in records[i : i + 8]], 3) + for i in range(0, 25, 8) + ] + ) + service = _service(tmp_path, records, llm) + outcome = service.search("p1", ["topic"], per_connector_limit=30) + + assert len(outcome.papers) == 20 + assert outcome.papers[0].title == "Paper 0" + assert len(service.store.list("paper", project_id="p1")) == 25 + assert any("deferred" in diagnostic for diagnostic in outcome.diagnostics) + event = next( + event["payload"] + for event in service.store.events("p1") + if event["event_type"] == "papers.search_completed" + ) + assert event["reader_workset_paper_ids"] == [paper.paper_id for paper in outcome.papers] + assert set(event["deferred_paper_ids"]) == { + paper["paper_id"] for paper in service.store.list("paper", project_id="p1") + if paper["paper_id"] not in event["reader_workset_paper_ids"] + } + + +def test_deferred_candidates_do_not_trigger_oa_or_materialization(tmp_path: Path): + records = [{"title": f"Paper {index}", "doi": f"10/paper-{index}"} for index in range(25)] + llm = BatchLLM( + [ + _batch_scores([record["doi"] for record in records[i : i + 8]], 3) + for i in range(0, 25, 8) + ] + ) + resolver = RecordingResolver() + materializer = RecordingMaterializer() + service = _service( + tmp_path, + records, + llm, + open_access=resolver, + materializer=materializer, + ) + outcome = service.search("p1", ["topic"], per_connector_limit=30) + + assert len(outcome.papers) == 20 + assert len(resolver.paper_ids) == 20 + assert resolver.paper_ids == materializer.paper_ids + event = next( + event["payload"] + for event in service.store.events("p1") + if event["event_type"] == "papers.search_completed" + ) + persisted_ids = {paper["paper_id"] for paper in service.store.list("paper", project_id="p1")} + assert len(event["deferred_paper_ids"]) == 5 + assert set(event["deferred_paper_ids"]) == ( + persisted_ids - set(event["reader_workset_paper_ids"]) + ) + assert set(event["deferred_paper_ids"]).isdisjoint(event["reader_workset_paper_ids"]) + assert event["reader_workset_paper_ids"] == resolver.paper_ids + + +def test_score_three_precedes_score_two_and_seeds_have_priority(tmp_path: Path): + records = [{"title": "Score two", "doi": "10/two"}, {"title": "Score three", "doi": "10/three"}] + seed = _paper("Explicit seed", doi="10/seed") + llm = BatchLLM([_batch_scores(["10/two", "10/three"], 2)]) + llm.responses[0] = { + "scores": [ + {"paper_id": "10/two", "relevance_score": 2}, + {"paper_id": "10/three", "relevance_score": 3}, + ] + } + outcome = _service(tmp_path, records, llm).search( + "p1", ["topic"], seed_papers=[seed], per_connector_limit=20 + ) + + assert [paper.title for paper in outcome.papers] == [ + "Explicit seed", + "Score three", + "Score two", + ] + + +def test_more_than_twenty_seeds_are_preserved_but_workset_is_bounded(tmp_path: Path): + seeds = [_paper(f"Seed {index}", doi=f"10/seed-{index}") for index in range(22)] + service = _service(tmp_path, [], None) + outcome = service.search("p1", [], seed_papers=seeds) + + assert len(outcome.papers) == 20 + assert len(service.store.list("paper", project_id="p1")) == 22 + assert any("explicit seed papers deferred" in diagnostic for diagnostic in outcome.diagnostics)