Feature: support self-hosted PaddleOCR-VL deployments (#19011)

PaddleOCR could only reach the hosted AI Studio service, whose
asynchronous job API (submit -> poll -> fetch) is unreachable in an
air-gapped install. A self-hosted PaddleOCR-VL exposes a PaddleX
pipeline that answers a single request synchronously instead; the two
protocols share nothing but the result schema.

Pick the protocol from the configured address rather than adding a
second provider, so existing configurations keep working and only the
URL decides: just paddleocr.aistudio-app.com serves the job API, and any
other host gets the synchronous /layout-parsing call. The address is
compared by hostname, so an AI Studio deployment on a sibling subdomain
is routed to the synchronous pipeline while a lookalike domain is not
mistaken for the hosted service. Self-hosted deployments usually run
unauthenticated, so the access token is now required only for the hosted
service.

Fix two defects this exposed on the way:

- A block whose content is nothing but a markdown image became an empty
section carrying a position tag, because emptiness was tested before the
images were stripped. Scanned pages that the layout model labels as one
image block hit this on every page.
- HTTP, JSON and errorCode failures on the synchronous path never
reached the callback, so a wrong path or a rejected token left the
document at zero chunks with no visible cause.

Closes #18757

deepdoc/parser/paddleocr_parser.py

@@ -14,6 +14,7 @@
 #
 from __future__ import annotations
 
+import base64
 import json
 import logging
 import os
@@ -25,6 +26,7 @@ from io import BytesIO
 from os import PathLike
 from pathlib import Path
 from typing import Any, Callable, ClassVar, Literal, Optional, Union, Tuple, List
+from urllib.parse import urlparse
 
 import numpy as np
 import pdfplumber
@@ -74,6 +76,26 @@ def _remove_images_from_markdown(markdown: str) -> str:
     return _MARKDOWN_IMAGE_PATTERN.sub("", markdown)
 
 
+_DEFAULT_BASE_URL = "https://paddleocr.aistudio-app.com"
+# Derived so the address and the host that selects the protocol cannot drift apart.
+_HOSTED_HOSTNAME = urlparse(_DEFAULT_BASE_URL).hostname
+
+
+def _is_hosted_service(base_url: str) -> bool:
+    """Tell the hosted service apart from a self-hosted PaddleX deployment.
+
+    The two speak different wire protocols — an asynchronous job API versus a
+    single synchronous pipeline call — and share nothing but the result schema.
+    Only the hosted address serves the job API, so the configured host decides
+    which protocol to use. An unset address falls back to the hosted default.
+    """
+    if not base_url:
+        return True
+    # A bare "host:port" would otherwise parse its host as the URL scheme.
+    parsed = urlparse(base_url if "//" in base_url else f"//{base_url}")
+    return (parsed.hostname or "").lower() == _HOSTED_HOSTNAME
+
+
 def _normalize_bbox(bbox: list[Any] | tuple[Any, ...]) -> tuple[float, float, float, float]:
     if len(bbox) < 4:
         return 0.0, 0.0, 0.0, 0.0
@@ -121,7 +143,7 @@ class PaddleOCRVLConfig:
 class PaddleOCRConfig:
     """Main configuration for PaddleOCR parser."""
 
-    base_url: str = "https://paddleocr.aistudio-app.com"
+    base_url: str = _DEFAULT_BASE_URL
     access_token: Optional[str] = None
     algorithm: AlgorithmType = "PaddleOCR-VL"
     request_timeout: int = 600
@@ -131,6 +153,11 @@ class PaddleOCRConfig:
     additional_params: dict[str, Any] = field(default_factory=dict)
     algorithm_config: dict[str, Any] = field(default_factory=dict)
 
+    @property
+    def local(self) -> bool:
+        """Whether the address points at a self-hosted, synchronous deployment."""
+        return not _is_hosted_service(self.base_url)
+
     @classmethod
     def from_dict(cls, config: Optional[dict[str, Any]]) -> "PaddleOCRConfig":
         """Create configuration from dictionary."""
@@ -173,14 +200,15 @@ class PaddleOCRConfig:
         return cls.from_dict(kwargs)
 
 
-_DEFAULT_BASE_URL = "https://paddleocr.aistudio-app.com"
-
-
 class PaddleOCRParser(RAGFlowPdfParser):
     """Parser for PDF documents using PaddleOCR API."""
 
     _ZOOMIN = 2
 
+    # Path the self-hosted PaddleX pipeline serves; the hosted service instead
+    # exposes the asynchronous job API under /api/v2/ocr/jobs.
+    _LOCAL_ENDPOINT_PATH = "layout-parsing"
+
     _COMMON_FIELD_MAPPING: ClassVar[dict[str, str]] = {
         "prettify_markdown": "prettifyMarkdown",
         "show_formula_number": "showFormulaNumber",
@@ -215,13 +243,6 @@ class PaddleOCRParser(RAGFlowPdfParser):
         "relevel_titles": "relevelTitles",
     }
 
-    _ALGORITHM_FIELD_MAPPINGS: ClassVar[dict[str, dict[str, str]]] = {
-        "PaddleOCR-VL": _VL_FIELD_MAPPING,
-        "PP-OCRv5": _VL_FIELD_MAPPING,
-        "PP-StructureV3": _VL_FIELD_MAPPING,
-        "PaddleOCR-VL-1.5": _VL_FIELD_MAPPING,
-    }
-
     def __init__(
         self,
         base_url: Optional[str] = None,
@@ -238,17 +259,24 @@ class PaddleOCRParser(RAGFlowPdfParser):
         self.request_timeout = request_timeout
         self.logger = logging.getLogger(self.__class__.__name__)
 
-        # Force PDF file type
-        self.file_type = 0
-
         # Initialize page images for cropping
         self.page_images: list[Image.Image] = []
         self.page_from = 0
 
+    @property
+    def local(self) -> bool:
+        """Whether the address points at a self-hosted, synchronous deployment."""
+        return not _is_hosted_service(self.base_url)
+
     # Public methods
     def check_installation(self) -> tuple[bool, str]:
         """Check if the parser is properly installed and configured."""
-        if not self.access_token:
+        if not self.base_url:
+            return False, "[PaddleOCR] Base URL not configured"
+
+        # A self-hosted deployment usually runs unauthenticated, so requiring a
+        # token there would reject valid configurations.
+        if not self.local and not self.access_token:
             return False, "[PaddleOCR] Access token not configured"
 
         return True, ""
@@ -408,8 +436,8 @@ class PaddleOCRParser(RAGFlowPdfParser):
 
         return source_path.read_bytes()
 
-    def _build_payload(self, data: bytes, file_type: int, config: PaddleOCRConfig) -> dict[str, Any]:
-        """Build optionalPayload for async Job API request."""
+    def _build_payload(self, config: PaddleOCRConfig) -> dict[str, Any]:
+        """Build the camelCase pipeline parameters shared by both wire protocols."""
         payload: dict[str, Any] = {}
 
         # Add common parameters
@@ -423,10 +451,9 @@ class PaddleOCRParser(RAGFlowPdfParser):
                 payload[api_param] = param_value
 
         # Add algorithm-specific parameters
-        algorithm_mapping = self._ALGORITHM_FIELD_MAPPINGS.get(config.algorithm, {})
         for param_key, param_value in config.algorithm_config.items():
-            if param_value is not None and param_key in algorithm_mapping:
-                api_param = algorithm_mapping[param_key]
+            if param_value is not None and param_key in self._VL_FIELD_MAPPING:
+                api_param = self._VL_FIELD_MAPPING[param_key]
                 payload[api_param] = param_value
 
         # Add any additional parameters
@@ -435,14 +462,105 @@ class PaddleOCRParser(RAGFlowPdfParser):
 
         return payload
 
+    @staticmethod
+    def _detect_file_type(data: bytes) -> int:
+        """Map the payload to the fileType the PaddleX pipeline expects: 0 = PDF, 1 = image."""
+        return 0 if data[:4] == b"%PDF" else 1
+
+    @classmethod
+    def _local_endpoint(cls, base_url: str) -> str:
+        """Append the pipeline path unless the configured address already carries it.
+
+        The address is meant to be the deployment root, but the endpoint is the
+        only URL PaddleX documents, so pasting it whole is the likelier mistake.
+        """
+        root = base_url.strip().rstrip("/")
+        if root.rsplit("/", 1)[-1] == cls._LOCAL_ENDPOINT_PATH:
+            return root
+        return f"{root}/{cls._LOCAL_ENDPOINT_PATH}"
+
+    def _send_local_request(self, data: bytes, config: PaddleOCRConfig, callback: Optional[Callable[[float, str], None]]) -> dict[str, Any]:
+        """Send request to a self-hosted PaddleX layout-parsing endpoint (single synchronous call).
+
+        Unlike the hosted service there is no job to submit and poll: the
+        pipeline answers with the parsed result directly, and the algorithm is
+        fixed by the server's startup arguments rather than chosen per request.
+        """
+        payload: dict[str, Any] = {
+            "file": base64.b64encode(data).decode("ascii"),
+            "fileType": self._detect_file_type(data),
+            # Markdown images are stripped from every section anyway, so asking
+            # the server to inline them as base64 would only inflate the response.
+            "returnMarkdownImages": False,
+        }
+        payload.update(self._build_payload(config))
+
+        url = self._local_endpoint(config.base_url)
+        headers: dict[str, str] = {"Content-Type": "application/json", "Client-Platform": "ragflow"}
+        if config.access_token:
+            # A bearer token on a plaintext connection is readable by anyone on
+            # the path, so drop it instead of leaking it.
+            if url.lower().startswith("https://"):
+                headers["Authorization"] = f"Bearer {config.access_token}"
+            else:
+                self.logger.warning("[PaddleOCR] access token not sent: the endpoint is not HTTPS")
+
+        self.logger.info(f"[PaddleOCR] local request: {url}")
+        if callback:
+            callback(0.1, "[PaddleOCR] submitting request")
+
+        def _fail(reason: str) -> RuntimeError:
+            # by_paddleocr() only logs the raised error, so a failure that never
+            # reaches the callback leaves the document at zero chunks with no
+            # visible cause.
+            if callback:
+                callback(-1, reason)
+            return RuntimeError(reason)
+
+        try:
+            resp = requests.post(url, json=payload, headers=headers, timeout=config.request_timeout)
+        except Exception as exc:
+            raise _fail(f"[PaddleOCR] request failed: {exc}")
+
+        if resp.status_code != 200:
+            raise _fail(f"[PaddleOCR] request failed: HTTP {resp.status_code} {resp.text[:500]}")
+
+        try:
+            body = resp.json()
+        except ValueError as exc:
+            raise _fail(f"[PaddleOCR] response is not JSON: {exc}")
+
+        error_code = body.get("errorCode")
+        if error_code:
+            raise _fail(f"[PaddleOCR] request failed: {body.get('errorMsg', 'Unknown error')} (errorCode: {error_code})")
+
+        if callback:
+            callback(0.8, "[PaddleOCR] result received")
+
+        result = body.get("result", {})
+        return {
+            "layoutParsingResults": result.get("layoutParsingResults", []),
+            "ocrResults": result.get("ocrResults", []),
+        }
+
     def _send_request(self, data: bytes, config: PaddleOCRConfig, callback: Optional[Callable[[float, str], None]]) -> dict[str, Any]:
-        """Send request to PaddleOCR async Job API (submit → poll → fetch)."""
-        optional_payload = self._build_payload(data, self.file_type, config)
+        """Send the parse request and return the result in a shape both protocols share.
+
+        Self-hosted deployments answer synchronously; the hosted service runs an
+        asynchronous job (submit → poll → fetch).
+        """
+        if config.local:
+            return self._send_local_request(data, config, callback)
+
+        optional_payload = self._build_payload(config)
 
         # Prepare headers
         headers: dict[str, str] = {"Client-Platform": "ragflow"}
         if config.access_token:
-            headers["Authorization"] = f"Bearer {config.access_token}"
+            if config.base_url.strip().lower().startswith("https://"):
+                headers["Authorization"] = f"Bearer {config.access_token}"
+            else:
+                self.logger.warning("[PaddleOCR] access token not sent: the endpoint is not HTTPS")
 
         jobs_url = f"{config.base_url.rstrip('/')}/api/v2/ocr/jobs"
         deadline = time.monotonic() + config.request_timeout
@@ -606,13 +724,13 @@ class PaddleOCRParser(RAGFlowPdfParser):
                 parsing_res_list = pruned_result.get("parsing_res_list", [])
 
                 for block in parsing_res_list:
-                    block_content = block.get("block_content", "").strip()
+                    # Strip images before testing for emptiness: a pure-image
+                    # block is all markup, so it would otherwise land in the
+                    # sections as an empty string carrying a position tag.
+                    block_content = _remove_images_from_markdown(block.get("block_content", "")).strip()
                     if not block_content:
                         continue
 
-                    # Remove images
-                    block_content = _remove_images_from_markdown(block_content)
-
                     label = block.get("block_label", "")
                     block_bbox = block.get("block_bbox", [0, 0, 0, 0])
                     left, top, right, bottom = _normalize_bbox(block_bbox)

test/unit_test/deepdoc/parser/test_paddleocr_parser.py

@@ -0,0 +1,381 @@
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import ModuleType
+from unittest.mock import Mock
+
+import pytest
+
+
+def _load_paddleocr_parser(monkeypatch):
+    repo_root = Path(__file__).resolve().parents[4]
+
+    deepdoc_mod = ModuleType("deepdoc")
+    deepdoc_mod.__path__ = [str(repo_root / "deepdoc")]
+    monkeypatch.setitem(sys.modules, "deepdoc", deepdoc_mod)
+
+    parser_mod = ModuleType("deepdoc.parser")
+    parser_mod.__path__ = [str(repo_root / "deepdoc" / "parser")]
+    monkeypatch.setitem(sys.modules, "deepdoc.parser", parser_mod)
+
+    pdf_parser_mod = ModuleType("deepdoc.parser.pdf_parser")
+
+    class _RAGFlowPdfParser:
+        pass
+
+    pdf_parser_mod.RAGFlowPdfParser = _RAGFlowPdfParser
+    monkeypatch.setitem(sys.modules, "deepdoc.parser.pdf_parser", pdf_parser_mod)
+
+    utils_mod = ModuleType("deepdoc.parser.utils")
+    utils_mod.extract_pdf_outlines = lambda *_args, **_kwargs: []
+    monkeypatch.setitem(sys.modules, "deepdoc.parser.utils", utils_mod)
+
+    module_name = "test_paddleocr_parser_unit_module"
+    module_path = repo_root / "deepdoc" / "parser" / "paddleocr_parser.py"
+    spec = importlib.util.spec_from_file_location(module_name, module_path)
+    module = importlib.util.module_from_spec(spec)
+    monkeypatch.setitem(sys.modules, module_name, module)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _clear_env(monkeypatch):
+    for key in ("PADDLEOCR_BASE_URL", "PADDLEOCR_ACCESS_TOKEN", "PADDLEOCR_ALGORITHM"):
+        monkeypatch.delenv(key, raising=False)
+
+
+def _local_response(markdown="hello", pruned=None):
+    resp = Mock()
+    resp.status_code = 200
+    resp.json.return_value = {
+        "logId": "test-log-id",
+        "errorCode": 0,
+        "errorMsg": "Success",
+        "result": {
+            "layoutParsingResults": [
+                {
+                    "prunedResult": pruned or {"parsing_res_list": []},
+                    "markdown": {"text": markdown},
+                }
+            ]
+        },
+    }
+    return resp
+
+
+@pytest.mark.p1
+@pytest.mark.parametrize(
+    # Not named "base_url": pytest-base-url ships a session-scoped fixture of
+    # that name, and shadowing it with a function-scoped parametrize argument
+    # makes its session-scoped consumer fail with ScopeMismatch.
+    ("address", "expect_local"),
+    [
+        # Only the hosted address serves the asynchronous job API.
+        ("https://paddleocr.aistudio-app.com", False),
+        ("https://paddleocr.aistudio-app.com/", False),
+        ("https://paddleocr.aistudio-app.com/api", False),
+        ("http://PaddleOCR.AIStudio-App.com", False),
+        (None, False),
+        ("http://localhost:8080", True),
+        ("http://127.0.0.1:8080/layout-parsing", True),
+        ("https://paddleocr.internal.example.com", True),
+        # An AI Studio deployment on a sibling subdomain serves the synchronous
+        # pipeline, so matching the parent domain rather than the exact host
+        # would route a working deployment to the wrong protocol.
+        ("https://o5r7debeac17pbt3.aistudio-app.com/layout-parsing", True),
+        # A lookalike host must not be mistaken for the hosted service.
+        ("https://paddleocr.aistudio-app.com.evil.example", True),
+        ("localhost:8080", True),
+    ],
+)
+def test_protocol_is_selected_by_address(monkeypatch, address, expect_local):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    parser = module.PaddleOCRParser(base_url=address)
+    assert parser.local is expect_local
+    assert module.PaddleOCRConfig(base_url=parser.base_url).local is expect_local
+
+
+@pytest.mark.p1
+def test_self_hosted_needs_base_url_not_token(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    # A self-hosted deployment usually runs unauthenticated, so the token must
+    # not be required; its address must be.
+    ok, reason = module.PaddleOCRParser(base_url="http://127.0.0.1:8080").check_installation()
+    assert ok is True
+    assert reason == ""
+
+    monkeypatch.setenv("PADDLEOCR_BASE_URL", "")
+    ok, reason = module.PaddleOCRParser().check_installation()
+    assert ok is False
+    assert "Base URL" in reason
+
+
+@pytest.mark.p1
+def test_hosted_still_requires_token(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    ok, reason = module.PaddleOCRParser().check_installation()
+    assert ok is False
+    assert "Access token" in reason
+
+    ok, _ = module.PaddleOCRParser(access_token="tok").check_installation()
+    assert ok is True
+
+
+@pytest.mark.p1
+def test_self_hosted_posts_to_layout_parsing_once(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+    # The synchronous endpoint must not be polled like the hosted job API.
+    monkeypatch.setattr(module.requests, "get", Mock(side_effect=AssertionError("must not poll")))
+
+    parser = module.PaddleOCRParser(base_url="http://127.0.0.1:8080/")
+    result = parser._send_request(b"\x89PNG binary", module.PaddleOCRConfig(base_url="http://127.0.0.1:8080"), None)
+
+    assert post.call_count == 1
+    assert post.call_args.args[0] == "http://127.0.0.1:8080/layout-parsing"
+    assert result["layoutParsingResults"][0]["markdown"]["text"] == "hello"
+    assert result["ocrResults"] == []
+
+
+@pytest.mark.p1
+@pytest.mark.parametrize(
+    ("configured", "expected"),
+    [
+        ("http://127.0.0.1:8080", "http://127.0.0.1:8080/layout-parsing"),
+        ("http://127.0.0.1:8080/", "http://127.0.0.1:8080/layout-parsing"),
+        # PaddleX documents the endpoint rather than the root, so the whole URL
+        # is the likelier thing to be pasted into the address field.
+        ("http://127.0.0.1:8080/layout-parsing", "http://127.0.0.1:8080/layout-parsing"),
+        ("http://127.0.0.1:8080/layout-parsing/", "http://127.0.0.1:8080/layout-parsing"),
+        ("http://127.0.0.1:8080/paddle/layout-parsing", "http://127.0.0.1:8080/paddle/layout-parsing"),
+    ],
+)
+def test_self_hosted_endpoint_is_not_duplicated(monkeypatch, configured, expected):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    parser = module.PaddleOCRParser(base_url=configured)
+    parser._send_request(b"%PDF-1.7 rest", module.PaddleOCRConfig(base_url=configured), None)
+
+    assert post.call_args.args[0] == expected
+
+
+@pytest.mark.p1
+@pytest.mark.parametrize(
+    ("data", "expected_type"),
+    [(b"%PDF-1.7 rest", 0), (b"\x89PNG\r\n\x1a\n", 1), (b"\xff\xd8\xff\xe0 jpeg", 1), (b"", 1)],
+)
+def test_file_type_detection(monkeypatch, data, expected_type):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    parser._send_request(data, module.PaddleOCRConfig(base_url="http://svc"), None)
+
+    assert post.call_args.kwargs["json"]["fileType"] == expected_type
+
+
+@pytest.mark.p1
+def test_self_hosted_sends_base64_and_skips_markdown_images(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    parser = module.PaddleOCRParser(base_url="https://svc", access_token="tok")
+    parser._send_request(b"%PDF-1.7 body", module.PaddleOCRConfig(base_url="https://svc", access_token="tok"), None)
+
+    payload = post.call_args.kwargs["json"]
+    assert payload["file"] == "JVBERi0xLjcgYm9keQ=="
+    # Section text has images stripped anyway, so inlining them would only
+    # inflate the response.
+    assert payload["returnMarkdownImages"] is False
+    assert post.call_args.kwargs["headers"]["Authorization"] == "Bearer tok"
+
+
+@pytest.mark.p1
+def test_self_hosted_omits_authorization_over_plain_http(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    parser = module.PaddleOCRParser(base_url="http://svc", access_token="tok")
+    parser._send_request(b"%PDF-1.7", module.PaddleOCRConfig(base_url="http://svc", access_token="tok"), None)
+
+    # A bearer token on a plaintext connection is readable by anyone on the path.
+    assert "Authorization" not in post.call_args.kwargs["headers"]
+
+
+@pytest.mark.p1
+def test_self_hosted_omits_authorization_without_token(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    post = Mock(return_value=_local_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    parser._send_request(b"%PDF-1.7", module.PaddleOCRConfig(base_url="http://svc"), None)
+
+    assert "Authorization" not in post.call_args.kwargs["headers"]
+
+
+@pytest.mark.p1
+@pytest.mark.parametrize(
+    ("response_attrs", "expected"),
+    [
+        ({"status_code": 404, "text": "Not Found"}, "HTTP 404"),
+        ({"status_code": 200, "json.return_value": {"errorCode": 1001, "errorMsg": "bad request"}}, "bad request"),
+        ({"status_code": 200, "json.side_effect": ValueError("no json")}, "not JSON"),
+    ],
+)
+def test_self_hosted_failures_reach_the_callback(monkeypatch, response_attrs, expected):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    resp = Mock()
+    resp.configure_mock(**response_attrs)
+    monkeypatch.setattr(module.requests, "post", Mock(return_value=resp))
+
+    reported: list[tuple[float, str]] = []
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    with pytest.raises(RuntimeError, match=expected):
+        parser._send_request(b"%PDF", module.PaddleOCRConfig(base_url="http://svc"), lambda p, m: reported.append((p, m)))
+
+    # by_paddleocr() only logs the raised error, so the callback is the only way
+    # the cause becomes visible instead of the document silently ending empty.
+    assert [msg for prog, msg in reported if prog == -1], reported
+    assert expected in [msg for prog, msg in reported if prog == -1][0]
+
+
+@pytest.mark.p1
+def test_self_hosted_connection_error_reaches_the_callback(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    monkeypatch.setattr(module.requests, "post", Mock(side_effect=OSError("connection refused")))
+
+    reported: list[tuple[float, str]] = []
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    with pytest.raises(RuntimeError, match="connection refused"):
+        parser._send_request(b"%PDF", module.PaddleOCRConfig(base_url="http://svc"), lambda p, m: reported.append((p, m)))
+
+    assert any(prog == -1 and "connection refused" in msg for prog, msg in reported), reported
+
+
+@pytest.mark.p1
+def test_parse_image_uses_self_hosted_protocol(monkeypatch, tmp_path):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    pruned = {"parsing_res_list": [{"block_content": "invoice total 42", "block_label": "text", "block_bbox": [1, 2, 3, 4]}]}
+    monkeypatch.setattr(module.requests, "post", Mock(return_value=_local_response(pruned=pruned)))
+
+    image = tmp_path / "sample.png"
+    image.write_bytes(b"\x89PNG\r\n\x1a\n fake")
+
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    assert parser.parse_image(str(image)) == "invoice total 42"
+
+
+@pytest.mark.p1
+def test_parse_pdf_sections_carry_position_tags(monkeypatch, tmp_path):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    pruned = {"parsing_res_list": [{"block_content": "chapter one", "block_label": "text", "block_bbox": [10, 20, 30, 40]}]}
+    monkeypatch.setattr(module.requests, "post", Mock(return_value=_local_response(pruned=pruned)))
+
+    pdf = tmp_path / "sample.pdf"
+    pdf.write_bytes(b"%PDF-1.7 fake")
+
+    parser = module.PaddleOCRParser(base_url="http://svc")
+    sections, tables = parser.parse_pdf(str(pdf))
+
+    # bbox survives the self-hosted response, which is what crop() relies on.
+    assert sections == [("chapter one", "@@1\t5.0\t15.0\t10.0\t20.0##")]
+    assert tables == []
+
+
+@pytest.mark.p1
+def test_pure_image_blocks_do_not_become_empty_sections(monkeypatch, tmp_path):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    # A scanned page the layout model classified as one image block: its content
+    # is nothing but markup, so it must drop out instead of yielding an empty
+    # section that still carries a position tag.
+    pruned = {
+        "parsing_res_list": [
+            {
+                "block_content": '<div style="text-align: center;"><img src="imgs/img.jpg" alt="Image" width="99%" /></div>\n',
+                "block_label": "image",
+                "block_bbox": [3, 235, 1190, 1431],
+            },
+            {"block_content": "real text", "block_label": "text", "block_bbox": [10, 20, 30, 40]},
+        ]
+    }
+    monkeypatch.setattr(module.requests, "post", Mock(return_value=_local_response(pruned=pruned)))
+
+    pdf = tmp_path / "scan.pdf"
+    pdf.write_bytes(b"%PDF-1.7 fake")
+
+    sections, _ = module.PaddleOCRParser(base_url="http://svc").parse_pdf(str(pdf))
+    assert sections == [("real text", "@@1\t5.0\t15.0\t10.0\t20.0##")]
+
+
+@pytest.mark.p1
+def test_hosted_payload_includes_vl_params_for_every_supported_algorithm(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    parser = module.PaddleOCRParser(access_token="tok")
+    for algorithm in module.SUPPORTED_PADDLEOCR_ALGORITHMS:
+        config = module.PaddleOCRConfig.from_dict({"algorithm": algorithm, "algorithm_config": {"max_new_tokens": 128}})
+        payload = parser._build_payload(config)
+        assert payload["maxNewTokens"] == 128, algorithm
+        assert payload["prettifyMarkdown"] is True, algorithm
+
+
+@pytest.mark.p1
+def test_hosted_protocol_still_submits_and_polls(monkeypatch):
+    _clear_env(monkeypatch)
+    module = _load_paddleocr_parser(monkeypatch)
+
+    submit = Mock()
+    submit.status_code = 200
+    submit.json.return_value = {"data": {"jobId": "job-1"}}
+    monkeypatch.setattr(module.requests, "post", Mock(return_value=submit))
+
+    poll = Mock()
+    poll.status_code = 200
+    poll.json.return_value = {"data": {"state": "done", "resultJsonUrl": "http://results/1.jsonl"}}
+    fetch = Mock()
+    fetch.status_code = 200
+    fetch.text = json.dumps({"result": {"layoutParsingResults": [{"markdown": {"text": "hosted"}}]}})
+    fetch.raise_for_status = Mock()
+    monkeypatch.setattr(module.requests, "get", Mock(side_effect=[poll, fetch]))
+
+    parser = module.PaddleOCRParser(access_token="tok")
+    result = parser._send_request(b"%PDF", module.PaddleOCRConfig(access_token="tok"), None)
+
+    assert result["layoutParsingResults"][0]["markdown"]["text"] == "hosted"