📄 feat: Add Opt-In DOCX Extraction Profile

.github/workflows/ci.yml

@@ -31,6 +31,7 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r requirements.txt
           pip install -r test_requirements.txt
+          pip install -r requirements.extraction.txt
 
       - name: Run unit tests
         env:

EXTRACTION.md

@@ -0,0 +1,56 @@
+# Opt-in document extraction contract
+
+`POST /v1/extract` implements a **document-v1** profile for DOCX only. It does not
+replace `/text`, alter ingestion/chunking, or invoke embeddings, OCR or the vector
+store. The route is disabled by default and requires a verified `JWT_SECRET`
+token with an `id` claim when enabled. The main RAG application **still initializes
+its vector store and embeddings at startup**. A parsing-only deployment remains a
+separate migration.
+
+Install the pinned optional engine in a custom image or Python environment,
+then opt in explicitly before starting (or restarting) the API. With the flag
+off, the route is not registered and does not parse multipart bodies:
+
+```sh
+pip install -r requirements.extraction.txt
+export RAG_EXTRACTION_API_ENABLED=true
+```
+
+Send multipart `file` and `profile=document-v1`. Use DOCX MIME or a `.docx`
+filename with a generic MIME. Success includes `text` (Markdown), `format`,
+`profile`, `completeness` (`complete`/`partial`), `may_omit_content`,
+`pages_needing_ocr` (empty until PDF support), `truncated` (always false), and
+`parser: {name, version}`. An embedded image marks the DOCX as *partial*. Never
+use partial text as proof of full content inspection. `complete` means the
+supported conversion finished without *known* omitted image entries, not that
+all information in the source is provably inspectable. No hosted OCR or
+second-parser fallback runs inside this endpoint.
+
+Failures use `detail.code`, not the native exception message:
+
+| Status | Codes | Action |
+|---|---|---|
+| 400/415 | `UNSUPPORTED_PROFILE`, `UNSUPPORTED_DOCUMENT_TYPE` | Select a supported profile/type |
+| 401/404 | `EXTRACTION_AUTH_REQUIRED`, `EXTRACTION_DISABLED` | Authenticate/opt in |
+| 413 | `PARSER_INPUT_LIMIT`, `PARSER_OUTPUT_LIMIT`, `ZIP_BOMB` | Hard refusal; never send the same bytes to another parser |
+| 422 | `ARCHIVE_INVALID`, `NO_DOCUMENT_TEXT`, `PARSE_FAILED` | Unusable archive or empty/unconvertible document |
+| 429 | `CONCURRENCY_LIMIT` | Retry later; not a reason to invoke paid OCR |
+| 503/504 | `PARSER_UNAVAILABLE`, `PARSER_CRASH`, `PARSER_TIMEOUT` | Retry or fix the service |
+
+The route checks the input limit of 15 MiB while staging the upload;
+serialized output is capped at 15 MiB before IPC. Starlette may have already
+spooled a multipart upload before the route runs: configure an upstream HTTP
+body-size limit as well for internet-facing deployments. The child checks
+*actual decompressed* ZIP entry bytes: at most
+25 MiB per entry, 100 MiB in total and 4,096 entries. Defaults are two active
+parses and six queued per API process. Set `RAG_EXTRACTION_CONCURRENT`,
+`RAG_EXTRACTION_QUEUED` and `RAG_EXTRACTION_TIMEOUT_SECONDS` to tune admission
+and the overall 30-second default deadline (queue wait, upload staging, parse).
+On cancellation/timeout the child is killed and reaped before its temp file is
+removed and its slot is reused.
+
+This is the first **service-side** slice. Existing LibreChat local parsing and
+RAG `/text` behavior remain in place until cross-service tests establish policy,
+authorization, preview, failure and compatibility behavior for each consumer.
+The real DOCX test fixture is copied from Marco's LibreChat AnyDoc PR #14701 at
+`fb7bbcd9cf75f4f78ecbd5a8780685c481600be2`.

app/models.py

@@ -2,7 +2,23 @@
 import hashlib
 from enum import Enum
 from pydantic import BaseModel
-from typing import Optional, List
+from typing import Optional, List, Literal
+
+
+class ParserProvenance(BaseModel):
+    name: Literal["anydoc"]
+    version: str
+
+
+class ExtractionResult(BaseModel):
+    profile: Literal["document-v1"]
+    text: str
+    format: Literal["markdown"]
+    completeness: Literal["complete", "partial"]
+    may_omit_content: bool
+    pages_needing_ocr: List[int]
+    truncated: bool
+    parser: ParserProvenance
 
 
 class DocumentResponse(BaseModel):

app/routes/extraction_routes.py

@@ -0,0 +1,124 @@
+"""Versioned document extraction. No embeddings, vector writes, or OCR calls."""
+
+import asyncio
+import math
+import os
+import tempfile
+from pathlib import Path
+
+import aiofiles
+from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
+
+from app.config import RAG_UPLOAD_DIR, logger
+from app.models import ExtractionResult
+from app.services.extraction import (
+    ExtractionAdmission,
+    ExtractionBusy,
+    ExtractionFailure,
+    run_worker,
+)
+
+router = APIRouter(prefix="/v1")
+DOCX_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+MAX_INPUT_BYTES = 15 * 1024 * 1024
+_DEFAULT_TIMEOUT = 30.0
+_admission: ExtractionAdmission | None = None
+
+
+def _error(code: str, status_code: int) -> HTTPException:
+    return HTTPException(status_code=status_code, detail={"code": code})
+
+
+def _get_admission() -> ExtractionAdmission:
+    global _admission
+    if _admission is None:
+        _admission = ExtractionAdmission(
+            concurrent=int(os.getenv("RAG_EXTRACTION_CONCURRENT", "2")),
+            queued=int(os.getenv("RAG_EXTRACTION_QUEUED", "6")),
+        )
+    return _admission
+
+
+async def _save_bounded(file: UploadFile, path: Path) -> None:
+    size = 0
+    async with aiofiles.open(path, "wb") as output:
+        while chunk := await file.read(64 * 1024):
+            size += len(chunk)
+            if size > MAX_INPUT_BYTES:
+                raise _error("PARSER_INPUT_LIMIT", 413)
+            await output.write(chunk)
+
+
+@router.post("/extract", response_model=ExtractionResult)
+async def extract_document(
+    request: Request,
+    file: UploadFile = File(...),
+    profile: str = Form(...),
+) -> ExtractionResult:
+    # Existing /text remains unchanged. An operator must explicitly enable
+    # and install this separate profile before moving any LibreChat caller.
+    if os.getenv("RAG_EXTRACTION_API_ENABLED", "false").lower() not in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }:
+        raise _error("EXTRACTION_DISABLED", 404)
+    # Legacy RAG deployments may run without auth; expensive extraction is
+    # never allowed anonymously, even when those older routes are public.
+    if not os.getenv("JWT_SECRET") or not getattr(request.state, "user", {}).get("id"):
+        raise _error("EXTRACTION_AUTH_REQUIRED", 401)
+    if profile != "document-v1":
+        raise _error("UNSUPPORTED_PROFILE", 400)
+    content_type = (file.content_type or "").split(";")[0].strip().lower()
+    extension = Path(file.filename or "").suffix.lower()
+    if content_type == "application/pdf" or not (
+        content_type == DOCX_TYPE
+        or (
+            content_type in {"application/octet-stream", "binary/octet-stream", ""}
+            and extension == ".docx"
+        )
+    ):
+        raise _error("UNSUPPORTED_DOCUMENT_TYPE", 415)
+    if file.size is not None and file.size > MAX_INPUT_BYTES:
+        raise _error("PARSER_INPUT_LIMIT", 413)
+
+    try:
+        admission = _get_admission()
+        timeout = float(
+            os.getenv("RAG_EXTRACTION_TIMEOUT_SECONDS", str(_DEFAULT_TIMEOUT))
+        )
+        if not math.isfinite(timeout) or timeout <= 0:
+            raise ValueError("Invalid extraction timeout")
+        async with asyncio.timeout(timeout):
+            async with admission.slot():
+                fd, filename = tempfile.mkstemp(
+                    prefix="rag-extract-", suffix=".docx", dir=RAG_UPLOAD_DIR
+                )
+                os.close(fd)
+                path = Path(filename)
+                try:
+                    await _save_bounded(file, path)
+                    return await run_worker(path)
+                finally:
+                    path.unlink(missing_ok=True)
+    except ExtractionBusy:
+        raise _error("CONCURRENCY_LIMIT", 429)
+    except ExtractionFailure as exc:
+        status_code = {
+            "ZIP_BOMB": 413,
+            "ARCHIVE_INVALID": 422,
+            "PARSER_OUTPUT_LIMIT": 413,
+            "NO_DOCUMENT_TEXT": 422,
+            "PARSE_FAILED": 422,
+            "PARSER_UNAVAILABLE": 503,
+            "PARSER_CRASH": 503,
+        }[exc.code]
+        raise _error(exc.code, status_code)
+    except TimeoutError:
+        raise _error("PARSER_TIMEOUT", 504)
+    except (OSError, ValueError) as exc:
+        logger.error(
+            "Extraction infrastructure unavailable | error=%s", type(exc).__name__
+        )
+        raise _error("PARSER_UNAVAILABLE", 503)

app/services/extraction.py

@@ -0,0 +1,100 @@
+"""Bounded, cancellable process boundary for opt-in document extraction."""
+
+import asyncio
+import json
+import sys
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import AsyncIterator
+
+from app.models import ExtractionResult
+
+MAX_IPC_BYTES = 15 * 1024 * 1024
+
+
+class ExtractionBusy(Exception):
+    pass
+
+
+class ExtractionFailure(Exception):
+    def __init__(self, code: str):
+        self.code = code
+
+
+class ExtractionAdmission:
+    """One per serving process: bound uploads waiting and native children running."""
+
+    def __init__(self, concurrent: int = 2, queued: int = 6):
+        if concurrent < 1 or queued < 0:
+            raise ValueError("Invalid extraction admission limits")
+        self._capacity = concurrent + queued
+        self._pending = 0
+        self._slots = asyncio.Semaphore(concurrent)
+
+    @asynccontextmanager
+    async def slot(self) -> AsyncIterator[None]:
+        # The serving process has one event loop. No await separates the check
+        # and increment, so concurrent requests cannot exceed the queue limit.
+        if self._pending >= self._capacity:
+            raise ExtractionBusy()
+        self._pending += 1
+        acquired = False
+        try:
+            await self._slots.acquire()
+            acquired = True
+            yield
+        finally:
+            self._pending -= 1
+            if acquired:
+                self._slots.release()
+
+
+def _command(path: Path) -> tuple[str, ...]:
+    return (sys.executable, "-m", "app.services.extraction_worker", str(path))
+
+
+async def run_worker(path: Path) -> ExtractionResult:
+    """Read a bounded child response; always reap a child before releasing its slot."""
+    process = await asyncio.create_subprocess_exec(
+        *_command(path),
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.DEVNULL,
+        cwd=str(Path(__file__).resolve().parents[2]),
+    )
+    try:
+        output = bytearray()
+        while chunk := await process.stdout.read(64 * 1024):
+            output.extend(chunk)
+            if len(output) > MAX_IPC_BYTES:
+                raise ExtractionFailure("PARSER_OUTPUT_LIMIT")
+        await process.wait()
+    except BaseException:
+        if process.returncode is None:
+            try:
+                process.kill()
+            except ProcessLookupError:
+                pass  # The child exited while cancellation was being delivered.
+        # Draining and reaping are necessary before the temporary file can be
+        # removed and admission can be granted to the next upload.
+        await process.communicate()
+        raise
+    if process.returncode != 0:
+        raise ExtractionFailure("PARSER_CRASH")
+    try:
+        message = json.loads(output)
+        if message.get("ok") is False:
+            code = message["code"]
+            if code in {
+                "ZIP_BOMB",
+                "ARCHIVE_INVALID",
+                "PARSER_UNAVAILABLE",
+                "PARSER_OUTPUT_LIMIT",
+                "NO_DOCUMENT_TEXT",
+                "PARSE_FAILED",
+            }:
+                raise ExtractionFailure(code)
+        return ExtractionResult.model_validate(message["result"])
+    except ExtractionFailure:
+        raise
+    except (AttributeError, KeyError, TypeError, ValueError) as exc:
+        raise ExtractionFailure("PARSER_CRASH") from exc