fix(file_processors): Preserve Docling chunk structure metadata (#6398)

# What does this PR do?

Fixes #6396.

Docling returns useful information for each chunk, such as section
headings and page numbers. OGX was reading some of this information from
the wrong place, so it was lost during file processing.

This PR fixes that for:

- local Docling;
- Docling Serve sync processing; and
- Docling Serve async processing.

The values are stored as existing chunk attributes. Headings and page
numbers are converted to strings so they work with the current vector
store search response.

For example:

```json
{
  "headings": "Installation > Database setup",
  "page_numbers": "4, 5"
}
```

The `>` separator keeps the heading order clear. A comma inside a
heading is not treated as a new heading.

The old nested Docling heading format is still supported as a fallback.

No new API field is added. Existing user attributes continue to work.
Applications can use the filename, heading, and page number to create
citations such as:

```text
manual.pdf, page 4, "Installation > Database setup"
```

## Related discussion

#6399 proposed returning a separate structured metadata field from
vector store search.

During review, we agreed to use the existing `attributes` field with
string values instead. This PR follows that decision, so it does not
depend on #6399.

## Test Plan

Run the focused unit tests:

```bash
uv run pytest -q \
  tests/unit/providers/file_processor/test_docling_serve.py \
  tests/unit/providers/file_processor/test_docling_metadata.py
```

Output:

```text
30 passed, 1 warning
```

The warning is an existing `AsyncMock` warning in the IBM SaaS
compatibility test.

Run the repository checks for the changed files:

```bash
uv run pre-commit run --files \
  src/ogx/providers/utils/files/structural_metadata.py \
  src/ogx/providers/inline/file_processor/docling/_metadata.py \
  src/ogx/providers/remote/file_processor/docling_serve/docling_serve.py \
  tests/unit/providers/file_processor/test_docling_metadata.py \
  tests/unit/providers/file_processor/test_docling_serve.py
```

All checks passed.

I also tested this with a running OGX stack using Docling Serve and
PGVector. A search result for text on the second page returned:

```json
{
  "file_id": "file-31dffd91a4be4f1196cac053379c9fbe",
  "filename": "precise-citation.pdf",
  "headings": "Precise Citations, Chapter Two",
  "page_numbers": "2",
  "verification": "citation-backward-compatibility"
}
```

This was enough to create the following document, section, and page
citation:

```text
precise-citation.pdf, page 2, "Precise Citations, Chapter Two"
```

---------

Signed-off-by: Peter Gustafsson <peter.gustafsson6@gmail.com>
Co-authored-by: Peter Gustafsson <peter.gustafsson6@gmail.com>

Author: pgustafs

src/ogx/providers/inline/file_processor/docling/_metadata.py

@@ -0,0 +1,31 @@
+# Copyright (c) The OGX Contributors.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from typing import Any
+
+from ogx.providers.utils.files.structural_metadata import structural_metadata_as_attributes
+
+
+def extract_structural_metadata(doc_chunk: Any) -> dict[str, Any]:
+    chunk_meta = getattr(doc_chunk, "meta", None)
+    if chunk_meta is None:
+        return {}
+
+    headings = getattr(chunk_meta, "headings", None)
+    legacy_headings = getattr(doc_chunk, "headings", None)
+    page_numbers = {
+        page_number
+        for doc_item in getattr(chunk_meta, "doc_items", [])
+        for provenance in (getattr(doc_item, "prov", None) or [])
+        if (page_number := getattr(provenance, "page_no", None)) is not None
+    }
+    metadata: dict[str, Any] = structural_metadata_as_attributes(
+        headings=headings,
+        page_numbers=sorted(page_numbers),
+    )
+    if not headings and legacy_headings:
+        metadata["headings"] = legacy_headings
+    return metadata

src/ogx/providers/inline/file_processor/docling/docling.py

@@ -34,6 +34,7 @@
     VectorStoreChunkingStrategy,
 )
 
+from ._metadata import extract_structural_metadata
 from .config import DoclingFileProcessorConfig
 
 log = get_logger(name=__name__, category="providers::file_processors")
@@ -277,7 +278,6 @@ def _create_chunks(
             if not text or not text.strip():
                 continue
 
-            headings = getattr(doc_chunk, "headings", None)
             chunk_window = f"{i}"
 
             chunk_id = generate_chunk_id(document_id, text, chunk_window)
@@ -286,8 +286,7 @@ def _create_chunks(
                 "document_id": document_id,
                 **document_metadata,
             }
-            if headings:
-                meta["headings"] = headings
+            meta.update(extract_structural_metadata(doc_chunk))
 
             chunks.append(
                 Chunk(

src/ogx/providers/remote/file_processor/docling_serve/docling_serve.py

@@ -19,6 +19,7 @@
 
 from ogx.log import get_logger
 from ogx.providers.utils.files.response import response_body_bytes
+from ogx.providers.utils.files.structural_metadata import structural_metadata_as_attributes
 from ogx.providers.utils.vector_io.vector_utils import generate_chunk_id
 from ogx_api.common.errors import InvalidParameterError
 from ogx_api.file_processors import ProcessFileRequest, ProcessFileResponse
@@ -339,9 +340,18 @@ async def _convert_and_chunk(
                 **document_metadata,
             }
 
-            headings = raw_chunk.get("meta", {}).get("headings", None)
-            if headings:
-                meta["headings"] = headings
+            legacy_meta = raw_chunk.get("meta") or {}
+            headings = raw_chunk.get("headings")
+            legacy_headings = legacy_meta.get("headings")
+            page_numbers = raw_chunk.get("page_numbers") or legacy_meta.get("page_numbers")
+            meta.update(
+                structural_metadata_as_attributes(
+                    headings=headings,
+                    page_numbers=page_numbers,
+                )
+            )
+            if not headings and legacy_headings:
+                meta["headings"] = legacy_headings
 
             chunks.append(
                 Chunk(
@@ -418,13 +428,18 @@ async def _convert_and_chunk_async(
                 **document_metadata,
             }
 
-            # Extract headings from meta object
-            headings = None
-            if hasattr(raw_chunk, "meta") and hasattr(raw_chunk.meta, "headings"):
-                headings = raw_chunk.meta.headings
-
-            if headings:
-                meta["headings"] = headings
+            legacy_meta = getattr(raw_chunk, "meta", None)
+            headings = getattr(raw_chunk, "headings", None)
+            legacy_headings = getattr(legacy_meta, "headings", None)
+            page_numbers = getattr(raw_chunk, "page_numbers", None) or getattr(legacy_meta, "page_numbers", None)
+            meta.update(
+                structural_metadata_as_attributes(
+                    headings=headings,
+                    page_numbers=page_numbers,
+                )
+            )
+            if not headings and legacy_headings:
+                meta["headings"] = legacy_headings
 
             chunks.append(
                 Chunk(

src/ogx/providers/utils/files/structural_metadata.py

@@ -0,0 +1,28 @@
+# Copyright (c) The OGX Contributors.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from typing import Any
+
+
+def _attribute_value(value: Any, separator: str) -> str:
+    if isinstance(value, list | tuple):
+        return separator.join(str(item) for item in value if str(item))
+    return str(value) if value is not None else ""
+
+
+def structural_metadata_as_attributes(*, headings: Any = None, page_numbers: Any = None) -> dict[str, str]:
+    """Convert structural chunk metadata to scalar vector-store attributes."""
+    metadata: dict[str, str] = {}
+
+    headings_value = _attribute_value(headings, " > ")
+    if headings_value:
+        metadata["headings"] = headings_value
+
+    page_numbers_value = _attribute_value(page_numbers, ", ")
+    if page_numbers_value:
+        metadata["page_numbers"] = page_numbers_value
+
+    return metadata

tests/unit/providers/file_processor/test_docling_metadata.py

@@ -0,0 +1,87 @@
+# Copyright (c) The OGX Contributors.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from types import SimpleNamespace
+
+from ogx.providers.inline.file_processor.docling._metadata import extract_structural_metadata
+from ogx.providers.utils.files.structural_metadata import structural_metadata_as_attributes
+from ogx_api.vector_io import VectorStoreContent, VectorStoreSearchResponse
+
+
+def test_extract_structural_metadata_from_native_docling_chunk():
+    doc_chunk = SimpleNamespace(
+        headings=["wrong location"],
+        meta=SimpleNamespace(
+            headings=["Introduction", "Architecture"],
+            doc_items=[
+                SimpleNamespace(prov=[SimpleNamespace(page_no=2), SimpleNamespace(page_no=1)]),
+                SimpleNamespace(prov=[SimpleNamespace(page_no=2)]),
+                SimpleNamespace(prov=None),
+            ],
+        ),
+    )
+
+    assert extract_structural_metadata(doc_chunk) == {
+        "headings": "Introduction > Architecture",
+        "page_numbers": "1, 2",
+    }
+
+
+def test_extract_structural_metadata_omits_empty_values():
+    doc_chunk = SimpleNamespace(meta=SimpleNamespace(headings=None, doc_items=[]))
+
+    assert extract_structural_metadata(doc_chunk) == {}
+
+
+def test_extract_structural_metadata_preserves_legacy_top_level_headings():
+    doc_chunk = SimpleNamespace(
+        headings=["Legacy heading"],
+        meta=SimpleNamespace(headings=None, doc_items=[]),
+    )
+
+    assert extract_structural_metadata(doc_chunk) == {
+        "headings": ["Legacy heading"],
+    }
+
+
+def test_structural_metadata_attributes_keep_commas_inside_headings():
+    assert structural_metadata_as_attributes(
+        headings=["Safety, Security", "Database setup"],
+        page_numbers=[4, 5],
+    ) == {
+        "headings": "Safety, Security > Database setup",
+        "page_numbers": "4, 5",
+    }
+
+
+def test_structural_metadata_attributes_preserve_existing_strings():
+    assert structural_metadata_as_attributes(
+        headings="Database setup",
+        page_numbers="4, 5",
+    ) == {
+        "headings": "Database setup",
+        "page_numbers": "4, 5",
+    }
+
+
+def test_structural_metadata_is_valid_in_vector_store_search_attributes():
+    metadata = structural_metadata_as_attributes(
+        headings=["Installation", "Database setup"],
+        page_numbers=[4, 5],
+    )
+
+    result = VectorStoreSearchResponse(
+        file_id="file-123",
+        filename="manual.pdf",
+        score=1.0,
+        attributes=metadata,
+        content=[VectorStoreContent(type="text", text="Database setup instructions")],
+    )
+
+    assert result.attributes == {
+        "headings": "Installation > Database setup",
+        "page_numbers": "4, 5",
+    }

tests/unit/providers/file_processor/test_docling_serve.py

@@ -42,9 +42,30 @@ def _make_httpx_response(json_body: dict, status_code: int = 200) -> httpx.Respo
 
 CHUNK_RESPONSE = {
     "chunks": [
-        {"text": "First chunk of text.", "meta": {"headings": ["Introduction"]}},
-        {"text": "Second chunk of text.", "meta": {}},
-        {"text": "Third chunk of text.", "meta": {"headings": ["Conclusion"]}},
+        {
+            "filename": "test.pdf",
+            "chunk_index": 0,
+            "text": "First chunk of text.",
+            "headings": ["Introduction"],
+            "doc_items": ["#/texts/0"],
+            "page_numbers": [1],
+        },
+        {
+            "filename": "test.pdf",
+            "chunk_index": 1,
+            "text": "Second chunk of text.",
+            "headings": None,
+            "doc_items": ["#/texts/1"],
+            "page_numbers": None,
+        },
+        {
+            "filename": "test.pdf",
+            "chunk_index": 2,
+            "text": "Third chunk of text.",
+            "headings": ["Safety, Security", "Conclusion"],
+            "doc_items": ["#/texts/2"],
+            "page_numbers": [2, 3],
+        },
     ],
 }
 
@@ -206,15 +227,36 @@ async def test_chunk_id_uniqueness(self, processor: DoclingServeFileProcessor, u
         ids = [c.chunk_id for c in response.chunks]
         assert len(ids) == len(set(ids))
 
-    async def test_headings_propagated(self, processor: DoclingServeFileProcessor, upload_file: UploadFile):
+    async def test_structural_metadata_propagated(self, processor: DoclingServeFileProcessor, upload_file: UploadFile):
         request = ProcessFileRequest(chunking_strategy=VectorStoreChunkingStrategyAuto())
 
         with patch("httpx.AsyncClient.post", return_value=_make_httpx_response(CHUNK_RESPONSE)):
             response = await processor.process_file(request, file=upload_file)
 
-        assert response.chunks[0].metadata["headings"] == ["Introduction"]
+        assert response.chunks[0].metadata["headings"] == "Introduction"
+        assert response.chunks[0].metadata["page_numbers"] == "1"
         assert "headings" not in response.chunks[1].metadata
-        assert response.chunks[2].metadata["headings"] == ["Conclusion"]
+        assert "page_numbers" not in response.chunks[1].metadata
+        assert response.chunks[2].metadata["headings"] == "Safety, Security > Conclusion"
+        assert response.chunks[2].metadata["page_numbers"] == "2, 3"
+
+    async def test_legacy_nested_headings_keep_existing_list_type(
+        self, processor: DoclingServeFileProcessor, upload_file: UploadFile
+    ):
+        request = ProcessFileRequest(chunking_strategy=VectorStoreChunkingStrategyAuto())
+        legacy_response = {
+            "chunks": [
+                {
+                    "text": "Legacy chunk shape.",
+                    "meta": {"headings": ["Legacy heading"]},
+                }
+            ]
+        }
+
+        with patch("httpx.AsyncClient.post", return_value=_make_httpx_response(legacy_response)):
+            response = await processor.process_file(request, file=upload_file)
+
+        assert response.chunks[0].metadata["headings"] == ["Legacy heading"]
 
     async def test_chunk_window_set(self, processor: DoclingServeFileProcessor, upload_file: UploadFile):
         request = ProcessFileRequest(chunking_strategy=VectorStoreChunkingStrategyAuto())
@@ -478,7 +520,14 @@ async def test_local_docker_allows_chunking(self, upload_file: UploadFile):
 
             # Mock submit_chunk() returning a successful job
             mock_job = AsyncMock()
-            mock_chunk = SimpleNamespace(text="Chunk content", meta=SimpleNamespace(headings=None))
+            mock_chunk = SimpleNamespace(
+                filename="test.pdf",
+                chunk_index=0,
+                text="Chunk content",
+                headings=["Introduction"],
+                doc_items=["#/texts/0"],
+                page_numbers=[1, 2],
+            )
             mock_response = SimpleNamespace(chunks=[mock_chunk])
             mock_job.result.return_value = mock_response
             mock_instance.submit_chunk.return_value = mock_job
@@ -493,4 +542,6 @@ async def test_local_docker_allows_chunking(self, upload_file: UploadFile):
             assert source.stream.getvalue() == b"%PDF-fake-content"
         assert result.chunks is not None
         assert len(result.chunks) > 0
+        assert result.chunks[0].metadata["headings"] == "Introduction"
+        assert result.chunks[0].metadata["page_numbers"] == "1, 2"
         assert result.metadata["conversion_method"] == "async"