eval/knowledge/run_eval.py

"""Knowledge-pipeline eval runner.

Scores an already-parsed document against `knowledge_gold.yaml` and writes a
timestamped result file. The 2026-08-24 figures were assembled by hand; this is
the reproducible version, so a number can be re-derived instead of trusted.

**Takes a ParsedDocument artifact, never a PDF.** MinerU is not invoked and is
not required — that is the seam doing its job. Point it at the `chunks.json`
written by `python -m src.knowledge_parsing.run`.

    # free: E1 recall, E2 compression, the funnel. No API calls, no spend.
    ./.venv/Scripts/python.exe -m eval.knowledge.run_eval <artifact.json>

    # add the paid branches: E3 schema fill and rule recall
    ./.venv/Scripts/python.exe -m eval.knowledge.run_eval <artifact.json> --extract

    # exercise every branch with no spend and no credentials (NOT scoreable)
    ./.venv/Scripts/python.exe -m eval.knowledge.run_eval <artifact.json> --extract --mock

Module mode only (`-m`). A file-path invocation breaks `sys.path` and the
`src.` imports below fail.

Two things this refuses to do, both for the same reason — a number that cannot
be compared is worse than no number:

* **It will not run on an artifact that does not declare `parser_backend`.**
  The same MinerU build emits different text from `pipeline` and `hybrid/high`,
  so results measured on different backends are not comparable and nothing
  downstream can tell them apart afterwards. `--allow-unlabelled-artifact`
  overrides, and stamps the result file so the caveat travels with it.
* **It never overwrites a result file.** Every run writes a new timestamped
  file beside the others; the frozen baseline stays frozen.
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any

from src.knowledge_extraction.adapter import parsed_doc_from_artifact
from src.knowledge_extraction.extract import MockExtractor, cacheable, prefix_tokens
from src.knowledge_extraction.service import (
    build_clusters,
    estimate_cost,
    extract_all,
    run_filters,
)
from src.knowledge_extraction.settings import EVIDENCE_K

from .score import GOLD_PATH, load_gold, score_glossary, score_rules, score_term_filter

_HERE = Path(__file__).resolve().parent
RESULTS_DIR = _HERE / "results"

# The kill lines. Each is a decision already taken, recorded in
# KNOWLEDGE_PIPELINE_CALIBRATION.md — not a target invented per run.
KILL_LINES = {"E1": 0.70, "E2": 2.0, "E3": 0.80}

# What the last committed run measured, for the delta column. Rule recall has
# no kill line: it is tracked, not gated, and 3/15 is the post-X22 scoring of
# the 2026-08-24 run — NOT the 1/15 in that file, which predates the scorer fix.
BASELINE = {
    "file": "v2_full_document_2026-08-24_093051.json",
    "E1": 0.8049,
    "E2": 2.81,
    "E3": 0.90,
    "rule_recall": 0.2,
}


def _azure_extractor():
    """Built lazily: importing it requires credentials to be configured."""
    from src.knowledge_extraction.extract import AzureExtractor

    try:
        return AzureExtractor()
    except Exception as exc:  # noqa: BLE001 - reported, not raised
        print(f"cannot build the Azure extractor: {exc!r}", file=sys.stderr)
        return None


def _latex_verification_counts(formulas: list[dict]) -> dict[str, int]:
    """How many stored formulas carry which guarantee.

    Cheap and free — the marker is set during validation, not by a model call —
    and it is the fastest read on whether an artifact's markup is rich enough
    to support the guard at all. A run that is all `unverified_operator` was
    parsed with a backend that cannot express multiplication.
    """
    counts = Counter(f.get("latex_verification") or "none" for f in formulas)
    return dict(sorted(counts.items()))


def _verdict(value: float, kill: float | None) -> str:
    if kill is None:
        return "—"
    return "PASS" if value >= kill else "FAIL"


def _delta(value: float, previous: float | None) -> str:
    if previous is None:
        return "—"
    diff = round(value - previous, 4)
    return f"{diff:+.4f}"


def build_report(
    doc,
    filtered,
    clustered,
    scores: dict[str, Any],
    result,
    args,
    started: datetime,
    seconds: float,
) -> dict:
    funnel = {
        "pages": doc.n_pages,
        "chunks": len(doc.chunks),
        "mentions_raw": len(filtered.mentions),
        "clusters": clustered.n_clusters,
        "compression_ratio": clustered.compression_ratio,
        "abbrev_pairs": len(filtered.abbrev_pairs),
        "rule_candidates": len(filtered.rule_candidates),
    }
    report: dict[str, Any] = {
        "run": {
            "created_at": started.strftime("%Y-%m-%d_%H%M%S"),
            "doc_id": doc.doc_id,
            "artifact": str(args.artifact),
            "content_hash": doc.content_hash,
            "parser": f"{doc.parser_name}/{doc.parser_version}",
            "parser_backend": doc.parser_backend or "UNLABELLED",
            "extracted": bool(args.extract),
            "mock": bool(args.mock),
            "limit": args.limit,
            "seconds": round(seconds, 1),
        },
        "funnel": funnel,
        "scores": scores,
        "baseline": BASELINE,
    }

    if args.allow_unlabelled_artifact and not doc.parser_backend:
        report["run"]["comparability"] = (
            "NOT COMPARABLE — the artifact does not declare parser_backend and the "
            "run was forced. Different MinerU backends emit different text."
        )
    if args.mock:
        report["run"]["comparability"] = (
            "NOT SCOREABLE — mock extractor. Branch plumbing only; the content is "
            "fabricated, so every quality figure here is meaningless."
        )

    if result is not None:
        # Which model actually served this run. A result file without it is a
        # measurement of an unknown: on 2026-08-27 identical input scored 6/7
        # one day and 0/12 the next under an unchanged deployment name, and
        # nothing recorded said whether the model had moved. See X25.
        report["run"]["served_by"] = {
            "model_version": sorted({u.model_version for u in result.usages if u.model_version}),
            "system_fingerprint": sorted(
                {u.system_fingerprint for u in result.usages if u.system_fingerprint}
            ),
        }
        prompt, cached, completion = result.total_tokens
        report["cost"] = {
            "prompt_tokens": prompt,
            "cached_tokens": cached,
            "completion_tokens": completion,
            "cache_hit_rate": round(cached / prompt, 3) if prompt else 0.0,
            "calls": len(result.usages),
        }
        report["quality"] = {
            "fields_rejected_by_span_check": len(result.rejected),
            "rejections": [r.model_dump(mode="json") for r in result.rejected[:20]],
            "glossary_entries": len(result.glossary),
            "rule_entries": len(result.rules),
            "formula_entries": len(result.formulas),
            "abstained_null_definition": sum(
                1 for e in result.glossary if not e.get("definition")
            ),
            "latex_verification": _latex_verification_counts(result.formulas),
            "links": result.links,
        }
    return report


def format_summary(report: dict) -> str:
    rows = []
    for key, score in report["scores"].items():
        value = score["value"]
        kill = KILL_LINES.get(key)
        rows.append(
            f"  {key:<12} {value:>8.4f}   kill {kill if kill else '  —  '!s:>5}   "
            f"{_verdict(value, kill):<5}  vs baseline {_delta(value, BASELINE.get(key))}"
            f"   {score['label']}"
        )
    funnel = report["funnel"]
    out = [
        "",
        f"  doc {report['run']['doc_id']}   backend {report['run']['parser_backend']}"
        f"   {funnel['pages']} pages -> {funnel['chunks']} chunks -> "
        f"{funnel['mentions_raw']} mentions -> {funnel['clusters']} clusters "
        f"({funnel['compression_ratio']}x)",
        "",
        *rows,
    ]
    if "cost" in report:
        cost = report["cost"]
        out += [
            "",
            f"  {cost['calls']} calls   {cost['prompt_tokens']} prompt tokens, "
            f"{cost['cached_tokens']} cached ({cost['cache_hit_rate']:.0%})",
            f"  {report['quality']['fields_rejected_by_span_check']} fields rejected"
            " by the span check",
            f"  latex_verification: {report['quality']['latex_verification']}",
        ]
    if "comparability" in report["run"]:
        out += ["", f"  !! {report['run']['comparability']}"]
    return "\n".join(out)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Score the knowledge pipeline against knowledge_gold.yaml"
    )
    parser.add_argument("artifact", type=Path, help="ParsedDocument JSON (chunks.json)")
    parser.add_argument("--doc-id", help="override the artifact's doc_id")
    parser.add_argument(
        "--extract", action="store_true", help="run the PAID branches (E3, rule recall)"
    )
    parser.add_argument(
        "--mock", action="store_true", help="mock extractor: no network, no spend, NOT scoreable"
    )
    parser.add_argument("--limit", type=int, help="cap items per branch (pilot)")
    parser.add_argument("--active-glossary", type=Path, help="glossary to diff against")
    parser.add_argument("--gold", type=Path, default=GOLD_PATH)
    parser.add_argument("--label", help="short note recorded in the result file")
    parser.add_argument(
        "--dry-run", action="store_true", help="token estimate only, zero API calls"
    )
    parser.add_argument(
        "--allow-unlabelled-artifact",
        action="store_true",
        help="run on an artifact with no parser_backend (results NOT comparable)",
    )
    args = parser.parse_args(argv)

    if not args.artifact.exists():
        print(f"artifact not found: {args.artifact}", file=sys.stderr)
        return 2

    raw = json.loads(args.artifact.read_text(encoding="utf-8"))
    doc = parsed_doc_from_artifact(raw, doc_id=args.doc_id, source_ref=str(args.artifact))

    if not doc.parser_backend and not args.allow_unlabelled_artifact:
        print(
            f"artifact does not declare parser_backend: {args.artifact}\n"
            "  A score measured on one MinerU backend is not comparable to one\n"
            "  measured on another, and nothing downstream can tell them apart\n"
            "  afterwards. Re-export from `src.knowledge_parsing.run`, or pass\n"
            "  --allow-unlabelled-artifact to record the result as uncomparable.",
            file=sys.stderr,
        )
        return 2

    started = datetime.now()
    gold = load_gold(args.gold)

    filtered = run_filters(doc)
    clustered = build_clusters(doc, filtered)

    if args.dry_run:
        est = estimate_cost(doc, clustered, filtered, args.limit)
        print("[dry-run] NO API CALLS MADE")
        for key, value in est.items():
            print(f"          {key}: {value}")
        for branch in ("glossary", "rule", "formula", "summary"):
            print(
                f"          prefix[{branch}]: {prefix_tokens(branch)} tokens, "
                f"cacheable={cacheable(branch)}"
            )
        return 0

    e1 = score_term_filter(gold, [m.surface for m in filtered.mentions])
    scores: dict[str, Any] = {
        "E1": {
            "label": "term-filter recall",
            "value": e1.recall,
            "detail": e1.as_dict(),
        },
        "E2": {
            "label": "review-burden compression",
            "value": clustered.compression_ratio,
            "detail": {
                "mentions": clustered.n_mentions,
                "clusters": clustered.n_clusters,
            },
        },
    }

    result = None
    if args.extract:
        extractor = MockExtractor() if args.mock else _azure_extractor()
        if extractor is None:
            return 3
        active = (
            json.loads(args.active_glossary.read_text(encoding="utf-8"))
            if args.active_glossary
            else None
        )
        result = extract_all(
            doc, clustered, filtered, extractor, args.limit, active
        )
        e3 = score_glossary(gold, result.glossary)
        rules = score_rules(gold, result.rules)
        scores["E3"] = {
            "label": "glossary schema fill (precision)",
            "value": e3.precision,
            "detail": e3.as_dict(),
        }
        scores["rule_recall"] = {
            "label": "rule-branch recall",
            "value": rules.recall,
            "detail": rules.as_dict(),
        }

    seconds = (datetime.now() - started).total_seconds()
    report = build_report(doc, filtered, clustered, scores, result, args, started, seconds)
    if args.label:
        report["run"]["label"] = args.label
    if not args.extract:
        report["run"]["note"] = (
            "Free stages only. E3 and rule recall require --extract, which spends."
        )

    print(format_summary(report))

    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    stem = "mock" if args.mock else ("full" if args.extract else "filters")
    out_path = RESULTS_DIR / f"v2_{stem}_{started:%Y-%m-%d_%H%M%S}.json"
    if out_path.exists():  # never overwrite a committed measurement
        print(f"refusing to overwrite {out_path}", file=sys.stderr)
        return 4
    out_path.write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"\n-> saved: {out_path.relative_to(_HERE.parent.parent)}")

    if EVIDENCE_K and args.extract and not args.mock:
        failed = [
            k for k, s in scores.items() if k in KILL_LINES and s["value"] < KILL_LINES[k]
        ]
        if failed:
            print(f"\nBELOW KILL LINE: {', '.join(failed)}", file=sys.stderr)
            return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())