learningcircuit--local-deep-research/tests/performance/_shared/run_full_search.py

"""End-to-end LDR pipeline test — runs a real Quick Summary via programmatic API.

Produces the same style of output as the "Quick Summary" mode in the web UI,
but from a single command so we can diff reports against one another (e.g.
same query with arxiv vs openalex). Useful for evaluating the relevance-
filter prompt and synthesis quality in situ.

Usage:
    LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true LDR_TESTING_WITH_MOCKS=false \\
    pdm run python tests/performance/_shared/run_full_search.py \\
        --query "LLM interpretability latest research" \\
        --engine openalex \\
        --output /tmp/ldr_report_openalex.md

Default model / URL mirror the other dev scripts so a bare invocation with
just --query also works.
"""

from __future__ import annotations

import argparse
import os
import sys
import tempfile
import time
from pathlib import Path

import requests

PROJECT_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(PROJECT_ROOT / "src"))

from local_deep_research.api import quick_summary  # noqa: E402
from local_deep_research.api.settings_utils import (  # noqa: E402
    create_settings_snapshot,
)
from local_deep_research.security import safe_get  # noqa: E402
from local_deep_research.security.file_write_verifier import (  # noqa: E402
    write_file_verified,
)

DEFAULT_OLLAMA_URL = os.environ.get(
    "LDR_TEST_OLLAMA_BASE_URL", "http://localhost:11434"
)
DEFAULT_OLLAMA_MODEL = os.environ.get("LDR_TEST_OLLAMA_MODEL", "qwen3.5:9b")

VALID_ENGINES = [
    "arxiv",
    "openalex",
    "wikipedia",
    "searxng",
]


def check_ollama(url: str) -> bool:
    try:
        resp = safe_get(
            f"{url.rstrip('/')}/api/tags",
            timeout=5,
            allow_private_ips=True,
        )
        return resp.status_code == 200
    except requests.RequestException:
        return False


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Run a Quick Summary end-to-end against a chosen search engine.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--query", required=True, help="Research query.")
    p.add_argument(
        "--engine",
        default="arxiv",
        choices=VALID_ENGINES,
        help="Search engine (search.tool).",
    )
    p.add_argument("--model", default=DEFAULT_OLLAMA_MODEL)
    p.add_argument("--ollama-url", default=DEFAULT_OLLAMA_URL)
    p.add_argument(
        "--iterations",
        type=int,
        default=1,
        help="search.iterations — 1 matches the REST API default for Quick Summary.",
    )
    p.add_argument(
        "--output",
        default=None,
        help="Path for the generated report. Defaults to /tmp/ldr_report_<engine>.md",
    )
    p.add_argument(
        "--verbose",
        action="store_true",
        help="Enable DEBUG logging (LDR_APP_DEBUG=true) — shows filter KEPT/REMOVED decisions.",
    )
    args = p.parse_args()
    if not args.output:
        args.output = str(
            Path(tempfile.gettempdir()) / f"ldr_report_{args.engine}.md"
        )
    return args


def main() -> int:
    args = parse_args()

    if args.verbose:
        os.environ["LDR_APP_DEBUG"] = "true"

    if not check_ollama(args.ollama_url):
        print(
            f"Ollama endpoint {args.ollama_url} not reachable. Start it or pass --ollama-url.",
            file=sys.stderr,
        )
        return 1

    settings = create_settings_snapshot(
        overrides={
            "llm.provider": "ollama",
            "llm.model": args.model,
            "llm.ollama.url": args.ollama_url,
            "search.tool": args.engine,
            "search.iterations": args.iterations,
            "api.allow_file_output": True,
        }
    )

    print(
        f"Running Quick Summary — engine={args.engine} model={args.model} "
        f"iterations={args.iterations}",
        file=sys.stderr,
    )
    print(f"  Query:  {args.query!r}", file=sys.stderr)
    print(f"  Ollama: {args.ollama_url}", file=sys.stderr)

    t0 = time.monotonic()
    result = quick_summary(
        args.query,
        settings_snapshot=settings,
        programmatic_mode=True,
    )
    elapsed = time.monotonic() - t0

    body = result.get("formatted_findings") or result.get("summary") or ""
    sources = result.get("sources", []) or []

    header = (
        f"# Research Results: {args.query}\n\n"
        f"- **Engine:** {args.engine}\n"
        f"- **Model:** {args.model}\n"
        f"- **Iterations:** {args.iterations}\n"
        f"- **Sources:** {len(sources)}\n"
        f"- **Elapsed:** {elapsed:.1f}s\n"
        f"- **Generated:** {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
        "---\n\n"
    )

    write_file_verified(
        args.output,
        header + body,
        "api.allow_file_output",
        context="dev Quick Summary eval",
        settings_snapshot=settings,
    )

    print(
        f"\nDone in {elapsed:.1f}s. {len(sources)} sources. Wrote: {args.output}",
        file=sys.stderr,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())