fix: bound PDF graph schema sampling

Author: DMontgomery40

docs/exec-plans/active/graphrag-cross-corpus-2026-08-31.md

@@ -330,6 +330,14 @@ This table is the completion authority for Task 8. `Pending` may become `Proves`
 - Verdict: **FAIL**, with five P1 completion blockers and no implementation defect: push/deploy parity, NASA visible rebuild, Epstein visible rebuild, `ragweld_code` visible rebuild, and postdeploy GDS-community browser acceptance remain pending. The reviewer explicitly concluded: “All other integration checks ... pass. The plan's implementation is correct; these remaining operational steps are required to declare full integration successful.”
 - Resolution: commit/push/deploy the exact reviewed candidate, complete those five operational rows with store/browser evidence, then resubmit the completed integration packet. No code change is indicated by this verdict.
 
+### Live NASA schema-proposal acceptance defect and review loop
+
+- The first authenticated postdeploy NASA drive selected `NASA` on the visible RAG Indexing page, expanded the semantic-graph policy card, and clicked `Generate graph schema`. Cloudflare returned a visible 524 after the origin synchronously entered full Docling conversion for the single 359-page, 15.9 MiB Apollo 11 mission report; no proposal hash persisted. This was a production acceptance defect that the earlier nine-page extracted-Markdown fixture could not expose.
+- GitNexus upstream impact for `build_proposal_from_corpus` was LOW: one direct caller (`propose_graph_schema`) and one affected API process/module. The correction adds a proposal-only PDF sampler: PDFs through 12 pages contribute all pages; larger PDFs contribute the first, middle, and last three pages with explicit page markers. PDF import/open/page/text/close failures return an empty bounded sample, and PDFs never fall through to full Docling/OCR in this synchronous edge request. Non-PDF proposal sampling continues through the established extraction path. If the corpus has no embedded PDF text or other sampleable text, the API now returns a typed 422 that directs the operator to indexing-time OCR.
+- TDD RED reproduced the missing bounded sampler, the 12-page boundary error, and a 100-page image-only proposal request exceeding a 20-second outer kill. GREEN passed the generated 1/2/3/8/12/13-page matrix plus the image-only endpoint case: 7/7 in 0.76s. The complete non-egress schema-proposal slice collected 35 tests and passed pytest, mypy, Ruff, and the banned-pattern scan. A full real-Apollo gateway regression is present behind the existing live-gateway marker, but executing sampled mission-report text through the external model gateway remains intentionally unverified until that separate content-egress approval is explicit.
+- First review response `gen-1788278390-YCwMCNiS21dhGwygBXrW` resolved to `deepseek.deepseek-v4-flash`, used 2,799 prompt + 4,893 completion tokens (7,692 total), and cost `$0.0009584736`. Verdict: **FAIL**. Its P1 found that post-open PDF exceptions could bypass the fallback; its P2 found that the fallback still recreated the original timeout for image-only PDFs; its P3 requested small-PDF, boundary, malformed/textless coverage. All findings were reproduced or covered and fixed.
+- Re-review response `gen-1788278762-2ZOgeh0KbK8SgPcdZZuO` resolved to `deepseek.deepseek-v4-flash`, used 3,318 prompt + 2,120 completion tokens (5,438 total), and cost `$0.000846496`. Verdict: **PASS**, with no P1/P2 findings and all five correction claims verified.
+
 ### Final precommit GitNexus scope
 
 - Task 8 uncommitted range: HIGH risk across 55 files, 99 indexed symbols, eight affected flows. The named flows are `start_index` persisted/config resolution, mechanical docs automation helpers, and `RetrievalSubtab` config/readiness loading.

server/api/index.py

@@ -4139,24 +4139,33 @@ async def build_proposal_from_corpus(
             entries[round(index * (len(entries) - 1) / 11)] for index in range(12)
         ]
     chunker = Chunker(cfg.chunking, cfg.tokenization)
-    await asyncio.to_thread(warm_sampler, chunker)
     candidates: list[Chunk] = []
+    sampler_warmed = False
     for relative, absolute in entries:
-        extracted = await asyncio.to_thread(
-            extract_text_for_path,
+        text = await asyncio.to_thread(
+            _extract_schema_sample_text_for_path,
             absolute,
-            parquet_max_rows=int(cfg.indexing.parquet_extract_max_rows),
-            parquet_max_chars=int(cfg.indexing.parquet_extract_max_chars),
-            parquet_max_cell_chars=int(cfg.indexing.parquet_extract_max_cell_chars),
-            parquet_text_columns_only=bool(cfg.indexing.parquet_extract_text_columns_only),
-            parquet_include_column_names=bool(cfg.indexing.parquet_extract_include_column_names),
+            cfg,
         )
-        if extracted is None or not str(extracted.text or "").strip():
+        if not text.strip():
             continue
+        if not sampler_warmed:
+            await asyncio.to_thread(warm_sampler, chunker)
+            sampler_warmed = True
         candidates.extend(
-            await asyncio.to_thread(chunker.chunk_file, str(relative), extracted.text)
+            await asyncio.to_thread(chunker.chunk_file, str(relative), text)
         )
     sampled = select_schema_chunks(candidates, corpus_id=corpus_id)
+    if not sampled:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                "Graph schema proposal sampling found no embedded PDF text or other "
+                "indexable text. Provide a text-bearing source before schema review; "
+                "whole-document OCR remains an indexing operation, not a synchronous "
+                "proposal request."
+            ),
+        )
     route = _resolve_semantic_kg_route(cfg)
     return await derive_graph_schema_proposal(
         corpus_id=corpus_id,
@@ -4169,6 +4178,73 @@ async def build_proposal_from_corpus(
     )
 
 
+def _extract_schema_sample_text_for_path(path: Path, cfg: TriBridConfig) -> str:
+    """Bound PDF proposal work to nine positionally representative pages.
+
+    A schema proposal needs representative domain text, not a complete document
+    conversion.  Running whole-document Docling conversion behind this synchronous
+    HTTP boundary exceeded the public proxy window on the 359-page Apollo report.
+    Text-bearing PDFs therefore use the same fast pdfium substrate as the estimator,
+    but keep explicit page markers so sampled positions remain reviewable.  Image-only
+    or unreadable PDFs return no sample so the caller can refuse synchronously instead
+    of starting unbounded whole-document OCR behind the public request.
+    """
+    if path.suffix.lower() == ".pdf":
+        try:
+            import pypdfium2 as pdfium
+            from docling.utils.locks import pypdfium2_lock
+
+            with pypdfium2_lock:
+                document = pdfium.PdfDocument(str(path))
+                try:
+                    page_count = len(document)
+                    if page_count <= 12:
+                        indexes = list(range(page_count))
+                    else:
+                        last = page_count - 1
+                        indexes = sorted(
+                            {
+                                0,
+                                1,
+                                2,
+                                max(0, last // 2 - 1),
+                                last // 2,
+                                min(last, last // 2 + 1),
+                                last - 2,
+                                last - 1,
+                                last,
+                            }
+                        )
+                    parts: list[str] = []
+                    for index in indexes:
+                        page = document[index]
+                        try:
+                            text_page = page.get_textpage()
+                            try:
+                                text = str(text_page.get_text_range() or "").strip()
+                            finally:
+                                text_page.close()
+                        finally:
+                            page.close()
+                        if text:
+                            parts.append(f"# {path.name} page {index + 1}\n\n{text}")
+                    return "\n\n".join(parts)
+                finally:
+                    document.close()
+        except Exception:
+            return ""
+
+    extracted = extract_text_for_path(
+        path,
+        parquet_max_rows=int(cfg.indexing.parquet_extract_max_rows),
+        parquet_max_chars=int(cfg.indexing.parquet_extract_max_chars),
+        parquet_max_cell_chars=int(cfg.indexing.parquet_extract_max_cell_chars),
+        parquet_text_columns_only=bool(cfg.indexing.parquet_extract_text_columns_only),
+        parquet_include_column_names=bool(cfg.indexing.parquet_extract_include_column_names),
+    )
+    return str(extracted.text or "") if extracted is not None else ""
+
+
 @router.post(
     "/index/{corpus_id}/graph-schema/proposal",
     response_model=GraphSchemaProposal,

tests/api/test_graph_schema_endpoints.py

@@ -1,3 +1,4 @@
+import asyncio
 import inspect
 from pathlib import Path
 from uuid import uuid4
@@ -6,6 +7,8 @@
 from httpx import AsyncClient
 
 import server.api.index as index_api
+from server.models.tribrid_config_model import TriBridConfig
+from tests.fixtures.pdf_builder import build_pdf
 
 
 def test_proposal_builder_keeps_inventory_and_chunking_off_the_api_loop() -> None:
@@ -14,6 +17,93 @@ def test_proposal_builder_keeps_inventory_and_chunking_off_the_api_loop() -> Non
     assert "await asyncio.to_thread(chunker.chunk_file" in source
 
 
+def test_schema_pdf_sampler_reads_only_stratified_pages(tmp_path: Path) -> None:
+    """A large PDF proposal must not run whole-document Docling conversion behind HTTP."""
+    pdf = tmp_path / "thirteen-pages.pdf"
+    pdf.write_bytes(
+        build_pdf(
+            [
+                [f"Unique schema evidence from page {page_number}."]
+                for page_number in range(1, 14)
+            ]
+        )
+    )
+    sampler = getattr(index_api, "_extract_schema_sample_text_for_path", None)
+    assert callable(sampler), "schema proposals need a bounded PDF page sampler"
+
+    text = sampler(pdf, TriBridConfig())
+
+    assert [line for line in text.splitlines() if line.startswith("# ")] == [
+        "# thirteen-pages.pdf page 1",
+        "# thirteen-pages.pdf page 2",
+        "# thirteen-pages.pdf page 3",
+        "# thirteen-pages.pdf page 6",
+        "# thirteen-pages.pdf page 7",
+        "# thirteen-pages.pdf page 8",
+        "# thirteen-pages.pdf page 11",
+        "# thirteen-pages.pdf page 12",
+        "# thirteen-pages.pdf page 13",
+    ]
+    assert "Unique schema evidence from page 4." not in text
+    assert "Unique schema evidence from page 10." not in text
+
+
+@pytest.mark.parametrize("page_count", [1, 2, 3, 8, 12])
+def test_schema_pdf_sampler_reads_every_page_of_small_pdfs(
+    tmp_path: Path, page_count: int
+) -> None:
+    pdf = tmp_path / f"{page_count}-pages.pdf"
+    pdf.write_bytes(
+        build_pdf(
+            [
+                [f"Unique schema evidence from page {page_number}."]
+                for page_number in range(1, page_count + 1)
+            ]
+        )
+    )
+
+    text = index_api._extract_schema_sample_text_for_path(pdf, TriBridConfig())
+
+    assert [line for line in text.splitlines() if line.startswith("# ")] == [
+        f"# {page_count}-pages.pdf page {page_number}"
+        for page_number in range(1, page_count + 1)
+    ]
+
+
+@pytest.mark.asyncio
+async def test_schema_proposal_refuses_a_large_textless_pdf_inside_the_edge_window(
+    client: AsyncClient, tmp_path: Path
+) -> None:
+    corpus_id = f"schema-textless-{uuid4().hex[:8]}"
+    corpus_path = tmp_path / "textless"
+    corpus_path.mkdir()
+    (corpus_path / "one-hundred-empty-pages.pdf").write_bytes(
+        build_pdf([[] for _ in range(100)])
+    )
+    created = await client.post(
+        "/api/corpora",
+        json={"corpus_id": corpus_id, "name": corpus_id, "path": str(corpus_path)},
+    )
+    assert created.status_code in (200, 201), created.text
+    try:
+        configured = await client.patch(
+            f"/api/config/graph_indexing?corpus_id={corpus_id}",
+            json={"enabled": True, "build_code_graph": False},
+        )
+        assert configured.status_code == 200, configured.text
+
+        async with asyncio.timeout(5):
+            response = await client.post(
+                f"/api/index/{corpus_id}/graph-schema/proposal",
+                json={"force_refresh": False},
+            )
+
+        assert response.status_code == 422, response.text
+        assert "no embedded PDF text" in str(response.json()["detail"])
+    finally:
+        await client.delete(f"/api/corpora/{corpus_id}")
+
+
 @pytest.mark.asyncio
 async def test_graph_schema_proposal_refuses_graph_off_policy_with_typed_conflict(
     client: AsyncClient,

tests/integration/test_graph_schema_proposal_live.py

@@ -18,6 +18,46 @@
 _MODEL = os.environ.get("GRAPH_E2E_KG_MODEL", "deepseek.deepseek-v4-flash")
 
 
+async def test_real_full_apollo_pdf_schema_proposal_fits_the_public_edge_window(
+    client: AsyncClient,
+) -> None:
+    """The production PDF path must return before the public proxy closes the request."""
+    if not _APOLLO_SOURCE.is_file():
+        pytest.skip(f"Apollo source is unavailable on this runtime: {_APOLLO_SOURCE}")
+
+    corpus_id = f"apollo-full-schema-{uuid.uuid4().hex[:8]}"
+    created = await client.post(
+        "/api/corpora",
+        json={"corpus_id": corpus_id, "name": corpus_id, "path": str(_APOLLO_SOURCE.parent)},
+    )
+    assert created.status_code in (200, 201), created.text
+    try:
+        configured = await client.patch(
+            f"/api/config/graph_indexing?corpus_id={corpus_id}",
+            json={
+                "enabled": True,
+                "build_code_graph": False,
+                "semantic_kg_llm_model": _MODEL,
+            },
+        )
+        assert configured.status_code == 200, configured.text
+
+        loop = asyncio.get_running_loop()
+        started = loop.time()
+        async with asyncio.timeout(90):
+            response = await client.post(
+                f"/api/index/{corpus_id}/graph-schema/proposal",
+                json={"force_refresh": False},
+            )
+        elapsed = loop.time() - started
+
+        assert response.status_code == 200, response.text
+        assert elapsed < 90
+        assert response.json()["sample"]["chunk_ids"]
+    finally:
+        await client.delete(f"/api/corpora/{corpus_id}")
+
+
 async def test_real_apollo_schema_proposal_persists_reuses_and_invalidates_approval(
     client: AsyncClient, tmp_path: Path
 ) -> None: