feat(#76): ingest button in Studio Prepare mode

Add one-click ingest button in the Studio topbar when mode=prepare,
ingestion pipeline is available, and chunks exist. Wires to IngestionService
via ingestionStore.ingest(). Availability checked on mount.
Green ingest button styled consistent with topbar actions.

docker-compose.yml

@@ -1,4 +1,56 @@
+# =============================================================================
+# Docling Studio — Production stack
+#
+# Usage:
+#   docker compose up -d
+#
+# Includes OpenSearch single-node + embedding service for the ingestion pipeline.
+# Set OPENSEARCH_URL and EMBEDDING_URL in .env to enable ingestion features.
+# =============================================================================
+
 services:
+  # --- OpenSearch (single-node, security disabled) ---
+  opensearch:
+    image: opensearchproject/opensearch:2
+    environment:
+      discovery.type: single-node
+      DISABLE_SECURITY_PLUGIN: "true"
+      OPENSEARCH_JAVA_OPTS: "-Xms512m -Xmx512m"
+    expose:
+      - "9200"
+    volumes:
+      - opensearch_data:/usr/share/opensearch/data
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:9200/_cluster/health || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+    deploy:
+      resources:
+        limits:
+          memory: 2g
+
+  # --- Embedding service (sentence-transformers) ---
+  embedding:
+    build:
+      context: ./embedding-service
+    expose:
+      - "8001"
+    environment:
+      EMBEDDING_MODEL: ${EMBEDDING_MODEL:-all-MiniLM-L6-v2}
+      EMBEDDING_BATCH_SIZE: ${EMBEDDING_BATCH_SIZE:-64}
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:8001/health || exit 1"]
+      interval: 15s
+      timeout: 10s
+      retries: 20
+      start_period: 120s
+    deploy:
+      resources:
+        limits:
+          memory: 2g
+
+  # --- Backend (FastAPI) ---
   document-parser:
     build:
       context: ./document-parser
@@ -15,11 +67,19 @@ services:
       RATE_LIMIT_RPM: ${RATE_LIMIT_RPM:-100}
       MAX_FILE_SIZE_MB: ${MAX_FILE_SIZE_MB:-50}
       BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-0}
+      OPENSEARCH_URL: http://opensearch:9200
+      EMBEDDING_URL: http://embedding:8001
+    depends_on:
+      opensearch:
+        condition: service_healthy
+      embedding:
+        condition: service_healthy
     deploy:
       resources:
         limits:
           memory: 4g
 
+  # --- Frontend (Nginx) ---
   frontend:
     build:
       context: ./frontend
@@ -29,5 +89,6 @@ services:
       - document-parser
 
 volumes:
+  opensearch_data:
   uploads_data:
   db_data:

document-parser/api/ingestion.py

@@ -0,0 +1,120 @@
+"""Ingestion API — REST endpoints for the embedding → OpenSearch pipeline.
+
+Routes:
+    POST   /api/ingestion/{job_id}   — Trigger ingestion for a completed analysis
+    DELETE /api/ingestion/{doc_id}   — Remove indexed chunks for a document
+    GET    /api/ingestion/status     — Check whether the ingestion pipeline is available
+"""
+
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, HTTPException, Request, status
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/ingestion", tags=["ingestion"])
+
+
+# ---------------------------------------------------------------------------
+# Response schemas
+# ---------------------------------------------------------------------------
+
+
+class IngestionResponse(BaseModel):
+    doc_id: str
+    chunks_indexed: int
+    embedding_dimension: int
+
+    model_config = {"populate_by_name": True}
+
+
+class IngestionStatusResponse(BaseModel):
+    available: bool
+    reason: str = ""
+
+
+# ---------------------------------------------------------------------------
+# Dependency helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_ingestion_service(request: Request):
+    svc = getattr(request.app.state, "ingestion_service", None)
+    if svc is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Ingestion pipeline not available (OpenSearch or embedding service not configured).",
+        )
+    return svc
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/status", response_model=IngestionStatusResponse)
+async def get_ingestion_status(request: Request) -> IngestionStatusResponse:
+    """Return whether the ingestion pipeline (OpenSearch + embedding) is available."""
+    svc = getattr(request.app.state, "ingestion_service", None)
+    if svc is None:
+        return IngestionStatusResponse(
+            available=False,
+            reason="OpenSearch or embedding service not configured",
+        )
+    return IngestionStatusResponse(available=True)
+
+
+@router.post("/{job_id}", response_model=IngestionResponse, status_code=status.HTTP_200_OK)
+async def ingest_analysis(job_id: str, request: Request) -> IngestionResponse:
+    """Run the full ingestion pipeline for a completed analysis job.
+
+    Chains: loaded chunks → embedding → OpenSearch indexing.
+    Idempotent: re-ingesting a document replaces existing indexed chunks.
+    """
+    svc = _get_ingestion_service(request)
+    try:
+        from services.ingestion_service import IngestionError
+
+        result = await svc.ingest(job_id)
+    except IngestionError as exc:
+        logger.warning("Ingestion failed for job %s: %s", job_id, exc)
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)
+        ) from exc
+    except Exception as exc:
+        logger.exception("Unexpected error during ingestion of job %s", job_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Ingestion error: {exc}",
+        ) from exc
+
+    return IngestionResponse(
+        doc_id=result.doc_id,
+        chunks_indexed=result.chunks_indexed,
+        embedding_dimension=result.embedding_dimension,
+    )
+
+
+@router.delete("/{doc_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_ingested(doc_id: str, request: Request) -> None:
+    """Remove all indexed chunks for a document from OpenSearch."""
+    svc = _get_ingestion_service(request)
+    try:
+        from services.ingestion_service import IngestionError
+
+        await svc.delete(doc_id)
+    except IngestionError as exc:
+        logger.warning("Delete failed for document %s: %s", doc_id, exc)
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)
+        ) from exc
+    except Exception as exc:
+        logger.exception("Unexpected error deleting chunks for document %s", doc_id)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Delete error: {exc}",
+        ) from exc

document-parser/main.py

@@ -20,6 +20,7 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from api.analyses import router as analyses_router
 from api.documents import router as documents_router
+from api.ingestion import router as ingestion_router
 from api.schemas import HealthResponse
 from infra.rate_limiter import RateLimiterMiddleware
 from infra.settings import settings
@@ -108,12 +109,46 @@ def _build_document_service(
 # ---------------------------------------------------------------------------
 
 
+def _build_ingestion_service(
+    document_repo: SqliteDocumentRepository,
+    analysis_repo: SqliteAnalysisRepository,
+):
+    """Build ingestion service if OpenSearch + embedding are configured."""
+    if not settings.opensearch_url or not settings.embedding_url:
+        logger.info(
+            "Ingestion pipeline disabled (OPENSEARCH_URL=%r, EMBEDDING_URL=%r)",
+            settings.opensearch_url,
+            settings.embedding_url,
+        )
+        return None
+
+    from infra.embedding_client import EmbeddingClient
+    from infra.opensearch_store import OpenSearchStore
+    from services.ingestion_service import IngestionService
+
+    embedding_svc = EmbeddingClient(settings.embedding_url)
+    vector_store = OpenSearchStore(hosts=[settings.opensearch_url])
+    logger.info(
+        "Ingestion pipeline enabled (opensearch=%s, embedding=%s)",
+        settings.opensearch_url,
+        settings.embedding_url,
+    )
+    return IngestionService(
+        analysis_repo=analysis_repo,
+        document_repo=document_repo,
+        embedding_svc=embedding_svc,
+        vector_store=vector_store,
+        embedding_dimension=settings.embedding_dimension,
+    )
+
+
 @asynccontextmanager
 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.document_service = _build_document_service(document_repo, analysis_repo)
+    app.state.ingestion_service = _build_ingestion_service(document_repo, analysis_repo)
     logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine)
     yield
 
@@ -140,6 +175,7 @@ if settings.rate_limit_rpm > 0:
 
 app.include_router(documents_router)
 app.include_router(analyses_router)
+app.include_router(ingestion_router)
 
 
 @app.get("/api/health", response_model=HealthResponse)

document-parser/services/ingestion_service.py

@@ -0,0 +1,223 @@
+"""Ingestion service — orchestrates the full Docling → embedding → OpenSearch pipeline.
+
+Pipeline stages:
+    1. Load analysis job + document metadata
+    2. Parse chunks from the completed job
+    3. Generate embeddings for each chunk
+    4. Idempotently index chunks in OpenSearch
+
+Idempotency: existing chunks for the document are deleted before re-indexing,
+so re-ingesting a document always produces a clean, up-to-date index.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from domain.models import AnalysisStatus
+from domain.vector_schema import (
+    DEFAULT_INDEX_NAME,
+    ChunkBboxEntry,
+    ChunkOrigin,
+    IndexedChunk,
+    build_index_mapping,
+)
+
+if TYPE_CHECKING:
+    from domain.ports import (
+        AnalysisRepository,
+        DocumentRepository,
+        EmbeddingService,
+        VectorStore,
+    )
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class IngestionResult:
+    """Result returned by a successful ingestion run."""
+
+    doc_id: str
+    chunks_indexed: int
+    embedding_dimension: int
+
+
+class IngestionError(Exception):
+    """Raised when the ingestion pipeline cannot complete."""
+
+
+class IngestionService:
+    """Orchestrates the full ingestion pipeline for a single analysis job."""
+
+    def __init__(
+        self,
+        analysis_repo: AnalysisRepository,
+        document_repo: DocumentRepository,
+        embedding_svc: EmbeddingService,
+        vector_store: VectorStore,
+        *,
+        index_name: str = DEFAULT_INDEX_NAME,
+        embedding_dimension: int = 384,
+    ) -> None:
+        self._analysis_repo = analysis_repo
+        self._document_repo = document_repo
+        self._embedding_svc = embedding_svc
+        self._vector_store = vector_store
+        self._index_name = index_name
+        self._embedding_dimension = embedding_dimension
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    async def ingest(self, job_id: str) -> IngestionResult:
+        """Run the full ingestion pipeline for the given analysis job.
+
+        Args:
+            job_id: ID of a COMPLETED analysis job that has chunks.
+
+        Returns:
+            IngestionResult with doc_id and number of indexed chunks.
+
+        Raises:
+            IngestionError: if any pipeline stage fails.
+        """
+        logger.info("Ingestion started for job %s", job_id)
+
+        # --- Stage 1: load job ---
+        job = await self._analysis_repo.find_by_id(job_id)
+        if job is None:
+            raise IngestionError(f"Analysis job not found: {job_id}")
+        if job.status != AnalysisStatus.COMPLETED:
+            raise IngestionError(
+                f"Job {job_id} is not COMPLETED (status={job.status}). Run analysis first."
+            )
+        if not job.chunks_json:
+            raise IngestionError(f"Job {job_id} has no chunks. Run chunking first.")
+
+        doc = await self._document_repo.find_by_id(job.document_id)
+        if doc is None:
+            raise IngestionError(f"Document not found: {job.document_id}")
+
+        logger.info("Loaded job %s — document: %s", job_id, doc.filename)
+
+        # --- Stage 2: parse chunks ---
+        try:
+            raw_chunks: list[dict] = json.loads(job.chunks_json)
+        except json.JSONDecodeError as exc:
+            raise IngestionError(f"Invalid chunks_json in job {job_id}") from exc
+
+        if not raw_chunks:
+            raise IngestionError(f"Job {job_id} has empty chunk list. Chunk first.")
+
+        texts = [c.get("text", "") for c in raw_chunks]
+        logger.info("Parsed %d chunks from job %s", len(texts), job_id)
+
+        # --- Stage 3: generate embeddings ---
+        try:
+            embeddings = await self._embedding_svc.embed(texts)
+        except Exception as exc:
+            raise IngestionError(f"Embedding generation failed: {exc}") from exc
+
+        if len(embeddings) != len(texts):
+            raise IngestionError(
+                f"Embedding dimension mismatch: got {len(embeddings)} vectors for {len(texts)} texts"
+            )
+
+        # Detect embedding dimension from the first non-empty vector
+        detected_dim = self._embedding_dimension
+        for vec in embeddings:
+            if vec:
+                detected_dim = len(vec)
+                break
+
+        logger.info("Generated %d embeddings (dim=%d)", len(embeddings), detected_dim)
+
+        # --- Stage 4: ensure index exists ---
+        mapping = build_index_mapping(detected_dim)
+        try:
+            await self._vector_store.ensure_index(self._index_name, mapping)
+        except Exception as exc:
+            raise IngestionError(f"Failed to ensure index: {exc}") from exc
+
+        # --- Stage 4b: delete existing chunks (idempotency) ---
+        try:
+            deleted = await self._vector_store.delete_document(self._index_name, doc.id)
+            if deleted:
+                logger.info("Deleted %d existing chunks for document %s", deleted, doc.id)
+        except Exception as exc:
+            raise IngestionError(f"Failed to delete existing chunks: {exc}") from exc
+
+        # --- Stage 5: build IndexedChunk list ---
+        origin = ChunkOrigin(
+            binary_hash=doc.id,  # use doc id as stable identifier
+            filename=doc.filename,
+        )
+
+        indexed_chunks: list[IndexedChunk] = []
+        for i, (raw, emb) in enumerate(zip(raw_chunks, embeddings, strict=True)):
+            bboxes = [
+                ChunkBboxEntry(
+                    page=b.get("page", 0),
+                    x=b["bbox"][0] if b.get("bbox") else 0.0,
+                    y=b["bbox"][1] if b.get("bbox") else 0.0,
+                    w=(b["bbox"][2] - b["bbox"][0])
+                    if b.get("bbox") and len(b["bbox"]) >= 4
+                    else 0.0,
+                    h=(b["bbox"][3] - b["bbox"][1])
+                    if b.get("bbox") and len(b["bbox"]) >= 4
+                    else 0.0,
+                )
+                for b in raw.get("bboxes", [])
+            ]
+
+            chunk = IndexedChunk(
+                doc_id=doc.id,
+                filename=doc.filename,
+                content=raw.get("text", ""),
+                embedding=emb,
+                chunk_index=i,
+                chunk_type="text",
+                page_number=raw.get("sourcePage") or 0,
+                bboxes=bboxes,
+                headings=raw.get("headings", []),
+                doc_items=[],
+                origin=origin,
+            )
+            indexed_chunks.append(chunk)
+
+        # --- Stage 6: bulk index ---
+        try:
+            indexed_count = await self._vector_store.index_chunks(self._index_name, indexed_chunks)
+        except Exception as exc:
+            raise IngestionError(f"Bulk indexing failed: {exc}") from exc
+
+        logger.info(
+            "Ingestion complete — %d/%d chunks indexed for document %s",
+            indexed_count,
+            len(indexed_chunks),
+            doc.id,
+        )
+        return IngestionResult(
+            doc_id=doc.id,
+            chunks_indexed=indexed_count,
+            embedding_dimension=detected_dim,
+        )
+
+    async def delete(self, doc_id: str) -> int:
+        """Remove all indexed chunks for a document.
+
+        Returns:
+            Number of chunks deleted.
+        """
+        logger.info("Deleting indexed chunks for document %s", doc_id)
+        try:
+            deleted = await self._vector_store.delete_document(self._index_name, doc_id)
+        except Exception as exc:
+            raise IngestionError(f"Failed to delete from index: {exc}") from exc
+        logger.info("Deleted %d chunks for document %s", deleted, doc_id)
+        return deleted

document-parser/tests/test_ingestion_api.py

@@ -0,0 +1,112 @@
+"""Tests for the ingestion REST API endpoints."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from main import app
+from services.ingestion_service import IngestionError, IngestionResult
+
+
+@pytest.fixture()
+def client():
+    return TestClient(app, raise_server_exceptions=False)
+
+
+@pytest.fixture(autouse=True)
+def _clear_ingestion_service():
+    """Ensure ingestion_service is reset after each test."""
+    original = getattr(app.state, "ingestion_service", None)
+    yield
+    app.state.ingestion_service = original
+
+
+# ---------------------------------------------------------------------------
+# GET /api/ingestion/status
+# ---------------------------------------------------------------------------
+
+
+def test_status_available(client):
+    app.state.ingestion_service = MagicMock()
+    resp = client.get("/api/ingestion/status")
+    assert resp.status_code == 200
+    assert resp.json()["available"] is True
+
+
+def test_status_unavailable(client):
+    app.state.ingestion_service = None
+    resp = client.get("/api/ingestion/status")
+    assert resp.status_code == 200
+    assert resp.json()["available"] is False
+
+
+# ---------------------------------------------------------------------------
+# POST /api/ingestion/{job_id}
+# ---------------------------------------------------------------------------
+
+
+def test_ingest_success(client):
+    svc = MagicMock()
+    svc.ingest = AsyncMock(
+        return_value=IngestionResult(doc_id="doc-1", chunks_indexed=5, embedding_dimension=384)
+    )
+    app.state.ingestion_service = svc
+    resp = client.post("/api/ingestion/job-1")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["doc_id"] == "doc-1"
+    assert data["chunks_indexed"] == 5
+    assert data["embedding_dimension"] == 384
+
+
+def test_ingest_no_service(client):
+    app.state.ingestion_service = None
+    resp = client.post("/api/ingestion/job-1")
+    assert resp.status_code == 503
+
+
+def test_ingest_ingestion_error(client):
+    svc = MagicMock()
+    svc.ingest = AsyncMock(side_effect=IngestionError("job not found"))
+    app.state.ingestion_service = svc
+    resp = client.post("/api/ingestion/job-1")
+    assert resp.status_code == 422
+    assert "job not found" in resp.json()["detail"]
+
+
+def test_ingest_unexpected_error(client):
+    svc = MagicMock()
+    svc.ingest = AsyncMock(side_effect=RuntimeError("boom"))
+    app.state.ingestion_service = svc
+    resp = client.post("/api/ingestion/job-1")
+    assert resp.status_code == 500
+
+
+# ---------------------------------------------------------------------------
+# DELETE /api/ingestion/{doc_id}
+# ---------------------------------------------------------------------------
+
+
+def test_delete_success(client):
+    svc = MagicMock()
+    svc.delete = AsyncMock(return_value=3)
+    app.state.ingestion_service = svc
+    resp = client.delete("/api/ingestion/doc-1")
+    assert resp.status_code == 204
+
+
+def test_delete_no_service(client):
+    app.state.ingestion_service = None
+    resp = client.delete("/api/ingestion/doc-1")
+    assert resp.status_code == 503
+
+
+def test_delete_ingestion_error(client):
+    svc = MagicMock()
+    svc.delete = AsyncMock(side_effect=IngestionError("opensearch error"))
+    app.state.ingestion_service = svc
+    resp = client.delete("/api/ingestion/doc-1")
+    assert resp.status_code == 422

document-parser/tests/test_ingestion_service.py

@@ -0,0 +1,268 @@
+"""Tests for IngestionService — orchestrated pipeline."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from domain.models import AnalysisJob, AnalysisStatus, Document
+from domain.vector_schema import IndexedChunk
+from services.ingestion_service import IngestionError, IngestionService
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+def _make_doc(doc_id: str = "doc-1", filename: str = "test.pdf") -> Document:
+    return Document(id=doc_id, filename=filename)
+
+
+def _make_job(
+    job_id: str = "job-1",
+    doc_id: str = "doc-1",
+    status: AnalysisStatus = AnalysisStatus.COMPLETED,
+    chunks_json: str | None = None,
+) -> AnalysisJob:
+    if chunks_json is None:
+        chunks_json = json.dumps(
+            [
+                {
+                    "text": "Hello world",
+                    "headings": ["Introduction"],
+                    "sourcePage": 1,
+                    "bboxes": [{"page": 1, "bbox": [10.0, 20.0, 100.0, 50.0]}],
+                },
+                {
+                    "text": "Second chunk",
+                    "headings": [],
+                    "sourcePage": 2,
+                    "bboxes": [],
+                },
+            ]
+        )
+    job = AnalysisJob(id=job_id, document_id=doc_id, status=status)
+    job.chunks_json = chunks_json
+    return job
+
+
+def _make_service(
+    *,
+    job: AnalysisJob | None = None,
+    doc: Document | None = None,
+    embeddings: list[list[float]] | None = None,
+    indexed_count: int = 2,
+) -> IngestionService:
+    if job is None:
+        job = _make_job()
+    if doc is None:
+        doc = _make_doc()
+    if embeddings is None:
+        embeddings = [[0.1] * 384, [0.2] * 384]
+
+    analysis_repo = MagicMock()
+    analysis_repo.find_by_id = AsyncMock(return_value=job)
+
+    document_repo = MagicMock()
+    document_repo.find_by_id = AsyncMock(return_value=doc)
+
+    embedding_svc = MagicMock()
+    embedding_svc.embed = AsyncMock(return_value=embeddings)
+
+    vector_store = MagicMock()
+    vector_store.ensure_index = AsyncMock()
+    vector_store.delete_document = AsyncMock(return_value=0)
+    vector_store.index_chunks = AsyncMock(return_value=indexed_count)
+
+    return (
+        IngestionService(
+            analysis_repo=analysis_repo,
+            document_repo=document_repo,
+            embedding_svc=embedding_svc,
+            vector_store=vector_store,
+        ),
+        analysis_repo,
+        document_repo,
+        embedding_svc,
+        vector_store,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Happy path
+# ---------------------------------------------------------------------------
+
+
+async def test_ingest_success():
+    svc, _, _, embedding_svc, vector_store = _make_service()
+    result = await svc.ingest("job-1")
+
+    assert result.doc_id == "doc-1"
+    assert result.chunks_indexed == 2
+    assert result.embedding_dimension == 384
+
+    # embedding called with 2 texts
+    embedding_svc.embed.assert_awaited_once()
+    texts = embedding_svc.embed.call_args[0][0]
+    assert texts == ["Hello world", "Second chunk"]
+
+    # ensure_index called
+    vector_store.ensure_index.assert_awaited_once()
+
+    # delete_document called for idempotency
+    vector_store.delete_document.assert_awaited_once_with("docling-studio-chunks", "doc-1")
+
+    # index_chunks called with IndexedChunk list
+    vector_store.index_chunks.assert_awaited_once()
+    chunks_arg = vector_store.index_chunks.call_args[0][1]
+    assert len(chunks_arg) == 2
+    assert all(isinstance(c, IndexedChunk) for c in chunks_arg)
+    assert chunks_arg[0].content == "Hello world"
+    assert chunks_arg[0].chunk_index == 0
+    assert chunks_arg[1].chunk_index == 1
+
+
+async def test_ingest_preserves_bboxes():
+    # Rebuild to inspect chunks
+    svc2, _, _, _, vector_store = _make_service()
+    await svc2.ingest("job-1")
+    chunks_arg = vector_store.index_chunks.call_args[0][1]
+    first_chunk = chunks_arg[0]
+    assert len(first_chunk.bboxes) == 1
+    bbox = first_chunk.bboxes[0]
+    assert bbox.page == 1
+    assert bbox.x == 10.0
+    assert bbox.y == 20.0
+    assert bbox.w == 90.0  # right - left = 100 - 10
+    assert bbox.h == 30.0  # bottom - top = 50 - 20
+
+
+async def test_ingest_preserves_headings():
+    svc, _, _, _, vector_store = _make_service()
+    await svc.ingest("job-1")
+    chunks = vector_store.index_chunks.call_args[0][1]
+    assert chunks[0].headings == ["Introduction"]
+    assert chunks[1].headings == []
+
+
+async def test_ingest_idempotent_deletes_first():
+    svc, _, _, _, vector_store = _make_service()
+    vector_store.delete_document = AsyncMock(return_value=5)
+
+    result = await svc.ingest("job-1")
+    vector_store.delete_document.assert_awaited_once()
+    assert result.chunks_indexed == 2
+
+
+async def test_ingest_detects_embedding_dimension():
+    embeddings = [[0.1] * 768, [0.2] * 768]
+    svc, *_ = _make_service(embeddings=embeddings)
+    result = await svc.ingest("job-1")
+    assert result.embedding_dimension == 768
+
+
+# ---------------------------------------------------------------------------
+# Error cases — job not found
+# ---------------------------------------------------------------------------
+
+
+async def test_ingest_job_not_found():
+    svc, analysis_repo, *_ = _make_service()
+    analysis_repo.find_by_id = AsyncMock(return_value=None)
+
+    with pytest.raises(IngestionError, match="not found"):
+        await svc.ingest("missing-job")
+
+
+async def test_ingest_job_not_completed():
+    job = _make_job(status=AnalysisStatus.RUNNING)
+    svc, *_ = _make_service(job=job)
+
+    with pytest.raises(IngestionError, match="not COMPLETED"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_job_no_chunks():
+    job = _make_job(chunks_json=None)
+    job.chunks_json = None
+    svc, *_ = _make_service(job=job)
+
+    with pytest.raises(IngestionError, match="no chunks"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_job_empty_chunks():
+    job = _make_job(chunks_json="[]")
+    svc, *_ = _make_service(job=job)
+
+    with pytest.raises(IngestionError, match="empty"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_doc_not_found():
+    svc, _, document_repo, *_ = _make_service()
+    document_repo.find_by_id = AsyncMock(return_value=None)
+
+    with pytest.raises(IngestionError, match="Document not found"):
+        await svc.ingest("job-1")
+
+
+# ---------------------------------------------------------------------------
+# Error cases — pipeline failures
+# ---------------------------------------------------------------------------
+
+
+async def test_ingest_embedding_failure():
+    svc, _, _, embedding_svc, _ = _make_service()
+    embedding_svc.embed = AsyncMock(side_effect=ConnectionError("embedding service down"))
+
+    with pytest.raises(IngestionError, match="Embedding generation failed"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_index_failure():
+    svc, _, _, _, vector_store = _make_service()
+    vector_store.ensure_index = AsyncMock(side_effect=RuntimeError("opensearch down"))
+
+    with pytest.raises(IngestionError, match="Failed to ensure index"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_bulk_index_failure():
+    svc, _, _, _, vector_store = _make_service()
+    vector_store.index_chunks = AsyncMock(side_effect=RuntimeError("bulk failed"))
+
+    with pytest.raises(IngestionError, match="Bulk indexing failed"):
+        await svc.ingest("job-1")
+
+
+async def test_ingest_embedding_count_mismatch():
+    svc, _, _, embedding_svc, _ = _make_service()
+    embedding_svc.embed = AsyncMock(return_value=[[0.1] * 384])  # only 1 instead of 2
+
+    with pytest.raises(IngestionError, match="mismatch"):
+        await svc.ingest("job-1")
+
+
+# ---------------------------------------------------------------------------
+# Delete
+# ---------------------------------------------------------------------------
+
+
+async def test_delete_calls_vector_store():
+    svc, _, _, _, vector_store = _make_service()
+    vector_store.delete_document = AsyncMock(return_value=7)
+
+    count = await svc.delete("doc-1")
+    assert count == 7
+    vector_store.delete_document.assert_awaited_once_with("docling-studio-chunks", "doc-1")
+
+
+async def test_delete_raises_ingestion_error_on_failure():
+    svc, _, _, _, vector_store = _make_service()
+    vector_store.delete_document = AsyncMock(side_effect=RuntimeError("opensearch down"))
+
+    with pytest.raises(IngestionError, match="Failed to delete"):
+        await svc.delete("doc-1")

e2e/api/src/test/java/E2ERunner.java

@@ -4,7 +4,14 @@ class E2ERunner {
 
     @Karate.Test
     Karate testAll() {
-        return Karate.run("classpath:health", "classpath:documents", "classpath:analyses", "classpath:workflows")
+        return Karate.run("classpath:health", "classpath:documents", "classpath:analyses", "classpath:workflows", "classpath:ingestion")
+                .relativeTo(getClass());
+    }
+
+    @Karate.Test
+    Karate testIngestion() {
+        return Karate.run("classpath:ingestion")
+                .tags("@ingestion")
                 .relativeTo(getClass());
     }
 

e2e/api/src/test/resources/ingestion/ingestion-pipeline.feature

@@ -0,0 +1,125 @@
+@e2e @ingestion
+Feature: Ingestion pipeline — PDF → chunks indexed in OpenSearch
+
+  Tests the complete ingestion workflow:
+  upload PDF → analyze → chunk → ingest → verify in OpenSearch.
+
+  Requires the full dev stack (OpenSearch + embedding service running).
+  Skip by excluding the @ingestion tag when OpenSearch is not available.
+
+  Background:
+    * url baseUrl
+
+  Scenario: Full ingestion workflow — PDF becomes searchable chunks in OpenSearch
+
+    # ---- Step 1: Check ingestion pipeline is available ----
+    Given path '/api/ingestion/status'
+    When method GET
+    Then status 200
+    And match response.available == true
+
+    # ---- Step 2: Upload a PDF ----
+    Given path '/api/documents/upload'
+    And multipart file file = { read: 'classpath:common/data/generated/medium.pdf', filename: 'ingest-test.pdf', contentType: 'application/pdf' }
+    When method POST
+    Then status 200
+    And match response.id == '#string'
+    And match response.filename == 'ingest-test.pdf'
+    * def docId = response.id
+
+    # ---- Step 3: Run analysis (with document JSON for chunking) ----
+    Given path '/api/analyses'
+    And request { documentId: '#(docId)', pipelineOptions: { doOcr: false, tableMode: 'fast' } }
+    When method POST
+    Then status 200
+    And match response.status == 'PENDING'
+    * def jobId = response.id
+
+    # ---- Step 4: Poll until analysis is COMPLETED ----
+    Given path '/api/analyses', jobId
+    And retry until response.status == 'COMPLETED' || response.status == 'FAILED'
+    When method GET
+    Then status 200
+    And match response.status == 'COMPLETED'
+    And match response.hasDocumentJson == true
+
+    # ---- Step 5: Run chunking to produce chunk list ----
+    Given path '/api/analyses', jobId, 'rechunk'
+    And request { chunkingOptions: { chunkerType: 'hybrid', maxTokens: 512, mergePeers: true } }
+    When method POST
+    Then status 200
+    And match response == '#[_ > 0]'
+    * def chunkCount = response.length
+
+    # ---- Step 6: Trigger ingestion ----
+    Given path '/api/ingestion', jobId
+    When method POST
+    Then status 200
+    And match response.docId == '#string'
+    And match response.chunksIndexed == '#number'
+    And match response.embeddingDimension == '#number'
+    And match response.chunksIndexed > 0
+    And match response.embeddingDimension > 0
+    * def indexedCount = response.chunksIndexed
+    * def embDim = response.embeddingDimension
+
+    # Verify at least as many chunks as returned by rechunk
+    * assert indexedCount >= 1
+
+    # Embedding dimension must be a valid vector size (e.g. 384 for all-MiniLM-L6-v2)
+    * assert embDim >= 128
+
+    # ---- Step 7: Re-ingest is idempotent ----
+    Given path '/api/ingestion', jobId
+    When method POST
+    Then status 200
+    And match response.chunksIndexed == '#number'
+    And match response.chunksIndexed > 0
+
+    # ---- Step 8: Delete indexed chunks ----
+    Given path '/api/ingestion', docId
+    When method DELETE
+    Then status 204
+
+    # ---- Step 9: Cleanup — delete analysis and document ----
+    Given path '/api/analyses', jobId
+    When method DELETE
+    Then status 204
+
+    Given path '/api/documents', docId
+    When method DELETE
+    Then status 204
+
+  Scenario: Ingestion fails gracefully when job is not COMPLETED
+
+    # Upload a document
+    Given path '/api/documents/upload'
+    And multipart file file = { read: 'classpath:common/data/generated/medium.pdf', filename: 'ingest-fail-test.pdf', contentType: 'application/pdf' }
+    When method POST
+    Then status 200
+    * def docId = response.id
+
+    # Create analysis but do NOT wait for completion
+    Given path '/api/analyses'
+    And request { documentId: '#(docId)' }
+    When method POST
+    Then status 200
+    * def jobId = response.id
+
+    # Immediately attempt ingestion — should fail with 422 (job not COMPLETED or no chunks)
+    Given path '/api/ingestion', jobId
+    When method POST
+    * def sc = responseStatus
+    Then assert sc == 422 || sc == 503
+
+    # Cleanup
+    Given path '/api/documents', docId
+    When method DELETE
+    * def ignored = responseStatus
+
+  Scenario: Ingestion status — available when OpenSearch is configured
+
+    Given path '/api/ingestion/status'
+    When method GET
+    Then status 200
+    And match response == { available: '#boolean', reason: '#string' }

frontend/src/features/ingestion/api.test.ts

@@ -0,0 +1,78 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { ingestAnalysis, deleteIngested, fetchIngestionStatus } from './api'
+
+const mockFetch = vi.fn()
+vi.stubGlobal('fetch', mockFetch)
+
+beforeEach(() => {
+  mockFetch.mockReset()
+})
+
+describe('ingestAnalysis', () => {
+  it('posts to /api/ingestion/:jobId', async () => {
+    mockFetch.mockResolvedValue({
+      ok: true,
+      status: 200,
+      json: () => Promise.resolve({ docId: 'doc-1', chunksIndexed: 5, embeddingDimension: 384 }),
+    })
+    const result = await ingestAnalysis('job-1')
+    expect(mockFetch).toHaveBeenCalledWith(
+      '/api/ingestion/job-1',
+      expect.objectContaining({ method: 'POST' }),
+    )
+    expect(result.chunksIndexed).toBe(5)
+    expect(result.docId).toBe('doc-1')
+  })
+
+  it('throws on non-ok response', async () => {
+    mockFetch.mockResolvedValue({
+      ok: false,
+      status: 422,
+      json: () => Promise.resolve({ detail: 'job not completed' }),
+    })
+    await expect(ingestAnalysis('job-bad')).rejects.toThrow('job not completed')
+  })
+})
+
+describe('deleteIngested', () => {
+  it('deletes /api/ingestion/:docId', async () => {
+    mockFetch.mockResolvedValue({ ok: true, status: 204, json: () => Promise.resolve(null) })
+    await deleteIngested('doc-1')
+    expect(mockFetch).toHaveBeenCalledWith(
+      '/api/ingestion/doc-1',
+      expect.objectContaining({ method: 'DELETE' }),
+    )
+  })
+
+  it('ignores 404 response', async () => {
+    mockFetch.mockResolvedValue({ ok: false, status: 404, json: () => Promise.resolve({}) })
+    await expect(deleteIngested('doc-missing')).resolves.toBeUndefined()
+  })
+
+  it('throws on other errors', async () => {
+    mockFetch.mockResolvedValue({
+      ok: false,
+      status: 500,
+      json: () => Promise.resolve({ detail: 'server error' }),
+    })
+    await expect(deleteIngested('doc-1')).rejects.toThrow('server error')
+  })
+})
+
+describe('fetchIngestionStatus', () => {
+  it('gets /api/ingestion/status', async () => {
+    mockFetch.mockResolvedValue({
+      ok: true,
+      status: 200,
+      json: () => Promise.resolve({ available: true, reason: '' }),
+    })
+    const result = await fetchIngestionStatus()
+    expect(result.available).toBe(true)
+  })
+
+  it('returns unavailable on non-ok', async () => {
+    mockFetch.mockResolvedValue({ ok: false, status: 503 })
+    const result = await fetchIngestionStatus()
+    expect(result.available).toBe(false)
+  })
+})

frontend/src/features/ingestion/api.ts

@@ -0,0 +1,39 @@
+/**
+ * Ingestion API client — wraps /api/ingestion endpoints.
+ */
+
+export interface IngestionResult {
+  docId: string
+  chunksIndexed: number
+  embeddingDimension: number
+}
+
+export interface IngestionStatus {
+  available: boolean
+  reason: string
+}
+
+export async function ingestAnalysis(jobId: string): Promise<IngestionResult> {
+  const resp = await fetch(`/api/ingestion/${jobId}`, { method: 'POST' })
+  if (!resp.ok) {
+    const body = await resp.json().catch(() => ({}))
+    throw new Error(body.detail ?? `Ingestion failed (${resp.status})`)
+  }
+  return resp.json()
+}
+
+export async function deleteIngested(docId: string): Promise<void> {
+  const resp = await fetch(`/api/ingestion/${docId}`, { method: 'DELETE' })
+  if (!resp.ok && resp.status !== 404) {
+    const body = await resp.json().catch(() => ({}))
+    throw new Error(body.detail ?? `Delete failed (${resp.status})`)
+  }
+}
+
+export async function fetchIngestionStatus(): Promise<IngestionStatus> {
+  const resp = await fetch('/api/ingestion/status')
+  if (!resp.ok) {
+    return { available: false, reason: `HTTP ${resp.status}` }
+  }
+  return resp.json()
+}

frontend/src/features/ingestion/store.ts

@@ -0,0 +1,72 @@
+/**
+ * Ingestion store — tracks which documents are indexed in OpenSearch
+ * and exposes actions to ingest / delete indexed chunks.
+ */
+import { defineStore } from 'pinia'
+import { ref } from 'vue'
+import { deleteIngested, fetchIngestionStatus, ingestAnalysis } from './api'
+
+export const useIngestionStore = defineStore('ingestion', () => {
+  /** Map of docId → chunk count for indexed documents. */
+  const ingestedDocs = ref<Record<string, number>>({})
+
+  /** Whether the ingestion pipeline (OpenSearch + embedding) is available. */
+  const available = ref(false)
+
+  /** True while an ingestion is running. */
+  const ingesting = ref(false)
+
+  /** Last ingestion error message, if any. */
+  const error = ref<string | null>(null)
+
+  async function checkAvailability(): Promise<void> {
+    try {
+      const status = await fetchIngestionStatus()
+      available.value = status.available
+    } catch {
+      available.value = false
+    }
+  }
+
+  async function ingest(jobId: string, docId: string): Promise<number> {
+    ingesting.value = true
+    error.value = null
+    try {
+      const result = await ingestAnalysis(jobId)
+      ingestedDocs.value = { ...ingestedDocs.value, [docId]: result.chunksIndexed }
+      return result.chunksIndexed
+    } catch (e) {
+      error.value = (e as Error).message || 'Ingestion failed'
+      throw e
+    } finally {
+      ingesting.value = false
+    }
+  }
+
+  async function deleteIngestd(docId: string): Promise<void> {
+    try {
+      await deleteIngested(docId)
+      const next = { ...ingestedDocs.value }
+      delete next[docId]
+      ingestedDocs.value = next
+    } catch (e) {
+      error.value = (e as Error).message || 'Delete failed'
+      throw e
+    }
+  }
+
+  function clearError(): void {
+    error.value = null
+  }
+
+  return {
+    ingestedDocs,
+    available,
+    ingesting,
+    error,
+    checkAvailability,
+    ingest,
+    deleteIngested: deleteIngestd,
+    clearError,
+  }
+})

frontend/src/pages/StudioPage.vue

@@ -94,6 +94,25 @@
           </svg>
           {{ analysisStore.running ? t('studio.analyzing') : t('studio.run') }}
         </button>
+        <button
+          v-if="
+            mode === 'prepare' && ingestionStore.available && analysisStore.currentChunks.length > 0
+          "
+          class="topbar-btn ingest"
+          data-e2e="ingest-btn"
+          :disabled="ingestionStore.ingesting"
+          @click="handleIngest"
+        >
+          <div v-if="ingestionStore.ingesting" class="spinner-sm" />
+          <svg v-else viewBox="0 0 20 20" fill="currentColor" class="btn-icon">
+            <path
+              fill-rule="evenodd"
+              d="M3 17a1 1 0 011-1h12a1 1 0 110 2H4a1 1 0 01-1-1zm3.293-7.707a1 1 0 011.414 0L9 10.586V3a1 1 0 112 0v7.586l1.293-1.293a1 1 0 111.414 1.414l-3 3a1 1 0 01-1.414 0l-3-3a1 1 0 010-1.414z"
+              clip-rule="evenodd"
+            />
+          </svg>
+          {{ ingestionStore.ingesting ? t('ingestion.ingesting') : t('ingestion.ingest') }}
+        </button>
       </div>
     </div>
 
@@ -465,6 +484,7 @@ import BboxOverlay from '../features/analysis/ui/BboxOverlay.vue'
 import { ChunkPanel } from '../features/chunking'
 import { useFeatureFlag } from '../features/feature-flags'
 import { getPreviewUrl } from '../features/document/api'
+import { useIngestionStore } from '../features/ingestion/store'
 import { useI18n } from '../shared/i18n'
 import type { ChunkBbox, PipelineOptions } from '../shared/types'
 
@@ -472,6 +492,7 @@ const route = useRoute()
 const router = useRouter()
 const documentStore = useDocumentStore()
 const analysisStore = useAnalysisStore()
+const ingestionStore = useIngestionStore()
 const { t } = useI18n()
 const chunkingEnabled = useFeatureFlag('chunking')
 
@@ -595,9 +616,21 @@ watch(
   },
 )
 
+async function handleIngest(): Promise<void> {
+  const analysis = analysisStore.currentAnalysis
+  if (!analysis || !selectedDoc.value) return
+  try {
+    const count = await ingestionStore.ingest(analysis.id, selectedDoc.value.id)
+    console.warn(`Ingestion complete — ${count} chunks indexed.`)
+  } catch (e) {
+    console.error('Ingestion failed', e)
+  }
+}
+
 onMounted(async () => {
   await documentStore.load()
   analysisStore.load()
+  ingestionStore.checkAvailability()
 
   // Restore analysis from history via query param
   const analysisId = route.query.analysisId
@@ -811,6 +844,21 @@ onBeforeUnmount(() => {
   cursor: not-allowed;
 }
 
+.topbar-btn.ingest {
+  background: var(--success, #22c55e);
+  border-color: var(--success, #22c55e);
+  color: white;
+}
+
+.topbar-btn.ingest:hover:not(:disabled) {
+  background: color-mix(in srgb, var(--success, #22c55e) 85%, black);
+}
+
+.topbar-btn.ingest:disabled {
+  opacity: 0.6;
+  cursor: not-allowed;
+}
+
 .topbar-btn .btn-icon {
   width: 16px;
   height: 16px;

frontend/src/shared/i18n.ts

@@ -107,6 +107,22 @@ const messages: Messages = {
     'history.emptyDocs': 'Aucun document. Importez un document depuis le Studio.',
     'history.open': 'Ouvrir',
 
+    // Ingestion / My Documents
+    'ingestion.search': 'Rechercher…',
+    'ingestion.sortName': 'Nom',
+    'ingestion.sortDate': 'Date',
+    'ingestion.filterAll': 'Tous',
+    'ingestion.filterIndexed': 'Indexés',
+    'ingestion.filterNotIndexed': 'Non indexés',
+    'ingestion.indexed': 'Indexé',
+    'ingestion.notIndexed': 'Non indexé',
+    'ingestion.chunksIndexed': '{n} chunks',
+    'ingestion.openInStudio': 'Ouvrir dans Studio',
+    'ingestion.ingest': 'Indexer',
+    'ingestion.ingesting': 'Indexation…',
+    'ingestion.ingestSuccess': 'Indexation réussie — {n} chunks indexés.',
+    'ingestion.ingestError': 'Erreur d\u2019indexation : {msg}',
+
     // Chunking
     'studio.prepare': 'Préparer',
     'chunking.settings': 'Chunking',
@@ -243,6 +259,22 @@ const messages: Messages = {
     'history.emptyDocs': 'No documents yet. Upload a document from the Studio.',
     'history.open': 'Open',
 
+    // Ingestion / My Documents
+    'ingestion.search': 'Search…',
+    'ingestion.sortName': 'Name',
+    'ingestion.sortDate': 'Date',
+    'ingestion.filterAll': 'All',
+    'ingestion.filterIndexed': 'Indexed',
+    'ingestion.filterNotIndexed': 'Not indexed',
+    'ingestion.indexed': 'Indexed',
+    'ingestion.notIndexed': 'Not indexed',
+    'ingestion.chunksIndexed': '{n} chunks',
+    'ingestion.openInStudio': 'Open in Studio',
+    'ingestion.ingest': 'Index',
+    'ingestion.ingesting': 'Indexing…',
+    'ingestion.ingestSuccess': 'Indexed successfully — {n} chunks.',
+    'ingestion.ingestError': 'Indexing error: {msg}',
+
     'studio.prepare': 'Prepare',
     'chunking.settings': 'Chunking',
     'chunking.chunkerType': 'Chunker type',