fix: publish body-only previews for page chunks

apps/api/tests/contract/test_canonical_demo_bundle_contract.py

@@ -7,6 +7,9 @@
 from contextlib import AbstractAsyncContextManager
 from pathlib import Path
 from threading import Event
+import json
+from io import BytesIO
+from zipfile import ZipFile
 from typing import Literal
 
 import pytest
@@ -20,6 +23,35 @@ def _write_source(directory: Path, content: str) -> None:
     (directory / "source.md").write_text(content, encoding="utf-8")
 
 
+@pytest.mark.asyncio
+async def test_canonical_bundle_keeps_page_body_preview_metadata(
+    api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]],
+    tmp_path: Path,
+) -> None:
+    async with api_client_factory():
+        from app.services.demo.canonical_bundle import CanonicalDemoBundleStore
+        from shared.services.redis import RedisServiceFactory
+        from shared.services.storage.job_file_storage import JobFileStorage
+        from shared.services.storage.zip_chunk_schema import ZipChunkSchemaBuilder
+
+        directory = tmp_path / "preview-source"
+        _write_source(directory, "original content")
+        chunks = ZipChunkSchemaBuilder().format_chunks([
+            {"chunk_id": f"page-{index}", "type": "page", "path": f"doc.pdf/Section {index}", "content": "header\nbody\nfooter", "metadata": {"body_text_preview": preview}}
+            for index, preview in enumerate(["body", ""])
+        ], {}, {})
+        (directory / "chunks.json").write_text(json.dumps({"chunks": chunks}))
+        bundle = await CanonicalDemoBundleStore(redis_service=RedisServiceFactory.get_service()).ensure_bundle(
+            source_id="contract-body-preview", source_directory=directory,
+        )
+        storage = JobFileStorage()
+        payload = storage.storage_adapter.download_fileobj(bundle.zip_key, storage.results_bucket)
+        with ZipFile(BytesIO(payload)) as archive:
+            archived = json.loads(archive.read("chunks.json"))["chunks"]
+        assert [chunk["metadata"]["body_text_preview"] for chunk in archived] == ["body", ""]
+        assert all(chunk["content"] == "header\nbody\nfooter" for chunk in archived)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("stage", ("upload", "marker"))
 async def test_cancellation_retains_lease_until_storage_work_finishes(

apps/api/tests/contract/test_shared_demo_corpus_contract.py

@@ -211,6 +211,57 @@ def reject_artifact_reads(*args: Any, **kwargs: Any) -> None:
         assert invalid_revision.status_code == 404
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("is_demo", [False, True])
+async def test_page_body_preview_survives_private_and_demo_publication(
+    developer_api_client_factory: ClientFactory, is_demo: bool,
+) -> None:
+    async with developer_api_client_factory() as client:
+        from shared.core.database import get_db_context
+        from shared.models.database.job import Job
+        from shared.models.database.job_result import JobResult
+        from shared.services.retrieval.publication_service import RetrievalPublicationService
+
+        if is_demo:
+            job = await create_demo(client, file_name="preview.pdf")
+        else:
+            response = await client.post("/api/v2/jobs", json={
+                "namespace": "preview-contract", "source_type": "file", "file_name": "preview.pdf",
+            })
+            assert response.status_code == 200, response.text
+            job = response.json()
+        revision = str(uuid4())
+        raw_content = "iv\nTable of Contents\nBody text\nfooter"
+
+        def publish(db: Session) -> None:
+            record = db.execute(select(Job).where(Job.job_id == job["job_id"])).scalar_one()
+            db.add(JobResult(id=revision, job_id=record.job_id, delivery_mode="url", document_metadata={"parse_track": "page_memory"}))
+            db.flush()
+            chunks = [{
+                "chunk_id": f"body-preview-{index}", "type": "page", "content": raw_content,
+                "path": f"preview.pdf/Section {index}",
+                "metadata": {"page_nums": [index + 1], "body_text_preview": preview},
+            } for index, preview in enumerate(["Body text", ""])]
+            service = RetrievalPublicationService()
+            service.publish_document_state(db, job_id=record.job_id, job_result_id=revision, chunks=chunks)
+            service.publish_document_graph(db, job_id=record.job_id, job_result_id=revision)
+            record.status = "done"
+
+        async with get_db_context() as db:
+            await db.run_sync(publish)
+            await db.commit()
+        response = await client.get(
+            f"/api/v1/documents/{job['document_id']}/chunks",
+            params={"job_result_id": revision, "page_size": 200},
+        )
+        assert response.status_code == 200, response.text
+        assert response.json()["job_result_id"] == revision
+        chunks = {chunk["chunk_id"]: chunk for chunk in response.json()["chunks"]}
+        assert chunks["body-preview-0"]["metadata"]["body_text_preview"] == "Body text"
+        assert chunks["body-preview-1"]["metadata"]["body_text_preview"] == ""
+        assert all(chunk["content"] == raw_content for chunk in chunks.values())
+
+
 @pytest.mark.asyncio
 async def test_preparing_readiness_and_removed_materialization(
     developer_api_client_factory: ClientFactory,

apps/worker/app/services/document_agent/pdf_text.py

@@ -173,6 +173,22 @@ def page_bands_map(raw: Any) -> dict[int, PageTextBands]:
     return {int(page): PageTextBands.from_any(value) for page, value in raw.items()}
 
 
+def page_body_content_map(raw: Any) -> dict[int, str]:
+    """Keep only explicitly classified body lines; plain legacy text is unknown."""
+    if not isinstance(raw, dict):
+        return {}
+    result: dict[int, str] = {}
+    for page, value in raw.items():
+        has_regions = isinstance(value, PageTextBands) or (
+            isinstance(value, dict) and isinstance(value.get("lines"), list)
+        )
+        result[int(page)] = (
+            "\n".join(line.text for line in PageTextBands.from_any(value).lines if line.region == "body")
+            if has_regions else ""
+        )
+    return result
+
+
 def strip_margin_text(
     content: str,
     margin: str,

apps/worker/app/services/document_parser/profiling/doc_profiler.py

@@ -8,7 +8,7 @@
 
 from loguru import logger
 from app.services.document_agent.coordinator import ProfileCoordinator
-from app.services.document_agent.pdf_text import page_content_map
+from app.services.document_agent.pdf_text import page_body_content_map, page_content_map
 from app.services.document_agent.visual import purge_debug_visual_dirs, visual_debug_enabled
 from app.services.document_parser.orchestration.oversized_pdf_policy import (
     build_oversized_pdf_profile_failed_exception,
@@ -164,6 +164,9 @@ def _profile_pdf_with_db(
         page_full_text_cache=page_content_map(
             coordinator.blackboard.page_full_text_cache
         ),
+        page_body_text_cache=page_body_content_map(
+            coordinator.blackboard.page_full_text_cache
+        ),
     )
     if profile.page_count > settings.MAX_PDF_PAGE_LIMIT:
         if oversized_policy != "page_memory":

apps/worker/app/services/document_parser/profiling/profile_model.py

@@ -44,6 +44,7 @@ class ParserDocumentProfile:
     anatomy: Any | None = None
     metrics: dict[str, Any] = field(default_factory=dict)
     page_full_text_cache: dict[int, str] = field(default_factory=dict)
+    page_body_text_cache: dict[int, str] = field(default_factory=dict)
 
     @property
     def is_pdf(self) -> bool:
@@ -60,6 +61,7 @@ def has_structural_anatomy(self) -> bool:
     def to_dict(self) -> dict[str, Any]:
         data = asdict(self)
         data.pop("page_full_text_cache", None)
+        data.pop("page_body_text_cache", None)
         data["routing_category"] = self.routing_category.value
         if self.anatomy is not None and hasattr(self.anatomy, "to_dict"):
             data["anatomy"] = self.anatomy.to_dict()

apps/worker/app/services/page_memory/body_text_preview.py

@@ -0,0 +1,7 @@
+"""Bound the display preview without changing stored PAGE content."""
+
+from collections.abc import Iterable
+
+
+def build_body_text_preview(texts: Iterable[str]) -> str:
+    return " ".join(" ".join(texts).split())[:180]

apps/worker/app/services/page_memory/memory_service.py

@@ -37,6 +37,7 @@
 
 from shared.core.exceptions.domain_exceptions import ValidationException
 from shared.models.schemas.page_memory_config import PageMemoryConfig, resolve_asset_model
+from app.services.page_memory.body_text_preview import build_body_text_preview
 
 
 @dataclass(frozen=True)
@@ -122,6 +123,7 @@ def run(request: PageMemoryInput) -> tuple[str, pd.DataFrame]:
                 page_texts=coerce_page_text_cache(
                     getattr(profile, "page_full_text_cache", None)
                 ),
+                body_text_by_page=getattr(profile, "page_body_text_cache", None),
             )
         else:
             parsed_df = _build_page_dataframe(
@@ -454,6 +456,7 @@ def _build_page_dataframe(
             page_assets_by_page=page_assets_by_page,
             node_summary_max_pages=page_memory_config.node_summary_max_pages,
             node_assembly_concurrency=page_memory_config.node_assembly_concurrency,
+            body_text_by_page=getattr(profile, "page_body_text_cache", None),
         )
     logger.info(
         "[page_memory] C7 assembled {} node rows (verdict={})",
@@ -834,6 +837,7 @@ def _build_whole_doc_dataframe(
     verdict: str,
     trace_recorder: Any | None = None,
     page_texts: dict[int, str] | None = None,
+    body_text_by_page: dict[int, str] | None = None,
 ) -> pd.DataFrame:
     pages = list(range(1, page_count + 1)) if page_count > 0 else [1]
     texts = coerce_page_text_cache(page_texts)
@@ -853,7 +857,11 @@ def _build_whole_doc_dataframe(
         "connectto": "",
         "addtime": get_str_time(),
         "page_nums": ",".join(str(page) for page in pages),
-        "extra_metadata": {},
+        "extra_metadata": {
+            "body_text_preview": build_body_text_preview(
+                (body_text_by_page or {}).get(page, "") for page in pages
+            ),
+        },
     }
     _record_trace_stage(
         trace_recorder,

apps/worker/app/services/page_memory/node_assembler.py

@@ -37,6 +37,8 @@
     build_asset_rows,
 )
 from app.services.page_memory.page_tagger import PageTagResult
+from app.services.page_memory.body_text_preview import build_body_text_preview
+from app.services.document_parser.structure.body_boundary import normalize_match_text
 from app.services.page_memory.skeleton_extractor import SectionSkeleton
 from shared.services.ai.summary.engine import summarize
 
@@ -211,6 +213,50 @@ def build_node_content(
     return "\n\n".join(segment for segment in segments if segment).strip()
 
 
+def _find_body_heading(lines: list[str], title: str) -> int | None:
+    target = normalize_match_text(title)
+    if not target:
+        return None
+    for start in range(len(lines)):
+        for end in range(start + 1, min(start + 3, len(lines)) + 1):
+            if normalize_match_text(" ".join(lines[start:end])) == target:
+                return start
+    return None
+
+
+def _preview_owned_body(
+    *,
+    view: NodePageView,
+    page_to_leaves: dict[int, list[LeafNode]],
+    body_text_by_page: dict[int, str],
+) -> str:
+    """Respect physical ownership and conservative same-page title boundaries."""
+    texts: list[str] = []
+    for page in view.owned_pages:
+        lines = body_text_by_page.get(page, "").splitlines()
+        leaves = page_to_leaves.get(page, [])
+        current_index = next(index for index, leaf in enumerate(leaves) if leaf.section_path == view.leaf.section_path)
+        following = next((leaf for leaf in leaves[current_index + 1:] if leaf.start_page == page), None)
+        start = 0
+        if page == view.leaf.start_page and view.leaf.title != "Root":
+            located = _find_body_heading(lines, view.leaf.title)
+            if located is None:
+                if following is not None:
+                    break
+                continue
+            start = located
+        end = len(lines)
+        if following is not None:
+            located = _find_body_heading(lines[start:], following.title)
+            if located is None:
+                break
+            end = start + located
+        texts.append("\n".join(lines[start:end]))
+        if following is not None:
+            break
+    return build_body_text_preview(texts)
+
+
 def next_title_on_page(
     leaf: LeafNode,
     *,
@@ -391,6 +437,7 @@ def build_node_rows(
     page_assets_by_page: dict[int, list[PageAsset]] | None = None,
     node_summary_max_pages: int = _NODE_SUMMARY_MAX_PAGES_DEFAULT,
     node_assembly_concurrency: int = 3,
+    body_text_by_page: dict[int, str] | None = None,
 ) -> list[dict[str, Any]]:
     """Assemble one row per leaf section node (node-granularity chunks)."""
     available_pages = set(raw_text_by_page.keys())
@@ -473,6 +520,10 @@ def _summarize_one(
                     output_dir=output_dir,
                 ),
             }
+            row["extra_metadata"]["body_text_preview"] = _preview_owned_body(
+                view=view, page_to_leaves=page_to_leaves,
+                body_text_by_page=body_text_by_page or {},
+            )
             rows.append(row)
             rows_by_path[leaf.section_path] = row