From 4c76101142ba2b453d4167801f7c16f3a1485c9b Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:37:51 +0200 Subject: [PATCH 01/45] fix: add document_timeout to PdfPipelineOptions (C1) Docling's native document_timeout is the only mechanism that can interrupt processing inside a blocked thread (OCR, table extraction). Without it, asyncio.wait_for cannot stop a frozen conversion. Configurable via DOCUMENT_TIMEOUT env var (default: 120s). Closes #57 (C1) --- document-parser/infra/local_converter.py | 2 ++ document-parser/infra/settings.py | 2 ++ document-parser/tests/test_pipeline_options.py | 1 + document-parser/tests/test_settings.py | 3 +++ 4 files changed, 8 insertions(+) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index ba8875d..b469de1 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -42,6 +42,7 @@ from domain.value_objects import ( PageElement, ) from infra.bbox import to_topleft_list +from infra.settings import settings logger = logging.getLogger(__name__) @@ -102,6 +103,7 @@ def _build_docling_converter(options: ConversionOptions) -> DoclingConverter: generate_page_images=options.generate_page_images, generate_picture_images=options.generate_picture_images, images_scale=options.images_scale, + document_timeout=settings.document_timeout, ) return DoclingConverter( diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 13ed454..b9f4b96 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -14,6 +14,7 @@ class Settings: docling_serve_url: str = "http://localhost:5001" docling_serve_api_key: str | None = None conversion_timeout: int = 900 + document_timeout: float = 120.0 # Docling-level per-document timeout (seconds) max_concurrent_analyses: int = 3 max_page_count: int = 0 # 0 = unlimited upload_dir: str = "./uploads" @@ -33,6 +34,7 @@ class Settings: docling_serve_url=os.environ.get("DOCLING_SERVE_URL", "http://localhost:5001"), docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"), conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "900")), + document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")), max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), diff --git a/document-parser/tests/test_pipeline_options.py b/document-parser/tests/test_pipeline_options.py index 2ce01f6..b52fdf4 100644 --- a/document-parser/tests/test_pipeline_options.py +++ b/document-parser/tests/test_pipeline_options.py @@ -53,6 +53,7 @@ class TestBuildConverter: assert opts.generate_page_images is False assert opts.generate_picture_images is False assert opts.images_scale == 1.0 + assert opts.document_timeout is not None def test_ocr_disabled(self): conv = build_converter(ConversionOptions(do_ocr=False)) diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index a6f600c..1b8837c 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -14,6 +14,7 @@ class TestSettingsDefaults: assert s.docling_serve_url == "http://localhost:5001" assert s.docling_serve_api_key is None assert s.conversion_timeout == 900 + assert s.document_timeout == 120.0 assert s.max_page_count == 0 assert s.upload_dir == "./uploads" assert s.db_path == "./data/docling_studio.db" @@ -36,6 +37,7 @@ class TestSettingsFromEnv: monkeypatch.setenv("DOCLING_SERVE_URL", "http://serve:9000") monkeypatch.setenv("DOCLING_SERVE_API_KEY", "secret-key") monkeypatch.setenv("CONVERSION_TIMEOUT", "120") + monkeypatch.setenv("DOCUMENT_TIMEOUT", "60.0") monkeypatch.setenv("MAX_PAGE_COUNT", "20") monkeypatch.setenv("UPLOAD_DIR", "/data/uploads") monkeypatch.setenv("DB_PATH", "/data/test.db") @@ -49,6 +51,7 @@ class TestSettingsFromEnv: assert s.docling_serve_url == "http://serve:9000" assert s.docling_serve_api_key == "secret-key" assert s.conversion_timeout == 120 + assert s.document_timeout == 60.0 assert s.max_page_count == 20 assert s.upload_dir == "/data/uploads" assert s.db_path == "/data/test.db" From c0fb1287183f6bf5ade6fbe015cfbc4c7773c435 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:38:17 +0200 Subject: [PATCH 02/45] fix: replace _converter_lock with timeout-based acquisition (C2) A frozen conversion holding the lock indefinitely blocks all subsequent jobs. Using lock.acquire(timeout=300) fails fast with a clear error instead of waiting forever. Ref #57 (C2) --- document-parser/infra/local_converter.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index b469de1..763b90d 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -46,8 +46,10 @@ from infra.settings import settings logger = logging.getLogger(__name__) -# Thread lock — DoclingConverter is not thread-safe +# Thread lock — DoclingConverter is not thread-safe. +# Uses a timeout to prevent a frozen conversion from blocking all others. _converter_lock = threading.Lock() +_LOCK_TIMEOUT = 300 # seconds — fail fast rather than wait forever # US Letter page dimensions (points) — fallback when page size is unknown _DEFAULT_PAGE_WIDTH = 612.0 @@ -212,9 +214,17 @@ def _process_content_item( def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: - with _converter_lock: + acquired = _converter_lock.acquire(timeout=_LOCK_TIMEOUT) + if not acquired: + raise TimeoutError( + f"Could not acquire converter lock within {_LOCK_TIMEOUT}s — " + "a previous conversion may be frozen" + ) + try: conv = _select_converter(options) result = conv.convert(file_path) + finally: + _converter_lock.release() doc = result.document page_count = len(doc.pages) From 6327b13614f592299d3f7f0ec583269137af4fd5 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:39:20 +0200 Subject: [PATCH 03/45] fix: pass max_num_pages and max_file_size to Docling convert (C3) Defense-in-depth: even if upload validation passes, Docling itself now enforces page count and file size limits. Configurable via MAX_PAGE_COUNT and MAX_FILE_SIZE env vars (0 = unlimited). Ref #57 (C3) --- document-parser/infra/local_converter.py | 7 ++++++- document-parser/infra/settings.py | 4 +++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index 763b90d..8d3eb75 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -222,7 +222,12 @@ def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResul ) try: conv = _select_converter(options) - result = conv.convert(file_path) + kwargs: dict = {} + if settings.max_page_count > 0: + kwargs["max_num_pages"] = settings.max_page_count + if settings.max_file_size > 0: + kwargs["max_file_size"] = settings.max_file_size + result = conv.convert(file_path, **kwargs) finally: _converter_lock.release() diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index b9f4b96..92c68b0 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -16,7 +16,8 @@ class Settings: conversion_timeout: int = 900 document_timeout: float = 120.0 # Docling-level per-document timeout (seconds) max_concurrent_analyses: int = 3 - max_page_count: int = 0 # 0 = unlimited + max_page_count: int = 0 # 0 = unlimited (upload validation) + max_file_size: int = 0 # 0 = unlimited (Docling-level, bytes) upload_dir: str = "./uploads" db_path: str = "./data/docling_studio.db" cors_origins: list[str] = field( @@ -37,6 +38,7 @@ class Settings: document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")), max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), + max_file_size=int(os.environ.get("MAX_FILE_SIZE", "0")), upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"), cors_origins=[o.strip() for o in cors_raw.split(",")], From f58b563a133a63b4328802eed14c0ff1a3479fed Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:40:11 +0200 Subject: [PATCH 04/45] fix: make default table_mode configurable via DEFAULT_TABLE_MODE (H1) TableFormerMode.ACCURATE is very expensive on CPU (~3-5x slower). The default can now be set to "fast" on resource-constrained environments (HF Spaces) via the DEFAULT_TABLE_MODE env var. User-specified table_mode in the request still takes precedence. Ref #57 (H1) --- document-parser/infra/settings.py | 2 ++ document-parser/services/analysis_service.py | 6 +++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 92c68b0..ebb596c 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -16,6 +16,7 @@ class Settings: conversion_timeout: int = 900 document_timeout: float = 120.0 # Docling-level per-document timeout (seconds) max_concurrent_analyses: int = 3 + default_table_mode: str = "accurate" # "accurate" or "fast" max_page_count: int = 0 # 0 = unlimited (upload validation) max_file_size: int = 0 # 0 = unlimited (Docling-level, bytes) upload_dir: str = "./uploads" @@ -37,6 +38,7 @@ class Settings: conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "900")), document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")), max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), + default_table_mode=os.environ.get("DEFAULT_TABLE_MODE", "accurate"), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), max_file_size=int(os.environ.get("MAX_FILE_SIZE", "0")), upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 93c556f..4b62eaa 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -15,6 +15,7 @@ from typing import TYPE_CHECKING from domain.models import AnalysisJob, AnalysisStatus from domain.value_objects import ChunkingOptions, ChunkResult, ConversionOptions, ConversionResult +from infra.settings import settings if TYPE_CHECKING: from domain.ports import DocumentChunker, DocumentConverter @@ -151,7 +152,10 @@ class AnalysisService: await analysis_repo.update_status(job) logger.info("Analysis started: %s (file: %s)", job_id, filename) - options = ConversionOptions(**(pipeline_options or {})) + opts_dict = pipeline_options or {} + if "table_mode" not in opts_dict: + opts_dict = {**opts_dict, "table_mode": settings.default_table_mode} + options = ConversionOptions(**opts_dict) result: ConversionResult = await asyncio.wait_for( self._converter.convert(file_path, options), From 4b1a15f49a1b20713529c4c6318610f489821b0e Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:41:00 +0200 Subject: [PATCH 05/45] fix: add retry tolerance to frontend polling (H3) A single transient network error (502, timeout) no longer kills the polling loop. The frontend now tolerates up to 3 consecutive errors before abandoning. Successful fetches reset the counter. Also aligns frontend polling timeout (15 min) with backend timeout. Ref #57 (H3, M5) --- frontend/src/features/analysis/store.test.ts | 34 +++++++++++++++++++- frontend/src/features/analysis/store.ts | 17 +++++++--- 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/frontend/src/features/analysis/store.test.ts b/frontend/src/features/analysis/store.test.ts index 6c6fb15..4806d00 100644 --- a/frontend/src/features/analysis/store.test.ts +++ b/frontend/src/features/analysis/store.test.ts @@ -154,17 +154,49 @@ describe('useAnalysisStore', () => { expect(store.currentAnalysis.status).toBe('FAILED') }) - it('polling stops on fetch error', async () => { + it('polling retries on transient errors and stops after MAX_POLL_RETRIES', async () => { const job = { id: 'j1', status: 'PENDING', documentId: 'd1' } api.createAnalysis.mockResolvedValue(job) api.fetchAnalysis.mockRejectedValue(new Error('network')) + vi.spyOn(console, 'warn').mockImplementation(() => {}) vi.spyOn(console, 'error').mockImplementation(() => {}) const store = useAnalysisStore() await store.run('d1') + // First two errors: still polling await vi.advanceTimersByTimeAsync(2000) + expect(store.running).toBe(true) + await vi.advanceTimersByTimeAsync(2000) + expect(store.running).toBe(true) + // Third error: stops polling + await vi.advanceTimersByTimeAsync(2000) + expect(store.running).toBe(false) + }) + + it('polling resets error count on successful fetch', async () => { + const job = { id: 'j1', status: 'PENDING', documentId: 'd1' } + api.createAnalysis.mockResolvedValue(job) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + // Fail once, succeed, fail once — should NOT stop polling + api.fetchAnalysis + .mockRejectedValueOnce(new Error('network')) + .mockResolvedValueOnce({ ...job, status: 'RUNNING' }) + .mockRejectedValueOnce(new Error('network')) + .mockResolvedValueOnce({ ...job, status: 'COMPLETED' }) + + const store = useAnalysisStore() + await store.run('d1') + + await vi.advanceTimersByTimeAsync(2000) // error 1 + expect(store.running).toBe(true) + await vi.advanceTimersByTimeAsync(2000) // success — resets counter + expect(store.running).toBe(true) + await vi.advanceTimersByTimeAsync(2000) // error 1 again + expect(store.running).toBe(true) + await vi.advanceTimersByTimeAsync(2000) // success — COMPLETED expect(store.running).toBe(false) }) }) diff --git a/frontend/src/features/analysis/store.ts b/frontend/src/features/analysis/store.ts index 8891cb7..6c509a6 100644 --- a/frontend/src/features/analysis/store.ts +++ b/frontend/src/features/analysis/store.ts @@ -10,7 +10,8 @@ export const useAnalysisStore = defineStore('analysis', () => { const error = ref(null) const pollingInterval = ref | null>(null) const pollingTimeout = ref | null>(null) - const MAX_POLLING_DURATION = 10 * 60 * 1000 // 10 minutes + const MAX_POLLING_DURATION = 15 * 60 * 1000 // 15 minutes — aligned with backend timeout + const MAX_POLL_RETRIES = 3 const currentPages = computed(() => { if (!currentAnalysis.value?.pagesJson) return [] @@ -87,9 +88,11 @@ export const useAnalysisStore = defineStore('analysis', () => { function startPolling(id: string): void { stopPolling() + let consecutiveErrors = 0 pollingInterval.value = setInterval(async () => { try { const updated = await api.fetchAnalysis(id) + consecutiveErrors = 0 currentAnalysis.value = updated const idx = analyses.value.findIndex((a) => a.id === id) if (idx !== -1) analyses.value[idx] = updated @@ -98,10 +101,14 @@ export const useAnalysisStore = defineStore('analysis', () => { running.value = false } } catch (e) { - error.value = (e as Error).message || 'Polling error' - console.error('Polling error', e) - stopPolling() - running.value = false + consecutiveErrors++ + console.warn(`Polling error (${consecutiveErrors}/${MAX_POLL_RETRIES})`, e) + if (consecutiveErrors >= MAX_POLL_RETRIES) { + error.value = (e as Error).message || 'Polling error' + console.error('Polling abandoned after retries', e) + stopPolling() + running.value = false + } } }, 2000) pollingTimeout.value = setTimeout(() => { From f89dc51661ec36c0b40e88dae85c13ea338700c4 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:41:20 +0200 Subject: [PATCH 06/45] fix: reset _default_converter on init failure (H5) If the lazy-init of the default converter fails (e.g. model download error), the singleton was left as None but subsequent calls would not retry. Now the failed state is cleared so the next request retries. Ref #57 (H5) --- document-parser/infra/local_converter.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index 8d3eb75..a126e7b 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -118,7 +118,11 @@ def _build_docling_converter(options: ConversionOptions) -> DoclingConverter: def _get_default_converter() -> DoclingConverter: global _default_converter if _default_converter is None: - _default_converter = _build_docling_converter(ConversionOptions()) + try: + _default_converter = _build_docling_converter(ConversionOptions()) + except Exception: + _default_converter = None + raise return _default_converter From c281b6c5510064fdacbfb59ca780a88388b918eb Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:41:50 +0200 Subject: [PATCH 07/45] fix: classify pipeline errors into user-friendly messages (M1) Raw PyTorch/Docling stack traces are no longer shown to the user. Common failures (missing compiler, OOM, lock contention, corrupted document) are mapped to actionable messages. Unknown errors are truncated to 200 chars. Ref #57 (M1) --- document-parser/services/analysis_service.py | 28 +++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 4b62eaa..f630e95 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -191,7 +191,33 @@ class AnalysisService: except Exception as e: logger.exception("Analysis failed: %s", job_id) - await _mark_failed(job_id, str(e)) + await _mark_failed(job_id, _classify_error(e)) + + +def _classify_error(exc: Exception) -> str: + """Return a user-friendly error message based on the exception type/content.""" + msg = str(exc).lower() + + if "invalidcxxcompiler" in msg or "no working c++ compiler" in msg: + return "Missing C++ compiler — set TORCHDYNAMO_DISABLE=1 to work around this" + + if "out of memory" in msg or "oom" in msg: + return "Out of memory — try a smaller document or disable table structure analysis" + + if "could not acquire converter lock" in msg: + return "Server busy — a previous conversion is still running. Please retry later" + + if "pipeline" in msg and "failed" in msg: + return "Document processing failed — the document may be corrupted or unsupported" + + if "timeout" in msg: + return "Processing took too long — try with fewer pages or simpler options" + + # Fallback: truncate raw error to something reasonable + raw = str(exc) + if len(raw) > 200: + raw = raw[:200] + "…" + return raw _background_tasks: set[asyncio.Task] = set() From 6177452de16d55434e31e5d73a94a5e341c1fcd6 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 14:58:26 +0200 Subject: [PATCH 08/45] test: add robustness tests for pipeline failure scenarios 25 tests covering all backend robustness fixes: - TestClassifyError (9): user-friendly error message mapping - TestDefaultTableMode (4): settings.default_table_mode injection - TestDocumentTimeout (2): document_timeout wired into pipeline - TestConverterLockTimeout (3): lock timeout + release guarantees - TestConvertSyncLimits (4): max_num_pages/max_file_size forwarding - TestGetDefaultConverterReset (3): converter singleton reset on failure Ref #57 --- document-parser/tests/test_robustness.py | 377 +++++++++++++++++++++++ 1 file changed, 377 insertions(+) create mode 100644 document-parser/tests/test_robustness.py diff --git a/document-parser/tests/test_robustness.py b/document-parser/tests/test_robustness.py new file mode 100644 index 0000000..9e33b29 --- /dev/null +++ b/document-parser/tests/test_robustness.py @@ -0,0 +1,377 @@ +"""Robustness tests — verify pipeline failure scenarios and protective mechanisms. + +Tests cover: document_timeout (C1), lock timeout (C2), convert limits (C3), +default table_mode (H1), converter reset on failure (H5), error classification (M1). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from domain.models import AnalysisJob +from domain.value_objects import ConversionOptions, ConversionResult, PageDetail +from infra.settings import Settings +from services.analysis_service import AnalysisService, _classify_error + +# --------------------------------------------------------------------------- +# M1 — _classify_error: user-friendly error messages +# --------------------------------------------------------------------------- + + +class TestClassifyError: + """Verify _classify_error maps exceptions to user-friendly messages.""" + + def test_cxx_compiler_error(self): + exc = RuntimeError("InvalidCxxCompiler: No working C++ compiler found") + assert "Missing C++ compiler" in _classify_error(exc) + + def test_no_working_compiler(self): + exc = RuntimeError("no working c++ compiler found in torch") + assert "Missing C++ compiler" in _classify_error(exc) + + def test_out_of_memory(self): + exc = MemoryError("Out of memory allocating tensor") + assert "Out of memory" in _classify_error(exc) + + def test_oom_shorthand(self): + exc = RuntimeError("OOM during inference on page 5") + assert "Out of memory" in _classify_error(exc) + + def test_lock_timeout(self): + exc = TimeoutError("Could not acquire converter lock within 300s") + assert "Server busy" in _classify_error(exc) + + def test_pipeline_failed(self): + exc = RuntimeError("Pipeline StandardPdfPipeline failed on page 3") + assert "Document processing failed" in _classify_error(exc) + + def test_timeout_generic(self): + exc = TimeoutError("timeout exceeded while processing") + assert "Processing took too long" in _classify_error(exc) + + def test_unknown_short_error(self): + exc = ValueError("something weird happened") + assert _classify_error(exc) == "something weird happened" + + def test_unknown_long_error_truncated(self): + long_msg = "x" * 300 + exc = ValueError(long_msg) + result = _classify_error(exc) + assert len(result) <= 201 + assert result.endswith("…") + + +# --------------------------------------------------------------------------- +# H1 — default table_mode injection +# --------------------------------------------------------------------------- + + +class TestDefaultTableMode: + """Verify _run_analysis_inner injects settings.default_table_mode.""" + + @pytest.fixture + def mock_job(self): + return AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf") + + def _make_result(self) -> ConversionResult: + return ConversionResult( + page_count=1, + content_markdown="# Test", + content_html="

Test

", + pages=[PageDetail(page_number=1, width=612.0, height=792.0)], + ) + + @patch("services.analysis_service.analysis_repo") + @patch("services.analysis_service.document_repo") + @pytest.mark.asyncio + async def test_default_table_mode_injected_when_missing( + self, mock_doc_repo, mock_analysis_repo, mock_job + ): + mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) + mock_analysis_repo.update_status = AsyncMock() + mock_doc_repo.update_page_count = AsyncMock() + + mock_converter = AsyncMock() + mock_converter.convert.return_value = self._make_result() + + svc = AnalysisService(converter=mock_converter) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {}) + + opts = mock_converter.convert.call_args[0][1] + assert opts.table_mode == "accurate" + + @patch("services.analysis_service.analysis_repo") + @patch("services.analysis_service.document_repo") + @pytest.mark.asyncio + async def test_default_table_mode_injected_when_none( + self, mock_doc_repo, mock_analysis_repo, mock_job + ): + mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) + mock_analysis_repo.update_status = AsyncMock() + mock_doc_repo.update_page_count = AsyncMock() + + mock_converter = AsyncMock() + mock_converter.convert.return_value = self._make_result() + + svc = AnalysisService(converter=mock_converter) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", None) + + opts = mock_converter.convert.call_args[0][1] + assert opts.table_mode == "accurate" + + @patch("services.analysis_service.analysis_repo") + @patch("services.analysis_service.document_repo") + @pytest.mark.asyncio + async def test_user_table_mode_preserved(self, mock_doc_repo, mock_analysis_repo, mock_job): + mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) + mock_analysis_repo.update_status = AsyncMock() + mock_doc_repo.update_page_count = AsyncMock() + + mock_converter = AsyncMock() + mock_converter.convert.return_value = self._make_result() + + svc = AnalysisService(converter=mock_converter) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"table_mode": "fast"}) + + opts = mock_converter.convert.call_args[0][1] + assert opts.table_mode == "fast" + + @patch("services.analysis_service.settings", Settings(default_table_mode="fast")) + @patch("services.analysis_service.analysis_repo") + @patch("services.analysis_service.document_repo") + @pytest.mark.asyncio + async def test_custom_default_from_settings(self, mock_doc_repo, mock_analysis_repo, mock_job): + mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job) + mock_analysis_repo.update_status = AsyncMock() + mock_doc_repo.update_page_count = AsyncMock() + + mock_converter = AsyncMock() + mock_converter.convert.return_value = self._make_result() + + svc = AnalysisService(converter=mock_converter) + await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {}) + + opts = mock_converter.convert.call_args[0][1] + assert opts.table_mode == "fast" + + +# --------------------------------------------------------------------------- +# Docling-dependent tests — skip if docling is not installed +# --------------------------------------------------------------------------- + +docling = pytest.importorskip("docling", reason="docling library not installed") + +from docling.datamodel.base_models import InputFormat # noqa: E402 + +from infra.local_converter import ( # noqa: E402 + _build_docling_converter as build_converter, +) +from infra.local_converter import ( # noqa: E402 + _convert_sync as convert_sync, +) +from infra.local_converter import ( # noqa: E402 + _get_default_converter as get_default_converter, +) + +# --------------------------------------------------------------------------- +# C1 — document_timeout in PdfPipelineOptions +# --------------------------------------------------------------------------- + + +class TestDocumentTimeout: + """Verify document_timeout from settings is wired into PdfPipelineOptions.""" + + def _get_pipeline_options(self, converter): + fmt_opt = converter.format_to_options[InputFormat.PDF] + return fmt_opt.pipeline_options + + def test_document_timeout_from_settings(self): + conv = build_converter(ConversionOptions()) + opts = self._get_pipeline_options(conv) + assert opts.document_timeout == 120.0 + + @patch("infra.local_converter.settings", Settings(document_timeout=45.0)) + def test_custom_document_timeout(self): + conv = build_converter(ConversionOptions()) + opts = self._get_pipeline_options(conv) + assert opts.document_timeout == 45.0 + + +# --------------------------------------------------------------------------- +# C2 — converter lock timeout +# --------------------------------------------------------------------------- + + +def _mock_docling_result(): + """Create a minimal mock that satisfies _convert_sync's result processing.""" + mock_result = MagicMock() + mock_result.document.pages = {} + mock_result.document.iterate_items.return_value = [] + mock_result.document.export_to_markdown.return_value = "" + mock_result.document.export_to_html.return_value = "" + mock_result.document.export_to_dict.return_value = {} + return mock_result + + +class TestConverterLockTimeout: + """Verify _convert_sync raises TimeoutError when lock cannot be acquired.""" + + @patch("infra.local_converter._converter_lock") + def test_lock_timeout_raises(self, mock_lock): + mock_lock.acquire.return_value = False + + with pytest.raises(TimeoutError, match="Could not acquire converter lock"): + convert_sync("/tmp/test.pdf", ConversionOptions()) + + mock_lock.release.assert_not_called() + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + def test_lock_released_on_success(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.return_value = _mock_docling_result() + mock_select.return_value = mock_conv + + convert_sync("/tmp/test.pdf", ConversionOptions()) + + mock_lock.release.assert_called_once() + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + def test_lock_released_on_error(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.side_effect = RuntimeError("boom") + mock_select.return_value = mock_conv + + with pytest.raises(RuntimeError, match="boom"): + convert_sync("/tmp/test.pdf", ConversionOptions()) + + mock_lock.release.assert_called_once() + + +# --------------------------------------------------------------------------- +# C3 — max_num_pages and max_file_size forwarding +# --------------------------------------------------------------------------- + + +class TestConvertSyncLimits: + """Verify _convert_sync forwards limit settings to conv.convert().""" + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + @patch("infra.local_converter.settings", Settings(max_page_count=10, max_file_size=0)) + def test_max_page_count_forwarded(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.return_value = _mock_docling_result() + mock_select.return_value = mock_conv + + convert_sync("/tmp/test.pdf", ConversionOptions()) + + kwargs = mock_conv.convert.call_args.kwargs + assert kwargs["max_num_pages"] == 10 + assert "max_file_size" not in kwargs + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + @patch("infra.local_converter.settings", Settings(max_page_count=0, max_file_size=5_000_000)) + def test_max_file_size_forwarded(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.return_value = _mock_docling_result() + mock_select.return_value = mock_conv + + convert_sync("/tmp/test.pdf", ConversionOptions()) + + kwargs = mock_conv.convert.call_args.kwargs + assert kwargs["max_file_size"] == 5_000_000 + assert "max_num_pages" not in kwargs + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + @patch( + "infra.local_converter.settings", + Settings(max_page_count=20, max_file_size=10_000_000), + ) + def test_both_limits_forwarded(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.return_value = _mock_docling_result() + mock_select.return_value = mock_conv + + convert_sync("/tmp/test.pdf", ConversionOptions()) + + kwargs = mock_conv.convert.call_args.kwargs + assert kwargs["max_num_pages"] == 20 + assert kwargs["max_file_size"] == 10_000_000 + + @patch("infra.local_converter._select_converter") + @patch("infra.local_converter._converter_lock") + @patch("infra.local_converter.settings", Settings(max_page_count=0, max_file_size=0)) + def test_zero_limits_not_forwarded(self, mock_lock, mock_select): + mock_lock.acquire.return_value = True + mock_conv = MagicMock() + mock_conv.convert.return_value = _mock_docling_result() + mock_select.return_value = mock_conv + + convert_sync("/tmp/test.pdf", ConversionOptions()) + + kwargs = mock_conv.convert.call_args.kwargs + assert "max_num_pages" not in kwargs + assert "max_file_size" not in kwargs + + +# --------------------------------------------------------------------------- +# H5 — _default_converter reset on init failure +# --------------------------------------------------------------------------- + + +class TestGetDefaultConverterReset: + """Verify _get_default_converter resets on failure and retries on next call.""" + + @pytest.fixture(autouse=True) + def reset_default_converter(self): + import infra.local_converter as mod + + original = mod._default_converter + mod._default_converter = None + yield + mod._default_converter = original + + @patch("infra.local_converter._build_docling_converter") + def test_init_failure_resets_to_none(self, mock_build): + import infra.local_converter as mod + + mock_build.side_effect = RuntimeError("torch not found") + + with pytest.raises(RuntimeError, match="torch not found"): + get_default_converter() + + assert mod._default_converter is None + + @patch("infra.local_converter._build_docling_converter") + def test_retry_after_failure_succeeds(self, mock_build): + mock_conv = MagicMock() + mock_build.side_effect = [RuntimeError("torch not found"), mock_conv] + + with pytest.raises(RuntimeError): + get_default_converter() + + result = get_default_converter() + assert result is mock_conv + assert mock_build.call_count == 2 + + @patch("infra.local_converter._build_docling_converter") + def test_success_caches_converter(self, mock_build): + mock_conv = MagicMock() + mock_build.return_value = mock_conv + + result1 = get_default_converter() + result2 = get_default_converter() + + assert result1 is result2 is mock_conv + mock_build.assert_called_once() From f04e5369eff3ab322582796eaeff4340716a465b Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 15:39:02 +0200 Subject: [PATCH 09/45] style: use single import style for infra.local_converter in tests Use `import infra.local_converter as lc_mod` consistently instead of mixing `import` and `from ... import` for the same module. Addresses CodeQL review comment on PR #58. --- document-parser/tests/test_robustness.py | 26 ++++++++---------------- 1 file changed, 9 insertions(+), 17 deletions(-) diff --git a/document-parser/tests/test_robustness.py b/document-parser/tests/test_robustness.py index 9e33b29..0f48f75 100644 --- a/document-parser/tests/test_robustness.py +++ b/document-parser/tests/test_robustness.py @@ -165,15 +165,11 @@ docling = pytest.importorskip("docling", reason="docling library not installed") from docling.datamodel.base_models import InputFormat # noqa: E402 -from infra.local_converter import ( # noqa: E402 - _build_docling_converter as build_converter, -) -from infra.local_converter import ( # noqa: E402 - _convert_sync as convert_sync, -) -from infra.local_converter import ( # noqa: E402 - _get_default_converter as get_default_converter, -) +import infra.local_converter as lc_mod # noqa: E402 + +build_converter = lc_mod._build_docling_converter +convert_sync = lc_mod._convert_sync +get_default_converter = lc_mod._get_default_converter # --------------------------------------------------------------------------- # C1 — document_timeout in PdfPipelineOptions @@ -335,23 +331,19 @@ class TestGetDefaultConverterReset: @pytest.fixture(autouse=True) def reset_default_converter(self): - import infra.local_converter as mod - - original = mod._default_converter - mod._default_converter = None + original = lc_mod._default_converter + lc_mod._default_converter = None yield - mod._default_converter = original + lc_mod._default_converter = original @patch("infra.local_converter._build_docling_converter") def test_init_failure_resets_to_none(self, mock_build): - import infra.local_converter as mod - mock_build.side_effect = RuntimeError("torch not found") with pytest.raises(RuntimeError, match="torch not found"): get_default_converter() - assert mod._default_converter is None + assert lc_mod._default_converter is None @patch("infra.local_converter._build_docling_converter") def test_retry_after_failure_succeeds(self, mock_build): From 2c254382c82e8f6d52e7d5c242da1f5ba67d87c1 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 16:59:37 +0200 Subject: [PATCH 10/45] fix: add __post_init__ validation to Settings dataclass (#65) Reject invalid configuration at startup: negative timeouts, zero concurrency, bad table mode. Reports all errors at once. --- document-parser/infra/settings.py | 21 +++++++++++ document-parser/tests/test_settings.py | 51 ++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index ebb596c..60cc3c5 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -25,6 +25,27 @@ class Settings: default_factory=lambda: ["http://localhost:3000", "http://localhost:5173"] ) + def __post_init__(self) -> None: + errors: list[str] = [] + if self.document_timeout <= 0: + errors.append(f"document_timeout must be > 0 (got {self.document_timeout})") + if self.conversion_timeout <= 0: + errors.append(f"conversion_timeout must be > 0 (got {self.conversion_timeout})") + if self.max_concurrent_analyses < 1: + errors.append( + f"max_concurrent_analyses must be >= 1 (got {self.max_concurrent_analyses})" + ) + if self.max_page_count < 0: + errors.append(f"max_page_count must be >= 0 (got {self.max_page_count})") + if self.max_file_size < 0: + errors.append(f"max_file_size must be >= 0 (got {self.max_file_size})") + if self.default_table_mode not in ("accurate", "fast"): + errors.append( + f"default_table_mode must be 'accurate' or 'fast' (got '{self.default_table_mode}')" + ) + if errors: + raise ValueError("Invalid settings:\n " + "\n ".join(errors)) + @classmethod def from_env(cls) -> Settings: """Build a Settings instance from environment variables.""" diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index 1b8837c..aa27233 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -29,6 +29,57 @@ class TestSettingsDefaults: s.upload_dir = "/other" # type: ignore[misc] +class TestSettingsValidation: + def test_negative_document_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout must be > 0"): + Settings(document_timeout=-1.0) + + def test_zero_document_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout must be > 0"): + Settings(document_timeout=0) + + def test_negative_conversion_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="conversion_timeout must be > 0"): + Settings(conversion_timeout=-1) + + def test_zero_max_concurrent_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_concurrent_analyses must be >= 1"): + Settings(max_concurrent_analyses=0) + + def test_negative_max_page_count_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_page_count must be >= 0"): + Settings(max_page_count=-1) + + def test_negative_max_file_size_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_file_size must be >= 0"): + Settings(max_file_size=-1) + + def test_invalid_table_mode_rejected(self): + import pytest + + with pytest.raises(ValueError, match="default_table_mode must be"): + Settings(default_table_mode="turbo") + + def test_multiple_errors_reported(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout") as exc_info: + Settings(document_timeout=-1, conversion_timeout=-1) + assert "conversion_timeout" in str(exc_info.value) + + class TestSettingsFromEnv: def test_reads_env_vars(self, monkeypatch): monkeypatch.setenv("APP_VERSION", "1.2.3") From 1657bce1f05f3ca927f913e9093db33301fe92f6 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:01:10 +0200 Subject: [PATCH 11/45] fix: make converter lock timeout configurable via LOCK_TIMEOUT env var (#61) Replace hardcoded _LOCK_TIMEOUT=300 with settings.lock_timeout, readable from LOCK_TIMEOUT environment variable. --- document-parser/infra/local_converter.py | 5 ++--- document-parser/infra/settings.py | 4 ++++ document-parser/tests/test_settings.py | 7 +++++++ 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index a126e7b..1a3b701 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -49,7 +49,6 @@ logger = logging.getLogger(__name__) # Thread lock — DoclingConverter is not thread-safe. # Uses a timeout to prevent a frozen conversion from blocking all others. _converter_lock = threading.Lock() -_LOCK_TIMEOUT = 300 # seconds — fail fast rather than wait forever # US Letter page dimensions (points) — fallback when page size is unknown _DEFAULT_PAGE_WIDTH = 612.0 @@ -218,10 +217,10 @@ def _process_content_item( def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: - acquired = _converter_lock.acquire(timeout=_LOCK_TIMEOUT) + acquired = _converter_lock.acquire(timeout=settings.lock_timeout) if not acquired: raise TimeoutError( - f"Could not acquire converter lock within {_LOCK_TIMEOUT}s — " + f"Could not acquire converter lock within {settings.lock_timeout}s — " "a previous conversion may be frozen" ) try: diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 60cc3c5..496e032 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -15,6 +15,7 @@ class Settings: docling_serve_api_key: str | None = None conversion_timeout: int = 900 document_timeout: float = 120.0 # Docling-level per-document timeout (seconds) + lock_timeout: int = 300 # converter lock acquisition timeout (seconds) max_concurrent_analyses: int = 3 default_table_mode: str = "accurate" # "accurate" or "fast" max_page_count: int = 0 # 0 = unlimited (upload validation) @@ -31,6 +32,8 @@ class Settings: errors.append(f"document_timeout must be > 0 (got {self.document_timeout})") if self.conversion_timeout <= 0: errors.append(f"conversion_timeout must be > 0 (got {self.conversion_timeout})") + if self.lock_timeout <= 0: + errors.append(f"lock_timeout must be > 0 (got {self.lock_timeout})") if self.max_concurrent_analyses < 1: errors.append( f"max_concurrent_analyses must be >= 1 (got {self.max_concurrent_analyses})" @@ -58,6 +61,7 @@ class Settings: docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"), conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "900")), document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")), + lock_timeout=int(os.environ.get("LOCK_TIMEOUT", "300")), max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), default_table_mode=os.environ.get("DEFAULT_TABLE_MODE", "accurate"), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index aa27233..9a683fd 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -15,6 +15,7 @@ class TestSettingsDefaults: assert s.docling_serve_api_key is None assert s.conversion_timeout == 900 assert s.document_timeout == 120.0 + assert s.lock_timeout == 300 assert s.max_page_count == 0 assert s.upload_dir == "./uploads" assert s.db_path == "./data/docling_studio.db" @@ -66,6 +67,12 @@ class TestSettingsValidation: with pytest.raises(ValueError, match="max_file_size must be >= 0"): Settings(max_file_size=-1) + def test_zero_lock_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="lock_timeout must be > 0"): + Settings(lock_timeout=0) + def test_invalid_table_mode_rejected(self): import pytest From 8ea0cebc1675614b1a73466abb241361619a44b0 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:04:09 +0200 Subject: [PATCH 12/45] fix: validate timeout cascade ordering in Settings (#62) Enforce document_timeout < lock_timeout < conversion_timeout at startup. Prevents inverted timeout configurations that cause unpredictable behavior. --- document-parser/infra/settings.py | 12 ++++++++++++ document-parser/tests/test_settings.py | 22 ++++++++++++++++++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 496e032..d8f1e48 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -46,6 +46,18 @@ class Settings: errors.append( f"default_table_mode must be 'accurate' or 'fast' (got '{self.default_table_mode}')" ) + # Timeout cascade: document_timeout < lock_timeout < conversion_timeout + if self.document_timeout > 0 and self.lock_timeout > 0 and self.conversion_timeout > 0: + if self.document_timeout >= self.lock_timeout: + errors.append( + f"document_timeout ({self.document_timeout}s) must be " + f"< lock_timeout ({self.lock_timeout}s)" + ) + if self.lock_timeout >= self.conversion_timeout: + errors.append( + f"lock_timeout ({self.lock_timeout}s) must be " + f"< conversion_timeout ({self.conversion_timeout}s)" + ) if errors: raise ValueError("Invalid settings:\n " + "\n ".join(errors)) diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index 9a683fd..604abac 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -79,6 +79,22 @@ class TestSettingsValidation: with pytest.raises(ValueError, match="default_table_mode must be"): Settings(default_table_mode="turbo") + def test_cascade_document_ge_lock_rejected(self): + import pytest + + with pytest.raises(ValueError, match=r"document_timeout.*< lock_timeout"): + Settings(document_timeout=400.0, lock_timeout=300, conversion_timeout=900) + + def test_cascade_lock_ge_conversion_rejected(self): + import pytest + + with pytest.raises(ValueError, match=r"lock_timeout.*< conversion_timeout"): + Settings(document_timeout=100.0, lock_timeout=900, conversion_timeout=900) + + def test_cascade_valid_ordering_accepted(self): + s = Settings(document_timeout=60.0, lock_timeout=300, conversion_timeout=900) + assert s.document_timeout < s.lock_timeout < s.conversion_timeout + def test_multiple_errors_reported(self): import pytest @@ -94,8 +110,9 @@ class TestSettingsFromEnv: monkeypatch.setenv("DEPLOYMENT_MODE", "huggingface") monkeypatch.setenv("DOCLING_SERVE_URL", "http://serve:9000") monkeypatch.setenv("DOCLING_SERVE_API_KEY", "secret-key") - monkeypatch.setenv("CONVERSION_TIMEOUT", "120") + monkeypatch.setenv("CONVERSION_TIMEOUT", "1200") monkeypatch.setenv("DOCUMENT_TIMEOUT", "60.0") + monkeypatch.setenv("LOCK_TIMEOUT", "600") monkeypatch.setenv("MAX_PAGE_COUNT", "20") monkeypatch.setenv("UPLOAD_DIR", "/data/uploads") monkeypatch.setenv("DB_PATH", "/data/test.db") @@ -108,7 +125,8 @@ class TestSettingsFromEnv: assert s.deployment_mode == "huggingface" assert s.docling_serve_url == "http://serve:9000" assert s.docling_serve_api_key == "secret-key" - assert s.conversion_timeout == 120 + assert s.conversion_timeout == 1200 + assert s.lock_timeout == 600 assert s.document_timeout == 60.0 assert s.max_page_count == 20 assert s.upload_dir == "/data/uploads" From 302b24cc4c75921346c68a66c4dd5dc29e1f1d10 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:06:16 +0200 Subject: [PATCH 13/45] fix: route _on_task_done errors through _classify_error (#63) Unhandled task exceptions now produce user-friendly messages instead of raw Python tracebacks in the error_message field. --- document-parser/services/analysis_service.py | 2 +- .../tests/test_analysis_service.py | 23 +++++++++++++++++-- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index f630e95..36c2f5a 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -232,7 +232,7 @@ def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: exc = task.exception() if exc: logger.error("Unhandled exception in analysis task %s: %s", job_id, exc, exc_info=True) - _schedule_mark_failed(job_id, str(exc)) + _schedule_mark_failed(job_id, _classify_error(exc)) def _schedule_mark_failed(job_id: str, error: str) -> None: diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py index 83f6be8..66cbaac 100644 --- a/document-parser/tests/test_analysis_service.py +++ b/document-parser/tests/test_analysis_service.py @@ -20,7 +20,7 @@ class TestOnTaskDone: # Create a task that raises async def failing_task(): - raise RuntimeError("boom") + raise RuntimeError("unexpected failure") task = asyncio.create_task(failing_task()) await asyncio.sleep(0) # let the task fail @@ -30,7 +30,26 @@ class TestOnTaskDone: # ensure_future schedules it; give the event loop a tick await asyncio.sleep(0) - mock_mark.assert_called_once_with(job_id, "boom") + mock_mark.assert_called_once_with(job_id, "unexpected failure") + + @pytest.mark.asyncio + async def test_exception_uses_classify_error(self): + """_on_task_done should route exceptions through _classify_error.""" + job_id = "job-classify" + + async def timeout_task(): + raise TimeoutError("timeout exceeded while processing") + + task = asyncio.create_task(timeout_task()) + await asyncio.sleep(0) + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + await asyncio.sleep(0) + + mock_mark.assert_called_once_with( + job_id, "Processing took too long — try with fewer pages or simpler options" + ) @pytest.mark.asyncio async def test_cancelled_task_marks_job_failed(self): From af10c200f0bfe615bde759539137ecaf14ef57ca Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:07:55 +0200 Subject: [PATCH 14/45] fix: cancel running conversion when analysis job is deleted (#64) Track background tasks per job ID. On delete, cancel the task to release the converter lock instead of letting phantom jobs run to completion. --- document-parser/services/analysis_service.py | 18 +++++- .../tests/test_analysis_service.py | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 36c2f5a..f003285 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -53,6 +53,7 @@ class AnalysisService: self._chunker = chunker self._conversion_timeout = conversion_timeout self._semaphore = asyncio.Semaphore(max_concurrent) + self._running_tasks: dict[str, asyncio.Task] = {} async def create( self, @@ -79,7 +80,8 @@ class AnalysisService: chunking_options, ) ) - task.add_done_callback(functools.partial(_on_task_done, job_id=job.id)) + self._running_tasks[job.id] = task + task.add_done_callback(functools.partial(self._on_task_done, job_id=job.id)) return job @@ -92,7 +94,11 @@ class AnalysisService: return await analysis_repo.find_by_id(job_id) async def delete(self, job_id: str) -> bool: - """Delete an analysis job. Returns True if it existed.""" + """Delete an analysis job, cancelling any running task. Returns True if it existed.""" + task = self._running_tasks.pop(job_id, None) + if task and not task.done(): + task.cancel() + logger.info("Cancelled running task for job %s", job_id) return await analysis_repo.delete(job_id) async def rechunk(self, job_id: str, chunking_options: dict) -> list[ChunkResult]: @@ -115,6 +121,11 @@ class AnalysisService: return chunks + def _on_task_done(self, task: asyncio.Task, *, job_id: str) -> None: + """Cleanup running tasks and delegate to module-level handler.""" + self._running_tasks.pop(job_id, None) + _on_task_done(task, job_id=job_id) + async def _run_analysis( self, job_id: str, @@ -235,6 +246,9 @@ def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: _schedule_mark_failed(job_id, _classify_error(exc)) +# Keep the module-level function as the default, but AnalysisService uses its own method. + + def _schedule_mark_failed(job_id: str, error: str) -> None: """Schedule _mark_failed as a tracked background task.""" t = asyncio.ensure_future(_mark_failed(job_id, error)) diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py index 66cbaac..a211dee 100644 --- a/document-parser/tests/test_analysis_service.py +++ b/document-parser/tests/test_analysis_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import functools from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -90,6 +91,60 @@ class TestOnTaskDone: mock_mark.assert_not_called() +class TestAnalysisServiceCancellation: + """Verify delete cancels running tasks.""" + + @pytest.mark.asyncio + async def test_delete_cancels_running_task(self): + """Deleting a job while running should cancel its task.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + blocker = asyncio.Event() + + async def slow_analysis(): + await blocker.wait() + + task = asyncio.create_task(slow_analysis()) + service._running_tasks["j1"] = task + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.delete = AsyncMock(return_value=True) + result = await service.delete("j1") + + assert result is True + assert task.cancelling() or task.cancelled() + assert "j1" not in service._running_tasks + + @pytest.mark.asyncio + async def test_delete_completed_job_no_error(self): + """Deleting a completed job should not raise even if no task tracked.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.delete = AsyncMock(return_value=True) + result = await service.delete("j-gone") + + assert result is True + + @pytest.mark.asyncio + async def test_task_cleaned_from_running_on_completion(self): + """After a task completes, it should be removed from _running_tasks.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + async def instant(): + pass + + task = asyncio.create_task(instant()) + service._running_tasks["j1"] = task + task.add_done_callback(functools.partial(service._on_task_done, job_id="j1")) + await task + + assert "j1" not in service._running_tasks + + class TestAnalysisServiceConcurrency: """Verify that the semaphore limits concurrent analysis jobs.""" From fc866ce229664aff636e5b40192b75f2b7e610fe Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:54:40 +0200 Subject: [PATCH 15/45] feat: batch large documents with page_range and progress reporting Use Docling's native page_range parameter to split large PDFs into sequential batches, preventing memory exhaustion and timeouts. Progress is reported via existing polling mechanism. Closes #56 --- document-parser/api/analyses.py | 2 + document-parser/api/schemas.py | 2 + document-parser/domain/models.py | 7 + document-parser/domain/ports.py | 2 + document-parser/infra/local_converter.py | 13 +- document-parser/infra/serve_converter.py | 15 +- document-parser/infra/settings.py | 4 + document-parser/persistence/analysis_repo.py | 17 +- document-parser/persistence/database.py | 2 + document-parser/services/analysis_service.py | 155 +++++++++++- .../tests/test_analysis_service.py | 232 +++++++++++++++++- document-parser/tests/test_serve_converter.py | 8 + document-parser/tests/test_settings.py | 17 ++ .../features/analysis/ui/AnalysisPanel.vue | 12 + .../src/features/analysis/ui/ResultTabs.vue | 44 ++++ frontend/src/shared/types.ts | 2 + 16 files changed, 521 insertions(+), 13 deletions(-) diff --git a/document-parser/api/analyses.py b/document-parser/api/analyses.py index 8e6112b..f54ae9d 100644 --- a/document-parser/api/analyses.py +++ b/document-parser/api/analyses.py @@ -39,6 +39,8 @@ def _to_response(job) -> AnalysisResponse: chunks_json=job.chunks_json, has_document_json=job.document_json is not None, error_message=job.error_message, + progress_current=job.progress_current, + progress_total=job.progress_total, started_at=str(job.started_at) if job.started_at else None, completed_at=str(job.completed_at) if job.completed_at else None, created_at=str(job.created_at), diff --git a/document-parser/api/schemas.py b/document-parser/api/schemas.py index ee602c6..e157df4 100644 --- a/document-parser/api/schemas.py +++ b/document-parser/api/schemas.py @@ -47,6 +47,8 @@ class AnalysisResponse(_CamelModel): chunks_json: str | None = None has_document_json: bool = False error_message: str | None = None + progress_current: int | None = None + progress_total: int | None = None started_at: str | datetime | None = None completed_at: str | datetime | None = None created_at: str | datetime diff --git a/document-parser/domain/models.py b/document-parser/domain/models.py index 3faec9a..036bc88 100644 --- a/document-parser/domain/models.py +++ b/document-parser/domain/models.py @@ -45,6 +45,8 @@ class AnalysisJob: document_json: str | None = None chunks_json: str | None = None error_message: str | None = None + progress_current: int | None = None + progress_total: int | None = None started_at: datetime | None = None completed_at: datetime | None = None created_at: datetime = field(default_factory=_utcnow) @@ -74,6 +76,11 @@ class AnalysisJob: self.chunks_json = chunks_json self.completed_at = _utcnow() + def update_progress(self, current: int, total: int) -> None: + """Update batch progress counters.""" + self.progress_current = current + self.progress_total = total + def mark_failed(self, error: str) -> None: """Transition to FAILED with an error message.""" self.status = AnalysisStatus.FAILED diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py index f4ea64d..1225638 100644 --- a/document-parser/domain/ports.py +++ b/document-parser/domain/ports.py @@ -28,6 +28,8 @@ class DocumentConverter(Protocol): self, file_path: str, options: ConversionOptions, + *, + page_range: tuple[int, int] | None = None, ) -> ConversionResult: ... diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index 1a3b701..521b415 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -216,7 +216,12 @@ def _process_content_item( # --------------------------------------------------------------------------- -def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: +def _convert_sync( + file_path: str, + options: ConversionOptions, + *, + page_range: tuple[int, int] | None = None, +) -> ConversionResult: acquired = _converter_lock.acquire(timeout=settings.lock_timeout) if not acquired: raise TimeoutError( @@ -230,6 +235,8 @@ def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResul kwargs["max_num_pages"] = settings.max_page_count if settings.max_file_size > 0: kwargs["max_file_size"] = settings.max_file_size + if page_range is not None: + kwargs["page_range"] = page_range result = conv.convert(file_path, **kwargs) finally: _converter_lock.release() @@ -275,5 +282,7 @@ class LocalConverter: self, file_path: str, options: ConversionOptions, + *, + page_range: tuple[int, int] | None = None, ) -> ConversionResult: - return await asyncio.to_thread(_convert_sync, file_path, options) + return await asyncio.to_thread(_convert_sync, file_path, options, page_range=page_range) diff --git a/document-parser/infra/serve_converter.py b/document-parser/infra/serve_converter.py index a6a0502..5013e1a 100644 --- a/document-parser/infra/serve_converter.py +++ b/document-parser/infra/serve_converter.py @@ -76,12 +76,14 @@ class ServeConverter: self, file_path: str, options: ConversionOptions, + *, + page_range: tuple[int, int] | None = None, ) -> ConversionResult: """Convert a document by uploading it to Docling Serve.""" path = Path(file_path) content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream" - form_data = _build_form_data(options) + form_data = _build_form_data(options, page_range=page_range) url = f"{self._base_url}{_API_PREFIX}/convert/file" async with httpx.AsyncClient(timeout=self._timeout) as client: @@ -112,13 +114,17 @@ class ServeConverter: return False -def _build_form_data(options: ConversionOptions) -> dict[str, str | list[str]]: +def _build_form_data( + options: ConversionOptions, + *, + page_range: tuple[int, int] | None = None, +) -> 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 repeated form keys (to_formats=md&to_formats=html&to_formats=json). """ - return { + data: dict[str, str | list[str]] = { "to_formats": ["md", "html", "json"], "do_ocr": str(options.do_ocr).lower(), "do_table_structure": str(options.do_table_structure).lower(), @@ -131,6 +137,9 @@ def _build_form_data(options: ConversionOptions) -> dict[str, str | list[str]]: "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]}" + return data def _parse_response(data: dict) -> ConversionResult: diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index d8f1e48..2ad8538 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -20,6 +20,7 @@ class Settings: default_table_mode: str = "accurate" # "accurate" or "fast" max_page_count: int = 0 # 0 = unlimited (upload validation) max_file_size: int = 0 # 0 = unlimited (Docling-level, bytes) + batch_page_size: int = 0 # 0 = disabled, > 0 = pages per batch upload_dir: str = "./uploads" db_path: str = "./data/docling_studio.db" cors_origins: list[str] = field( @@ -42,6 +43,8 @@ class Settings: errors.append(f"max_page_count must be >= 0 (got {self.max_page_count})") if self.max_file_size < 0: errors.append(f"max_file_size must be >= 0 (got {self.max_file_size})") + if self.batch_page_size < 0: + errors.append(f"batch_page_size must be >= 0 (got {self.batch_page_size})") if self.default_table_mode not in ("accurate", "fast"): errors.append( f"default_table_mode must be 'accurate' or 'fast' (got '{self.default_table_mode}')" @@ -78,6 +81,7 @@ class Settings: default_table_mode=os.environ.get("DEFAULT_TABLE_MODE", "accurate"), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), max_file_size=int(os.environ.get("MAX_FILE_SIZE", "0")), + batch_page_size=int(os.environ.get("BATCH_PAGE_SIZE", "0")), upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"), cors_origins=[o.strip() for o in cors_raw.split(",")], diff --git a/document-parser/persistence/analysis_repo.py b/document-parser/persistence/analysis_repo.py index deb08b6..d6d8c7b 100644 --- a/document-parser/persistence/analysis_repo.py +++ b/document-parser/persistence/analysis_repo.py @@ -27,6 +27,8 @@ def _row_to_job(row) -> AnalysisJob: document_json=row["document_json"] if "document_json" in keys else None, chunks_json=row["chunks_json"] if "chunks_json" in keys else None, error_message=row["error_message"], + progress_current=row["progress_current"] if "progress_current" in keys else None, + progress_total=row["progress_total"] if "progress_total" in keys else None, started_at=_parse_dt(row["started_at"]), completed_at=_parse_dt(row["completed_at"]), created_at=_parse_dt(row["created_at"]) or datetime.now(UTC), @@ -78,7 +80,8 @@ async def update_status(job: AnalysisJob) -> None: """UPDATE analysis_jobs SET status = ?, content_markdown = ?, content_html = ?, pages_json = ?, document_json = ?, chunks_json = ?, - error_message = ?, started_at = ?, completed_at = ? + error_message = ?, progress_current = ?, progress_total = ?, + started_at = ?, completed_at = ? WHERE id = ?""", ( job.status.value, @@ -88,6 +91,8 @@ async def update_status(job: AnalysisJob) -> None: job.document_json, job.chunks_json, job.error_message, + job.progress_current, + job.progress_total, str(job.started_at) if job.started_at else None, str(job.completed_at) if job.completed_at else None, job.id, @@ -96,6 +101,16 @@ async def update_status(job: AnalysisJob) -> None: await db.commit() +async def update_progress(job_id: str, current: int, total: int) -> None: + """Update only the progress columns for a running analysis.""" + async with get_connection() as db: + await db.execute( + "UPDATE analysis_jobs SET progress_current = ?, progress_total = ? WHERE id = ?", + (current, total, job_id), + ) + await db.commit() + + async def update_chunks(job_id: str, chunks_json: str) -> bool: """Update only the chunks_json column for a completed analysis.""" async with get_connection() as db: diff --git a/document-parser/persistence/database.py b/document-parser/persistence/database.py index 10d82d3..b32e694 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -50,6 +50,8 @@ CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at); _MIGRATIONS = [ ("document_json", "ALTER TABLE analysis_jobs ADD COLUMN document_json TEXT"), ("chunks_json", "ALTER TABLE analysis_jobs ADD COLUMN chunks_json TEXT"), + ("progress_current", "ALTER TABLE analysis_jobs ADD COLUMN progress_current INTEGER"), + ("progress_total", "ALTER TABLE analysis_jobs ADD COLUMN progress_total INTEGER"), ] diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index f003285..1cffe2d 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -10,11 +10,21 @@ import asyncio import functools import json import logging +import math +import re from dataclasses import asdict from typing import TYPE_CHECKING +import pypdfium2 as pdfium + from domain.models import AnalysisJob, AnalysisStatus -from domain.value_objects import ChunkingOptions, ChunkResult, ConversionOptions, ConversionResult +from domain.value_objects import ( + ChunkingOptions, + ChunkResult, + ConversionOptions, + ConversionResult, + PageDetail, +) from infra.settings import settings if TYPE_CHECKING: @@ -38,6 +48,67 @@ def _chunk_to_dict(c: ChunkResult) -> dict: # Maximum number of concurrent analysis jobs to prevent resource exhaustion. _DEFAULT_MAX_CONCURRENT = 3 +# Regex to extract content from Docling's well-formed HTML output. +_BODY_RE = re.compile(r"]*>(.*)", re.DOTALL | re.IGNORECASE) + + +def _count_pdf_pages(file_path: str) -> int: + """Count pages in a PDF. Returns 0 if the file is not a valid PDF.""" + try: + pdf = pdfium.PdfDocument(file_path) + count = len(pdf) + pdf.close() + return count + except Exception: + logger.debug("Cannot open %s as PDF, batching disabled", file_path) + return 0 + + +def _extract_html_body(html: str) -> str: + """Extract content between tags. + + Docling produces well-formed HTML — regex is safe for this controlled output. + Returns raw html as fallback if no tag is found. + """ + match = _BODY_RE.search(html) + return match.group(1).strip() if match else html + + +def _merge_results(results: list[ConversionResult]) -> ConversionResult: + """Merge multiple batch ConversionResults into a single consolidated result. + + document_json is intentionally set to None: merging DoclingDocument's internal + tree structure across batches is error-prone. Re-chunking is disabled for + batched conversions (robustness decision for 0.3.1). + """ + if not results: + return ConversionResult(page_count=0, content_markdown="", content_html="", pages=[]) + + all_pages: list[PageDetail] = [] + all_md: list[str] = [] + html_bodies: list[str] = [] + total_skipped = 0 + + for r in results: + all_pages.extend(r.pages) + all_md.append(r.content_markdown) + html_bodies.append(_extract_html_body(r.content_html)) + total_skipped += r.skipped_items + + merged_body = "\n".join(html_bodies) + merged_html = ( + f'{merged_body}' + ) + + return ConversionResult( + page_count=sum(r.page_count for r in results), + content_markdown="\n\n".join(all_md), + content_html=merged_html, + pages=all_pages, + skipped_items=total_skipped, + document_json=None, + ) + class AnalysisService: """Orchestrates document analysis using an injected converter.""" @@ -121,6 +192,66 @@ class AnalysisService: return chunks + async def _run_batched_conversion( + self, + job_id: str, + file_path: str, + options: ConversionOptions, + total_pages: int, + batch_size: int, + ) -> ConversionResult | None: + """Convert a document in batches using page_range. + + Returns None if the job was deleted mid-batch (caller should abort). + Raises on batch failure (fail-fast: entire job fails). + """ + num_batches = math.ceil(total_pages / batch_size) + await analysis_repo.update_progress(job_id, 0, total_pages) + logger.info( + "Batched conversion: %d pages in %d batches of %d for job %s", + total_pages, + num_batches, + batch_size, + job_id, + ) + + results: list[ConversionResult] = [] + for batch_idx in range(num_batches): + start = batch_idx * batch_size + 1 + end = min(start + batch_size - 1, total_pages) + + if not await analysis_repo.find_by_id(job_id): + logger.info( + "Job %s deleted during batch %d/%d, aborting", + job_id, + batch_idx + 1, + num_batches, + ) + return None + + try: + batch_result = await asyncio.wait_for( + self._converter.convert(file_path, options, page_range=(start, end)), + timeout=self._conversion_timeout, + ) + except Exception as exc: + raise RuntimeError( + f"Batch {batch_idx + 1}/{num_batches} (pages {start}-{end}) failed: {exc}" + ) from exc + + results.append(batch_result) + await analysis_repo.update_progress(job_id, end, total_pages) + logger.info( + "Batch %d/%d done (pages %d-%d) for job %s", + batch_idx + 1, + num_batches, + start, + end, + job_id, + ) + + return _merge_results(results) + def _on_task_done(self, task: asyncio.Task, *, job_id: str) -> None: """Cleanup running tasks and delegate to module-level handler.""" self._running_tasks.pop(job_id, None) @@ -168,10 +299,24 @@ class AnalysisService: opts_dict = {**opts_dict, "table_mode": settings.default_table_mode} options = ConversionOptions(**opts_dict) - result: ConversionResult = await asyncio.wait_for( - self._converter.convert(file_path, options), - timeout=self._conversion_timeout, - ) + total_pages = _count_pdf_pages(file_path) + batch_size = settings.batch_page_size + + if batch_size > 0 and total_pages > batch_size: + result = await self._run_batched_conversion( + job_id, + file_path, + options, + total_pages, + batch_size, + ) + if result is None: + return # job was deleted mid-batch + else: + result = await asyncio.wait_for( + self._converter.convert(file_path, options), + timeout=self._conversion_timeout, + ) pages_json = json.dumps([asdict(p) for p in result.pages]) diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py index a211dee..09edf4a 100644 --- a/document-parser/tests/test_analysis_service.py +++ b/document-parser/tests/test_analysis_service.py @@ -1,4 +1,4 @@ -"""Tests for AnalysisService — callbacks, concurrency, and orchestration.""" +"""Tests for AnalysisService — callbacks, concurrency, orchestration, and batching.""" from __future__ import annotations @@ -8,7 +8,14 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from services.analysis_service import AnalysisService, _on_task_done +from domain.value_objects import ConversionResult, PageDetail +from services.analysis_service import ( + AnalysisService, + _count_pdf_pages, + _extract_html_body, + _merge_results, + _on_task_done, +) class TestOnTaskDone: @@ -186,3 +193,224 @@ class TestAnalysisServiceConcurrency: # Both should have completed assert call_order.count("start") == 2 assert call_order.count("end") == 2 + + +# --------------------------------------------------------------------------- +# Batch helper tests +# --------------------------------------------------------------------------- + + +class TestCountPdfPages: + def test_valid_pdf(self, tmp_path): + """A real (minimal) PDF should return its page count.""" + # Create a minimal valid 1-page PDF using pypdfium2 + import pypdfium2 as pdfium + + pdf = pdfium.PdfDocument.new() + pdf.new_page(612, 792) + path = tmp_path / "test.pdf" + pdf.save(str(path)) + pdf.close() + + assert _count_pdf_pages(str(path)) == 1 + + def test_non_pdf_file(self, tmp_path): + """A non-PDF file should return 0.""" + path = tmp_path / "test.docx" + path.write_bytes(b"PK\x03\x04 not a pdf") + assert _count_pdf_pages(str(path)) == 0 + + def test_nonexistent_file(self): + """A nonexistent file should return 0.""" + assert _count_pdf_pages("/no/such/file.pdf") == 0 + + def test_empty_file(self, tmp_path): + """An empty file should return 0.""" + path = tmp_path / "empty.pdf" + path.write_bytes(b"") + assert _count_pdf_pages(str(path)) == 0 + + +class TestExtractHtmlBody: + def test_extracts_body(self): + html = '

Hello

' + assert _extract_html_body(html) == "

Hello

" + + def test_no_body_tag_returns_raw(self): + html = "

No body tag

" + assert _extract_html_body(html) == html + + def test_empty_body(self): + html = "" + assert _extract_html_body(html) == "" + + +class TestMergeResults: + def test_empty_list(self): + result = _merge_results([]) + assert result.page_count == 0 + assert result.content_markdown == "" + assert result.pages == [] + assert result.document_json is None + + def test_single_result_passthrough(self): + r = ConversionResult( + page_count=3, + content_markdown="# Page 1", + content_html="

Page 1

", + pages=[PageDetail(page_number=1, width=612, height=792)], + document_json='{"pages": {}}', + ) + merged = _merge_results([r]) + assert merged.page_count == 3 + assert merged.content_markdown == "# Page 1" + assert merged.pages == [PageDetail(page_number=1, width=612, height=792)] + assert merged.document_json is None # intentionally dropped + + def test_merges_multiple_results(self): + r1 = ConversionResult( + page_count=2, + content_markdown="# Batch 1", + content_html="

B1

", + pages=[ + PageDetail(page_number=1, width=612, height=792), + PageDetail(page_number=2, width=612, height=792), + ], + skipped_items=1, + ) + r2 = ConversionResult( + page_count=2, + content_markdown="# Batch 2", + content_html="

B2

", + pages=[ + PageDetail(page_number=3, width=612, height=792), + PageDetail(page_number=4, width=612, height=792), + ], + skipped_items=2, + ) + merged = _merge_results([r1, r2]) + assert merged.page_count == 4 + assert merged.content_markdown == "# Batch 1\n\n# Batch 2" + assert len(merged.pages) == 4 + assert merged.pages[0].page_number == 1 + assert merged.pages[3].page_number == 4 + assert merged.skipped_items == 3 + assert merged.document_json is None + assert "

B1

" in merged.content_html + assert "

B2

" in merged.content_html + # Valid HTML structure + assert merged.content_html.startswith("") + assert "" in merged.content_html + + +# --------------------------------------------------------------------------- +# Batched conversion integration tests +# --------------------------------------------------------------------------- + + +class TestBatchedConversion: + @pytest.mark.asyncio + async def test_batched_conversion_produces_merged_result(self): + """When batch_page_size is set and document exceeds it, results are merged.""" + converter = AsyncMock() + + # Simulate 2 batches: pages 1-5 and 6-8 + converter.convert.side_effect = [ + ConversionResult( + page_count=5, + content_markdown="# Batch 1", + content_html="

B1

", + pages=[PageDetail(page_number=i, width=612, height=792) for i in range(1, 6)], + ), + ConversionResult( + page_count=3, + content_markdown="# Batch 2", + content_html="

B2

", + pages=[PageDetail(page_number=i, width=612, height=792) for i in range(6, 9)], + ), + ] + + service = AnalysisService(converter=converter, conversion_timeout=60) + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.find_by_id = AsyncMock(return_value=MagicMock()) + mock_repo.update_progress = AsyncMock() + + result = await service._run_batched_conversion( + "job-1", + "/fake.pdf", + MagicMock(), + total_pages=8, + batch_size=5, + ) + + assert result is not None + assert result.page_count == 8 + assert len(result.pages) == 8 + assert result.document_json is None + assert converter.convert.call_count == 2 + + # Verify page_range was passed correctly + call1_kwargs = converter.convert.call_args_list[0].kwargs + call2_kwargs = converter.convert.call_args_list[1].kwargs + assert call1_kwargs["page_range"] == (1, 5) + assert call2_kwargs["page_range"] == (6, 8) + + @pytest.mark.asyncio + async def test_batch_failure_raises_with_enriched_message(self): + """If a batch fails, RuntimeError is raised with batch info.""" + converter = AsyncMock() + converter.convert.side_effect = [ + ConversionResult( + page_count=5, + content_markdown="ok", + content_html="ok", + pages=[PageDetail(page_number=i, width=612, height=792) for i in range(1, 6)], + ), + RuntimeError("OOM"), + ] + + service = AnalysisService(converter=converter, conversion_timeout=60) + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.find_by_id = AsyncMock(return_value=MagicMock()) + mock_repo.update_progress = AsyncMock() + + with pytest.raises(RuntimeError, match=r"Batch 2/2 \(pages 6-8\) failed: OOM"): + await service._run_batched_conversion( + "job-fail", + "/fake.pdf", + MagicMock(), + total_pages=8, + batch_size=5, + ) + + @pytest.mark.asyncio + async def test_job_deleted_mid_batch_returns_none(self): + """If the job is deleted between batches, conversion aborts with None.""" + converter = AsyncMock() + converter.convert.return_value = ConversionResult( + page_count=5, + content_markdown="ok", + content_html="ok", + pages=[PageDetail(page_number=i, width=612, height=792) for i in range(1, 6)], + ) + + service = AnalysisService(converter=converter, conversion_timeout=60) + + # First check returns job, second returns None (deleted) + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.find_by_id = AsyncMock(side_effect=[MagicMock(), None]) + mock_repo.update_progress = AsyncMock() + + result = await service._run_batched_conversion( + "job-del", + "/fake.pdf", + MagicMock(), + total_pages=10, + batch_size=5, + ) + + assert result is None + # Only first batch should have been converted + assert converter.convert.call_count == 1 diff --git a/document-parser/tests/test_serve_converter.py b/document-parser/tests/test_serve_converter.py index 58c5ecf..1f073c6 100644 --- a/document-parser/tests/test_serve_converter.py +++ b/document-parser/tests/test_serve_converter.py @@ -56,6 +56,14 @@ class TestBuildFormData: assert data["images_scale"] == "2.0" assert data["include_images"] == "true" + def test_page_range_included_when_set(self): + data = _build_form_data(ConversionOptions(), page_range=(11, 20)) + assert data["page_range"] == "11-20" + + def test_page_range_absent_when_none(self): + data = _build_form_data(ConversionOptions()) + assert "page_range" not in data + # --------------------------------------------------------------------------- # Unit tests — response parsing diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index 604abac..20dafad 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -17,6 +17,7 @@ class TestSettingsDefaults: assert s.document_timeout == 120.0 assert s.lock_timeout == 300 assert s.max_page_count == 0 + assert s.batch_page_size == 0 assert s.upload_dir == "./uploads" assert s.db_path == "./data/docling_studio.db" assert "http://localhost:3000" in s.cors_origins @@ -67,6 +68,20 @@ class TestSettingsValidation: with pytest.raises(ValueError, match="max_file_size must be >= 0"): Settings(max_file_size=-1) + def test_negative_batch_page_size_rejected(self): + import pytest + + with pytest.raises(ValueError, match="batch_page_size must be >= 0"): + Settings(batch_page_size=-1) + + def test_zero_batch_page_size_accepted(self): + s = Settings(batch_page_size=0) + assert s.batch_page_size == 0 + + def test_positive_batch_page_size_accepted(self): + s = Settings(batch_page_size=10) + assert s.batch_page_size == 10 + def test_zero_lock_timeout_rejected(self): import pytest @@ -114,6 +129,7 @@ class TestSettingsFromEnv: monkeypatch.setenv("DOCUMENT_TIMEOUT", "60.0") monkeypatch.setenv("LOCK_TIMEOUT", "600") monkeypatch.setenv("MAX_PAGE_COUNT", "20") + monkeypatch.setenv("BATCH_PAGE_SIZE", "15") monkeypatch.setenv("UPLOAD_DIR", "/data/uploads") monkeypatch.setenv("DB_PATH", "/data/test.db") monkeypatch.setenv("CORS_ORIGINS", "http://a.com, http://b.com") @@ -129,6 +145,7 @@ class TestSettingsFromEnv: assert s.lock_timeout == 600 assert s.document_timeout == 60.0 assert s.max_page_count == 20 + assert s.batch_page_size == 15 assert s.upload_dir == "/data/uploads" assert s.db_path == "/data/test.db" assert s.cors_origins == ["http://a.com", "http://b.com"] diff --git a/frontend/src/features/analysis/ui/AnalysisPanel.vue b/frontend/src/features/analysis/ui/AnalysisPanel.vue index 2f464e3..7e714cd 100644 --- a/frontend/src/features/analysis/ui/AnalysisPanel.vue +++ b/frontend/src/features/analysis/ui/AnalysisPanel.vue @@ -38,6 +38,18 @@
{{ analysisStore.currentAnalysis.status }} +
diff --git a/frontend/src/features/analysis/ui/ResultTabs.vue b/frontend/src/features/analysis/ui/ResultTabs.vue index 2369fb4..0ac9329 100644 --- a/frontend/src/features/analysis/ui/ResultTabs.vue +++ b/frontend/src/features/analysis/ui/ResultTabs.vue @@ -105,6 +105,18 @@