[NOTICKET] feat(knowledge-parsing): document parsing half of the knowledge pipeline

Implements §4 B2/B5/B6 and closes the four open seam items (S2, S3, S4, S6b/c/d)
against the contracts.py draft circulated 2026-08-19. Additive and flag-gated:
src/knowledge/ (Tesseract -> pgvector) is untouched.

Seam — src/knowledge_parsing/contracts.py, pydantic per S6e:
- ParsedDocument envelope: content_hash, n_pages, version, parser_name/version/
backend/config, schema_version, raw_output_dir. parser_backend is read from
MinerU's _middle.json, so it records what actually ran, not what was configured
— a MinerU change and a prompt change stay distinguishable (S2).
- Heading structure is READ, not derived. Verified in MinerU's source: pipeline
and vlm run identical text_level logic, and bbox is emitted by both. Heading
availability is document-dependent, not backend-dependent — the earlier claim
came from sampling the McGraw-Hill handbook, which has no numbered headings (S3).
- source_wording on TermRecord; heading carried verbatim (S4).
- page_idx/page_idxs, 0-based, no conversion anywhere — 1-based is the UI's job,
done once at display time (S6b).

Parsing:
- Content-addressed parse cache keyed on file hash + settings + MinerU version.
Parsing is the expensive stage (~19 s/page CPU); the normalizer can be iterated
without re-parsing. Version is part of the key so numbers stay comparable.
- Batch with --resume and per-document failure isolation (failures.jsonl); one
bad document does not stop the run.
- manifest.json + report.py --scale N produce the throughput numbers directly.
Runs under 20 pages are refused for extrapolation: model load is a fixed cost,
so 1 page reads ~2.8x slower per page than 20 (54.0 vs 19.0 s/page measured).

Normalizer, from KNOWLEDGE_PIPELINE_CALIBRATION.md §4:
- Breadcrumb headings. Documents reprint their heading path on every page; a
naive splitter reopens the section and shatters it. A heading already on the
stack is a breadcrumb — the section continues and its page range extends.
- The heading line is the first segment of Chunk.text, verbatim. In the BUMA
standard the heading names the term and the body opens with "Adalah ..."
without repeating it, so a split-out heading leaves the defining chunk with no
mention of its own term — ranking below sections that merely use it. It also
makes Chunk.text the single place a span check must look.
- Heading length capped at 90 chars and chunks at ~1500 tokens, per §4.

Quality checks (checks.py) — warnings, never failures:
- Numbers in extracted formulas are cross-checked against the source PDF's text
layer. This is the only check that can catch silent corruption: the known real
defect turned 5600 into 55600, leaving valid-looking LaTeX and no error. No
output-only check can see that. Threshold 3 digits — at 2 the real defect was
still caught but six false positives came with it.

Dependencies: MinerU is an optional extra ("knowledge-parsing"), not a main
dependency. It is imported inside parse.py at call time, so importing the package
does not pull torch and the deployed Space builds without it. Verified: importing
src.knowledge_parsing leaves mineru and torch absent from sys.modules.

Verification: ruff clean (T201 scoped to the two CLI files, following the
existing eval/** precedent); import main OK; full run reproduced in this repo
against the existing parse cache. Output paths (data/knowledge_*) are gitignored.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.gitignore

@@ -54,4 +54,8 @@ docs/specs/tabular_parquet_contract.md
 docs/specs/tabular_parquet.md
 
 # Personal / local working docs (not for the shared repo) — archived out of root
-docs/_archive/
+docs/_archive/
+# Knowledge pipeline output — data, not code (parse cache + run artifacts)
+data/knowledge_cache/
+data/knowledge_runs/
+data/knowledge_docs/

pyproject.toml

@@ -106,6 +106,18 @@ dev = [
     "pre-commit==4.0.1",
 ]
 
+# Document parsing for the knowledge pipeline (src/knowledge_parsing/).
+#
+# Deliberately an EXTRA, not a main dependency: MinerU pulls torch and its model
+# tail (GBs), and the agent service never parses documents at request time —
+# parsing is an offline, admin-triggered batch job. Keeping it here means the
+# deployed Space does not build or ship any of it.
+#
+#     pip install -e ".[parsing]"
+knowledge-parsing = [
+    "mineru==3.4.4",
+]
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/agent_service"]
 
@@ -131,6 +143,10 @@ ignore = [
 # Same rule, same reason: this is an operator-run CLI, and its printed output IS
 # the deliverable.
 "src/knowledge_extraction/cli.py" = ["T201"]
+# Same again for the knowledge-parsing CLI entry points. Scoped to the two files
+# rather than the package, so the library modules stay print-free.
+"src/knowledge_parsing/run.py" = ["T201"]
+"src/knowledge_parsing/report.py" = ["T201"]
 
 [tool.mypy]
 python_version = "3.12"

src/knowledge_parsing/__init__.py

@@ -0,0 +1,22 @@
+"""Document parsing — the first half of the knowledge pipeline.
+
+PDF/DOCX in, a versioned `ParsedDocument` out. The extraction half consumes that
+artifact and never the parser itself, which is what keeps MinerU swappable.
+
+Additive and flag-gated: the existing unstructured path (`src/knowledge/`,
+Tesseract OCR -> chunk -> pgvector) is untouched and keeps running as-is.
+
+Heavy dependencies (MinerU, torch) are an optional extra — see `pyproject.toml`.
+Importing this package does not import them; only `parse.py` does, at call time,
+so the agent service starts without them installed.
+"""
+
+from .contracts import SCHEMA_VERSION, Chunk, Mention, ParsedDocument, TermRecord
+
+__all__ = [
+    "SCHEMA_VERSION",
+    "Chunk",
+    "Mention",
+    "ParsedDocument",
+    "TermRecord",
+]

src/knowledge_parsing/checks.py

@@ -0,0 +1,144 @@
+"""Pemeriksaan mutu hasil parse — menangkap kerusakan yang TIDAK bersuara.
+
+Dasarnya temuan di `Hasil-Uji-MinerU.md`: formula EOQ hasil pipeline CPU
+angkanya hilang tanpa error, tanpa peringatan. Kerusakan seperti itu tidak akan
+pernah ketahuan dari status "selesai" — harus diperiksa sendiri.
+
+Semua temuan di sini sifatnya PERINGATAN, bukan kegagalan: dokumen tetap
+diproses, tapi peringatannya tercatat di manifest supaya tidak ada mutu yang
+turun diam-diam di antara ratusan dokumen.
+"""
+
+from __future__ import annotations
+
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+# Angka di dalam latex. Formula teknis hampir selalu punya angka atau pembagi.
+_ADA_ANGKA = re.compile(r"\d")
+
+# Panjang minimal deret angka yang dicocokkan ke sumber.
+#
+# Diuji ke dokumen nyata: dengan ambang 2 digit, bug asli (55600) tertangkap
+# tapi ikut muncul 6 peringatan palsu dari angka pendek — lapisan teks PDF-nya
+# memang rusak sebagian, jadi angka pendek sering tidak ketemu padahal benar.
+# Dengan 3 digit, bug tetap tertangkap dan derau hilang.
+MIN_DIGIT = 3
+
+_SEMUA_ANGKA = re.compile(r"\d+")
+
+
+def _deret(teks: str, min_digit: int) -> set[str]:
+    return {a for a in _SEMUA_ANGKA.findall(teks) if len(a) >= min_digit}
+
+
+def _angka_dari_latex(latex: str, min_digit: int = MIN_DIGIT) -> set[str]:
+    """MinerU menulis latex berspasi ('5 5 6 0 0'), jadi spasi dibuang dulu."""
+    return _deret(re.sub(r"\s+", "", latex), min_digit)
+
+
+def _angka_per_halaman(pdf: Path, min_digit: int = MIN_DIGIT) -> dict[int, set[str]] | None:
+    """Deret angka pada lapisan teks PDF asli, per halaman.
+
+    Dipakai sebagai pembanding. Kalau PDF tidak terbaca, kembalikan None dan
+    pemeriksaan ini dilewati diam-diam — lebih baik tidak memeriksa daripada
+    memberi peringatan palsu.
+    """
+    try:
+        from pypdf import PdfReader
+        reader = PdfReader(str(pdf))
+    except Exception:
+        return None
+
+    hasil: dict[int, set[str]] = {}
+    for i, hal in enumerate(reader.pages):
+        try:
+            teks = hal.extract_text() or ""
+        except Exception:
+            teks = ""
+        # buang pemisah ribuan & spasi supaya "$5,600" dan "5 600" -> "5600"
+        bersih = re.sub(r"[, \s]", "", teks)
+        hasil[i] = _deret(bersih, min_digit)
+    return hasil
+
+
+def periksa_items(
+    items: list[dict[str, Any]],
+    min_latex_len: int = 8,
+    pdf_sumber: Path | None = None,
+    offset_halaman: int = 0,
+) -> dict[str, Any]:
+    """Kembalikan ringkasan + daftar peringatan untuk satu dokumen.
+
+    `pdf_sumber` mengaktifkan pencocokan angka ke PDF asli — satu-satunya cara
+    menangkap angka yang berubah diam-diam (bug EOQ: 5600 -> 55600).
+    """
+    per_tipe = Counter(x.get("type") for x in items)
+    halaman = {x.get("page_idx") for x in items if x.get("page_idx") is not None}
+
+    peringatan: list[dict[str, Any]] = []
+
+    # 1. Halaman tanpa teks sama sekali -> kemungkinan gagal dibaca
+    teks_per_halaman = Counter(
+        x.get("page_idx") for x in items
+        if x.get("type") in {"text", "table", "equation"} and (x.get("text") or x.get("table_body"))
+    )
+    for p in sorted(halaman):
+        if teks_per_halaman.get(p, 0) == 0:
+            peringatan.append({"jenis": "halaman_kosong", "page": p})
+
+    # 2. Formula terpotong / kehilangan angka  <- bug EOQ
+    for i, x in enumerate(items):
+        if x.get("type") != "equation":
+            continue
+        latex = (x.get("text") or "").strip()
+        if len(latex) < min_latex_len:
+            peringatan.append({"jenis": "latex_terlalu_pendek", "item": i,
+                               "page": x.get("page_idx"), "isi": latex[:60]})
+        elif not _ADA_ANGKA.search(latex) and "\\frac" not in latex:
+            peringatan.append({"jenis": "latex_tanpa_angka", "item": i,
+                               "page": x.get("page_idx"), "isi": latex[:60]})
+
+    # 3. ⭐ Angka di formula tidak ada di PDF asli -> berubah diam-diam.
+    #    Ini SATU-SATUNYA pemeriksaan yang bisa menangkap bug EOQ; pemeriksaan
+    #    lain di berkas ini hanya melihat output sendiri, dan angka yang salah
+    #    tetap menghasilkan latex yang sah sepenuhnya.
+    angka_asli = _angka_per_halaman(pdf_sumber) if pdf_sumber else None
+    if angka_asli is not None:
+        for i, x in enumerate(items):
+            if x.get("type") != "equation":
+                continue
+            p = x.get("page_idx", 0) + offset_halaman
+            acuan = angka_asli.get(p)
+            if not acuan:          # halaman tanpa lapisan teks -> tidak bisa dinilai
+                continue
+            for angka in sorted(_angka_dari_latex(x.get("text") or "")):
+                if angka not in acuan:
+                    peringatan.append({
+                        "jenis": "angka_tidak_ada_di_sumber", "item": i, "page": p,
+                        "angka": angka,
+                        "catatan": "angka pada formula tidak ditemukan di teks PDF asli",
+                    })
+
+    # 4. Tabel tanpa isi
+    for i, x in enumerate(items):
+        if x.get("type") == "table" and not (x.get("table_body") or "").strip():
+            peringatan.append({"jenis": "tabel_kosong", "item": i, "page": x.get("page_idx")})
+
+    # 5. Chart tanpa konteks apa pun (tidak bisa dipakai tahap berikutnya)
+    for i, x in enumerate(items):
+        if x.get("type") == "chart":
+            punya = (x.get("content") or "").strip() or (x.get("chart_caption") or [])
+            if not punya:
+                peringatan.append({"jenis": "chart_tanpa_konteks", "item": i,
+                                   "page": x.get("page_idx")})
+
+    return {
+        "item": len(items),
+        "halaman_terdeteksi": len(halaman),
+        "per_tipe": dict(per_tipe),
+        "peringatan": peringatan,
+        "jumlah_peringatan": len(peringatan),
+    }

src/knowledge_parsing/config.py

@@ -0,0 +1,121 @@
+"""Satu-satunya tempat pengaturan pipeline.
+
+⭐ Titik flip GPU ada di sini: ubah `backend` dari "pipeline" ke "vlm" dan tidak
+ada berkas lain yang perlu disentuh.
+
+    pipeline  - murni CPU. Terbukti jalan di laptop tanpa GPU (19,0 dtk/hal @ 20 hal).
+                Dipakai sementara supaya pipeline bisa dibangun & diuji sekarang.
+    vlm       - target resmi (akurasi 95,30). BUTUH GPU, VRAM >= 8 GB.
+    hybrid    - butuh GPU juga. Di T4 kecepatannya setara pipeline-CPU, jadi
+                GPU di jalur ini membeli mutu, bukan kecepatan.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+Backend = Literal["pipeline", "vlm", "hybrid"]
+
+# Akar penyimpanan: root repo (src/knowledge_parsing/ -> ../../).
+# Semua keluaran (cache/, runs/) adalah DATA, bukan kode — di-gitignore.
+ROOT = Path(__file__).resolve().parent.parent.parent
+
+
+@dataclass
+class PipelineConfig:
+    # --- yang paling sering diubah ---
+    backend: Backend = "pipeline"
+
+    # ⚠️ MinerU TIDAK punya kode bahasa Indonesia. Dokumen berbahasa Indonesia
+    # memakai "ch", yang cakupannya "Chinese, English, Japanese, Chinese
+    # Traditional, LATIN" — aksara Latin itu yang mencakup bahasa Indonesia.
+    # Mengisi "id" akan gagal: ValueError: Language id not supported.
+    lang: str = "ch"
+
+    # --- lokasi ---
+    input_dir: Path = ROOT / "data" / "knowledge_docs"
+    cache_dir: Path = ROOT / "data" / "knowledge_cache"     # hasil parse, dikunci isi dokumen
+    runs_dir: Path = ROOT / "data" / "knowledge_runs"       # hasil normalize + manifest per run
+
+    # --- opsi MinerU ---
+    formula_enable: bool = True
+    table_enable: bool = True
+    effort: str = "medium"               # hanya dipakai backend hybrid
+    start_page: int = 0
+    end_page: int | None = None
+
+    # --- perilaku ---
+    resume: bool = True                  # lewati dokumen yang sudah selesai
+    write_debug_pdf: bool = True         # _layout.pdf & _span.pdf: cek mutu paling cepat
+
+    # --- ambang pemeriksaan mutu (lihat checks.py) ---
+    min_latex_len: int = 8               # latex lebih pendek dari ini dicurigai terpotong
+
+    extra: dict[str, Any] = field(default_factory=dict)
+
+    def lang_kanonik(self) -> str:
+        """Kode bahasa yang benar-benar dipakai MinerU.
+
+        `validate_public_ocr_lang` sekaligus memvalidasi dan mengkanonikkan:
+        alias 'latin' / 'en' / 'japan' -> 'ch', dan kode tak dikenal ditolak.
+        Ini fungsi yang sama yang memunculkan pesan "Language id not supported".
+
+        Dipakai juga oleh sidik jari cache: tanpa dikanonikkan, `--lang latin`
+        dan `--lang ch` menghasilkan kunci cache berbeda padahal hasil
+        parsingnya identik — dokumen yang sama akan diparse dua kali.
+
+        Catatan: hanya ImportError yang ditangkap (MinerU belum terpasang).
+        ValueError sengaja DIBIARKAN naik — itu justru kode bahasa yang salah,
+        dan menelannya berarti kesalahan baru ketahuan setelah model dimuat.
+        """
+        try:
+            from mineru.utils.ocr_language import validate_public_ocr_lang
+        except ImportError:
+            return self.lang
+        return validate_public_ocr_lang(self.lang)
+
+    def periksa(self) -> None:
+        """Gagal cepat kalau pengaturannya salah.
+
+        Tanpa ini, kode bahasa yang keliru baru ketahuan setelah ~30 detik
+        memuat model — dan pada batch besar, setelah dokumen ke sekian.
+        """
+        try:
+            self.lang_kanonik()
+        except ValueError as e:
+            raise SystemExit(
+                f"{e}\n"
+                "  -> Dokumen berbahasa Indonesia memakai 'ch' "
+                "(mencakup aksara Latin). MinerU tidak punya kode 'id'."
+            ) from e
+
+    def fingerprint(self) -> str:
+        """Sidik jari pengaturan yang MEMPENGARUHI hasil parse.
+
+        Dipakai sebagai bagian kunci cache. Opsi yang tidak mengubah hasil
+        (resume, runs_dir, ambang pemeriksaan) sengaja tidak ikut, supaya
+        mengubahnya tidak membatalkan cache yang masih sah.
+        """
+        bahan = {
+            "backend": self.backend,
+            "lang": self.lang_kanonik(),
+            "formula_enable": self.formula_enable,
+            "table_enable": self.table_enable,
+            "effort": self.effort if self.backend == "hybrid" else None,
+            "start_page": self.start_page,
+            "end_page": self.end_page,
+        }
+        payload = json.dumps(bahan, sort_keys=True).encode()
+        return hashlib.sha256(payload).hexdigest()[:12]
+
+    def to_dict(self) -> dict[str, Any]:
+        d = asdict(self)
+        for k, v in d.items():
+            if isinstance(v, Path):
+                d[k] = str(v)
+        d["fingerprint"] = self.fingerprint()
+        return d

src/knowledge_parsing/contracts.py

@@ -0,0 +1,185 @@
+"""Data shapes handed between pipeline stages — the parsing/extraction seam.
+
+Extraction consumes `ParsedDocument`, never a file path and never the parser's
+API. That is what keeps the parser swappable: a Tesseract or Azure Document
+Intelligence path emits the same artifact, and the extraction half never learns
+which one ran.
+
+Design note — these fields were derived from real MinerU output, not designed on
+paper. Verified against actual `content_list.json` output and cross-checked
+against MinerU's source for both the `pipeline` and `vlm` backends.
+
+Two things worth knowing before reviewing:
+
+1. `bbox` and `text_level` are emitted by **both** backends — verified in
+   MinerU's source, where `pipeline` and `vlm` run identical logic:
+
+       elif para_type == BlockType.TITLE:
+           title_level = get_title_level(para_block)
+           if title_level != 0:
+               para_content['text_level'] = title_level
+
+   So heading structure is read from the document when MinerU detects titles,
+   and only derived from numbering patterns as a fallback.
+
+2. Heading availability is **document-dependent, not backend-dependent**. A
+   document with explicit numbered headings (e.g. the BUMA standard: `2.` /
+   `2.1.` / `2.1.1.`) yields a clean hierarchy. A document of mid-chapter pages
+   with no headings yields none — which is why `section_no`, `heading` and
+   `heading_path` stay Optional rather than required.
+
+⚠️ `Chunk.text` must stay VERBATIM from the source document. The extraction-side
+guardrail locates LLM-quoted spans literally inside this text; if the text is
+ever cleaned up (lines rejoined, whitespace normalised), the lookup fails and
+the field is silently set to null instead of raising. The failure looks like a
+bad LLM, but the cause would be here.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+ChunkKind = Literal["text", "table", "chart", "equation"]
+
+SCHEMA_VERSION = "0.2.0"
+
+
+class Chunk(BaseModel):
+    """One unit of a parsed document, ready for the extraction stage."""
+
+    chunk_id: str
+    doc_id: str
+    kind: ChunkKind
+
+    # Content. VERBATIM — never reflowed or normalised. See note above.
+    #
+    # When a section has a heading, its heading line is the FIRST segment of
+    # this text, verbatim. Not for tidiness: in the BUMA standard the heading
+    # names the term ("2.1.3. Physical of Availability (PA)") and the body then
+    # opens with "Adalah ..." without repeating it. With the heading split out,
+    # the chunk that actually *defines* a term contains no mention of it, and
+    # ranks below calculation sections that merely use it repeatedly.
+    #
+    # It also makes this field the single place a span check has to look:
+    # `TermRecord.source_wording` is findable here, not in a second field.
+    text: str
+
+    # Location in the source document.
+    #
+    # Named `page_idx` deliberately: these are 0-BASED, exactly as MinerU reports
+    # them, with no conversion anywhere in the pipeline. Page numbers eventually
+    # reach a human review queue, and an off-by-one there is invisible until an
+    # expert opens the wrong page. Converting to 1-based is the UI's job, done
+    # once at display time — never here, so the artifact always matches the raw
+    # MinerU output kept alongside it.
+    page_idx: int
+    page_idxs: list[int] = Field(default_factory=list)
+
+    # Structural context. All Optional — many documents carry no headings.
+    section_no: str | None = None  # e.g. "2.1.3", when the document is numbered
+
+    # This chunk's own section title, VERBATIM as the document writes it —
+    # including wording a reader may be tempted to "fix". The BUMA standard says
+    # "Physical of Availability (PA)", not "Physical Availability"; that exact
+    # string must reach the extraction model, or it gets silently normalised and
+    # the discrepancy never reaches the expert.
+    heading: str | None = None
+
+    # Running headers of every page this chunk spans, in page order. A list
+    # rather than a single value: a chunk crossing a chapter boundary would
+    # otherwise silently keep only the first page's chapter.
+    chapters: list[str] = Field(default_factory=list)
+
+    # Breadcrumb of enclosing headings, outermost first, built from MinerU's
+    # `text_level` hierarchy. Example from the BUMA standard:
+    #   ["2. PENJELASAN PARAMETER", "2.1. Production Parameter", "2.1.1. Production"]
+    heading_path: list[str] = Field(default_factory=list)
+
+    # Flags for cheap filtering downstream
+    has_formula: bool = False
+    is_tabular: bool = False
+
+    # Trace back to the source: item indices in MinerU's content_list.json
+    source_items: list[int] = Field(default_factory=list)
+
+    # Position on the page of the first source item, as MinerU reports it.
+    # Carried through for a curation UI that highlights where on the page a
+    # definition came from. Nothing in the pipeline reasons about it — ordering
+    # uses item sequence, never coordinates.
+    bbox: list[int] | None = None
+
+    # Non-text attachments (formula images, table/chart crops)
+    images: list[str] = Field(default_factory=list)
+
+
+class ParsedDocument(BaseModel):
+    """The artifact itself — one parsed document, self-describing.
+
+    The chunk list alone is not enough to hand across the seam: an artifact that
+    travels to the extraction half must carry its own provenance. Without
+    `parser_name` / `parser_version` / `parser_backend`, a MinerU upgrade and a
+    prompt change are indistinguishable when extraction results shift.
+
+    `content_hash` is the hash of the SOURCE FILE, so re-parsing the same
+    document is detectable and a changed document forces a new artifact version.
+    """
+
+    doc_id: str
+    chunks: list[Chunk]
+
+    # Identity of the source
+    source_path: str
+    content_hash: str                    # sha256 of the source file
+    n_pages: int
+
+    # Which parser produced this, and how
+    parser_name: str = "mineru"
+    parser_version: str | None = None    # e.g. "3.4.4"
+    parser_backend: str | None = None    # "pipeline" | "vlm" | "hybrid", as MinerU recorded it
+    parser_config: str | None = None     # fingerprint of the settings that affect output
+
+    # Version of THIS ARTIFACT for this document — bumped when the document is
+    # re-parsed (new source content, new parser version, or changed settings).
+    # Distinct from `schema_version`, which versions the contract itself.
+    version: int = 1
+
+    schema_version: str = SCHEMA_VERSION
+    created_at: str | None = None
+
+    # Where the untouched MinerU output for this document lives
+    raw_output_dir: str | None = None
+
+
+# --- Seam for the downstream stages (declared, not yet used) ---
+# Written here so the handoff shape is visible from the start. The extraction
+# half owns the final form of these two — see `KNOWLEDGE_PIPELINE_TODO.md` §5.
+
+
+class Mention(BaseModel):
+    """One occurrence of a term inside a chunk."""
+
+    term: str
+    chunk_id: str
+    page_idx: int
+
+
+class TermRecord(BaseModel):
+    """Extracted knowledge for one term (one cluster of mentions)."""
+
+    term: str
+    full_name: str | None = None
+
+    # What the document literally says, before any normalisation — e.g.
+    # "Physical of Availability (PA)" where `full_name` may read "Physical
+    # Availability". Span-checked against `Chunk.text` like every other field,
+    # so the discrepancy surfaces to the expert instead of being quietly fixed.
+    source_wording: str | None = None
+
+    definition: str | None = None
+    formula_latex: str | None = None
+    subdomain_tags: list[str] = Field(default_factory=list)
+    mention_count: int = 0
+    provenance: dict = Field(default_factory=dict)
+    extraction_status: str = "ok"

src/knowledge_parsing/manifest.py

@@ -0,0 +1,86 @@
+"""Catatan satu kali jalan — bahan mentah untuk angka justifikasi biaya.
+
+Yang WAJIB tercatat dan gampang terlupa: backend yang benar-benar dipakai dan
+versi MinerU. Tanpa keduanya, angka CPU hari ini tidak sah dibandingkan dengan
+angka GPU bulan depan — padahal perbandingan itulah inti argumennya.
+
+Backend dibaca dari `_middle.json` keluaran MinerU (`_backend`), bukan dari
+config, supaya yang tercatat adalah kenyataan, bukan niat.
+"""
+
+from __future__ import annotations
+
+import json
+import platform
+from dataclasses import asdict, dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class CatatanDokumen:
+    doc_id: str
+    source: str
+    status: str                      # "ok" | "gagal"
+    pages: int = 0
+    detik_parse: float = 0.0
+    detik_normalisasi: float = 0.0
+    dari_cache: bool = False
+    jumlah_chunk: int = 0
+    backend_tercatat: str | None = None
+    pemeriksaan: dict[str, Any] = field(default_factory=dict)
+    galat: str | None = None
+
+    @property
+    def detik_per_halaman(self) -> float | None:
+        return self.detik_parse / self.pages if self.pages else None
+
+
+@dataclass
+class Manifest:
+    run_id: str
+    dimulai: str
+    config: dict[str, Any]
+    mineru_version: str | None = None
+    mesin: dict[str, str] = field(default_factory=dict)
+    dokumen: list[CatatanDokumen] = field(default_factory=list)
+
+    @classmethod
+    def baru(cls, config: dict[str, Any]) -> Manifest:
+        return cls(
+            run_id=datetime.now().strftime("%Y%m%d-%H%M%S"),
+            dimulai=datetime.now().isoformat(timespec="seconds"),
+            config=config,
+            mesin={
+                "os": platform.platform(),
+                "python": platform.python_version(),
+                "prosesor": platform.processor() or "?",
+            },
+        )
+
+    def ringkasan(self) -> dict[str, Any]:
+        ok = [d for d in self.dokumen if d.status == "ok"]
+        halaman = sum(d.pages for d in ok)
+        # Hanya run NYATA yang dihitung untuk kecepatan — hasil dari cache
+        # akan membuat angkanya terlihat jauh lebih cepat dari kenyataan.
+        nyata = [d for d in ok if not d.dari_cache]
+        detik = sum(d.detik_parse for d in nyata)
+        halaman_nyata = sum(d.pages for d in nyata)
+        return {
+            "dokumen_ok": len(ok),
+            "dokumen_gagal": len(self.dokumen) - len(ok),
+            "total_halaman": halaman,
+            "halaman_terukur": halaman_nyata,
+            "detik_parse_terukur": round(detik, 1),
+            "detik_per_halaman": round(detik / halaman_nyata, 2) if halaman_nyata else None,
+            "total_peringatan": sum(
+                d.pemeriksaan.get("jumlah_peringatan", 0) for d in self.dokumen
+            ),
+        }
+
+    def simpan(self, path: Path) -> None:
+        isi = asdict(self)
+        isi["ringkasan"] = self.ringkasan()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(isi, indent=2, ensure_ascii=False), encoding="utf-8")

src/knowledge_parsing/normalize.py

@@ -0,0 +1,248 @@
+"""Tahap 3 — ubah daftar item datar MinerU menjadi chunk yang bermakna.
+
+Kenapa tahap ini ada: MinerU mengeluarkan daftar item DATAR, bukan unit
+bermakna. Item-item itu perlu digabung jadi chunk per section.
+
+MinerU memberi penanda judul `text_level` (2 / 2.1 / 2.1.1 -> level 2 / 3 / 4)
+di KEDUA backend — dicek langsung ke source-nya, `pipeline` dan `vlm` memakai
+logika yang sama. Tapi penanda itu hanya muncul kalau dokumennya memang punya
+judul yang terdeteksi:
+
+  - dokumen standar BUMA (judul bernomor eksplisit) -> hierarki lengkap
+  - handbook McGraw-Hill (kumpulan halaman tengah bab) -> nyaris tidak ada,
+    1 dari 84 item, dan itu pun caption tabel
+
+Jadi ketersediaan judul itu sifat DOKUMEN, bukan sifat backend. Karena itu
+`text_level` dipakai sebagai sinyal utama, dengan pola penomoran sebagai
+cadangan, dan hasilnya tetap boleh kosong.
+
+Yang dilakukan:
+  - `page_number` dibuang (nomor halaman, bukan isi)
+  - `header` dipakai sebagai konteks bab, bukan judul section
+    (di dokumen uji semuanya header berjalan di bbox y~57-72, satu per halaman)
+  - judul section dari `text_level`, cadangan pola penomoran ("2.1.3 Judul"),
+    lalu disusun jadi `heading_path` (jejak judul induk sampai judul sendiri)
+  - `table` dan `chart` jadi chunk sendiri; `equation` menempel ke chunk berjalan
+    dan menyalakan `has_formula`
+
+⚠️ Teks TIDAK PERNAH dirapikan di sini — tidak digabung barisnya, tidak
+diseragamkan spasinya. Lihat alasannya di contracts.py.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+from .contracts import Chunk
+
+# "2.1.3 Judul" / "2.1.3. Judul" / "4 Judul"
+_POLA_NOMOR = re.compile(r"^(\d+(?:\.\d+)*)\.?\s+(\S.*)$")
+
+# Judul biasanya pendek. Ambang ini mencegah paragraf yang kebetulan diawali
+# angka ikut dianggap judul. Nilainya diselaraskan dengan konstanta terkalibrasi
+# di KNOWLEDGE_PIPELINE_CALIBRATION.md §4: baris lebih panjang dari ini adalah
+# kalimat atau baris formula, bukan judul.
+_MAKS_PANJANG_JUDUL = 90
+
+_DIBUANG = {"page_number"}
+
+# Batas ukuran chunk. Judul tetap batas utama; ini cuma pengaman untuk dokumen
+# yang judulnya tidak terdeteksi sama sekali — tanpa batas, satu chunk bisa
+# menelan seluruh dokumen, yang membuat evidence ranking tumpul dan menggelembungkan
+# porsi token cabang summary. ~6000 karakter kira-kira setara 1.500 token.
+MAKS_KARAKTER_CHUNK = 6000
+
+
+def _judul(item: dict[str, Any]) -> tuple[str | None, str | None, int | None]:
+    """Kembalikan (nomor_section, judul, level) kalau item ini judul section.
+
+    `text_level` dari MinerU adalah sinyal utama — dipakai kalau ada, karena itu
+    hasil deteksi judul MinerU sendiri dan membawa tingkat hierarkinya
+    (2 / 2.1 / 2.1.1 -> level 2 / 3 / 4). Pola penomoran cuma cadangan untuk
+    dokumen yang judulnya tidak terdeteksi MinerU.
+    """
+    if item.get("type") not in {"text", "title"}:
+        return None, None, None
+    teks = (item.get("text") or "").strip()
+    if not teks or len(teks) > _MAKS_PANJANG_JUDUL:
+        return None, None, None
+
+    level = item.get("text_level")
+    if level:                              # MinerU yakin ini judul
+        m = _POLA_NOMOR.match(teks)
+        return (m.group(1), m.group(2), int(level)) if m else (None, teks, int(level))
+
+    m = _POLA_NOMOR.match(teks)
+    if m and not teks.endswith((".", ":", ";")):
+        # Tanpa text_level, tingkat diperkirakan dari kedalaman penomoran:
+        # "2" -> 1, "2.1" -> 2, "2.1.1" -> 3
+        return m.group(1), m.group(2), m.group(1).count(".") + 1
+    return None, None, None
+
+
+def _teks_tabel(item: dict[str, Any]) -> str:
+    bagian = list(item.get("table_caption") or [])
+    if item.get("table_body"):
+        bagian.append(item["table_body"])
+    bagian += list(item.get("table_footnote") or [])
+    return "\n\n".join(b for b in bagian if b)
+
+
+def _teks_chart(item: dict[str, Any]) -> str:
+    bagian = list(item.get("chart_caption") or [])
+    if (item.get("content") or "").strip():
+        bagian.append(item["content"])
+    bagian += list(item.get("chart_footnote") or [])
+    return "\n\n".join(b for b in bagian if b)
+
+
+def normalisasi(items: list[dict[str, Any]], doc_id: str) -> list[Chunk]:
+    # Konteks bab per halaman, dari header berjalan
+    bab_per_halaman: dict[int, str] = {}
+    for x in items:
+        if x.get("type") == "header" and (x.get("text") or "").strip():
+            bab_per_halaman.setdefault(x.get("page_idx", 0), x["text"].strip())
+
+    chunks: list[Chunk] = []
+    berjalan: Chunk | None = None
+    potongan: list[str] = []
+
+    # Tumpukan judul yang sedang berlaku: [(level, teks_judul), ...].
+    # Judul level N menutup semua judul level >= N sebelumnya.
+    tumpukan: list[tuple[int, str]] = []
+
+    def dorong_judul(level: int, teks: str) -> None:
+        while tumpukan and tumpukan[-1][0] >= level:
+            tumpukan.pop()
+        tumpukan.append((level, teks))
+
+    def tutup() -> None:
+        nonlocal berjalan, potongan
+        if berjalan is not None:
+            berjalan.text = "\n\n".join(potongan).strip()
+            if berjalan.text or berjalan.images:
+                berjalan.page_idxs = sorted(set(berjalan.page_idxs))
+                chunks.append(berjalan)
+        berjalan, potongan = None, []
+
+    def buka(kind: str, page: int, section_no=None, heading=None) -> Chunk:
+        return Chunk(
+            chunk_id=f"{doc_id}::{len(chunks):04d}",
+            doc_id=doc_id, kind=kind, text="",
+            page_idx=page, page_idxs=[page],
+            section_no=section_no, heading=heading,
+            chapters=[bab_per_halaman[page]] if page in bab_per_halaman else [],
+            heading_path=[t for _, t in tumpukan],
+        )
+
+    for i, item in enumerate(items):
+        tipe = item.get("type")
+        if tipe in _DIBUANG or tipe == "header":
+            continue
+        page = item.get("page_idx", 0)
+
+        if tipe == "table":
+            tutup()
+            c = buka("table", page)
+            c.text = _teks_tabel(item)
+            c.is_tabular = True
+            c.source_items = [i]
+            c.bbox = item.get("bbox")
+            if item.get("img_path"):
+                c.images = [item["img_path"]]
+            if c.text or c.images:
+                chunks.append(c)
+            continue
+
+        if tipe == "chart":
+            tutup()
+            c = buka("chart", page)
+            c.text = _teks_chart(item)
+            c.source_items = [i]
+            c.bbox = item.get("bbox")
+            if item.get("img_path"):
+                c.images = [item["img_path"]]
+            chunks.append(c)
+            continue
+
+        if tipe == "equation":
+            latex = (item.get("text") or "").strip()
+            if berjalan is None:
+                berjalan = buka("text", page)
+            berjalan.has_formula = True
+            berjalan.source_items.append(i)
+            berjalan.page_idxs.append(page)
+            if item.get("img_path"):
+                berjalan.images.append(item["img_path"])
+            if latex:
+                potongan.append(latex)
+            continue
+
+        # sisanya: teks
+        teks = (item.get("text") or "").strip()
+        if not teks:
+            continue
+
+        nomor, judul, level = _judul(item)
+        if judul is not None:
+            # BREADCRUMB: banyak dokumen mencetak ulang jalur judulnya di atas
+            # setiap halaman (BUMA mengulang "2. PENJELASAN PARAMETER /
+            # 2.1. Production Parameter" di hal. 2-8). Judul yang SUDAH ada di
+            # tumpukan berarti section yang sedang berjalan atau induknya —
+            # bukan section baru. Tanpa aturan ini, section yang sama pecah
+            # berkali-kali dan semua angka di hilir ikut rusak.
+            if any(teks == t for _, t in tumpukan):
+                if berjalan is not None:
+                    berjalan.page_idxs.append(page)   # section berlanjut, halaman meluas
+                continue
+            tutup()
+            # Judul didorong SEBELUM chunk dibuka, supaya chunk isinya membawa
+            # judulnya sendiri di ujung heading_path.
+            dorong_judul(level or 1, teks)
+            berjalan = buka("text", page, section_no=nomor, heading=judul)
+            berjalan.source_items = [i]
+            berjalan.bbox = item.get("bbox")
+            # Baris judul ikut masuk `text`, verbatim, sebagai potongan pertama.
+            #
+            # Alasannya bukan kerapian. Di standar BUMA, judul menyebut istilahnya
+            # ("2.1.3. Physical of Availability (PA)") lalu badan section mulai
+            # dengan "Adalah ..." TANPA mengulang istilahnya. Kalau judul dipisah
+            # dari teks, chunk yang justru mendefinisikan sebuah istilah tidak
+            # memuat istilah itu sama sekali — jadi kalah ranking dari section
+            # perhitungan yang cuma menyebutnya berkali-kali.
+            #
+            # Menaruhnya di sini juga membuat `text` jadi satu-satunya tempat
+            # pengecekan span: `source_wording` bisa dicari harfiah di dalamnya,
+            # tidak perlu melihat dua field.
+            potongan.append(teks)
+            continue
+
+        if berjalan is None:
+            berjalan = buka("text", page)
+            berjalan.bbox = item.get("bbox")
+        berjalan.source_items.append(i)
+        berjalan.page_idxs.append(page)
+        bab = bab_per_halaman.get(page)
+        if bab and bab not in berjalan.chapters:
+            berjalan.chapters.append(bab)
+        potongan.append(teks)          # verbatim, tanpa dirapikan
+
+        # Pengaman ukuran: hanya berlaku untuk dokumen tanpa judul terdeteksi.
+        # Chunk dipotong di batas item, jadi teksnya tetap verbatim.
+        if sum(len(x) for x in potongan) >= MAKS_KARAKTER_CHUNK:
+            lanjutan_dari = berjalan
+            tutup()
+            berjalan = buka("text", page,
+                            section_no=lanjutan_dari.section_no,
+                            heading=lanjutan_dari.heading)
+
+    tutup()
+    return chunks
+
+
+def normalisasi_dari_berkas(content_list: Path, doc_id: str) -> list[Chunk]:
+    items = json.loads(content_list.read_text(encoding="utf-8"))
+    return normalisasi(items, doc_id)

src/knowledge_parsing/parse.py

@@ -0,0 +1,167 @@
+"""Tahap 2 — parsing dokumen dengan MinerU, dengan cache berbasis isi.
+
+Kenapa cache-nya dikunci ke ISI dokumen, bukan ke nomor run: parsing adalah
+tahap termahal (19 dtk/halaman di CPU). Kalau cache-nya per run, tiap kali
+normalizer diubah dan pipeline dijalankan ulang, parsing ikut terulang percuma.
+Dengan kunci isi+pengaturan+versi, dokumen yang sama hanya diparse sekali —
+mau normalizer diubah 20 kali sekalipun.
+
+Kunci cache = sha256(isi berkas) + sidik jari pengaturan + versi MinerU.
+Versi ikut supaya angka hasil MinerU lama tidak diam-diam tercampur dengan
+hasil versi baru — ini penting karena angkanya dipakai untuk justifikasi biaya.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+from .config import PipelineConfig
+
+
+@dataclass
+class ParseResult:
+    doc_id: str
+    source: Path
+    cache_dir: Path          # folder berisi output MinerU apa adanya
+    content_list: Path       # *_content_list.json
+    middle_json: Path | None
+    pages: int
+    seconds: float
+    from_cache: bool
+    backend_tercatat: str | None    # dibaca dari _middle.json, BUKAN dari config
+    mineru_version: str | None
+
+
+def versi_mineru() -> str:
+    try:
+        from mineru.version import __version__  # type: ignore
+        return str(__version__)
+    except Exception:
+        try:
+            from importlib.metadata import version
+            return version("mineru")
+        except Exception:
+            return "unknown"
+
+
+def hash_berkas(path: Path, potong: int = 16) -> str:
+    h = hashlib.sha256()
+    with path.open("rb") as f:
+        for blok in iter(lambda: f.read(1 << 20), b""):
+            h.update(blok)
+    return h.hexdigest()[:potong]
+
+
+def kunci_cache(source: Path, cfg: PipelineConfig) -> str:
+    return f"{hash_berkas(source)}-{cfg.fingerprint()}-{versi_mineru()}"
+
+
+def _cari_output(root: Path) -> tuple[Path | None, Path | None]:
+    """Cari content_list & middle json di bawah root.
+
+    Sengaja pakai glob, bukan path tetap: struktur subfolder MinerU berbeda
+    antar backend ('auto' untuk pipeline, lain untuk vlm). Glob membuat modul
+    ini tidak perlu diubah saat backend diganti.
+    """
+    content = next(iter(sorted(root.rglob("*_content_list.json"))), None)
+    middle = next(iter(sorted(root.rglob("*_middle.json"))), None)
+    return content, middle
+
+
+def _baca_middle(middle: Path | None) -> tuple[str | None, str | None]:
+    """Ambil backend & versi yang BENAR-BENAR dipakai, dari output MinerU."""
+    if not middle or not middle.exists():
+        return None, None
+    try:
+        d = json.loads(middle.read_text(encoding="utf-8"))
+        return d.get("_backend"), d.get("_version_name")
+    except Exception:
+        return None, None
+
+
+def parse_dokumen(source: Path, cfg: PipelineConfig) -> ParseResult:
+    """Parse satu dokumen. Kalau sudah ada di cache, tidak dijalankan ulang."""
+    doc_id = source.stem
+    tujuan = cfg.cache_dir / "parse" / kunci_cache(source, cfg)
+    penanda = tujuan / ".selesai"
+
+    if penanda.exists():
+        content, middle = _cari_output(tujuan)
+        if content:
+            meta = json.loads(penanda.read_text(encoding="utf-8"))
+            backend_tercatat, versi = _baca_middle(middle)
+            return ParseResult(
+                doc_id=doc_id, source=source, cache_dir=tujuan,
+                content_list=content, middle_json=middle,
+                pages=meta.get("pages", 0), seconds=meta.get("seconds", 0.0),
+                from_cache=True,
+                backend_tercatat=backend_tercatat, mineru_version=versi,
+            )
+        shutil.rmtree(tujuan, ignore_errors=True)   # cache rusak, ulangi
+
+    from mineru.cli.common import do_parse, read_fn
+
+    # Playground melakukan ini sebelum parsing; alias seperti "latin"/"en"
+    # diterjemahkan ke kode kanonik. Disamakan supaya hasilnya identik.
+    lang = cfg.lang_kanonik()
+
+    sedang = tujuan.with_suffix(".sedang")
+    shutil.rmtree(sedang, ignore_errors=True)
+    sedang.mkdir(parents=True, exist_ok=True)
+
+    pdf_bytes = read_fn(source)
+    mulai = time.perf_counter()
+    do_parse(
+        output_dir=str(sedang),
+        pdf_file_names=[doc_id],
+        pdf_bytes_list=[pdf_bytes],
+        p_lang_list=[lang],
+        backend=cfg.backend,
+        formula_enable=cfg.formula_enable,
+        table_enable=cfg.table_enable,
+        f_draw_layout_bbox=cfg.write_debug_pdf,
+        f_draw_span_bbox=cfg.write_debug_pdf,
+        start_page_id=cfg.start_page,
+        end_page_id=cfg.end_page,
+        **({"effort": cfg.effort} if cfg.backend == "hybrid" else {}),
+    )
+    detik = time.perf_counter() - mulai
+
+    content, middle = _cari_output(sedang)
+    if content is None:
+        raise RuntimeError(
+            f"MinerU selesai tapi *_content_list.json tidak ditemukan di {sedang}"
+        )
+
+    halaman = _hitung_halaman(source, cfg)
+    sedang.rename(tujuan)
+    content = tujuan / content.relative_to(sedang)
+    middle = tujuan / middle.relative_to(sedang) if middle else None
+
+    penanda.write_text(
+        json.dumps({"pages": halaman, "seconds": detik, "doc_id": doc_id}),
+        encoding="utf-8",
+    )
+    backend_tercatat, versi = _baca_middle(middle)
+
+    return ParseResult(
+        doc_id=doc_id, source=source, cache_dir=tujuan,
+        content_list=content, middle_json=middle,
+        pages=halaman, seconds=detik, from_cache=False,
+        backend_tercatat=backend_tercatat, mineru_version=versi,
+    )
+
+
+def _hitung_halaman(source: Path, cfg: PipelineConfig) -> int:
+    if cfg.end_page is not None:
+        return cfg.end_page - cfg.start_page + 1
+    try:
+        from pypdf import PdfReader
+        return len(PdfReader(str(source)).pages)
+    except Exception:
+        return 0

src/knowledge_parsing/report.py

@@ -0,0 +1,117 @@
+"""Rangkum semua run jadi angka siap pakai untuk justifikasi resource.
+
+    py -m src.knowledge_parsing.report
+    py -m src.knowledge_parsing.report --scale 6800
+
+`--scale` mengekstrapolasi ke jumlah halaman tertentu (mis. skala BUMA
+±6.700-6.800 halaman) memakai detik/halaman yang TERUKUR — dokumen yang
+diambil dari cache tidak ikut dihitung, karena akan membuat angkanya
+terlihat jauh lebih cepat dari kenyataan.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from .config import PipelineConfig
+
+# Di bawah ini, ongkos tetap pemuatan model mendominasi sehingga detik/halaman
+# jauh lebih besar dari kenyataan pada dokumen besar. Terukur di mesin ini:
+# 1 hal -> 54,0 dtk/hal, 5 hal -> 33,2, 20 hal -> 19,0.
+MIN_HALAMAN_SAHIH = 20
+
+
+def _jam(detik: float) -> str:
+    if detik < 90:
+        return f"{detik:.0f} dtk"
+    if detik < 5400:
+        return f"{detik/60:.1f} mnt"
+    return f"{detik/3600:.1f} jam"
+
+
+def kumpulkan(runs_dir: Path) -> list[dict]:
+    hasil = []
+    for m in sorted(runs_dir.glob("*/manifest.json")):
+        try:
+            hasil.append(json.loads(m.read_text(encoding="utf-8")))
+        except (OSError, json.JSONDecodeError) as e:
+            # Manifest rusak/separuh tertulis (run yang mati di tengah). Dilewati,
+            # tapi disebutkan — laporan yang diam-diam kehilangan satu run akan
+            # memberi angka detik/halaman yang salah tanpa ada tandanya.
+            print(f"  ! lewati manifest tidak terbaca: {m} ({type(e).__name__})")
+    return hasil
+
+
+def cetak(runs: list[dict], scale: int | None) -> None:
+    if not runs:
+        print("Belum ada run. Jalankan: py -m src.knowledge_parsing.run --input dokumen/")
+        return
+
+    print(f"{'run':<17} {'backend':<9} {'ver':<8} {'dok':>4} {'hal':>5} "
+          f"{'dtk/hal':>8} {'peringatan':>11}")
+    print("-" * 68)
+
+    for r in runs:
+        s = r.get("ringkasan", {})
+        cfg = r.get("config", {})
+        # backend yang tercatat di output MinerU lebih dipercaya daripada config
+        tercatat = {d.get("backend_tercatat") for d in r.get("dokumen", [])} - {None}
+        backend = "/".join(sorted(tercatat)) if tercatat else cfg.get("backend", "?")
+        dph = s.get("detik_per_halaman")
+        print(f"{r.get('run_id',''):<17} {backend:<9} "
+              f"{str(r.get('mineru_version') or '?')[:7]:<8} "
+              f"{s.get('dokumen_ok',0):>4} {s.get('total_halaman',0):>5} "
+              f"{(f'{dph:.2f}' if dph else '-'):>8} {s.get('total_peringatan',0):>11}")
+
+    terukur = [r for r in runs if (r.get("ringkasan") or {}).get("detik_per_halaman")]
+    if not terukur:
+        print("\n(belum ada run yang benar-benar diparse — semua dari cache)")
+        return
+
+    print("\n--- ekstrapolasi ---")
+    target = scale or 6800
+    meragukan = False
+    for r in terukur:
+        s = r["ringkasan"]
+        dph = s["detik_per_halaman"]
+        hal = s.get("halaman_terukur", 0)
+        cfg = r.get("config", {})
+        tanda = ""
+        if hal < MIN_HALAMAN_SAHIH:
+            tanda = f"  ⚠ hanya {hal} hal."
+            meragukan = True
+        print(f"  {cfg.get('backend','?'):<9} {dph:>6.2f} dtk/hal  →  "
+              f"{target} halaman = {_jam(dph * target)}{tanda}")
+
+    if meragukan:
+        print(
+            f"\n  ⚠ Run dengan < {MIN_HALAMAN_SAHIH} halaman TIDAK layak diekstrapolasi.\n"
+            "    Pemuatan model adalah ongkos tetap, jadi dokumen kecil terlihat jauh\n"
+            "    lebih lambat per halaman. Terukur sebelumnya di mesin ini:\n"
+            "      1 hal -> 54,0 dtk/hal | 5 hal -> 33,2 | 20 hal -> 19,0\n"
+            "    Pakai dokumen >= 20 halaman sebelum angkanya dibawa ke luar."
+        )
+
+    if len(terukur) >= 2:
+        cepat = min(terukur, key=lambda r: r["ringkasan"]["detik_per_halaman"])
+        lambat = max(terukur, key=lambda r: r["ringkasan"]["detik_per_halaman"])
+        a = lambat["ringkasan"]["detik_per_halaman"]
+        b = cepat["ringkasan"]["detik_per_halaman"]
+        if b:
+            print(f"\n  selisih tercepat vs terlambat: {a/b:.1f}×")
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="Ringkasan angka dari semua run")
+    p.add_argument("--runs-dir", type=Path, default=None)
+    p.add_argument("--scale", type=int, default=None,
+                   help="ekstrapolasi ke berapa halaman (mis. 6800)")
+    a = p.parse_args()
+    cfg = PipelineConfig()
+    cetak(kumpulkan(a.runs_dir or cfg.runs_dir), a.scale)
+
+
+if __name__ == "__main__":
+    main()

src/knowledge_parsing/run.py

@@ -0,0 +1,168 @@
+"""Perangkai tahap 1-3: satu folder dokumen masuk, chunk + manifest keluar.
+
+    py -m src.knowledge_parsing.run --input dokumen/
+    py -m src.knowledge_parsing.run --input dokumen/ --backend vlm     # setelah GPU ada
+
+Sifat yang disengaja:
+  - satu dokumen gagal TIDAK menghentikan sisanya (dicatat di failures.jsonl)
+  - bisa dilanjut dari tengah: dokumen yang sudah selesai dilewati (--no-resume
+    untuk memaksa ulang)
+  - output MinerU asli tidak disentuh, hanya ditunjuk dari runs/
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+import traceback
+from datetime import datetime
+from pathlib import Path
+
+from .checks import periksa_items
+from .config import PipelineConfig
+from .contracts import ParsedDocument
+from .manifest import CatatanDokumen, Manifest
+from .normalize import normalisasi
+from .parse import hash_berkas, parse_dokumen, versi_mineru
+
+EKSTENSI = {".pdf", ".docx", ".pptx", ".doc", ".ppt"}
+
+
+def kumpulkan_dokumen(folder: Path) -> list[Path]:
+    return sorted(p for p in folder.rglob("*") if p.suffix.lower() in EKSTENSI)
+
+
+def jalankan(cfg: PipelineConfig) -> Manifest:
+    cfg.periksa()
+    dokumen = kumpulkan_dokumen(cfg.input_dir)
+    if not dokumen:
+        raise SystemExit(f"Tidak ada dokumen di {cfg.input_dir}")
+
+    manifest = Manifest.baru(cfg.to_dict())
+    manifest.mineru_version = versi_mineru()
+
+    run_dir = cfg.runs_dir / manifest.run_id
+    run_dir.mkdir(parents=True, exist_ok=True)
+    failures = run_dir / "failures.jsonl"
+
+    print(f"Run {manifest.run_id} | backend={cfg.backend} | {len(dokumen)} dokumen")
+    print(f"Hasil ke: {run_dir}\n")
+
+    for n, source in enumerate(dokumen, 1):
+        doc_id = source.stem
+        tujuan = run_dir / doc_id
+        chunk_file = tujuan / "chunks.json"
+
+        if cfg.resume and chunk_file.exists():
+            print(f"[{n}/{len(dokumen)}] {doc_id} — dilewati (sudah ada)")
+            continue
+
+        print(f"[{n}/{len(dokumen)}] {doc_id} … ", end="", flush=True)
+        catatan = CatatanDokumen(doc_id=doc_id, source=str(source), status="ok")
+        try:
+            hasil = parse_dokumen(source, cfg)
+            catatan.pages = hasil.pages
+            catatan.detik_parse = hasil.seconds
+            catatan.dari_cache = hasil.from_cache
+            catatan.backend_tercatat = hasil.backend_tercatat
+
+            items = json.loads(hasil.content_list.read_text(encoding="utf-8"))
+            # PDF asli ikut dikirim: tanpa pembanding, angka yang berubah
+            # diam-diam (bug EOQ) tidak akan pernah terdeteksi.
+            catatan.pemeriksaan = periksa_items(
+                items, cfg.min_latex_len,
+                pdf_sumber=source, offset_halaman=cfg.start_page,
+            )
+
+            t0 = time.perf_counter()
+            chunks = normalisasi(items, doc_id)
+            catatan.detik_normalisasi = time.perf_counter() - t0
+            catatan.jumlah_chunk = len(chunks)
+
+            artifact = ParsedDocument(
+                doc_id=doc_id,
+                chunks=chunks,
+                source_path=str(source),
+                content_hash=hash_berkas(source, potong=64),
+                n_pages=hasil.pages,
+                parser_version=hasil.mineru_version or versi_mineru(),
+                # backend dari _middle.json (yang benar-benar jalan), bukan dari config
+                parser_backend=hasil.backend_tercatat or cfg.backend,
+                parser_config=cfg.fingerprint(),
+                created_at=datetime.now().isoformat(timespec="seconds"),
+                raw_output_dir=str(hasil.cache_dir),
+            )
+            tujuan.mkdir(parents=True, exist_ok=True)
+            chunk_file.write_text(
+                artifact.model_dump_json(indent=2),
+                encoding="utf-8",
+            )
+            # Penunjuk ke output MinerU asli — tidak disalin supaya tidak dobel.
+            (tujuan / "sumber-mineru.txt").write_text(
+                str(hasil.cache_dir), encoding="utf-8"
+            )
+
+            tanda = " (cache)" if hasil.from_cache else f" {hasil.seconds:.1f}s"
+            peringatan = catatan.pemeriksaan.get("jumlah_peringatan", 0)
+            tanda += f" | {len(chunks)} chunk"
+            if peringatan:
+                tanda += f" | ⚠ {peringatan} peringatan"
+            print("ok" + tanda)
+
+        except Exception as e:  # satu gagal tidak menghentikan sisanya
+            catatan.status = "gagal"
+            catatan.galat = f"{type(e).__name__}: {e}"
+            print(f"GAGAL — {catatan.galat}")
+            with failures.open("a", encoding="utf-8") as f:
+                f.write(json.dumps({
+                    "doc_id": doc_id, "source": str(source),
+                    "galat": catatan.galat, "trace": traceback.format_exc(),
+                }, ensure_ascii=False) + "\n")
+
+        manifest.dokumen.append(catatan)
+        manifest.simpan(run_dir / "manifest.json")
+
+    r = manifest.ringkasan()
+    print("\n--- ringkasan ---")
+    print(f"  ok / gagal      : {r['dokumen_ok']} / {r['dokumen_gagal']}")
+    print(f"  total halaman   : {r['total_halaman']}")
+    if r["detik_per_halaman"]:
+        print(f"  detik/halaman   : {r['detik_per_halaman']}  "
+              f"(dari {r['halaman_terukur']} hal. yang benar-benar diparse)")
+    if r["total_peringatan"]:
+        print(f"  ⚠ peringatan mutu: {r['total_peringatan']} — lihat manifest.json")
+    print(f"  manifest        : {run_dir / 'manifest.json'}")
+    return manifest
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="Pipeline parsing dokumen (MinerU)")
+    p.add_argument("--input", type=Path, help="folder berisi dokumen")
+    p.add_argument("--backend", choices=["pipeline", "vlm", "hybrid"],
+                   help="pipeline=CPU, vlm=butuh GPU (target resmi)")
+    p.add_argument("--lang", default=None)
+    p.add_argument("--no-resume", action="store_true", help="proses ulang semua")
+    p.add_argument("--start-page", type=int, default=None)
+    p.add_argument("--end-page", type=int, default=None)
+    a = p.parse_args()
+
+    cfg = PipelineConfig()
+    if a.input:
+        cfg.input_dir = a.input
+    if a.backend:
+        cfg.backend = a.backend
+    if a.lang:
+        cfg.lang = a.lang
+    if a.no_resume:
+        cfg.resume = False
+    if a.start_page is not None:
+        cfg.start_page = a.start_page
+    if a.end_page is not None:
+        cfg.end_page = a.end_page
+
+    jalankan(cfg)
+
+
+if __name__ == "__main__":
+    main()