test(chunking): add doc_items field to ChunkResult serialization test

The `doc_items` field was added to `ChunkResult` earlier in the
feature branch (used by ChunkWriter for DERIVED_FROM edges), but the
test fixture was never updated. CI caught it now that the PR is open.

Fixes: tests/test_chunking.py::TestChunkResult::test_serializable

.env.example

@@ -48,3 +48,8 @@
 
 # Embedding vector dimension (default: 384 for Granite Embedding 30M / all-MiniLM-L6-v2)
 # EMBEDDING_DIMENSION=384
+
+# Neo4j — graph-native document structure (used by docker-compose.dev.yml)
+# NEO4J_URI=bolt://neo4j:7687
+# NEO4J_USER=neo4j
+# NEO4J_PASSWORD=changeme

.github/workflows/ci.yml

@@ -29,7 +29,7 @@ jobs:
         with:
           python-version: "3.12"
           cache: pip
-          cache-dependency-path: document-parser/requirements.txt
+          cache-dependency-path: document-parser/requirements-test.txt
 
       - name: Install system dependencies
         run: sudo apt-get update && sudo apt-get install -y --no-install-recommends poppler-utils
@@ -37,8 +37,8 @@ jobs:
       - name: Install Python dependencies
         run: |
           pip install --upgrade pip
-          pip install -r requirements.txt
+          pip install -r requirements-test.txt
-          pip install pytest pytest-asyncio httpx ruff
+          pip install httpx ruff
 
       - name: Lint
         run: ruff check .

README.md

@@ -33,6 +33,7 @@ Upload a PDF, configure the extraction pipeline, and visualize the results — t
 - **Per-page results** — right panel syncs with the current PDF page
 - **Chunking** — split extracted content into semantic chunks (hierarchical, hybrid, or page-based) with configurable token limits and inline editing
 - **Ingestion pipeline** — Docling → chunking → embedding → OpenSearch vector indexing (one-click from Studio)
+- **Graph storage (Neo4j)** — full DoclingDocument tree (sections, paragraphs, tables, pages, chunks) mirrored as a graph with `PARENT_OF`, `NEXT`, `ON_PAGE`, `HAS_CHUNK`, `DERIVED_FROM` relations, with an in-app graph view powered by Cytoscape.js
 - **Markdown & HTML export** of extracted content
 - **Document management** — upload, list, delete, search, filter by indexing status
 - **Analysis history** — re-visit and open past analyses
@@ -244,6 +245,69 @@ When ingestion is enabled, the UI shows:
 | `EMBEDDING_URL` | — | Embedding service endpoint (empty = ingestion disabled) |
 | `EMBEDDING_DIMENSION` | `384` | Vector dimension (must match embedding model) |
 
+## Graph storage with Neo4j (opt-in)
+
+Docling Studio can mirror the full **DoclingDocument tree** into a [Neo4j](https://neo4j.com/) graph: sections, paragraphs, tables, figures, pages, and chunks all become first-class nodes connected by `HAS_ROOT`, `PARENT_OF`, `NEXT`, `ON_PAGE`, `HAS_CHUNK`, and `DERIVED_FROM` edges. This enables queries that are impossible with a flat chunk store — navigating a document's outline, finding all tables under a given section, or tracing a chunk back to its source elements.
+
+Enable Neo4j with the ingestion profile (it ships alongside OpenSearch):
+
+```bash
+docker compose --profile ingestion \
+  -f docker-compose.yml -f docker-compose.ingestion.yml \
+  up --build
+```
+
+The Neo4j Browser is available at <http://localhost:7474> (user `neo4j`, password `changeme` by default).
+
+### Schema at a glance
+
+```mermaid
+graph TD
+    D[Document] -->|HAS_ROOT| SH[SectionHeader]
+    D -->|HAS_CHUNK| C[Chunk]
+    SH -->|PARENT_OF| P[Paragraph]
+    SH -->|PARENT_OF| T[Table]
+    P -->|NEXT| T
+    P -->|ON_PAGE| PG[Page]
+    T -->|ON_PAGE| PG
+    C -->|DERIVED_FROM| P
+    C -->|DERIVED_FROM| T
+```
+
+### Example Cypher queries
+
+Find all "Methods" sections across documents (impossible in vector-only stores):
+
+```cypher
+MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(s:SectionHeader)
+WHERE toLower(s.text) CONTAINS 'method'
+RETURN d.title, s.text, s.level
+```
+
+Get the parent section and sibling elements of a chunk (context for RAG):
+
+```cypher
+MATCH (c:Chunk {id: $chunk_id})-[:DERIVED_FROM]->(e:Element)
+MATCH (e)<-[:PARENT_OF]-(parent:Element)-[:PARENT_OF]->(sibling:Element)
+RETURN parent, collect(sibling) AS siblings
+```
+
+List all tables from documents ingested from an `invoices/` path:
+
+```cypher
+MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(t:Table)
+WHERE d.source_uri CONTAINS 'invoices/'
+RETURN d.title, t.caption, t.cells_json
+```
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `NEO4J_URI` | — | Neo4j Bolt endpoint (empty = graph storage disabled) |
+| `NEO4J_USER` | `neo4j` | Neo4j username |
+| `NEO4J_PASSWORD` | `changeme` | Neo4j password |
+
+The in-app **Graph** tab (under *Results*) renders the per-document graph with [Cytoscape.js](https://js.cytoscape.org/) (see [ADR-001](docs/architecture/adrs/ADR-001-graph-visualization-library.md) for the library choice). Documents with more than **200 pages** return `HTTP 413` from `GET /api/documents/{id}/graph`; pagination ships in v0.6.
+
 ## CI / Release
 
 GitHub Actions pipelines (see [`.github/workflows/`](.github/workflows/)):

docker-compose.dev.yml

@@ -9,6 +9,26 @@
 # =============================================================================
 
 services:
+  # --- Neo4j (graph-native document structure) ---
+  neo4j:
+    image: neo4j:5.15-community
+    environment:
+      NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme}
+      NEO4J_server_memory_heap_initial__size: 512m
+      NEO4J_server_memory_heap_max__size: 1g
+    ports:
+      - "7474:7474"
+      - "7687:7687"
+    volumes:
+      - neo4j_data:/data
+      - neo4j_logs:/logs
+    healthcheck:
+      test: ["CMD-SHELL", "cypher-shell -u $${NEO4J_USER:-neo4j} -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+      start_period: 30s
+
   # --- OpenSearch (single-node, security disabled for local dev) ---
   opensearch:
     image: opensearchproject/opensearch:2
@@ -77,12 +97,17 @@ services:
       BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-10}
       OPENSEARCH_URL: http://opensearch:9200
       EMBEDDING_URL: http://embedding:8001
+      NEO4J_URI: bolt://neo4j:7687
+      NEO4J_USER: ${NEO4J_USER:-neo4j}
+      NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme}
     command: ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
     depends_on:
       opensearch:
         condition: service_healthy
       embedding:
         condition: service_healthy
+      neo4j:
+        condition: service_healthy
     deploy:
       resources:
         limits:
@@ -105,6 +130,8 @@ services:
 
 volumes:
   opensearch_data:
+  neo4j_data:
+  neo4j_logs:
   uploads_data:
   db_data:
   frontend_node_modules:

docker-compose.yml

@@ -1,4 +1,22 @@
 services:
+  # --- Neo4j (graph-native document structure) ---
+  neo4j:
+    profiles: ["ingestion"]
+    image: neo4j:5.15-community
+    environment:
+      NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme}
+      NEO4J_server_memory_heap_initial__size: 512m
+      NEO4J_server_memory_heap_max__size: 1g
+    volumes:
+      - neo4j_data:/data
+      - neo4j_logs:/logs
+    healthcheck:
+      test: ["CMD-SHELL", "cypher-shell -u $${NEO4J_USER:-neo4j} -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+      start_period: 30s
+
   # --- OpenSearch (single-node, security disabled) ---
   opensearch:
     profiles: ["ingestion"]
@@ -53,6 +71,9 @@ services:
       BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-10}
       OPENSEARCH_URL: ${OPENSEARCH_URL:-}
       EMBEDDING_URL: ${EMBEDDING_URL:-}
+      NEO4J_URI: ${NEO4J_URI:-}
+      NEO4J_USER: ${NEO4J_USER:-neo4j}
+      NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme}
     deploy:
       resources:
         limits:
@@ -69,5 +90,7 @@ services:
 
 volumes:
   opensearch_data:
+  neo4j_data:
+  neo4j_logs:
   uploads_data:
   db_data:

docs/architecture/adrs/ADR-001-graph-visualization-library.md

@@ -0,0 +1,142 @@
+# ADR-001: Graph visualization library for the Neo4j graph view
+
+**Date**: 2026-04-17
+**Status**: Proposed
+**Deciders**: Pier-Jean Malandrino
+
+## Context
+
+v0.5.0 introduces Neo4j as a graph-native storage layer for parsed documents
+(see [docs/design/neo4j-integration.md](../../design/neo4j-integration.md)
+and [#186](https://github.com/scub-france/Docling-Studio/issues/186)). We need
+an in-app visualization of that graph: the `DoclingDocument` tree as rendered
+in Neo4j, with nodes colored by element type (`SectionHeader`, `Paragraph`,
+`Table`, `Figure`, `ListItem`, `Formula`) and edges (`PARENT_OF`, `NEXT`,
+`ON_PAGE`, `HAS_CHUNK`, `DERIVED_FROM`).
+
+The view lives in the existing Vue 3 debug panel. It is the **primary demo
+artifact** for the Hackernoon hackathon (Neo4j partner), so polish matters as
+much as correctness.
+
+### Constraints
+
+- Vue 3 + Vite frontend, no framework change
+- Must render the full tree of a 200-page document (worst case ≈ a few
+  thousand nodes; see graph endpoint cap in the design doc §8.4)
+- Needs a **clean hierarchical layout** — documents are trees, not arbitrary
+  graphs; a good tree layout is the single biggest UX lever
+- Needs per-node styling (shape + color by label), click, hover, zoom, pan
+- Must be installable without Java/Python-side changes
+- License compatible with the repo (MIT-ish preferred)
+
+### Non-goals for v0.5.0
+
+- 3D rendering
+- Force-directed simulation as the primary layout (we have a tree)
+- Editing nodes in place (read-only view)
+- Rendering millions of nodes
+
+## Decision
+
+Use **Cytoscape.js** via a thin Vue wrapper (`vue-cytoscape` or a bespoke
+`GraphView.vue` that imports `cytoscape` directly and uses the
+`dagre`/`breadthfirst` layouts).
+
+## Consequences
+
+### Positive
+
+- Battle-tested library (13k+ GitHub stars, maintained since 2013, used by
+  Neo4j's own "Bloom"-style visualizations in the community)
+- First-class support for hierarchical layouts via `cytoscape-dagre` (hub-and-
+  spoke / tree) and built-in `breadthfirst` — both map naturally to our
+  `PARENT_OF` structure
+- CSS-like selector syntax for styling (`node[label = "Table"] { ... }`),
+  which is pleasant to evolve as we add node types
+- Permissive licensing (MIT)
+- Headless mode available, so it can be tested outside a DOM (Jest + jsdom
+  works cleanly)
+- Active ecosystem: `cytoscape-cola`, `cytoscape-klay`, `cytoscape-popper` for
+  tooltips, all maintained
+- Bundle size is reasonable for a demo: ~300 KB min+gz for core + dagre, well
+  below our current frontend budget
+
+### Negative
+
+- Styling DSL is powerful but has its own syntax to learn; not plain CSS
+- Large graphs (>10k nodes) benefit from canvas+WebGL libraries
+  (sigma.js, reagraph) — we are explicitly not in that regime for v0.5, but
+  we would need to swap if we later visualize the cross-document graph
+- No Vue 3 component library that is both maintained and popular — we wrap it
+  ourselves in `GraphView.vue` (the wrapper is ~50 LOC, so this is minor)
+
+### Neutral
+
+- Not "Neo4j-branded": we do not use Neovis.js, which is a thin Cytoscape
+  wrapper around the Bolt protocol. Our graph API already returns shaped
+  JSON, so the Neovis convenience is not worth the lock-in
+- We take on one runtime dependency (`cytoscape` + `cytoscape-dagre`)
+
+## Alternatives Considered
+
+### Alternative 1: vis-network (vis.js)
+
+- **Pros**: Very easy to get started, built-in physics, shipped by Neo4j
+  Browser historically
+- **Cons**: Maintenance has been rocky (original vis.js split into several
+  forks; `vis-network` is the maintained branch but releases are sparse);
+  hierarchical layout is OK but less configurable than dagre; styling API is
+  less expressive; TypeScript types lag behind the JS API
+- **Why rejected**: Hierarchical layout quality is the single most important
+  criterion for a document tree, and vis-network is clearly a notch below
+  Cytoscape + dagre here. Maintenance trajectory is also a concern for a
+  release we want to keep shipping on
+
+### Alternative 2: Neovis.js
+
+- **Pros**: Built by Neo4j Labs, connects directly to a Bolt endpoint, nice
+  out-of-the-box "Neo4j look"
+- **Cons**: Wraps Cytoscape anyway, so everything it can do we can do with
+  Cytoscape directly; expects the browser to talk Bolt, which forces us to
+  expose Neo4j creds in the frontend OR to proxy Bolt through the backend
+  (both worse than our current "backend returns JSON" design); limited
+  customization compared to raw Cytoscape
+- **Why rejected**: The auth story is a non-starter for a hackathon demo we
+  want to show publicly, and we lose nothing vs. Cytoscape by going one
+  layer lower
+
+### Alternative 3: D3 (d3-hierarchy + d3-force)
+
+- **Pros**: Maximum flexibility; beautiful, publication-grade output; full
+  SVG control
+- **Cons**: Much more code for the same result — layout, zoom, pan, hover,
+  selection all hand-rolled; steeper learning curve for future contributors
+  to the project; no built-in graph data model
+- **Why rejected**: We're building a product feature, not a data-viz
+  artefact. The time budget (1 day of Day 3) doesn't fit a D3 build-your-own
+
+### Alternative 4: Reagraph / react-force-graph / sigma.js (WebGL)
+
+- **Pros**: Scales to tens of thousands of nodes at 60 FPS; good for future
+  cross-document visualization
+- **Cons**: Optimized for force-directed layouts, weaker hierarchical
+  support; Reagraph is React-only (requires a React island inside Vue);
+  sigma.js's tree layout is immature
+- **Why rejected**: Wrong regime for a single-document tree. Worth
+  reconsidering if/when we visualize the full corpus graph in a later release
+
+### Alternative 5: Mermaid
+
+- **Pros**: Trivial to embed, already used in docs
+- **Cons**: Static rendering, no interactivity, not designed for thousands of
+  nodes, no per-node click/hover
+- **Why rejected**: A viewer, not a visualizer. We need interactivity
+
+## References
+
+- [Neo4j integration design doc](../../design/neo4j-integration.md) §8.3
+- [Issue #186 — Neo4j integration](https://github.com/scub-france/Docling-Studio/issues/186)
+- [Cytoscape.js](https://js.cytoscape.org/)
+- [cytoscape-dagre](https://github.com/cytoscape/cytoscape.js-dagre)
+- [vis-network](https://visjs.github.io/vis-network/docs/network/)
+- [Neovis.js](https://github.com/neo4j-contrib/neovis.js)

docs/design/neo4j-integration.md

@@ -0,0 +1,435 @@
+# Neo4j integration — Docling-Studio v0.5.0
+
+Design doc for Neo4j integration targeting release 0.5.0.
+Target: Hackernoon hackathon demo (Neo4j partner).
+
+---
+
+## 1. Context and goals
+
+### Already in Docling-Studio
+- Ingestion pipeline: Docling parser → chunking (HybridChunker) → embedding → OpenSearch (vector index)
+- Vue 3 + FastAPI UI
+- Debug view to inspect/edit chunks before retrieval
+- Docker compose with existing services
+
+### What we add in v0.5.0
+- Neo4j as **graph-native storage** of the document structure
+- A new ingestion layer that stores the DoclingDocument tree faithfully as a graph
+- Minimal UI to visualize the graph (demo value to the judges)
+- Compose pipeline with Neo4j
+
+### Why graph-native (hackathon positioning)
+> Most document AI tools store parsed content as flat chunks in a vector DB.
+> Docling-Studio v0.5 introduces a graph-native storage layer on top of Neo4j,
+> preserving the full hierarchical structure of documents as first-class citizens.
+> This unlocks hybrid retrieval, agentic navigation, and structural debugging —
+> impossible with chunk-only stores.
+
+### Out of scope for v0.5.0 (roadmap mention only)
+- EnrichmentWriter (entities / summaries / keywords via docling-agent) — v0.6.0
+- Agent reasoning trace viewer — v0.6.0
+- RAG hybrid (graph traversal + vector) — v0.7.0
+- Document versioning — v0.7.0+
+
+---
+
+## 2. Architectural principles
+
+### Port & adapter, with nuance
+
+**Write side**: one `Writer` port, **composable stages** (not alternative adapters).
+Pipelines A and B are additive, not exclusive.
+
+```
+  CORE (always)          Pipeline A (RAG)       Pipeline B (agent-ready, v0.6+)
+  ┌─────────────┐        ┌────────────────┐    ┌───────────────────┐
+  │ TreeWriter  │ ─────▶ │ ChunkWriter    │    │ EnrichmentWriter  │
+  │             │        │ (existing      │    │ (via docling-     │
+  │             │        │  OpenSearch +  │    │  agent, v0.6+)    │
+  │             │        │  adds chunks   │    │                   │
+  │             │        │  to Neo4j)     │    │                   │
+  └─────────────┘        └────────────────┘    └───────────────────┘
+```
+
+```python
+# docling_studio/ingestion/pipeline.py
+class Writer(Protocol):
+    def write(self, doc: DoclingDocument, ctx: IngestionContext) -> None: ...
+
+# Explicit composition per use case
+def build_pipeline(config: PipelineConfig) -> list[Writer]:
+    writers = [TreeWriter(neo4j_driver)]
+    if config.rag_enabled:
+        writers.append(ChunkWriter(neo4j_driver, chunker, embedder, opensearch))
+    if config.enrichment_enabled:  # v0.6.0+
+        writers.append(EnrichmentWriter(neo4j_driver, docling_agent))
+    return writers
+```
+
+**Read side**: two distinct ports (same Neo4j backend, different queries).
+
+```python
+class RAGRetrievalPort(Protocol):
+    def search(self, query: str, k: int) -> list[Chunk]: ...
+    def similar(self, chunk_id: str, k: int) -> list[Chunk]: ...
+
+class TreeNavigationPort(Protocol):  # v0.6.0+
+    def get_outline(self, doc_id: str) -> Tree: ...
+    def read_node(self, ref: str) -> Element: ...
+    def list_children(self, ref: str) -> list[Element]: ...
+    def walk(self, ref: str, depth: int) -> SubTree: ...
+```
+
+---
+
+## 3. Neo4j schema
+
+### Constraints & indexes (created at boot)
+
+```cypher
+// Uniqueness
+CREATE CONSTRAINT document_id IF NOT EXISTS
+  FOR (d:Document) REQUIRE d.id IS UNIQUE;
+
+CREATE CONSTRAINT element_composite IF NOT EXISTS
+  FOR (e:Element) REQUIRE (e.doc_id, e.self_ref) IS UNIQUE;
+
+CREATE CONSTRAINT page_composite IF NOT EXISTS
+  FOR (p:Page) REQUIRE (p.doc_id, p.page_no) IS UNIQUE;
+
+CREATE CONSTRAINT chunk_id IF NOT EXISTS
+  FOR (c:Chunk) REQUIRE c.id IS UNIQUE;
+
+// Full-text index (element text search)
+CREATE FULLTEXT INDEX element_text IF NOT EXISTS
+  FOR (e:Element) ON EACH [e.text];
+
+// Simple indexes for per-doc queries
+CREATE INDEX element_doc IF NOT EXISTS FOR (e:Element) ON (e.doc_id);
+CREATE INDEX chunk_doc  IF NOT EXISTS FOR (c:Chunk)   ON (c.doc_id);
+```
+
+### Data model
+
+```cypher
+// Root document
+(:Document {
+  id: string,                    // UUID or PDF hash
+  title: string,
+  source_uri: string,            // path or S3
+  ingested_at: datetime,
+  docling_version: string,
+  stages_applied: list<string>,  // ["tree", "chunks"] etc.
+  last_tree_write: datetime,
+  last_chunk_write: datetime,
+  tenant_id: string              // simple multi-tenancy
+})
+
+// All tree elements (shared :Element label + specific label)
+(:Element:SectionHeader {doc_id, self_ref, text, level, prov_page, prov_bbox})
+(:Element:Paragraph    {doc_id, self_ref, text, prov_page, prov_bbox})
+(:Element:Table        {doc_id, self_ref, caption, cells_json, prov_page, prov_bbox})
+(:Element:Figure       {doc_id, self_ref, caption, image_uri, prov_page, prov_bbox})
+(:Element:ListItem     {doc_id, self_ref, text, marker, prov_page, prov_bbox})
+(:Element:Formula      {doc_id, self_ref, latex, text, prov_page, prov_bbox})
+
+// Page for layout provenance
+(:Page {doc_id, page_no, width, height})
+
+// Chunks (Pipeline A)
+(:Chunk {
+  id, doc_id,
+  text,
+  chunk_index,
+  embedding_ref,   // id in OpenSearch (no inline duplication)
+  token_count
+})
+```
+
+### Relations
+
+```cypher
+// Hierarchical structure
+(:Document)-[:HAS_ROOT]->(:Element)
+(:Element)-[:PARENT_OF {order: int}]->(:Element)  // order preserves sequence
+(:Element)-[:NEXT]->(:Element)                     // DFS pre-order reading
+
+// Layout
+(:Element)-[:ON_PAGE]->(:Page)
+
+// Pipeline A (chunking)
+(:Document)-[:HAS_CHUNK]->(:Chunk)
+(:Chunk)-[:DERIVED_FROM]->(:Element)  // back-reference; a chunk can span multiple elements
+```
+
+### Decisions
+
+| Decision | Choice | Rationale |
+|----------|-------|---------------|
+| Element composite key | `(doc_id, self_ref)` | self_ref not unique across docs |
+| Multi-tenancy | `tenant_id` property on Document | Simple, filterable, migrable to multi-db later |
+| Table cells | `cells_json` property | v0.5 KISS. May model `(Table)-[:HAS_CELL]->(Cell)` in v0.6+ |
+| Reading order | `[:NEXT]` chain + `{order}` on `PARENT_OF` | Both views useful |
+| Versioning | None (replace strategy on re-upload) | v0.5 KISS |
+| APOC | Not required | Pure Cypher is sufficient for v0.5 |
+
+### Re-ingestion strategy
+
+```cypher
+// Before ingesting, wipe existing
+MATCH (d:Document {id: $doc_id})
+OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK]->()
+DETACH DELETE d
+// Then re-walk cleanly
+```
+
+---
+
+## 4. Implementation plan (3 days)
+
+### Day 1 — Infra + schema
+
+- [ ] Add `neo4j` service to `docker-compose.yml` (`neo4j:5.15-community`, persistent volume, healthcheck)
+- [ ] Add env vars (`NEO4J_URI`, `NEO4J_USER`, `NEO4J_PASSWORD`) to `.env.example`
+- [ ] Create module `docling_studio/storage/neo4j/`:
+  - `driver.py` — neo4j-python driver wrapper (connection pool, context manager)
+  - `schema.py` — idempotent `bootstrap_schema()` (CREATE CONSTRAINT / INDEX at startup)
+  - `__init__.py` with exports
+- [ ] Hook `bootstrap_schema()` in FastAPI startup
+- [ ] Basic integration tests:
+  - Driver connection
+  - Schema bootstrap (idempotence verified)
+  - Simple round-trip: write Document, read Document, delete Document
+
+**Deliverable:** docker compose boots with healthy Neo4j, schema in place at init.
+
+### Day 2 — TreeWriter (write + read)
+
+- [ ] `storage/neo4j/tree_writer.py` — `DoclingDocument → Neo4j` walker
+  - `write_document(doc, tenant_id, driver)` in a transaction
+  - DFS pre-order, batched `MERGE` for perf
+  - Pages first, then Elements, then `PARENT_OF` / `NEXT` / `ON_PAGE` relations
+  - Dynamic labels based on `node.label` (`SectionHeader`, `Paragraph`, …)
+- [ ] `storage/neo4j/tree_reader.py` — inverse walker `Neo4j → DoclingDocument`
+  - `read_document(doc_id, driver) -> DoclingDocument`
+  - Loads all Elements + Pages, rebuilds the Pydantic structure
+  - Prerequisite for v0.6 (feeding docling-agent from Neo4j)
+- [ ] Integrate into existing ingestion pipeline:
+  - Add TreeWriter as first stage of `IngestionPipeline`
+  - `neo4j_enabled: bool` config toggle
+- [ ] Round-trip tests:
+  - 3–4 varied PDFs (academic, invoice, report)
+  - Assertion: `doc_original == read_document(write_document(doc_original))`
+  - Beware dates, bbox floats (tolerance)
+
+**Deliverable:** A PDF uploaded to Docling-Studio is fully present in Neo4j and rebuildable.
+
+### Day 3 — UI + ChunkWriter + packaging
+
+- [ ] `storage/neo4j/chunk_writer.py`:
+  - After existing chunking, push each Chunk to Neo4j
+  - Create `(:Chunk)-[:DERIVED_FROM]->(:Element)` via source element `self_ref`
+  - Do NOT duplicate embeddings (stay in OpenSearch, keep `embedding_ref`)
+- [ ] Frontend: new "Graph view" tab in debug panel
+  - Vue component with `cytoscape` (lighter, better layout API — see [ADR-001](../architecture/adrs/ADR-001-graph-visualization-library.md))
+  - FastAPI endpoint `/api/documents/{doc_id}/graph` returns full nodes + edges for the document, **capped at 200 pages** (HTTP 413 beyond; pagination deferred to v0.6). The endpoint must include a `truncated: bool` flag and `node_count` / `edge_count` in the response envelope so the UI can warn the user cleanly.
+  - View: vertical tree, colors per node type, click-to-zoom, hover details
+- [ ] Per-document "Graph-ready" / "RAG-ready" badge in list
+- [ ] README update:
+  - "Graph storage with Neo4j" section
+  - Schema diagram (Mermaid or image)
+  - 2–3 Cypher examples like "find all paragraphs under section X that mention Y"
+  - Neo4j badge in features list
+- [ ] (bonus if time) "Query explorer" dev tab for live demo: Cypher editor + results
+
+**Deliverable:** release 0.5.0 with Neo4j visible, functional, documented.
+
+---
+
+## 5. Proposed code structure
+
+```
+docling_studio/
+├── storage/
+│   ├── neo4j/
+│   │   ├── __init__.py
+│   │   ├── driver.py           # connection management
+│   │   ├── schema.py           # bootstrap_schema()
+│   │   ├── tree_writer.py      # DoclingDocument -> Neo4j
+│   │   ├── tree_reader.py      # Neo4j -> DoclingDocument
+│   │   ├── chunk_writer.py     # Chunks -> Neo4j
+│   │   └── queries.py          # shared Cypher queries
+│   ├── opensearch/             # (existing)
+│   └── ports.py                # Writer, RAGRetrievalPort protocols
+├── ingestion/
+│   └── pipeline.py             # IngestionPipeline composing Writers
+├── api/
+│   └── graph.py                # /api/documents/{id}/graph
+└── frontend/
+    └── components/
+        └── GraphView.vue       # cytoscape + graph API fetch
+```
+
+---
+
+## 6. Docker compose (added excerpt)
+
+```yaml
+services:
+  neo4j:
+    image: neo4j:5.15-community
+    environment:
+      NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme}
+      NEO4J_PLUGINS: '["apoc"]'
+      NEO4J_server_memory_heap_initial__size: 512m
+      NEO4J_server_memory_heap_max__size: 1g
+    ports:
+      - "7474:7474"   # Browser UI (demo)
+      - "7687:7687"   # Bolt protocol
+    volumes:
+      - neo4j_data:/data
+      - neo4j_logs:/logs
+    healthcheck:
+      test: ["CMD-SHELL", "cypher-shell -u neo4j -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+
+  docling-studio-backend:
+    depends_on:
+      neo4j:
+        condition: service_healthy
+    environment:
+      NEO4J_URI: bolt://neo4j:7687
+      NEO4J_USER: neo4j
+      NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme}
+
+volumes:
+  neo4j_data:
+  neo4j_logs:
+```
+
+---
+
+## 7. Tests
+
+### Unit tests
+- `tests/storage/neo4j/test_schema.py` — bootstrap is idempotent
+- `tests/storage/neo4j/test_tree_writer.py` — round-trip on synthetic DoclingDocument
+- `tests/storage/neo4j/test_chunk_writer.py` — chunks written with correct `DERIVED_FROM`
+
+### Integration tests
+- `tests/integration/test_ingestion_pipeline.py` — full pipeline on a real PDF
+- PDF fixtures: 1 academic (complex heading hierarchy), 1 invoice (tables), 1 report (lists)
+
+### E2E (bonus)
+- Upload PDF via UI → check structure in Neo4j Browser
+
+---
+
+## 8. Open decisions to settle before coding
+
+1. **Neo4j edition**: Community (free) or AuraDB (managed) ?
+   - Rec: Community in Docker for v0.5.0 dev/demo. AuraDB mentioned as prod option.
+
+2. **Chunks: duplicate embeddings in Neo4j or OpenSearch ref ?**
+   - Rec: OpenSearch ref (avoid duplication; OpenSearch remains source of truth for vectors). In v0.6+, consider native Neo4j vector index.
+
+3. **Graph view UI: cytoscape or vis-network ?**
+   - Decided: **Cytoscape.js** — see [ADR-001](../architecture/adrs/ADR-001-graph-visualization-library.md) for the full analysis.
+
+4. **Graph endpoint: return full doc or paginate ?**
+   - Decided: full doc for v0.5, **hard cap at 200 pages**. Beyond the cap, the endpoint returns HTTP 413 with a `truncated: true` flag; the UI shows "Graph too large to render — reduce scope". Pagination ships in v0.6.
+
+5. **Error strategy**: if Neo4j is down at ingestion, fail or degrade gracefully ?
+   - Rec: **fail fast** for v0.5 (avoid silent inconsistencies). `neo4j_required: bool` config option.
+
+---
+
+## 9. Hooks for later (v0.6.0+ — don't implement but prepare)
+
+**EnrichmentWriter (v0.6)** — will need:
+- The reader (Neo4j → DoclingDocument) to re-materialize the doc, feed docling-agent, re-patch enrichments
+- A stage addable to `IngestionPipeline` without touching other stages
+- An `:Entity` label (not created in v0.5 but schema-compatible)
+
+**Agent reasoning trace viewer (v0.6)** — will need:
+- An event stream (WebSocket) that v0.5 already prepares via the reactive UI
+- A node_ref ↔ Element correlation in Neo4j (our composite `self_ref` key is enough)
+
+**TreeNavigationPort (v0.7)** — will need:
+- Optimized Cypher queries for descendant/ancestor walk (indexes already provisioned)
+
+---
+
+## 10. v0.5.0 success criteria
+
+**Must have:**
+- [ ] A PDF uploaded to Docling-Studio is in Neo4j with structure preserved
+- [ ] Neo4j Browser shows the graph and is manually explorable
+- [ ] A graph visual in the Docling-Studio UI works
+- [ ] `docker compose up` works zero-config
+- [ ] README mentions Neo4j and describes the schema
+
+**Nice to have (decreasing priority):**
+- [ ] Graph-ready / RAG-ready badge per doc
+- [ ] Live query explorer in the UI
+- [ ] 2–3 example queries in README that do something impossible with vector-only
+
+**For the hackathon (post-release):**
+- [ ] 60s video: upload PDF → structure in Neo4j → cross-doc query impossible in vector-only
+- [ ] HackerNoon post explaining "graph-native documents" positioning
+- [ ] Explicit Neo4j partnership mention
+
+---
+
+## 11. Fundamental architectural decisions recap
+
+| Question | Answer |
+|----------|---------|
+| Is Neo4j source of truth or cache ? | **Source of truth** for structure. OpenSearch remains source of truth for embeddings. |
+| Does chunking go away ? | No, v0.5.0 keeps existing chunking. "Chunkless" is Pipeline B, v0.6+. |
+| Can it be toggled per doc ? | Yes — `stages_applied` on Document + pipeline config |
+| What about OpenSearch ? | Stays, stores vectors. Neo4j tracks `(:Chunk)-[:DERIVED_FROM]->(:Element)` links. |
+| Multi-tenancy ? | `tenant_id` property on Document, Cypher filter |
+| Versioning ? | None for v0.5.0 — replace strategy on re-upload |
+
+---
+
+## Appendix — Demo queries
+
+### Query 1 — All "Methods" sections across documents
+```cypher
+MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(s:SectionHeader)
+WHERE toLower(s.text) CONTAINS 'method'
+RETURN d.title, s.text, s.level
+```
+
+### Query 2 — Context of a chunk (parent + siblings)
+```cypher
+MATCH (c:Chunk {id: $chunk_id})-[:DERIVED_FROM]->(e:Element)
+MATCH (e)<-[:PARENT_OF]-(parent:Element)
+MATCH (parent)-[:PARENT_OF]->(sibling:Element)
+RETURN parent, collect(sibling) AS siblings
+```
+
+### Query 3 — All tables from a document type
+```cypher
+MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(t:Table)
+WHERE d.source_uri CONTAINS 'invoices/'
+RETURN d.title, t.caption, t.cells_json
+```
+
+### Query 4 — Direct children of a section (ordered)
+```cypher
+MATCH (s:Element {doc_id: $doc_id, self_ref: $section_ref})
+MATCH (s)-[pof:PARENT_OF]->(child)
+RETURN child
+ORDER BY pof.order
+```
+
+---
+
+*Single reference doc for Neo4j v0.5.0 implementation.
+Read this first in the implementation thread.*

document-parser/api/analyses.py

@@ -155,7 +155,7 @@ async def delete_chunk(job_id: str, chunk_index: int, service: ServiceDep) -> li
     ]
 
 
-@router.delete("/{job_id}", status_code=204)
+@router.delete("/{job_id}", status_code=204, response_model=None)
 async def delete_analysis(job_id: str, service: ServiceDep) -> None:
     """Delete an analysis job."""
     deleted = await service.delete(job_id)

document-parser/api/documents.py

@@ -85,7 +85,7 @@ async def get_document(doc_id: str, service: ServiceDep) -> DocumentResponse:
     return _to_response(doc)
 
 
-@router.delete("/{doc_id}", status_code=204)
+@router.delete("/{doc_id}", status_code=204, response_model=None)
 async def delete_document(doc_id: str, service: ServiceDep) -> None:
     """Delete a document and its file."""
     deleted = await service.delete(doc_id)

document-parser/api/graph.py

@@ -0,0 +1,76 @@
+"""Graph API — returns a cytoscape-shaped view of the Neo4j graph for a doc.
+
+v0.5 contract:
+- Returns the **full** graph for the document (see design §8.4)
+- Hard cap at 200 pages; beyond that, HTTP 413 with `truncated: true`
+- No pagination (ships in v0.6)
+"""
+
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+
+from infra.neo4j import fetch_graph
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/documents", tags=["graph"])
+
+MAX_PAGES = 200
+
+
+class GraphNode(BaseModel):
+    id: str
+    group: str
+    label: str | None = None
+
+    model_config = {"extra": "allow"}
+
+
+class GraphEdge(BaseModel):
+    id: str
+    source: str
+    target: str
+    type: str
+    order: int | None = None
+
+
+class GraphResponse(BaseModel):
+    doc_id: str
+    nodes: list[GraphNode]
+    edges: list[GraphEdge]
+    node_count: int
+    edge_count: int
+    truncated: bool
+    page_count: int
+
+
+@router.get("/{doc_id}/graph", response_model=GraphResponse)
+async def get_document_graph(doc_id: str, request: Request) -> GraphResponse:
+    neo = getattr(request.app.state, "neo4j", None)
+    if neo is None:
+        raise HTTPException(status_code=503, detail="Neo4j is not configured")
+
+    payload = await fetch_graph(neo, doc_id, max_pages=MAX_PAGES)
+    if payload is None:
+        raise HTTPException(status_code=404, detail=f"No graph for document {doc_id}")
+    if payload.truncated:
+        raise HTTPException(
+            status_code=413,
+            detail=(
+                f"Graph too large: document has {payload.page_count} pages "
+                f"(cap {MAX_PAGES}). Pagination ships in v0.6."
+            ),
+        )
+
+    return GraphResponse(
+        doc_id=payload.doc_id,
+        nodes=[GraphNode(**n) for n in payload.nodes],
+        edges=[GraphEdge(**e) for e in payload.edges],
+        node_count=payload.node_count,
+        edge_count=payload.edge_count,
+        truncated=payload.truncated,
+        page_count=payload.page_count,
+    )

document-parser/api/ingestion.py

@@ -74,7 +74,7 @@ async def ingest_analysis(
     )
 
 
-@router.delete("/{doc_id}", status_code=204)
+@router.delete("/{doc_id}", status_code=204, response_model=None)
 async def delete_ingested_document(doc_id: str, ingestion: IngestionDep) -> None:
     """Delete all indexed chunks for a document."""
     await ingestion.delete_document(doc_id)

document-parser/domain/value_objects.py

@@ -8,6 +8,10 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
+# US Letter page dimensions (points) — fallback when page size is unknown
+DEFAULT_PAGE_WIDTH: float = 612.0
+DEFAULT_PAGE_HEIGHT: float = 792.0
+
 
 @dataclass(frozen=True)
 class PageElement:
@@ -71,6 +75,14 @@ class ChunkBbox:
     bbox: list[float]  # [left, top, right, bottom] in TOPLEFT origin
 
 
+@dataclass(frozen=True)
+class ChunkDocItem:
+    """Source element referenced by a chunk. Enables Neo4j DERIVED_FROM edges."""
+
+    self_ref: str
+    label: str
+
+
 @dataclass(frozen=True)
 class ChunkResult:
     text: str
@@ -78,3 +90,4 @@ class ChunkResult:
     source_page: int | None = None
     token_count: int = 0
     bboxes: list[ChunkBbox] = field(default_factory=list)
+    doc_items: list[ChunkDocItem] = field(default_factory=list)

document-parser/infra/local_chunker.py

@@ -15,7 +15,7 @@ from docling_core.transforms.chunker import HierarchicalChunker
 from docling_core.transforms.chunker.hybrid_chunker import HybridChunker
 from docling_core.types.doc.document import DoclingDocument
 
-from domain.value_objects import ChunkBbox, ChunkingOptions, ChunkResult
+from domain.value_objects import ChunkBbox, ChunkDocItem, ChunkingOptions, ChunkResult
 from infra.bbox import EMPTY_BBOX, to_topleft_list
 
 logger = logging.getLogger(__name__)
@@ -39,9 +39,18 @@ def _chunk_sync(document_json: str, options: ChunkingOptions) -> list[ChunkResul
         source_page = None
         token_count = 0
         bboxes: list[ChunkBbox] = []
+        doc_items: list[ChunkDocItem] = []
 
         if hasattr(chunk, "meta") and chunk.meta and chunk.meta.doc_items:
             for doc_item in chunk.meta.doc_items:
+                ref = getattr(doc_item, "self_ref", None)
+                if ref:
+                    doc_items.append(
+                        ChunkDocItem(
+                            self_ref=ref,
+                            label=str(getattr(doc_item, "label", "") or ""),
+                        )
+                    )
                 if not hasattr(doc_item, "prov") or not doc_item.prov:
                     continue
                 for prov in doc_item.prov:
@@ -67,6 +76,7 @@ def _chunk_sync(document_json: str, options: ChunkingOptions) -> list[ChunkResul
                 source_page=source_page,
                 token_count=token_count,
                 bboxes=bboxes,
+                doc_items=doc_items,
             )
         )
 

document-parser/infra/local_converter.py

@@ -36,6 +36,8 @@ from docling_core.types.doc import (
 )
 
 from domain.value_objects import (
+    DEFAULT_PAGE_HEIGHT,
+    DEFAULT_PAGE_WIDTH,
     ConversionOptions,
     ConversionResult,
     PageDetail,
@@ -50,10 +52,6 @@ logger = logging.getLogger(__name__)
 # Uses a timeout to prevent a frozen conversion from blocking all others.
 _converter_lock = threading.Lock()
 
-# US Letter page dimensions (points) — fallback when page size is unknown
-_DEFAULT_PAGE_WIDTH = 612.0
-_DEFAULT_PAGE_HEIGHT = 792.0
-
 # Default converter (lazy-init on first request)
 _default_converter: DoclingConverter | None = None
 
@@ -175,11 +173,11 @@ def _process_content_item(
                 logger.warning(
                     "Page %d not found in document metadata — using US Letter fallback (%sx%s pt)",
                     page_no,
-                    _DEFAULT_PAGE_WIDTH,
+                    DEFAULT_PAGE_WIDTH,
-                    _DEFAULT_PAGE_HEIGHT,
+                    DEFAULT_PAGE_HEIGHT,
                 )
                 pages[page_no] = PageDetail(
-                    page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT
+                    page_number=page_no, width=DEFAULT_PAGE_WIDTH, height=DEFAULT_PAGE_HEIGHT
                 )
 
             page_height = pages[page_no].height
@@ -248,10 +246,10 @@ def _convert_sync(
         pages_detail = [
             PageDetail(
                 page_number=i + 1,
-                width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH,
+                width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else DEFAULT_PAGE_WIDTH,
                 height=doc.pages[i + 1].size.height
                 if (i + 1) in doc.pages
-                else _DEFAULT_PAGE_HEIGHT,
+                else DEFAULT_PAGE_HEIGHT,
             )
             for i in range(page_count)
         ]

document-parser/infra/neo4j/__init__.py

@@ -0,0 +1,31 @@
+"""Neo4j storage adapter — graph-native document structure.
+
+Provides a thin driver wrapper, idempotent schema bootstrap, and
+walkers between DoclingDocument and the graph model.
+"""
+
+from infra.neo4j.chunk_writer import ChunkWriteResult, write_chunks
+from infra.neo4j.driver import Neo4jDriver, close_driver, get_driver
+from infra.neo4j.queries import fetch_graph
+from infra.neo4j.schema import bootstrap_schema
+from infra.neo4j.tree_reader import (
+    delete_document,
+    document_exists,
+    read_document_json,
+)
+from infra.neo4j.tree_writer import TreeWriteResult, write_document
+
+__all__ = [
+    "ChunkWriteResult",
+    "Neo4jDriver",
+    "TreeWriteResult",
+    "bootstrap_schema",
+    "close_driver",
+    "delete_document",
+    "document_exists",
+    "fetch_graph",
+    "get_driver",
+    "read_document_json",
+    "write_chunks",
+    "write_document",
+]

document-parser/infra/neo4j/chunk_writer.py

@@ -0,0 +1,134 @@
+"""ChunkWriter — push chunk nodes and DERIVED_FROM edges to Neo4j.
+
+Embeddings stay in OpenSearch. Each :Chunk node carries a chunk_index so the
+OpenSearch entry can be retrieved via (doc_id, chunk_index). The
+`embedding_ref` property is reserved for a future vector-store id (not used
+in v0.5 — OpenSearch indexes by doc_id+chunk_index already).
+
+When chunks carry `doc_items` provenance (list of `self_ref` strings), we
+create `(:Chunk)-[:DERIVED_FROM]->(:Element)` links so that queries can go
+from a chunk back to its source elements. Chunks without doc_items get no
+back-links but are still persisted.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from infra.neo4j.driver import Neo4jDriver
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ChunkWriteResult:
+    doc_id: str
+    chunks_written: int
+    derived_from_edges: int
+
+
+def _chunk_id(doc_id: str, index: int) -> str:
+    return f"{doc_id}::chunk::{index}"
+
+
+async def write_chunks(
+    neo: Neo4jDriver,
+    *,
+    doc_id: str,
+    chunks_json: str,
+) -> ChunkWriteResult:
+    """Persist chunks for `doc_id`. Wipes prior chunks first (idempotent)."""
+    chunks: list[dict[str, Any]] = json.loads(chunks_json)
+    active = [c for c in chunks if not c.get("deleted")]
+
+    chunk_rows: list[dict[str, Any]] = []
+    derived_rows: list[dict[str, Any]] = []
+    for idx, c in enumerate(active):
+        cid = _chunk_id(doc_id, idx)
+        chunk_rows.append(
+            {
+                "id": cid,
+                "doc_id": doc_id,
+                "text": c.get("text") or "",
+                "chunk_index": idx,
+                "token_count": c.get("tokenCount") or 0,
+                "embedding_ref": "",
+            }
+        )
+        for item in c.get("docItems") or []:
+            ref = item.get("selfRef") if isinstance(item, dict) else None
+            if ref:
+                derived_rows.append({"chunk_id": cid, "doc_id": doc_id, "self_ref": ref})
+
+    async with (
+        neo.driver.session(database=neo.database) as session,
+        await session.begin_transaction() as tx,
+    ):
+        # Replace existing chunks.
+        await tx.run(
+            """
+                MATCH (d:Document {id: $doc_id})-[:HAS_CHUNK]->(c:Chunk)
+                DETACH DELETE c
+                """,
+            doc_id=doc_id,
+        )
+        await tx.run("MATCH (c:Chunk {doc_id: $doc_id}) DETACH DELETE c", doc_id=doc_id)
+
+        if chunk_rows:
+            await tx.run(
+                """
+                    MATCH (d:Document {id: $doc_id})
+                    UNWIND $rows AS r
+                    CREATE (c:Chunk {
+                      id: r.id,
+                      doc_id: r.doc_id,
+                      text: r.text,
+                      chunk_index: r.chunk_index,
+                      token_count: r.token_count,
+                      embedding_ref: r.embedding_ref
+                    })
+                    MERGE (d)-[:HAS_CHUNK]->(c)
+                    """,
+                doc_id=doc_id,
+                rows=chunk_rows,
+            )
+
+        if derived_rows:
+            await tx.run(
+                """
+                    UNWIND $rows AS r
+                    MATCH (c:Chunk {id: r.chunk_id})
+                    MATCH (e:Element {doc_id: r.doc_id, self_ref: r.self_ref})
+                    MERGE (c)-[:DERIVED_FROM]->(e)
+                    """,
+                rows=derived_rows,
+            )
+
+        # Flag the Document with the new stage.
+        await tx.run(
+            """
+                MATCH (d:Document {id: $doc_id})
+                SET d.stages_applied = [s IN coalesce(d.stages_applied, []) WHERE s <> 'chunks']
+                                       + ['chunks'],
+                    d.last_chunk_write = datetime()
+                """,
+            doc_id=doc_id,
+        )
+
+        await tx.commit()
+
+    logger.info(
+        "Neo4j: wrote %d chunks (%d DERIVED_FROM) for doc %s",
+        len(chunk_rows),
+        len(derived_rows),
+        doc_id,
+    )
+    return ChunkWriteResult(
+        doc_id=doc_id,
+        chunks_written=len(chunk_rows),
+        derived_from_edges=len(derived_rows),
+    )

document-parser/infra/neo4j/driver.py

@@ -0,0 +1,48 @@
+"""Async Neo4j driver wrapper.
+
+Owns a single `AsyncDriver` per process. Callers acquire it via
+`get_driver()` and must call `close_driver()` at shutdown.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from neo4j import AsyncDriver, AsyncGraphDatabase
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class Neo4jDriver:
+    driver: AsyncDriver
+    database: str = "neo4j"
+
+
+_instance: Neo4jDriver | None = None
+
+
+async def get_driver(uri: str, user: str, password: str, database: str = "neo4j") -> Neo4jDriver:
+    """Return the process-wide driver, creating it on first call.
+
+    Verifies connectivity once at creation — raises if the server is unreachable.
+    """
+    global _instance
+    if _instance is not None:
+        return _instance
+
+    driver = AsyncGraphDatabase.driver(uri, auth=(user, password))
+    await driver.verify_connectivity()
+    logger.info("Neo4j driver connected to %s (db=%s)", uri, database)
+    _instance = Neo4jDriver(driver=driver, database=database)
+    return _instance
+
+
+async def close_driver() -> None:
+    global _instance
+    if _instance is None:
+        return
+    await _instance.driver.close()
+    _instance = None
+    logger.info("Neo4j driver closed")

document-parser/infra/neo4j/queries.py

@@ -0,0 +1,230 @@
+"""Reusable Cypher queries — kept out of the API layer for reuse + testing."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from infra.neo4j.driver import Neo4jDriver
+
+
+@dataclass
+class GraphPayload:
+    doc_id: str
+    nodes: list[dict[str, Any]]
+    edges: list[dict[str, Any]]
+    node_count: int
+    edge_count: int
+    truncated: bool
+    page_count: int
+
+
+# Full graph for one doc: Document + Elements + Pages + Chunks and their edges.
+# Each node/edge type is collected inside its own CALL {} subquery so every
+# block contributes a single row — avoids the cartesian product that chained
+# OPTIONAL MATCH on 6+ edge types would produce (hangs on multi-page docs).
+# See: https://neo4j.com/developer/kb/using-subqueries-to-control-the-scope-of-aggregations/
+_FETCH_GRAPH = """
+MATCH (d:Document {id: $doc_id})
+CALL { WITH d MATCH (e:Element {doc_id: d.id}) RETURN collect(e) AS elements }
+CALL { WITH d MATCH (p:Page {doc_id: d.id})    RETURN collect(p) AS pages }
+CALL { WITH d MATCH (c:Chunk {doc_id: d.id})   RETURN collect(c) AS chunks }
+CALL {
+  WITH d
+  MATCH (pe:Element {doc_id: d.id})-[r:PARENT_OF]->(ce:Element)
+  RETURN collect({from: pe.self_ref, to: ce.self_ref, order: r.order, type: 'PARENT_OF'}) AS parent_edges
+}
+CALL {
+  WITH d
+  MATCH (a:Element {doc_id: d.id})-[:NEXT]->(b:Element)
+  RETURN collect({from: a.self_ref, to: b.self_ref, type: 'NEXT'}) AS next_edges
+}
+CALL {
+  WITH d
+  MATCH (er:Element {doc_id: d.id})-[:ON_PAGE]->(pr:Page)
+  RETURN collect({from: er.self_ref, to: pr.page_no, type: 'ON_PAGE'}) AS on_page_edges
+}
+CALL {
+  WITH d
+  MATCH (d)-[:HAS_ROOT]->(rr:Element)
+  RETURN collect({from: d.id, to: rr.self_ref, type: 'HAS_ROOT'}) AS has_root_edges
+}
+CALL {
+  WITH d
+  MATCH (d)-[:HAS_CHUNK]->(rc:Chunk)
+  RETURN collect({from: d.id, to: rc.id, type: 'HAS_CHUNK'}) AS has_chunk_edges
+}
+CALL {
+  WITH d
+  MATCH (cc:Chunk {doc_id: d.id})-[:DERIVED_FROM]->(ee:Element)
+  RETURN collect({from: cc.id, to: ee.self_ref, type: 'DERIVED_FROM'}) AS derived_from_edges
+}
+RETURN d AS document, elements, pages, chunks,
+       parent_edges, next_edges, on_page_edges,
+       has_root_edges, has_chunk_edges, derived_from_edges
+"""
+
+
+def _element_node(doc_id: str, e: dict[str, Any]) -> dict[str, Any]:
+    # Determine the specific element label: Neo4j returns it via labels(e) on the
+    # driver side; when we project nodes via RETURN, the driver wraps them as Node
+    # objects, so we convert below.
+    return {
+        "id": f"elem::{e.get('self_ref')}",
+        "group": "element",
+        "docling_label": e.get("docling_label"),
+        "self_ref": e.get("self_ref"),
+        "text": (e.get("text") or "")[:200],
+        "prov_page": e.get("prov_page"),
+        "level": e.get("level"),
+        "doc_id": doc_id,
+    }
+
+
+def _page_node(doc_id: str, p: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": f"page::{p.get('page_no')}",
+        "group": "page",
+        "page_no": p.get("page_no"),
+        "width": p.get("width"),
+        "height": p.get("height"),
+        "doc_id": doc_id,
+    }
+
+
+def _chunk_node(p: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": f"chunk::{p.get('id')}",
+        "group": "chunk",
+        "chunk_index": p.get("chunk_index"),
+        "text": (p.get("text") or "")[:200],
+        "token_count": p.get("token_count"),
+    }
+
+
+def _edge_id(from_id: str, to_id: str, edge_type: str) -> str:
+    return f"{edge_type}::{from_id}::{to_id}"
+
+
+async def fetch_graph(
+    neo: Neo4jDriver,
+    doc_id: str,
+    *,
+    max_pages: int = 200,
+) -> GraphPayload | None:
+    """Return the full graph for a document, or None if the document is unknown.
+
+    Enforces the page cap from design §8.4: beyond `max_pages`, returns a
+    `truncated=True` payload with empty node/edge lists so the caller can
+    surface a clean error (HTTP 413) to the UI.
+    """
+    async with neo.driver.session(database=neo.database) as session:
+        page_count_result = await session.run(
+            "MATCH (p:Page {doc_id: $doc_id}) RETURN count(p) AS n",
+            doc_id=doc_id,
+        )
+        pc_record = await page_count_result.single()
+        if pc_record is None:
+            return None
+        page_count = int(pc_record["n"])
+
+        exists_result = await session.run(
+            "MATCH (d:Document {id: $doc_id}) RETURN count(d) AS n",
+            doc_id=doc_id,
+        )
+        exists_record = await exists_result.single()
+        if not exists_record or exists_record["n"] == 0:
+            return None
+
+        if page_count > max_pages:
+            return GraphPayload(
+                doc_id=doc_id,
+                nodes=[],
+                edges=[],
+                node_count=0,
+                edge_count=0,
+                truncated=True,
+                page_count=page_count,
+            )
+
+        result = await session.run(_FETCH_GRAPH, doc_id=doc_id)
+        record = await result.single()
+
+    nodes: list[dict[str, Any]] = []
+    edges: list[dict[str, Any]] = []
+
+    if record is None:
+        return None
+
+    # Document node.
+    doc_node = record["document"]
+    if doc_node is not None:
+        nodes.append(
+            {
+                "id": f"doc::{doc_id}",
+                "group": "document",
+                "doc_id": doc_id,
+                "title": doc_node.get("title"),
+                "stages_applied": doc_node.get("stages_applied"),
+            }
+        )
+
+    # Element nodes, keeping the specific label (:SectionHeader, etc.).
+    for e in record["elements"] or []:
+        if e is None:
+            continue
+        labels = [label for label in e.labels if label != "Element"]
+        node = _element_node(doc_id, dict(e))
+        node["label"] = labels[0] if labels else "TextElement"
+        nodes.append(node)
+
+    # Pages.
+    for p in record["pages"] or []:
+        if p is None:
+            continue
+        nodes.append(_page_node(doc_id, dict(p)))
+
+    # Chunks.
+    for c in record["chunks"] or []:
+        if c is None:
+            continue
+        nodes.append(_chunk_node(dict(c)))
+
+    # Edges — filter out rows whose from/to is null (OPTIONAL MATCH can yield them).
+    def _push_element_edge(e: dict[str, Any], from_prefix: str, to_prefix: str) -> None:
+        frm, to = e.get("from"), e.get("to")
+        if frm is None or to is None:
+            return
+        edges.append(
+            {
+                "id": _edge_id(f"{from_prefix}{frm}", f"{to_prefix}{to}", e["type"]),
+                "source": f"{from_prefix}{frm}",
+                "target": f"{to_prefix}{to}",
+                "type": e["type"],
+                "order": e.get("order"),
+            }
+        )
+
+    for e in record["parent_edges"] or []:
+        _push_element_edge(e, "elem::", "elem::")
+    for e in record["next_edges"] or []:
+        _push_element_edge(e, "elem::", "elem::")
+    for e in record["on_page_edges"] or []:
+        _push_element_edge(e, "elem::", "page::")
+    for e in record["has_root_edges"] or []:
+        _push_element_edge(e, "doc::", "elem::")
+    for e in record["has_chunk_edges"] or []:
+        _push_element_edge(e, "doc::", "chunk::")
+    for e in record["derived_from_edges"] or []:
+        _push_element_edge(e, "chunk::", "elem::")
+
+    return GraphPayload(
+        doc_id=doc_id,
+        nodes=nodes,
+        edges=edges,
+        node_count=len(nodes),
+        edge_count=len(edges),
+        truncated=False,
+        page_count=page_count,
+    )

document-parser/infra/neo4j/schema.py

@@ -0,0 +1,50 @@
+"""Idempotent Neo4j schema bootstrap.
+
+Runs at backend startup. All statements use `IF NOT EXISTS`, so calling
+this multiple times is safe — it's the contract integration tests rely on.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from infra.neo4j.driver import Neo4jDriver
+
+logger = logging.getLogger(__name__)
+
+
+CONSTRAINTS: tuple[str, ...] = (
+    "CREATE CONSTRAINT document_id IF NOT EXISTS FOR (d:Document) REQUIRE d.id IS UNIQUE",
+    "CREATE CONSTRAINT element_composite IF NOT EXISTS "
+    "FOR (e:Element) REQUIRE (e.doc_id, e.self_ref) IS UNIQUE",
+    "CREATE CONSTRAINT page_composite IF NOT EXISTS "
+    "FOR (p:Page) REQUIRE (p.doc_id, p.page_no) IS UNIQUE",
+    "CREATE CONSTRAINT chunk_id IF NOT EXISTS FOR (c:Chunk) REQUIRE c.id IS UNIQUE",
+)
+
+INDEXES: tuple[str, ...] = (
+    "CREATE INDEX element_doc IF NOT EXISTS FOR (e:Element) ON (e.doc_id)",
+    "CREATE INDEX chunk_doc IF NOT EXISTS FOR (c:Chunk) ON (c.doc_id)",
+)
+
+FULLTEXT_INDEXES: tuple[str, ...] = (
+    "CREATE FULLTEXT INDEX element_text IF NOT EXISTS FOR (e:Element) ON EACH [e.text]",
+)
+
+
+async def bootstrap_schema(neo: Neo4jDriver) -> None:
+    """Create constraints and indexes required by the graph model.
+
+    Idempotent: safe to call on every startup.
+    """
+    async with neo.driver.session(database=neo.database) as session:
+        for stmt in (*CONSTRAINTS, *INDEXES, *FULLTEXT_INDEXES):
+            await session.run(stmt)
+    logger.info(
+        "Neo4j schema bootstrapped (%d constraints, %d indexes, %d fulltext)",
+        len(CONSTRAINTS),
+        len(INDEXES),
+        len(FULLTEXT_INDEXES),
+    )

document-parser/infra/neo4j/tree_reader.py

@@ -0,0 +1,64 @@
+"""TreeReader — fetch a DoclingDocument back from Neo4j.
+
+v0.5.0 implementation relies on the verbatim `document_json` property stored
+on the Document node by TreeWriter. Reconstruction by walking Element nodes
+is deferred to v0.6 (EnrichmentWriter prerequisite), where we may need to
+rebuild the DoclingDocument after enrichments have been patched on graph
+nodes directly.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from infra.neo4j.driver import Neo4jDriver
+
+logger = logging.getLogger(__name__)
+
+
+async def read_document_json(neo: Neo4jDriver, doc_id: str) -> str | None:
+    """Return the stored DoclingDocument JSON for `doc_id`, or None if absent."""
+    async with neo.driver.session(database=neo.database) as session:
+        result = await session.run(
+            "MATCH (d:Document {id: $doc_id}) RETURN d.document_json AS json",
+            doc_id=doc_id,
+        )
+        record = await result.single()
+    if record is None:
+        return None
+    return record["json"]
+
+
+async def document_exists(neo: Neo4jDriver, doc_id: str) -> bool:
+    async with neo.driver.session(database=neo.database) as session:
+        result = await session.run(
+            "MATCH (d:Document {id: $doc_id}) RETURN count(d) AS n",
+            doc_id=doc_id,
+        )
+        record = await result.single()
+    return bool(record and record["n"] > 0)
+
+
+async def delete_document(neo: Neo4jDriver, doc_id: str) -> int:
+    """Wipe everything related to a doc_id. Returns nodes removed."""
+    async with neo.driver.session(database=neo.database) as session:
+        result = await session.run(
+            """
+            MATCH (d:Document {id: $doc_id})
+            OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK*0..]->(n)
+            WITH d, collect(DISTINCT n) AS children
+            DETACH DELETE d
+            WITH children
+            UNWIND children AS c
+            DETACH DELETE c
+            RETURN size(children) + 1 AS removed
+            """,
+            doc_id=doc_id,
+        )
+        record = await result.single()
+        # Also clean up orphan elements and pages tagged with this doc_id.
+        await session.run("MATCH (e:Element {doc_id: $doc_id}) DETACH DELETE e", doc_id=doc_id)
+        await session.run("MATCH (p:Page {doc_id: $doc_id}) DETACH DELETE p", doc_id=doc_id)
+    return int(record["removed"]) if record else 0

document-parser/infra/neo4j/tree_writer.py

@@ -0,0 +1,335 @@
+"""TreeWriter — persist a DoclingDocument as a graph in Neo4j.
+
+v0.5.0 strategy: replace-on-write. For a given doc_id, all existing
+Document/Element/Page/Chunk nodes are wiped before re-ingestion. The full
+serialized `DoclingDocument` JSON is stored as a property on the Document
+node so that `TreeReader` can round-trip it verbatim — reconstruction from
+graph nodes is deferred to v0.6 (see docs/design/neo4j-integration.md §2).
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from infra.neo4j.driver import Neo4jDriver
+
+logger = logging.getLogger(__name__)
+
+
+# Docling label → specific Neo4j label. Every node also carries :Element.
+_LABEL_MAP: dict[str, str] = {
+    "section_header": "SectionHeader",
+    "title": "SectionHeader",
+    "paragraph": "Paragraph",
+    "text": "Paragraph",
+    "list_item": "ListItem",
+    "list": "ListItem",
+    "table": "Table",
+    "picture": "Figure",
+    "formula": "Formula",
+    "code": "Code",
+    "caption": "Caption",
+    "footnote": "Footnote",
+    "page_header": "PageHeader",
+    "page_footer": "PageFooter",
+}
+_DEFAULT_LABEL = "TextElement"
+
+
+def _element_label(docling_label: str) -> str:
+    return _LABEL_MAP.get(docling_label.lower(), _DEFAULT_LABEL)
+
+
+@dataclass
+class TreeWriteResult:
+    doc_id: str
+    elements_written: int
+    pages_written: int
+
+
+def _iter_items(doc_data: dict[str, Any]):
+    """Yield every item from texts/tables/pictures/groups with its source list."""
+    for key in ("texts", "tables", "pictures", "groups"):
+        for item in doc_data.get(key, []) or []:
+            yield key, item
+
+
+def _first_prov(item: dict[str, Any]) -> tuple[int | None, list[float] | None]:
+    prov = item.get("prov") or []
+    if not prov:
+        return None, None
+    p0 = prov[0]
+    bbox = p0.get("bbox")
+    bbox_list: list[float] | None = None
+    if isinstance(bbox, dict):
+        bbox_list = [bbox.get("l", 0.0), bbox.get("t", 0.0), bbox.get("r", 0.0), bbox.get("b", 0.0)]
+    elif isinstance(bbox, list):
+        bbox_list = list(bbox)
+    return p0.get("page_no"), bbox_list
+
+
+def _parent_ref(item: dict[str, Any]) -> str | None:
+    parent = item.get("parent")
+    if isinstance(parent, dict):
+        return parent.get("$ref") or parent.get("cref")
+    return None
+
+
+def _element_props(item: dict[str, Any], doc_id: str) -> dict[str, Any]:
+    page, bbox = _first_prov(item)
+    props: dict[str, Any] = {
+        "doc_id": doc_id,
+        "self_ref": item.get("self_ref") or "",
+        "docling_label": (item.get("label") or "").lower(),
+        "text": item.get("text") or "",
+        "prov_page": page,
+        "prov_bbox": bbox,
+    }
+    # Type-specific extras.
+    if "level" in item:
+        props["level"] = item.get("level")
+    if "caption" in item and isinstance(item.get("caption"), str):
+        props["caption"] = item.get("caption")
+    if item.get("data") and isinstance(item["data"], dict):
+        # Tables carry cell layout under data; stringify to keep the schema flat.
+        with contextlib.suppress(TypeError, ValueError):
+            props["cells_json"] = json.dumps(item["data"])
+    return props
+
+
+def _dfs_order(doc_data: dict[str, Any]) -> list[str]:
+    """Return self_refs in reading order (DFS pre-order from body)."""
+    by_ref: dict[str, dict[str, Any]] = {}
+    for _, item in _iter_items(doc_data):
+        ref = item.get("self_ref")
+        if ref:
+            by_ref[ref] = item
+    body = doc_data.get("body") or {}
+    order: list[str] = []
+
+    def walk(children: list[dict[str, Any]] | None) -> None:
+        if not children:
+            return
+        for ch in children:
+            ref = ch.get("$ref") or ch.get("cref")
+            if not ref:
+                continue
+            order.append(ref)
+            child = by_ref.get(ref)
+            if child:
+                walk(child.get("children"))
+
+    walk(body.get("children"))
+    return order
+
+
+async def write_document(
+    neo: Neo4jDriver,
+    *,
+    doc_id: str,
+    filename: str,
+    document_json: str,
+    tenant_id: str = "default",
+    source_uri: str | None = None,
+    docling_version: str | None = None,
+) -> TreeWriteResult:
+    """Persist the full DoclingDocument tree to Neo4j.
+
+    Idempotent: wipes any existing graph for doc_id before writing.
+    Fails fast (exception propagates) if Neo4j is unavailable — per design §8.5.
+    """
+    doc_data = json.loads(document_json)
+    ingested_at = datetime.now(tz=UTC).isoformat()
+
+    elements: list[dict[str, Any]] = []
+    for _, item in _iter_items(doc_data):
+        ref = item.get("self_ref")
+        if not ref:
+            continue
+        specific = _element_label(item.get("label") or "")
+        elements.append(
+            {
+                "specific_label": specific,
+                "parent_ref": _parent_ref(item),
+                **_element_props(item, doc_id),
+            }
+        )
+
+    pages: list[dict[str, Any]] = []
+    for page_no_str, page_obj in (doc_data.get("pages") or {}).items():
+        try:
+            page_no = int(page_no_str)
+        except (TypeError, ValueError):
+            continue
+        size = page_obj.get("size") or {}
+        pages.append(
+            {
+                "doc_id": doc_id,
+                "page_no": page_no,
+                "width": size.get("width"),
+                "height": size.get("height"),
+            }
+        )
+
+    reading_order = _dfs_order(doc_data)
+
+    async with (
+        neo.driver.session(database=neo.database) as session,
+        await session.begin_transaction() as tx,
+    ):
+        # 1. Wipe existing graph for this doc_id (replace strategy).
+        await tx.run(
+            "MATCH (d:Document {id: $doc_id}) "
+            "OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK*0..]->(n) "
+            "DETACH DELETE d, n",
+            doc_id=doc_id,
+        )
+        # Also wipe orphan elements/chunks that may still reference this doc.
+        await tx.run("MATCH (e:Element {doc_id: $doc_id}) DETACH DELETE e", doc_id=doc_id)
+        await tx.run("MATCH (p:Page {doc_id: $doc_id}) DETACH DELETE p", doc_id=doc_id)
+
+        # 2. Document node (carries the verbatim JSON for TreeReader).
+        await tx.run(
+            """
+                CREATE (d:Document {
+                  id: $doc_id,
+                  title: $title,
+                  source_uri: $source_uri,
+                  ingested_at: datetime($ingested_at),
+                  docling_version: $docling_version,
+                  stages_applied: ['tree'],
+                  last_tree_write: datetime($ingested_at),
+                  tenant_id: $tenant_id,
+                  document_json: $document_json
+                })
+                """,
+            doc_id=doc_id,
+            title=filename,
+            source_uri=source_uri or "",
+            ingested_at=ingested_at,
+            docling_version=docling_version or "",
+            tenant_id=tenant_id,
+            document_json=document_json,
+        )
+
+        # 3. Page nodes.
+        if pages:
+            await tx.run(
+                "UNWIND $pages AS p "
+                "CREATE (:Page {doc_id: p.doc_id, page_no: p.page_no, "
+                "width: p.width, height: p.height})",
+                pages=pages,
+            )
+
+        # 4. Element nodes — use dynamic :Element:<specific> labels via APOC-free trick.
+        # We split by specific label so the CREATE statement is static (no APOC).
+        by_specific: dict[str, list[dict[str, Any]]] = {}
+        for e in elements:
+            by_specific.setdefault(e["specific_label"], []).append(e)
+        for specific, batch in by_specific.items():
+            await tx.run(
+                f"""
+                    UNWIND $batch AS e
+                    CREATE (n:Element:{specific} {{
+                      doc_id: e.doc_id,
+                      self_ref: e.self_ref,
+                      docling_label: e.docling_label,
+                      text: e.text,
+                      prov_page: e.prov_page,
+                      prov_bbox: e.prov_bbox,
+                      level: e.level,
+                      caption: e.caption,
+                      cells_json: e.cells_json
+                    }})
+                    """,
+                batch=batch,
+            )
+
+        # 5. PARENT_OF relations (tree structure). Order tracked inline.
+        parent_rows = [
+            {
+                "doc_id": doc_id,
+                "parent_ref": e["parent_ref"],
+                "child_ref": e["self_ref"],
+                "order": idx,
+            }
+            for idx, e in enumerate(elements)
+            if e["parent_ref"] and e["parent_ref"] != "#/body"
+        ]
+        if parent_rows:
+            await tx.run(
+                """
+                    UNWIND $rows AS r
+                    MATCH (p:Element {doc_id: r.doc_id, self_ref: r.parent_ref})
+                    MATCH (c:Element {doc_id: r.doc_id, self_ref: r.child_ref})
+                    MERGE (p)-[rel:PARENT_OF]->(c)
+                    SET rel.order = r.order
+                    """,
+                rows=parent_rows,
+            )
+
+        # 6. HAS_ROOT for top-level children of the document body.
+        root_rows = [
+            {"doc_id": doc_id, "child_ref": e["self_ref"]}
+            for e in elements
+            if e["parent_ref"] == "#/body"
+        ]
+        if root_rows:
+            await tx.run(
+                """
+                    UNWIND $rows AS r
+                    MATCH (d:Document {id: r.doc_id})
+                    MATCH (c:Element {doc_id: r.doc_id, self_ref: r.child_ref})
+                    MERGE (d)-[:HAS_ROOT]->(c)
+                    """,
+                rows=root_rows,
+            )
+
+        # 7. ON_PAGE from first provenance.
+        on_page_rows = [
+            {"doc_id": doc_id, "self_ref": e["self_ref"], "page_no": e["prov_page"]}
+            for e in elements
+            if e["prov_page"] is not None
+        ]
+        if on_page_rows:
+            await tx.run(
+                """
+                    UNWIND $rows AS r
+                    MATCH (e:Element {doc_id: r.doc_id, self_ref: r.self_ref})
+                    MATCH (p:Page {doc_id: r.doc_id, page_no: r.page_no})
+                    MERGE (e)-[:ON_PAGE]->(p)
+                    """,
+                rows=on_page_rows,
+            )
+
+        # 8. NEXT chain in DFS pre-order.
+        if len(reading_order) > 1:
+            pairs = [
+                {"doc_id": doc_id, "a": reading_order[i], "b": reading_order[i + 1]}
+                for i in range(len(reading_order) - 1)
+            ]
+            await tx.run(
+                """
+                    UNWIND $pairs AS p
+                    MATCH (a:Element {doc_id: p.doc_id, self_ref: p.a})
+                    MATCH (b:Element {doc_id: p.doc_id, self_ref: p.b})
+                    MERGE (a)-[:NEXT]->(b)
+                    """,
+                pairs=pairs,
+            )
+
+        await tx.commit()
+
+    logger.info(
+        "Neo4j: wrote doc %s (%d elements, %d pages)",
+        doc_id,
+        len(elements),
+        len(pages),
+    )
+    return TreeWriteResult(doc_id=doc_id, elements_written=len(elements), pages_written=len(pages))

document-parser/infra/opensearch_store.py

@@ -69,13 +69,14 @@ class OpenSearchStore:
         verify_certs: Whether to verify TLS certificates.
     """
 
-    def __init__(self, url: str, *, verify_certs: bool = False) -> None:
+    def __init__(self, url: str, *, verify_certs: bool = False, default_limit: int = 1000) -> None:
         self._client = AsyncOpenSearch(
             hosts=[url],
             use_ssl=url.startswith("https"),
             verify_certs=verify_certs,
             ssl_show_warn=False,
         )
+        self._default_limit = default_limit
 
     # -- lifecycle -------------------------------------------------------------
 
@@ -147,9 +148,11 @@ class OpenSearchStore:
         index_name: str,
         doc_id: str,
         *,
-        limit: int = 1000,
+        limit: int | None = None,
     ) -> list[SearchResult]:
         """Retrieve all indexed chunks for a document, ordered by chunk_index."""
+        if limit is None:
+            limit = self._default_limit
         resp = await self._client.search(
             index=index_name,
             body={

document-parser/infra/serve_converter.py

@@ -21,6 +21,8 @@ import httpx
 from docling_core.types.doc.base import BoundingBox, CoordOrigin
 
 from domain.value_objects import (
+    DEFAULT_PAGE_HEIGHT,
+    DEFAULT_PAGE_WIDTH,
     ConversionOptions,
     ConversionResult,
     PageDetail,
@@ -31,7 +33,6 @@ from infra.bbox import to_topleft_list
 logger = logging.getLogger(__name__)
 
 _API_PREFIX = "/v1"
-_DEFAULT_TIMEOUT = 600.0
 
 # Docling Serve label → our element type
 _LABEL_MAP = {
@@ -60,7 +61,7 @@ class ServeConverter:
         self,
         base_url: str,
         api_key: str | None = None,
-        timeout: float = _DEFAULT_TIMEOUT,
+        timeout: float = 600.0,
     ):
         self._base_url = base_url.rstrip("/")
         self._api_key = api_key
@@ -95,6 +96,13 @@ class ServeConverter:
                     headers=self._headers(),
                 )
 
+        if response.status_code >= 400:
+            logger.error(
+                "Docling Serve error %d: %s (form_data=%s)",
+                response.status_code,
+                response.text[:500],
+                {k: v for k, v in form_data.items()},
+            )
         response.raise_for_status()
         result_data = response.json()
 
@@ -121,8 +129,12 @@ def _build_form_data(
 ) -> dict[str, str | list[str]]:
     """Build form fields matching Docling Serve's multipart form contract.
 
-    Array fields (to_formats) are sent as lists — httpx encodes them as
+    Serve uses FastAPI's ``Form()`` parsing — list/tuple fields are sent
-    repeated form keys (to_formats=md&to_formats=html&to_formats=json).
+    as **repeated form keys** (httpx encodes Python lists this way
+    automatically: ``to_formats=md&to_formats=html&to_formats=json``).
+
+    Note: ``generate_page_images`` is a PdfPipelineOptions field, NOT a
+    ConvertDocumentsOptions field — sending it causes a 422.
     """
     data: dict[str, str | list[str]] = {
         "to_formats": ["md", "html", "json"],
@@ -134,11 +146,12 @@ def _build_form_data(
         "do_picture_classification": str(options.do_picture_classification).lower(),
         "do_picture_description": str(options.do_picture_description).lower(),
         "include_images": str(options.generate_picture_images).lower(),
-        "generate_page_images": str(options.generate_page_images).lower(),
         "images_scale": str(options.images_scale),
     }
     if page_range is not None:
-        data["page_range"] = f"{page_range[0]}-{page_range[1]}"
+        # Serve expects page_range as two repeated form fields:
+        # page_range=1&page_range=10
+        data["page_range"] = [str(page_range[0]), str(page_range[1])]
     return data
 
 
@@ -192,8 +205,8 @@ def _extract_pages_from_docling_document(doc: dict) -> list[PageDetail]:
         size = page_data.get("size", {})
         pages_dict[page_no] = PageDetail(
             page_number=page_no,
-            width=size.get("width", 612.0),
+            width=size.get("width", DEFAULT_PAGE_WIDTH),
-            height=size.get("height", 792.0),
+            height=size.get("height", DEFAULT_PAGE_HEIGHT),
         )
 
     # Process all element arrays
@@ -220,8 +233,8 @@ def _add_element(item: dict, pages: dict[int, PageDetail]) -> None:
         if page_no not in pages:
             pages[page_no] = PageDetail(
                 page_number=page_no,
-                width=612.0,
+                width=DEFAULT_PAGE_WIDTH,
-                height=792.0,
+                height=DEFAULT_PAGE_HEIGHT,
             )
 
         bbox_data = prov.get("bbox", {})

document-parser/infra/settings.py

@@ -25,6 +25,10 @@ class Settings:
     batch_page_size: int = 0  # 0 = disabled, > 0 = pages per batch
     opensearch_url: str = ""  # empty = disabled
     embedding_url: str = ""  # empty = disabled (e.g. http://localhost:8001)
+    neo4j_uri: str = ""  # empty = disabled (e.g. bolt://neo4j:7687)
+    neo4j_user: str = "neo4j"
+    neo4j_password: str = "changeme"
+    opensearch_default_limit: int = 1000  # max chunks returned by get_chunks
     embedding_dimension: int = 384  # Granite Embedding 30M / all-MiniLM-L6-v2
     upload_dir: str = "./uploads"
     db_path: str = "./data/docling_studio.db"
@@ -54,6 +58,10 @@ class Settings:
             errors.append(f"rate_limit_rpm must be >= 0 (got {self.rate_limit_rpm})")
         if self.batch_page_size < 0:
             errors.append(f"batch_page_size must be >= 0 (got {self.batch_page_size})")
+        if self.opensearch_default_limit < 1:
+            errors.append(
+                f"opensearch_default_limit must be >= 1 (got {self.opensearch_default_limit})"
+            )
         if self.embedding_dimension < 1:
             errors.append(f"embedding_dimension must be >= 1 (got {self.embedding_dimension})")
         if self.default_table_mode not in ("accurate", "fast"):
@@ -97,6 +105,10 @@ class Settings:
             batch_page_size=int(os.environ.get("BATCH_PAGE_SIZE", "10")),
             opensearch_url=os.environ.get("OPENSEARCH_URL", ""),
             embedding_url=os.environ.get("EMBEDDING_URL", ""),
+            neo4j_uri=os.environ.get("NEO4J_URI", ""),
+            neo4j_user=os.environ.get("NEO4J_USER", "neo4j"),
+            neo4j_password=os.environ.get("NEO4J_PASSWORD", "changeme"),
+            opensearch_default_limit=int(os.environ.get("OPENSEARCH_DEFAULT_LIMIT", "1000")),
             embedding_dimension=int(os.environ.get("EMBEDDING_DIMENSION", "384")),
             upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"),
             db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"),

document-parser/main.py

@@ -47,6 +47,7 @@ def _build_converter():
         return ServeConverter(
             base_url=settings.docling_serve_url,
             api_key=settings.docling_serve_api_key,
+            timeout=settings.conversion_timeout,
         )
     else:
         from infra.local_converter import LocalConverter
@@ -56,12 +57,15 @@ def _build_converter():
 
 
 def _build_chunker():
-    """Build the chunker adapter — only available in local mode."""
+    """Build the chunker adapter.
-    if settings.conversion_engine == "local":
-        from infra.local_chunker import LocalChunker
 
-        return LocalChunker()
+    Uses LocalChunker in all modes — in remote mode it chunks the
-    return None
+    DoclingDocument JSON returned by Docling Serve, so docling-core
+    (lightweight) is the only local dependency needed.
+    """
+    from infra.local_chunker import LocalChunker
+
+    return LocalChunker()
 
 
 def _build_repos() -> tuple[SqliteDocumentRepository, SqliteAnalysisRepository]:
@@ -71,6 +75,7 @@ def _build_repos() -> tuple[SqliteDocumentRepository, SqliteAnalysisRepository]:
 def _build_analysis_service(
     document_repo: SqliteDocumentRepository,
     analysis_repo: SqliteAnalysisRepository,
+    neo4j_driver=None,
 ) -> AnalysisService:
     converter = _build_converter()
     chunker = _build_chunker()
@@ -86,10 +91,33 @@ def _build_analysis_service(
         conversion_timeout=settings.conversion_timeout,
         max_concurrent=settings.max_concurrent_analyses,
         config=config,
+        neo4j_driver=neo4j_driver,
     )
 
 
-def _build_ingestion_service() -> IngestionService | None:
+async def _init_neo4j():
+    """Initialize the Neo4j driver and bootstrap schema — skip if not configured."""
+    if not settings.neo4j_uri:
+        logger.info("Neo4j disabled (NEO4J_URI not set)")
+        return None
+
+    from infra.neo4j import bootstrap_schema, get_driver
+
+    try:
+        neo = await get_driver(
+            settings.neo4j_uri,
+            settings.neo4j_user,
+            settings.neo4j_password,
+        )
+        await bootstrap_schema(neo)
+        logger.info("Neo4j ready (uri=%s)", settings.neo4j_uri)
+        return neo
+    except Exception:
+        logger.exception("Neo4j init failed — continuing without graph storage")
+        return None
+
+
+def _build_ingestion_service(neo4j_driver=None) -> IngestionService | None:
     """Build the ingestion service — only if embedding + opensearch are configured."""
     if not settings.embedding_url or not settings.opensearch_url:
         logger.info("Ingestion disabled (EMBEDDING_URL or OPENSEARCH_URL not set)")
@@ -99,7 +127,10 @@ def _build_ingestion_service() -> IngestionService | None:
     from infra.opensearch_store import OpenSearchStore
 
     embedding = EmbeddingClient(settings.embedding_url)
-    vector_store = OpenSearchStore(settings.opensearch_url)
+    vector_store = OpenSearchStore(
+        settings.opensearch_url,
+        default_limit=settings.opensearch_default_limit,
+    )
     config = IngestionConfig(
         embedding_dimension=settings.embedding_dimension,
     )
@@ -108,7 +139,7 @@ def _build_ingestion_service() -> IngestionService | None:
         settings.embedding_url,
         settings.opensearch_url,
     )
-    return IngestionService(embedding, vector_store, config)
+    return IngestionService(embedding, vector_store, config, neo4j_driver=neo4j_driver)
 
 
 def _build_document_service(
@@ -136,15 +167,24 @@ def _build_document_service(
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     await init_db()
     document_repo, analysis_repo = _build_repos()
-    app.state.analysis_service = _build_analysis_service(document_repo, analysis_repo)
+    app.state.neo4j = await _init_neo4j()
+    app.state.analysis_service = _build_analysis_service(
+        document_repo, analysis_repo, neo4j_driver=app.state.neo4j
+    )
     app.state.document_service = _build_document_service(document_repo, analysis_repo)
-    ingestion_service = _build_ingestion_service()
+    ingestion_service = _build_ingestion_service(neo4j_driver=app.state.neo4j)
     app.state.ingestion_service = ingestion_service
     if ingestion_service is not None:
         app.include_router(ingestion_router)
         logger.info("Ingestion router mounted")
     logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine)
-    yield
+    try:
+        yield
+    finally:
+        if app.state.neo4j is not None:
+            from infra.neo4j import close_driver
+
+            await close_driver()
 
 
 app = FastAPI(
@@ -170,6 +210,11 @@ if settings.rate_limit_rpm > 0:
 app.include_router(documents_router)
 app.include_router(analyses_router)
 
+# Graph view — mounted regardless; individual requests 503 if Neo4j is absent.
+from api.graph import router as graph_router  # noqa: E402
+
+app.include_router(graph_router)
+
 
 @app.get("/api/health", response_model=HealthResponse)
 async def health() -> HealthResponse:

document-parser/persistence/database.py

@@ -9,11 +9,9 @@ from contextlib import asynccontextmanager
 
 import aiosqlite
 
-from infra.settings import settings
-
 logger = logging.getLogger(__name__)
 
-DB_PATH = settings.db_path
+DB_PATH = os.environ.get("DB_PATH", "./data/docling_studio.db")
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS documents (

document-parser/requirements-test.txt

@@ -0,0 +1,4 @@
+-r requirements.txt
+pytest>=8.0.0,<9.0.0
+pytest-asyncio>=0.23.0,<1.0.0
+pytestarch>=2.0.0,<3.0.0

document-parser/requirements.txt

@@ -1,4 +1,4 @@
-docling-core>=2.0.0,<3.0.0
+docling-core[chunking]>=2.0.0,<3.0.0
 fastapi>=0.115.0,<1.0.0
 uvicorn[standard]>=0.32.0,<1.0.0
 python-multipart>=0.0.12
@@ -8,3 +8,4 @@ aiosqlite>=0.20.0,<1.0.0
 httpx>=0.27.0,<1.0.0
 pypdfium2>=4.0.0,<5.0.0
 opensearch-py[async]>=2.6.0,<3.0.0
+neo4j>=5.15.0,<6.0.0

document-parser/services/analysis_service.py

@@ -44,6 +44,7 @@ def _chunk_to_dict(c: ChunkResult) -> dict:
         "sourcePage": c.source_page,
         "tokenCount": c.token_count,
         "bboxes": [{"page": b.page, "bbox": b.bbox} for b in c.bboxes],
+        "docItems": [{"selfRef": d.self_ref, "label": d.label} for d in c.doc_items],
     }
 
 
@@ -69,6 +70,7 @@ class AnalysisConfig:
 
     default_table_mode: str = "accurate"
     batch_page_size: int = 0
+    neo4j_required: bool = False  # if True, ingestion fails when Neo4j write fails
 
 
 class AnalysisService:
@@ -83,6 +85,7 @@ class AnalysisService:
         conversion_timeout: int = 600,
         max_concurrent: int = _DEFAULT_MAX_CONCURRENT,
         config: AnalysisConfig | None = None,
+        neo4j_driver=None,
     ):
         self._converter = converter
         self._chunker = chunker
@@ -93,6 +96,7 @@ class AnalysisService:
         self._running_tasks: dict[str, asyncio.Task] = {}
         self._background_tasks: set[asyncio.Task] = set()
         self._config = config or AnalysisConfig()
+        self._neo4j = neo4j_driver
 
     async def create(
         self,
@@ -324,11 +328,18 @@ class AnalysisService:
         file_path: str,
         options: ConversionOptions,
     ) -> ConversionResult | None:
-        """Run batched or single conversion. Returns None if the job was deleted mid-batch."""
+        """Run batched or single conversion. Returns None if the job was deleted mid-batch.
+
+        Batching is only used for local mode — it limits memory usage when
+        Docling runs in-process.  In remote mode the Serve instance manages
+        its own resources, and batching would discard document_json (needed
+        for chunking).
+        """
         total_pages = _count_pdf_pages(file_path)
         batch_size = self._config.batch_page_size
+        is_remote = self._is_remote_converter()
 
-        if batch_size > 0 and total_pages > batch_size:
+        if batch_size > 0 and total_pages > batch_size and not is_remote:
             return await self._run_batched_conversion(
                 job_id, file_path, options, total_pages, batch_size
             )
@@ -337,6 +348,15 @@ class AnalysisService:
             timeout=self._conversion_timeout,
         )
 
+    def _is_remote_converter(self) -> bool:
+        """Check if the converter is a remote (Serve) adapter."""
+        try:
+            from infra.serve_converter import ServeConverter
+
+            return isinstance(self._converter, ServeConverter)
+        except ImportError:
+            return False
+
     async def _finalize_analysis(
         self,
         job_id: str,
@@ -370,8 +390,32 @@ class AnalysisService:
         if result.page_count:
             await self._document_repo.update_page_count(job.document_id, result.page_count)
 
+        await self._write_tree_to_neo4j(job, result.document_json)
+
         logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count)
 
+    async def _write_tree_to_neo4j(self, job, document_json: str | None) -> None:
+        """Mirror the DoclingDocument tree into Neo4j if configured.
+
+        Silent no-op when Neo4j isn't wired in. Logs but does not fail the
+        analysis when the write fails, unless `config.neo4j_required` is set.
+        """
+        if self._neo4j is None or not document_json:
+            return
+        try:
+            from infra.neo4j import write_document
+
+            await write_document(
+                self._neo4j,
+                doc_id=job.document_id,
+                filename=job.document_filename or job.document_id,
+                document_json=document_json,
+            )
+        except Exception:
+            logger.exception("Neo4j TreeWriter failed for doc %s", job.document_id)
+            if self._config.neo4j_required:
+                raise
+
     async def _run_analysis_inner(
         self,
         job_id: str,

document-parser/services/ingestion_service.py

@@ -54,10 +54,12 @@ class IngestionService:
         embedding_service: EmbeddingService,
         vector_store: VectorStore,
         config: IngestionConfig | None = None,
+        neo4j_driver=None,
     ) -> None:
         self._embedding = embedding_service
         self._vector_store = vector_store
         self._config = config or IngestionConfig()
+        self._neo4j = neo4j_driver
 
     async def ensure_index(self) -> None:
         """Ensure the vector index exists with the correct mapping."""
@@ -139,6 +141,15 @@ class IngestionService:
         indexed = await self._vector_store.index_chunks(self._config.index_name, indexed_chunks)
         logger.info("Indexed %d/%d chunks for doc %s", indexed, len(indexed_chunks), doc_id)
 
+        # 5. Mirror chunks in Neo4j if configured (with DERIVED_FROM edges).
+        if self._neo4j is not None:
+            try:
+                from infra.neo4j import write_chunks
+
+                await write_chunks(self._neo4j, doc_id=doc_id, chunks_json=chunks_json)
+            except Exception:
+                logger.exception("Neo4j ChunkWriter failed for doc %s", doc_id)
+
         return IngestionResult(
             doc_id=doc_id,
             chunks_indexed=indexed,

document-parser/tests/neo4j/conftest.py

@@ -0,0 +1,40 @@
+"""Shared fixtures for Neo4j integration tests.
+
+These tests are skipped unless a live Neo4j is reachable via NEO4J_TEST_URI
+(defaulting to bolt://localhost:7687). CI spins up `neo4j:5.15-community`
+alongside the job; locally run `docker compose -f docker-compose.dev.yml up neo4j`.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+# Skip the entire module cleanly when the neo4j driver package is absent
+# (e.g. local dev without the dependency installed).
+pytest.importorskip("neo4j")
+
+from infra.neo4j import close_driver, get_driver
+
+
+def _cfg() -> tuple[str, str, str]:
+    return (
+        os.environ.get("NEO4J_TEST_URI", "bolt://localhost:7687"),
+        os.environ.get("NEO4J_TEST_USER", "neo4j"),
+        os.environ.get("NEO4J_TEST_PASSWORD", "changeme"),
+    )
+
+
+@pytest.fixture
+async def neo4j_driver():
+    uri, user, password = _cfg()
+    try:
+        neo = await get_driver(uri, user, password)
+    except Exception as exc:
+        pytest.skip(f"Neo4j not reachable at {uri}: {exc}")
+    # Wipe DB before each test — integration tests assume an empty graph.
+    async with neo.driver.session(database=neo.database) as session:
+        await session.run("MATCH (n) DETACH DELETE n")
+    yield neo
+    await close_driver()

document-parser/tests/neo4j/test_chunk_writer.py

@@ -0,0 +1,111 @@
+"""ChunkWriter creates Chunk nodes + DERIVED_FROM links.
+
+Builds on the tree_writer fixture — writes the tree first so that DERIVED_FROM
+has Elements to link against.
+"""
+
+from __future__ import annotations
+
+import json
+
+from infra.neo4j import fetch_graph, write_chunks, write_document
+from infra.neo4j.schema import bootstrap_schema
+from tests.neo4j.test_tree_writer import FIXTURE
+
+CHUNKS = [
+    {
+        "text": "Introduction. First paragraph on page 1.",
+        "sourcePage": 1,
+        "tokenCount": 8,
+        "docItems": [
+            {"selfRef": "#/texts/0", "label": "section_header"},
+            {"selfRef": "#/texts/1", "label": "paragraph"},
+        ],
+    },
+    {
+        "text": "Continued on page 2.",
+        "sourcePage": 2,
+        "tokenCount": 4,
+        "docItems": [{"selfRef": "#/texts/2", "label": "paragraph"}],
+        "deleted": False,
+    },
+    # soft-deleted chunk: must be ignored
+    {"text": "gone", "deleted": True, "docItems": []},
+]
+
+
+async def test_write_chunks_and_derived_from(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=json.dumps(FIXTURE),
+    )
+
+    result = await write_chunks(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        chunks_json=json.dumps(CHUNKS),
+    )
+
+    assert result.chunks_written == 2
+    assert result.derived_from_edges == 3
+
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as s:
+        count = await (
+            await s.run(
+                "MATCH (:Document {id: $id})-[:HAS_CHUNK]->(c:Chunk) RETURN count(c) AS n",
+                id="doc-fixture",
+            )
+        ).single()
+        assert count["n"] == 2
+
+        # First chunk derives from 2 elements, second from 1.
+        for idx, expected in [(0, 2), (1, 1)]:
+            cnt = await (
+                await s.run(
+                    "MATCH (c:Chunk {id: $cid})-[:DERIVED_FROM]->(e:Element) RETURN count(e) AS n",
+                    cid=f"doc-fixture::chunk::{idx}",
+                )
+            ).single()
+            assert cnt["n"] == expected
+
+        stages = await (
+            await s.run(
+                "MATCH (d:Document {id: $id}) RETURN d.stages_applied AS s", id="doc-fixture"
+            )
+        ).single()
+        assert "chunks" in stages["s"]
+
+
+async def test_fetch_graph_returns_full_payload(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=json.dumps(FIXTURE),
+    )
+    await write_chunks(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        chunks_json=json.dumps(CHUNKS),
+    )
+
+    payload = await fetch_graph(neo4j_driver, "doc-fixture")
+    assert payload is not None
+    assert payload.truncated is False
+    assert payload.page_count == 2
+
+    groups = {n["group"] for n in payload.nodes}
+    assert groups == {"document", "element", "page", "chunk"}
+
+    edge_types = {e["type"] for e in payload.edges}
+    # Every edge kind written by TreeWriter and ChunkWriter should be present.
+    assert {"HAS_ROOT", "PARENT_OF", "NEXT", "ON_PAGE", "HAS_CHUNK", "DERIVED_FROM"} <= edge_types
+
+
+async def test_fetch_graph_missing_doc_returns_none(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    assert await fetch_graph(neo4j_driver, "no-such-doc") is None

document-parser/tests/neo4j/test_document_roundtrip.py

@@ -0,0 +1,32 @@
+"""Minimal Document node round-trip — validates the driver + schema end-to-end."""
+
+from __future__ import annotations
+
+from infra.neo4j.schema import bootstrap_schema
+
+
+async def test_document_write_read_delete(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as session:
+        await session.run(
+            "CREATE (d:Document {id: $id, title: $title, tenant_id: $tenant})",
+            id="doc-42",
+            title="Round-trip fixture",
+            tenant="default",
+        )
+
+        result = await session.run(
+            "MATCH (d:Document {id: $id}) RETURN d.title AS title, d.tenant_id AS tenant",
+            id="doc-42",
+        )
+        record = await result.single()
+        assert record is not None
+        assert record["title"] == "Round-trip fixture"
+        assert record["tenant"] == "default"
+
+        await session.run("MATCH (d:Document {id: $id}) DETACH DELETE d", id="doc-42")
+        gone = await (
+            await session.run("MATCH (d:Document {id: $id}) RETURN d", id="doc-42")
+        ).single()
+        assert gone is None

document-parser/tests/neo4j/test_driver.py

@@ -0,0 +1,10 @@
+"""Neo4j driver connectivity smoke test."""
+
+from __future__ import annotations
+
+
+async def test_driver_connects_and_runs_cypher(neo4j_driver):
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as session:
+        result = await session.run("RETURN 1 AS x")
+        record = await result.single()
+        assert record["x"] == 1

document-parser/tests/neo4j/test_schema.py

@@ -0,0 +1,38 @@
+"""Schema bootstrap is idempotent and produces the expected constraints/indexes."""
+
+from __future__ import annotations
+
+from infra.neo4j.schema import CONSTRAINTS, FULLTEXT_INDEXES, INDEXES, bootstrap_schema
+
+
+async def _count_schema(neo4j_driver) -> tuple[int, int]:
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as session:
+        constraints = await (await session.run("SHOW CONSTRAINTS")).data()
+        indexes = await (await session.run("SHOW INDEXES")).data()
+    return len(constraints), len(indexes)
+
+
+async def test_bootstrap_is_idempotent(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    first = await _count_schema(neo4j_driver)
+
+    # Running a second time must not duplicate anything.
+    await bootstrap_schema(neo4j_driver)
+    second = await _count_schema(neo4j_driver)
+
+    assert first == second
+    # Sanity: we created at least what we declared.
+    assert first[0] >= len(CONSTRAINTS)
+    assert first[1] >= len(INDEXES) + len(FULLTEXT_INDEXES)
+
+
+async def test_document_id_is_unique(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as session:
+        await session.run("CREATE (d:Document {id: 'doc-1', title: 'first'})")
+        with_err: Exception | None = None
+        try:
+            await session.run("CREATE (d:Document {id: 'doc-1', title: 'dup'})")
+        except Exception as exc:
+            with_err = exc
+        assert with_err is not None, "unique constraint on Document.id must reject duplicates"

document-parser/tests/neo4j/test_tree_writer.py

@@ -0,0 +1,196 @@
+"""TreeWriter round-trip + structural sanity checks.
+
+Fixture is a hand-crafted DoclingDocument JSON with: one section containing
+two paragraphs and a table, spanning two pages. Tests verify that the graph
+mirrors the structure (HAS_ROOT, PARENT_OF, ON_PAGE, NEXT) and that
+re-writing the same doc is an idempotent replace.
+"""
+
+from __future__ import annotations
+
+import json
+
+from infra.neo4j import read_document_json, write_document
+from infra.neo4j.schema import bootstrap_schema
+
+FIXTURE = {
+    "name": "fixture.pdf",
+    "pages": {
+        "1": {"page_no": 1, "size": {"width": 595, "height": 842}},
+        "2": {"page_no": 2, "size": {"width": 595, "height": 842}},
+    },
+    "body": {
+        "self_ref": "#/body",
+        "children": [
+            {"$ref": "#/texts/0"},
+            {"$ref": "#/texts/1"},
+            {"$ref": "#/texts/2"},
+            {"$ref": "#/tables/0"},
+        ],
+    },
+    "texts": [
+        {
+            "self_ref": "#/texts/0",
+            "parent": {"$ref": "#/body"},
+            "label": "section_header",
+            "text": "Introduction",
+            "level": 1,
+            "prov": [{"page_no": 1, "bbox": {"l": 10, "t": 10, "r": 100, "b": 30}}],
+        },
+        {
+            "self_ref": "#/texts/1",
+            "parent": {"$ref": "#/body"},
+            "label": "paragraph",
+            "text": "First paragraph on page 1.",
+            "prov": [{"page_no": 1, "bbox": {"l": 10, "t": 40, "r": 500, "b": 80}}],
+        },
+        {
+            "self_ref": "#/texts/2",
+            "parent": {"$ref": "#/body"},
+            "label": "paragraph",
+            "text": "Continued on page 2.",
+            "prov": [{"page_no": 2, "bbox": {"l": 10, "t": 40, "r": 500, "b": 80}}],
+        },
+    ],
+    "tables": [
+        {
+            "self_ref": "#/tables/0",
+            "parent": {"$ref": "#/body"},
+            "label": "table",
+            "text": "",
+            "data": {"num_rows": 2, "num_cols": 2, "grid": [[1, 2], [3, 4]]},
+            "prov": [{"page_no": 2, "bbox": {"l": 10, "t": 90, "r": 500, "b": 200}}],
+        }
+    ],
+    "pictures": [],
+    "groups": [],
+}
+
+
+async def _count(session, cypher: str, **params) -> int:
+    r = await session.run(cypher, **params)
+    rec = await r.single()
+    return int(rec["n"]) if rec else 0
+
+
+async def test_write_creates_expected_structure(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    doc_json = json.dumps(FIXTURE)
+
+    result = await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=doc_json,
+    )
+
+    assert result.elements_written == 4
+    assert result.pages_written == 2
+
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as s:
+        assert (
+            await _count(
+                s,
+                "MATCH (d:Document {id: $id}) RETURN count(d) AS n",
+                id="doc-fixture",
+            )
+            == 1
+        )
+        assert (
+            await _count(
+                s,
+                "MATCH (:Document {id: $id})-[:HAS_ROOT]->(e:Element) RETURN count(e) AS n",
+                id="doc-fixture",
+            )
+            == 4
+        )
+        assert (
+            await _count(
+                s,
+                "MATCH (e:Element:SectionHeader {doc_id: $id, self_ref: '#/texts/0'}) "
+                "RETURN count(e) AS n",
+                id="doc-fixture",
+            )
+            == 1
+        )
+        assert (
+            await _count(
+                s,
+                "MATCH (e:Element:Table {doc_id: $id}) RETURN count(e) AS n",
+                id="doc-fixture",
+            )
+            == 1
+        )
+        # Reading-order chain: 3 NEXT edges for 4 elements.
+        assert (
+            await _count(
+                s,
+                "MATCH (a:Element {doc_id: $id})-[:NEXT]->(b:Element {doc_id: $id}) "
+                "RETURN count(*) AS n",
+                id="doc-fixture",
+            )
+            == 3
+        )
+        # ON_PAGE: one per element with prov.
+        assert (
+            await _count(
+                s,
+                "MATCH (:Element {doc_id: $id})-[:ON_PAGE]->(:Page {doc_id: $id}) "
+                "RETURN count(*) AS n",
+                id="doc-fixture",
+            )
+            == 4
+        )
+
+
+async def test_rewrite_is_idempotent_replace(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    doc_json = json.dumps(FIXTURE)
+
+    await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=doc_json,
+    )
+    # Second write with the same id must not duplicate anything.
+    await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=doc_json,
+    )
+
+    async with neo4j_driver.driver.session(database=neo4j_driver.database) as s:
+        assert (
+            await _count(s, "MATCH (d:Document {id: $id}) RETURN count(d) AS n", id="doc-fixture")
+            == 1
+        )
+        assert (
+            await _count(
+                s,
+                "MATCH (e:Element {doc_id: $id}) RETURN count(e) AS n",
+                id="doc-fixture",
+            )
+            == 4
+        )
+
+
+async def test_reader_returns_verbatim_json(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    doc_json = json.dumps(FIXTURE, sort_keys=True)
+    await write_document(
+        neo4j_driver,
+        doc_id="doc-fixture",
+        filename="fixture.pdf",
+        document_json=doc_json,
+    )
+
+    read_back = await read_document_json(neo4j_driver, "doc-fixture")
+    assert read_back is not None
+    assert json.loads(read_back) == json.loads(doc_json)
+
+
+async def test_reader_missing_doc_returns_none(neo4j_driver):
+    await bootstrap_schema(neo4j_driver)
+    assert await read_document_json(neo4j_driver, "no-such-doc") is None

document-parser/tests/test_architecture.py

@@ -0,0 +1,208 @@
+"""Hexagonal architecture tests — enforce layer dependency rules.
+
+Uses pytestarch for inter-layer dependency rules and ast-based import
+scanning for external (third-party) dependency constraints.
+
+Rules enforced:
+- domain   -> no import from api, services, infra, persistence
+- services -> no import from api, infra, persistence
+- api      -> no import from infra, persistence
+- infra    -> no import from api, services
+- persistence -> no import from api, services, infra
+- domain   -> no import of fastapi, sqlalchemy, httpx, opensearchpy
+- services -> no import of fastapi
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+from pytestarch import Rule, get_evaluable_architecture
+
+# ---------------------------------------------------------------------------
+# pytestarch evaluable (project root = document-parser/)
+# ---------------------------------------------------------------------------
+
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+# pytestarch uses the directory name as module prefix when given absolute paths.
+# We use the directory name to build qualified module references.
+_PREFIX = _PROJECT_ROOT.name  # "document-parser"
+
+_evaluable = get_evaluable_architecture(str(_PROJECT_ROOT), str(_PROJECT_ROOT))
+
+
+def _mod(layer: str) -> str:
+    """Return the fully-qualified pytestarch module name for a layer."""
+    return f"{_PREFIX}.{layer}"
+
+
+# ---------------------------------------------------------------------------
+# Helper: collect top-level imports from all .py files in a package
+# ---------------------------------------------------------------------------
+
+
+def _collect_imports(package: str) -> set[str]:
+    """Return the set of top-level module names imported by *package*."""
+    pkg_path = Path(_PROJECT_ROOT) / package
+    imports: set[str] = set()
+    for py_file in pkg_path.rglob("*.py"):
+        tree = ast.parse(py_file.read_text(), filename=str(py_file))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    imports.add(alias.name.split(".")[0])
+            elif isinstance(node, ast.ImportFrom) and node.module:
+                imports.add(node.module.split(".")[0])
+    return imports
+
+
+# ---------------------------------------------------------------------------
+# Inter-layer dependency rules (pytestarch)
+# ---------------------------------------------------------------------------
+
+
+class TestDomainLayerIsolation:
+    """domain must not depend on any other layer."""
+
+    @pytest.mark.parametrize("forbidden", ["api", "services", "infra", "persistence"])
+    def test_domain_does_not_import(self, forbidden: str):
+        rule = (
+            Rule()
+            .modules_that()
+            .are_sub_modules_of(_mod("domain"))
+            .should_not()
+            .import_modules_that()
+            .are_sub_modules_of(_mod(forbidden))
+        )
+        rule.assert_applies(_evaluable)
+
+
+class TestServicesLayerIsolation:
+    """services may import domain only."""
+
+    @pytest.mark.parametrize("forbidden", ["api", "infra", "persistence"])
+    def test_services_does_not_import(self, forbidden: str):
+        rule = (
+            Rule()
+            .modules_that()
+            .are_sub_modules_of(_mod("services"))
+            .should_not()
+            .import_modules_that()
+            .are_sub_modules_of(_mod(forbidden))
+        )
+        rule.assert_applies(_evaluable)
+
+
+class TestApiLayerIsolation:
+    """api may import services and domain, but not infra or persistence."""
+
+    @pytest.mark.parametrize("forbidden", ["infra", "persistence"])
+    def test_api_does_not_import(self, forbidden: str):
+        rule = (
+            Rule()
+            .modules_that()
+            .are_sub_modules_of(_mod("api"))
+            .should_not()
+            .import_modules_that()
+            .are_sub_modules_of(_mod(forbidden))
+        )
+        rule.assert_applies(_evaluable)
+
+
+class TestInfraLayerIsolation:
+    """infra may import domain (ports), but not api or services."""
+
+    @pytest.mark.parametrize("forbidden", ["api", "services"])
+    def test_infra_does_not_import(self, forbidden: str):
+        rule = (
+            Rule()
+            .modules_that()
+            .are_sub_modules_of(_mod("infra"))
+            .should_not()
+            .import_modules_that()
+            .are_sub_modules_of(_mod(forbidden))
+        )
+        rule.assert_applies(_evaluable)
+
+
+class TestPersistenceLayerIsolation:
+    """persistence may import domain, but not api, services, or infra."""
+
+    @pytest.mark.parametrize("forbidden", ["api", "services", "infra"])
+    def test_persistence_does_not_import(self, forbidden: str):
+        rule = (
+            Rule()
+            .modules_that()
+            .are_sub_modules_of(_mod("persistence"))
+            .should_not()
+            .import_modules_that()
+            .are_sub_modules_of(_mod(forbidden))
+        )
+        rule.assert_applies(_evaluable)
+
+
+# ---------------------------------------------------------------------------
+# External dependency rules (ast-based)
+# ---------------------------------------------------------------------------
+
+_DOMAIN_FORBIDDEN_EXTERNALS = {"fastapi", "sqlalchemy", "httpx", "opensearchpy"}
+_SERVICES_FORBIDDEN_EXTERNALS = {"fastapi"}
+
+
+class TestDomainExternalDependencies:
+    """domain must not import infrastructure-specific third-party libraries."""
+
+    @pytest.mark.parametrize("lib", sorted(_DOMAIN_FORBIDDEN_EXTERNALS))
+    def test_domain_does_not_import_external(self, lib: str):
+        imports = _collect_imports("domain")
+        assert lib not in imports, f"domain imports forbidden external library '{lib}'"
+
+
+class TestServicesExternalDependencies:
+    """services must not import web-framework libraries."""
+
+    @pytest.mark.parametrize("lib", sorted(_SERVICES_FORBIDDEN_EXTERNALS))
+    def test_services_does_not_import_external(self, lib: str):
+        imports = _collect_imports("services")
+        assert lib not in imports, f"services imports forbidden external library '{lib}'"
+
+
+# ---------------------------------------------------------------------------
+# Convention: ports live exclusively in domain.ports
+# ---------------------------------------------------------------------------
+
+
+class TestPortConvention:
+    """Protocol definitions (ports) must live in domain.ports only."""
+
+    def test_no_protocol_outside_domain_ports(self):
+        """No Protocol subclass should be defined outside domain/ports.py."""
+        ports_file = Path(_PROJECT_ROOT) / "domain" / "ports.py"
+        for py_file in Path(_PROJECT_ROOT).rglob("*.py"):
+            if py_file == ports_file:
+                continue
+            # Skip test files and __pycache__
+            if "tests" in py_file.parts or "__pycache__" in py_file.parts:
+                continue
+            tree = ast.parse(py_file.read_text(), filename=str(py_file))
+            for node in ast.walk(tree):
+                if isinstance(node, ast.ClassDef):
+                    for base in node.bases:
+                        base_name = _get_name(base)
+                        if base_name == "Protocol":
+                            pytest.fail(
+                                f"Protocol '{node.name}' defined in {py_file.relative_to(_PROJECT_ROOT)}"
+                                f" — ports must live in domain/ports.py"
+                            )
+
+
+def _get_name(node: ast.expr) -> str:
+    """Extract a simple name from an AST expression node."""
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        return node.attr
+    return ""

document-parser/tests/test_chunking.py

@@ -66,6 +66,7 @@ class TestChunkResult:
             "source_page": 1,
             "token_count": 10,
             "bboxes": [],
+            "doc_items": [],
         }
 
 
@@ -465,3 +466,68 @@ class TestRechunkEndpoint:
             },
         )
         assert resp.status_code == 422
+
+
+# ---------------------------------------------------------------------------
+# Remote chunking path — hybrid local chunking from Serve's document_json
+# ---------------------------------------------------------------------------
+
+
+class TestRemoteChunkingPath:
+    """Verify that chunking works on document_json produced by Serve (remote mode)."""
+
+    @pytest.mark.asyncio
+    async def test_rechunk_with_serve_document_json(self):
+        """AnalysisService.rechunk() works with a LocalChunker even in remote mode."""
+        from infra.local_chunker import LocalChunker
+        from services.analysis_service import AnalysisService
+
+        chunker = LocalChunker()
+        analysis_repo = AsyncMock()
+        document_repo = AsyncMock()
+        converter = AsyncMock()  # ServeConverter mock — not used for rechunking
+
+        service = AnalysisService(
+            converter=converter,
+            analysis_repo=analysis_repo,
+            document_repo=document_repo,
+            chunker=chunker,
+        )
+
+        # Simulate a completed job with document_json from Serve
+        job = AnalysisJob(id="j-remote", document_id="d1")
+        job.mark_running()
+        job.mark_completed(
+            markdown="# Title\nParagraph text here.",
+            html="<h1>Title</h1><p>Paragraph text here.</p>",
+            pages_json="[]",
+            document_json=json.dumps({
+                "schema_name": "DoclingDocument",
+                "version": "1.0.0",
+                "name": "test",
+                "origin": {
+                    "mimetype": "application/pdf",
+                    "filename": "test.pdf",
+                    "binary_hash": 0,
+                },
+                "furniture": {"self_ref": "#/furniture", "children": [], "content_layer": "furniture"},
+                "body": {"self_ref": "#/body", "children": [], "content_layer": "body"},
+                "groups": [],
+                "texts": [],
+                "pictures": [],
+                "tables": [],
+                "key_value_items": [],
+                "form_items": [],
+                "pages": {},
+            }),
+        )
+        analysis_repo.find_by_id = AsyncMock(return_value=job)
+        analysis_repo.update_chunks = AsyncMock(return_value=True)
+
+        chunks = await service.rechunk(
+            "j-remote",
+            {"chunker_type": "hybrid", "max_tokens": 512},
+        )
+
+        assert isinstance(chunks, list)
+        analysis_repo.update_chunks.assert_called_once()

document-parser/tests/test_serve_converter.py

@@ -39,7 +39,6 @@ class TestBuildFormData:
         assert data["do_picture_classification"] == "false"
         assert data["do_picture_description"] == "false"
         assert data["include_images"] == "false"
-        assert data["generate_page_images"] == "false"
         assert data["images_scale"] == "1.0"
         assert set(data["to_formats"]) == {"md", "html", "json"}
 
@@ -56,9 +55,14 @@ class TestBuildFormData:
         assert data["images_scale"] == "2.0"
         assert data["include_images"] == "true"
 
-    def test_page_range_included_when_set(self):
+    def test_no_generate_page_images_field(self):
+        """generate_page_images is a PdfPipelineOptions field, not a Serve field."""
+        data = _build_form_data(ConversionOptions())
+        assert "generate_page_images" not in data
+
+    def test_page_range_as_repeated_fields(self):
         data = _build_form_data(ConversionOptions(), page_range=(11, 20))
-        assert data["page_range"] == "11-20"
+        assert data["page_range"] == ["11", "20"]
 
     def test_page_range_absent_when_none(self):
         data = _build_form_data(ConversionOptions())
@@ -402,11 +406,12 @@ class TestServeConverterConvert:
         assert len(result.pages[0].elements) == 1
         assert result.pages[0].elements[0].type == "title"
 
-        # Verify form fields sent as dict with list for repeated keys
+        # Verify form fields sent correctly
         call_kwargs = mock_client.post.call_args
         sent_data = call_kwargs.kwargs.get("data", {})
         assert sent_data["do_ocr"] == "true"
         assert set(sent_data["to_formats"]) == {"md", "html", "json"}
+        assert "generate_page_images" not in sent_data
 
     @pytest.mark.asyncio
     async def test_http_error_raises(self, tmp_path):
@@ -414,6 +419,8 @@ class TestServeConverterConvert:
         test_file.write_bytes(b"%PDF-1.4 fake content")
 
         mock_response = MagicMock()
+        mock_response.status_code = 500
+        mock_response.text = "Internal Server Error"
         mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
             "Server Error",
             request=MagicMock(),
@@ -510,3 +517,17 @@ class TestConverterWiring:
             converter = _build_converter()
         assert isinstance(converter, ServeConverter)
         assert converter._api_key == "my-key"
+
+    def test_remote_engine_builds_chunker(self):
+        """Chunker must be available in remote mode (hybrid local chunking)."""
+        from infra.local_chunker import LocalChunker
+        from infra.settings import Settings
+
+        with patch(
+            "main.settings",
+            Settings(conversion_engine="remote", docling_serve_url="http://serve:5001"),
+        ):
+            from main import _build_chunker
+
+            chunker = _build_chunker()
+        assert isinstance(chunker, LocalChunker)

document-parser/tests/test_settings.py

@@ -19,6 +19,7 @@ class TestSettingsDefaults:
         assert s.max_page_count == 0
         assert s.max_file_size_mb == 50
         assert s.batch_page_size == 0
+        assert s.opensearch_default_limit == 1000
         assert s.upload_dir == "./uploads"
         assert s.db_path == "./data/docling_studio.db"
         assert "http://localhost:3000" in s.cors_origins
@@ -103,6 +104,12 @@ class TestSettingsValidation:
         with pytest.raises(ValueError, match="lock_timeout must be > 0"):
             Settings(lock_timeout=0)
 
+    def test_zero_opensearch_default_limit_rejected(self):
+        import pytest
+
+        with pytest.raises(ValueError, match="opensearch_default_limit must be >= 1"):
+            Settings(opensearch_default_limit=0)
+
     def test_invalid_table_mode_rejected(self):
         import pytest
 
@@ -146,6 +153,7 @@ class TestSettingsFromEnv:
         monkeypatch.setenv("MAX_PAGE_COUNT", "20")
         monkeypatch.setenv("MAX_FILE_SIZE_MB", "100")
         monkeypatch.setenv("BATCH_PAGE_SIZE", "15")
+        monkeypatch.setenv("OPENSEARCH_DEFAULT_LIMIT", "500")
         monkeypatch.setenv("UPLOAD_DIR", "/data/uploads")
         monkeypatch.setenv("DB_PATH", "/data/test.db")
         monkeypatch.setenv("CORS_ORIGINS", "http://a.com, http://b.com")
@@ -163,6 +171,7 @@ class TestSettingsFromEnv:
         assert s.max_page_count == 20
         assert s.max_file_size_mb == 100
         assert s.batch_page_size == 15
+        assert s.opensearch_default_limit == 500
         assert s.upload_dir == "/data/uploads"
         assert s.db_path == "/data/test.db"
         assert s.cors_origins == ["http://a.com", "http://b.com"]

frontend/package-lock.json

@@ -1,13 +1,15 @@
 {
   "name": "docling-studio",
-  "version": "0.3.1",
+  "version": "0.4.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "docling-studio",
-      "version": "0.3.1",
+      "version": "0.4.0",
       "dependencies": {
+        "cytoscape": "^3.30.0",
+        "cytoscape-dagre": "^2.5.0",
         "dompurify": "^3.3.3",
         "marked": "^17.0.4",
         "pinia": "^2.3.0",
@@ -16,6 +18,8 @@
       },
       "devDependencies": {
         "@eslint/js": "^9.0.0",
+        "@types/cytoscape": "^3.21.4",
+        "@types/cytoscape-dagre": "^2.3.3",
         "@types/dompurify": "^3.2.0",
         "@vitejs/plugin-vue": "^6.0.5",
         "@vitest/mocker": "^4.1.2",
@@ -1021,6 +1025,23 @@
         "assertion-error": "^2.0.1"
       }
     },
+    "node_modules/@types/cytoscape": {
+      "version": "3.21.9",
+      "resolved": "https://registry.npmjs.org/@types/cytoscape/-/cytoscape-3.21.9.tgz",
+      "integrity": "sha512-JyrG4tllI6jvuISPjHK9j2Xv/LTbnLekLke5otGStjFluIyA9JjgnvgZrSBsp8cEDpiTjwgZUZwpPv8TSBcoLw==",
+      "dev": true,
+      "license": "MIT"
+    },
+    "node_modules/@types/cytoscape-dagre": {
+      "version": "2.3.4",
+      "resolved": "https://registry.npmjs.org/@types/cytoscape-dagre/-/cytoscape-dagre-2.3.4.tgz",
+      "integrity": "sha512-uOGXuPfPLFoKZaegjHl9oj4tqONNJuhUl180FiJgRZ35rVijBs6J4UP1Ah6mA6S46h+7pv4ICqpgfdC3EADZlw==",
+      "dev": true,
+      "license": "MIT",
+      "dependencies": {
+        "cytoscape": "^3.31"
+      }
+    },
     "node_modules/@types/deep-eql": {
       "version": "4.0.2",
       "resolved": "https://registry.npmjs.org/@types/deep-eql/-/deep-eql-4.0.2.tgz",
@@ -1824,6 +1845,37 @@
       "resolved": "https://registry.npmjs.org/csstype/-/csstype-3.2.3.tgz",
       "integrity": "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ=="
     },
+    "node_modules/cytoscape": {
+      "version": "3.33.2",
+      "resolved": "https://registry.npmjs.org/cytoscape/-/cytoscape-3.33.2.tgz",
+      "integrity": "sha512-sj4HXd3DokGhzZAdjDejGvTPLqlt84vNFN8m7bGsOzDY5DyVcxIb2ejIXat2Iy7HxWhdT/N1oKyheJ5YdpsGuw==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=0.10"
+      }
+    },
+    "node_modules/cytoscape-dagre": {
+      "version": "2.5.0",
+      "resolved": "https://registry.npmjs.org/cytoscape-dagre/-/cytoscape-dagre-2.5.0.tgz",
+      "integrity": "sha512-VG2Knemmshop4kh5fpLO27rYcyUaaDkRw+6PiX4bstpB+QFt0p2oauMrsjVbUamGWQ6YNavh7x2em2uZlzV44g==",
+      "license": "MIT",
+      "dependencies": {
+        "dagre": "^0.8.5"
+      },
+      "peerDependencies": {
+        "cytoscape": "^3.2.22"
+      }
+    },
+    "node_modules/dagre": {
+      "version": "0.8.5",
+      "resolved": "https://registry.npmjs.org/dagre/-/dagre-0.8.5.tgz",
+      "integrity": "sha512-/aTqmnRta7x7MCCpExk7HQL2O4owCT2h8NT//9I1OQ9vt29Pa0BzSAkR5lwFUcQ7491yVi/3CXU9jQ5o0Mn2Sw==",
+      "license": "MIT",
+      "dependencies": {
+        "graphlib": "^2.1.8",
+        "lodash": "^4.17.15"
+      }
+    },
     "node_modules/de-indent": {
       "version": "1.0.2",
       "resolved": "https://registry.npmjs.org/de-indent/-/de-indent-1.0.2.tgz",
@@ -2252,6 +2304,15 @@
         "url": "https://github.com/sponsors/sindresorhus"
       }
     },
+    "node_modules/graphlib": {
+      "version": "2.1.8",
+      "resolved": "https://registry.npmjs.org/graphlib/-/graphlib-2.1.8.tgz",
+      "integrity": "sha512-jcLLfkpoVGmH7/InMC/1hIvOPSUh38oJtGhvrOFGzioE1DZ+0YW16RgmOJhHiuWTvGiJQ9Z1Ik43JvkRPRvE+A==",
+      "license": "MIT",
+      "dependencies": {
+        "lodash": "^4.17.15"
+      }
+    },
     "node_modules/has-flag": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/has-flag/-/has-flag-4.0.0.tgz",
@@ -2401,8 +2462,7 @@
     "node_modules/lodash": {
       "version": "4.17.23",
       "resolved": "https://registry.npmjs.org/lodash/-/lodash-4.17.23.tgz",
-      "integrity": "sha512-LgVTMpQtIopCi79SJeDiP0TfWi5CNEc/L/aRdTh3yIvmZXTnheWpKjSZhnvMl8iXbC1tFg9gdHHDMLoV7CnG+w==",
+      "integrity": "sha512-LgVTMpQtIopCi79SJeDiP0TfWi5CNEc/L/aRdTh3yIvmZXTnheWpKjSZhnvMl8iXbC1tFg9gdHHDMLoV7CnG+w=="
-      "dev": true
     },
     "node_modules/lodash.merge": {
       "version": "4.6.2",

frontend/package.json

@@ -16,6 +16,8 @@
     "format:check": "prettier --check src/"
   },
   "dependencies": {
+    "cytoscape": "^3.30.0",
+    "cytoscape-dagre": "^2.5.0",
     "dompurify": "^3.3.3",
     "marked": "^17.0.4",
     "pinia": "^2.3.0",
@@ -24,6 +26,8 @@
   },
   "devDependencies": {
     "@eslint/js": "^9.0.0",
+    "@types/cytoscape": "^3.21.4",
+    "@types/cytoscape-dagre": "^2.3.3",
     "@vitest/mocker": "^4.1.2",
     "@types/dompurify": "^3.2.0",
     "@vitejs/plugin-vue": "^6.0.5",

frontend/src/features/analysis/graphApi.ts

@@ -0,0 +1,40 @@
+import { apiFetch } from '../../shared/api/http'
+
+export interface GraphNode {
+  id: string
+  group: 'document' | 'element' | 'page' | 'chunk'
+  label?: string
+  docling_label?: string
+  self_ref?: string
+  text?: string
+  prov_page?: number | null
+  level?: number | null
+  page_no?: number
+  chunk_index?: number
+  title?: string
+  doc_id?: string
+  token_count?: number
+  [key: string]: unknown
+}
+
+export interface GraphEdge {
+  id: string
+  source: string
+  target: string
+  type: 'HAS_ROOT' | 'PARENT_OF' | 'NEXT' | 'ON_PAGE' | 'HAS_CHUNK' | 'DERIVED_FROM'
+  order?: number | null
+}
+
+export interface GraphPayload {
+  doc_id: string
+  nodes: GraphNode[]
+  edges: GraphEdge[]
+  node_count: number
+  edge_count: number
+  truncated: boolean
+  page_count: number
+}
+
+export function fetchDocumentGraph(docId: string): Promise<GraphPayload> {
+  return apiFetch<GraphPayload>(`/api/documents/${encodeURIComponent(docId)}/graph`)
+}

frontend/src/features/analysis/ui/GraphView.vue

@@ -0,0 +1,342 @@
+<template>
+  <div class="graph-view" data-e2e="graph-view">
+    <div v-if="loading" class="graph-placeholder">
+      <div class="spinner-large" />
+      <span>{{ t('results.graphLoading') }}</span>
+    </div>
+    <div v-else-if="error" class="graph-placeholder error" data-e2e="graph-error">
+      <span>{{ error }}</span>
+      <button class="retry-btn" @click="load">{{ t('results.retry') }}</button>
+    </div>
+    <div v-else-if="empty" class="graph-placeholder">
+      <span>{{ t('results.graphEmpty') }}</span>
+    </div>
+    <template v-else>
+      <div class="graph-toolbar">
+        <span class="graph-stats">
+          {{ payload?.node_count }} nodes · {{ payload?.edge_count }} edges ·
+          {{ payload?.page_count }} pages
+        </span>
+        <span class="graph-legend">
+          <span class="legend-chip legend-document">Document</span>
+          <span class="legend-chip legend-section">Section</span>
+          <span class="legend-chip legend-paragraph">Paragraph</span>
+          <span class="legend-chip legend-table">Table</span>
+          <span class="legend-chip legend-figure">Figure</span>
+          <span class="legend-chip legend-page">Page</span>
+          <span class="legend-chip legend-chunk">Chunk</span>
+        </span>
+      </div>
+      <div ref="containerRef" class="graph-canvas" data-e2e="graph-canvas" />
+    </template>
+  </div>
+</template>
+
+<script setup lang="ts">
+import { onMounted, onBeforeUnmount, ref, watch, nextTick } from 'vue'
+import { useI18n } from '../../../shared/i18n'
+import { fetchDocumentGraph, type GraphPayload } from '../graphApi'
+
+const props = defineProps<{ docId: string | null }>()
+const { t } = useI18n()
+
+const containerRef = ref<HTMLDivElement | null>(null)
+const payload = ref<GraphPayload | null>(null)
+const loading = ref(false)
+const error = ref<string | null>(null)
+const empty = ref(false)
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+let cy: any | null = null
+
+const NODE_COLORS: Record<string, string> = {
+  document: '#1E293B',
+  SectionHeader: '#F97316',
+  Paragraph: '#3B82F6',
+  TextElement: '#3B82F6',
+  Table: '#8B5CF6',
+  Figure: '#22C55E',
+  ListItem: '#06B6D4',
+  Formula: '#EC4899',
+  Code: '#14B8A6',
+  Caption: '#EAB308',
+  Page: '#94A3B8',
+  Chunk: '#DC2626',
+}
+
+function nodeColor(n: GraphPayload['nodes'][number]): string {
+  if (n.group === 'document') return NODE_COLORS.document
+  if (n.group === 'page') return NODE_COLORS.Page
+  if (n.group === 'chunk') return NODE_COLORS.Chunk
+  return NODE_COLORS[n.label || 'TextElement'] || NODE_COLORS.TextElement
+}
+
+function nodeLabel(n: GraphPayload['nodes'][number]): string {
+  if (n.group === 'document') return n.title || n.id
+  if (n.group === 'page') return `p.${n.page_no}`
+  if (n.group === 'chunk') return `chunk #${n.chunk_index}`
+  const txt = (n.text || '').slice(0, 40)
+  return txt || n.label || n.docling_label || n.self_ref || n.id
+}
+
+async function load(): Promise<void> {
+  if (!props.docId) {
+    empty.value = true
+    return
+  }
+  loading.value = true
+  error.value = null
+  empty.value = false
+  try {
+    payload.value = await fetchDocumentGraph(props.docId)
+    if (!payload.value.nodes.length) {
+      empty.value = true
+      return
+    }
+    // Flip loading off so the canvas <div> mounts, then wait a tick before init.
+    loading.value = false
+    await nextTick()
+    await renderGraph()
+  } catch (e) {
+    error.value = (e as Error).message || 'Failed to load graph'
+    console.error('Failed to load graph', e)
+  } finally {
+    loading.value = false
+  }
+}
+
+async function renderGraph(): Promise<void> {
+  if (!containerRef.value || !payload.value) return
+  // Dynamic import keeps cytoscape out of the main chunk.
+  const [{ default: cytoscape }, { default: dagre }] = await Promise.all([
+    import('cytoscape'),
+    import('cytoscape-dagre'),
+  ])
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  ;(cytoscape as any).use(dagre)
+
+  if (cy) {
+    cy.destroy()
+    cy = null
+  }
+
+  const elements = [
+    ...payload.value.nodes.map((n) => ({
+      data: {
+        id: n.id,
+        label: nodeLabel(n),
+        bg: nodeColor(n),
+        group: n.group,
+        raw: n,
+      },
+    })),
+    ...payload.value.edges.map((e) => ({
+      data: {
+        id: e.id,
+        source: e.source,
+        target: e.target,
+        type: e.type,
+      },
+    })),
+  ]
+
+  cy = cytoscape({
+    container: containerRef.value,
+    elements,
+    style: [
+      {
+        selector: 'node',
+        style: {
+          'background-color': 'data(bg)',
+          label: 'data(label)',
+          color: '#0F172A',
+          'font-size': 10,
+          'text-wrap': 'ellipsis',
+          'text-max-width': '140px',
+          'text-valign': 'center',
+          'text-halign': 'center',
+          width: 28,
+          height: 28,
+          'border-width': 1,
+          'border-color': '#0F172A',
+        },
+      },
+      {
+        selector: 'node[group = "document"]',
+        style: { shape: 'round-rectangle', width: 60, height: 36, color: '#F8FAFC' },
+      },
+      {
+        selector: 'node[group = "page"]',
+        style: { shape: 'round-rectangle', width: 40, height: 24 },
+      },
+      {
+        selector: 'node[group = "chunk"]',
+        style: { shape: 'diamond', color: '#F8FAFC' },
+      },
+      {
+        selector: 'edge',
+        style: {
+          width: 1,
+          'line-color': '#94A3B8',
+          'target-arrow-color': '#94A3B8',
+          'target-arrow-shape': 'triangle',
+          'curve-style': 'bezier',
+          'font-size': 8,
+          color: '#64748B',
+        },
+      },
+      {
+        selector: 'edge[type = "PARENT_OF"]',
+        style: { 'line-color': '#1E293B', 'target-arrow-color': '#1E293B', width: 1.5 },
+      },
+      {
+        selector: 'edge[type = "NEXT"]',
+        style: { 'line-style': 'dashed', 'line-color': '#64748B' },
+      },
+      {
+        selector: 'edge[type = "ON_PAGE"]',
+        style: { 'line-color': '#CBD5E1', width: 1 },
+      },
+      {
+        selector: 'edge[type = "DERIVED_FROM"]',
+        style: { 'line-color': '#DC2626', 'target-arrow-color': '#DC2626' },
+      },
+    ],
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    layout: {
+      name: 'dagre',
+      rankDir: 'TB',
+      nodeSep: 30,
+      edgeSep: 10,
+      rankSep: 40,
+    } as any,
+    wheelSensitivity: 0.15,
+  })
+}
+
+function disposeGraph(): void {
+  if (cy) {
+    cy.destroy()
+    cy = null
+  }
+}
+
+onMounted(load)
+onBeforeUnmount(disposeGraph)
+watch(
+  () => props.docId,
+  () => {
+    disposeGraph()
+    load()
+  },
+)
+</script>
+
+<style scoped>
+.graph-view {
+  display: flex;
+  flex-direction: column;
+  height: 100%;
+  overflow: hidden;
+}
+
+.graph-toolbar {
+  display: flex;
+  justify-content: space-between;
+  align-items: center;
+  padding: 8px 12px;
+  border-bottom: 1px solid var(--border);
+  gap: 12px;
+  flex-wrap: wrap;
+}
+
+.graph-stats {
+  font-family: 'IBM Plex Mono', monospace;
+  font-size: 11px;
+  color: var(--text-muted);
+}
+
+.graph-legend {
+  display: flex;
+  gap: 6px;
+  flex-wrap: wrap;
+}
+
+.legend-chip {
+  font-size: 10px;
+  font-weight: 600;
+  padding: 2px 8px;
+  border-radius: 10px;
+  color: #f8fafc;
+}
+
+.legend-document {
+  background: #1e293b;
+}
+.legend-section {
+  background: #f97316;
+}
+.legend-paragraph {
+  background: #3b82f6;
+}
+.legend-table {
+  background: #8b5cf6;
+}
+.legend-figure {
+  background: #22c55e;
+}
+.legend-page {
+  background: #94a3b8;
+  color: #0f172a;
+}
+.legend-chunk {
+  background: #dc2626;
+}
+
+.graph-canvas {
+  flex: 1;
+  min-height: 0;
+  background: var(--bg);
+}
+
+.graph-placeholder {
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  justify-content: center;
+  height: 100%;
+  gap: 12px;
+  color: var(--text-muted);
+  font-size: 14px;
+  padding: 32px;
+  text-align: center;
+}
+
+.graph-placeholder.error {
+  color: var(--error);
+}
+
+.retry-btn {
+  background: var(--accent);
+  color: white;
+  border: none;
+  padding: 6px 16px;
+  border-radius: var(--radius-sm);
+  cursor: pointer;
+  font-size: 12px;
+}
+
+.spinner-large {
+  width: 32px;
+  height: 32px;
+  border: 3px solid var(--border-light);
+  border-top-color: var(--accent);
+  border-radius: 50%;
+  animation: spin 0.8s linear infinite;
+}
+
+@keyframes spin {
+  to {
+    transform: rotate(360deg);
+  }
+}
+</style>

frontend/src/features/feature-flags/store.test.ts

@@ -29,12 +29,12 @@ describe('useFeatureFlagStore', () => {
     expect(store.isEnabled('chunking')).toBe(true)
   })
 
-  it('disables chunking when engine is remote', async () => {
+  it('enables chunking when engine is remote', async () => {
     mockApiFetch.mockResolvedValue({ status: 'ok', engine: 'remote' })
     const store = useFeatureFlagStore()
     await store.load()
     expect(store.engine).toBe('remote')
-    expect(store.isEnabled('chunking')).toBe(false)
+    expect(store.isEnabled('chunking')).toBe(true)
   })
 
   it('enables disclaimer when deploymentMode is huggingface', async () => {

frontend/src/features/feature-flags/store.ts

@@ -32,7 +32,7 @@ interface FeatureFlagContext {
 const featureRegistry: Record<FeatureFlag, FeatureFlagDef> = {
   chunking: {
     description: 'Document chunking for RAG preparation',
-    isEnabled: (ctx) => ctx.engine === 'local',
+    isEnabled: (ctx) => ctx.engine !== null,
   },
   disclaimer: {
     description: 'Show shared-instance disclaimer banner',

frontend/src/features/feature-flags/useFeatureFlag.test.ts

@@ -23,6 +23,6 @@ describe('useFeatureFlag', () => {
     expect(flag.value).toBe(true)
 
     store.$patch({ engine: 'remote' })
-    expect(flag.value).toBe(false)
+    expect(flag.value).toBe(true)
   })
 })

frontend/src/pages/StudioPage.vue

@@ -84,6 +84,27 @@
             </svg>
             {{ t('studio.ingest') }}
           </button>
+          <button
+            v-if="chunkingEnabled && ingestionEnabled && ingestionStore.available"
+            class="toggle-btn"
+            data-e2e="toggle-btn maintain-btn"
+            :class="{ active: mode === 'maintain' }"
+            @click="mode = 'maintain'"
+            :disabled="!analysisStore.currentAnalysis"
+          >
+            <svg class="toggle-icon" viewBox="0 0 20 20" fill="currentColor">
+              <path
+                d="M10 3.5a6.5 6.5 0 100 13 6.5 6.5 0 000-13zM6 10a4 4 0 118 0 4 4 0 01-8 0zm4-2a2 2 0 100 4 2 2 0 000-4z"
+              />
+              <path
+                d="M10 1v2M10 17v2M1 10h2M17 10h2M3.5 3.5l1.4 1.4M15.1 15.1l1.4 1.4M3.5 16.5l1.4-1.4M15.1 4.9l1.4-1.4"
+                stroke="currentColor"
+                stroke-width="1.5"
+                fill="none"
+              />
+            </svg>
+            {{ t('studio.maintain') }}
+          </button>
         </div>
       </div>
       <div class="topbar-actions">
@@ -475,6 +496,11 @@
             :chunk-count="analysisStore.currentChunks?.length ?? 0"
           />
         </div>
+
+        <!-- MAINTAIN MODE -->
+        <div v-if="mode === 'maintain'" class="maintain-panel">
+          <GraphView :doc-id="analysisStore.currentAnalysis?.documentId ?? null" />
+        </div>
       </div>
     </div>
   </div>
@@ -489,6 +515,7 @@ import { useIngestionStore } from '../features/ingestion/store'
 import { DocumentUpload, DocumentList } from '../features/document/index'
 import { ResultTabs } from '../features/analysis/index'
 import BboxOverlay from '../features/analysis/ui/BboxOverlay.vue'
+import GraphView from '../features/analysis/ui/GraphView.vue'
 import { ChunkPanel } from '../features/chunking'
 import { IngestPanel } from '../features/ingestion'
 import { useFeatureFlag } from '../features/feature-flags'
@@ -1404,10 +1431,11 @@ onBeforeUnmount(() => {
   padding-top: 16px;
 }
 
-/* Verify / Prepare / Ingest panels */
+/* Verify / Prepare / Ingest / Maintain panels */
 .verify-panel,
 .prepare-panel,
-.ingest-panel-wrapper {
+.ingest-panel-wrapper,
+.maintain-panel {
   height: 100%;
   overflow: hidden;
   display: flex;

frontend/src/shared/i18n.ts

@@ -81,6 +81,10 @@ const messages: Messages = {
     'results.elements': 'Éléments',
     'results.markdown': 'Markdown',
     'results.images': 'Images',
+    'results.graph': 'Graphe',
+    'results.graphLoading': 'Chargement du graphe…',
+    'results.graphEmpty': 'Pas encore de graphe pour ce document (activez Neo4j).',
+    'results.retry': 'Réessayer',
     'results.pageOf': 'Page {current} sur {total}',
     'results.noElements': 'Aucun élément détecté sur cette page',
     'results.noImages': 'Aucune image détectée dans ce document',
@@ -110,6 +114,7 @@ const messages: Messages = {
     // Chunking
     'studio.prepare': 'Préparer',
     'studio.ingest': 'Ingérer',
+    'studio.maintain': 'Maintenir',
     'chunking.settings': 'Chunking',
     'chunking.chunkerType': 'Type de chunker',
     'chunking.maxTokens': 'Tokens max',
@@ -253,6 +258,10 @@ const messages: Messages = {
     'results.elements': 'Elements',
     'results.markdown': 'Markdown',
     'results.images': 'Images',
+    'results.graph': 'Graph',
+    'results.graphLoading': 'Loading graph…',
+    'results.graphEmpty': 'No graph yet for this document (enable Neo4j).',
+    'results.retry': 'Retry',
     'results.pageOf': 'Page {current} of {total}',
     'results.noElements': 'No elements detected on this page',
     'results.noImages': 'No images detected in this document',
@@ -279,6 +288,7 @@ const messages: Messages = {
 
     'studio.prepare': 'Prepare',
     'studio.ingest': 'Ingest',
+    'studio.maintain': 'Maintain',
     'chunking.settings': 'Chunking',
     'chunking.chunkerType': 'Chunker type',
     'chunking.maxTokens': 'Max tokens',