@samitouri / QOSami_Agents_Zero / commits / b2ead06a

fix(document_query): isolate LiteParse parsing

Run LiteParse in a subprocess so native parser crashes cannot take down the Web UI process. Bound parser concurrency and LiteParse workers for multi-chat stability, seed Q&A context with leading document chunks for title/abstract grounding, and keep a small-document fallback when vector search returns no chunks.

Alessandro committed May 29, 2026 at 15:51 UTC b2ead06a4eef18d02a83846792d4aec156122f32

7 files changed +306 -21

plugins/_document_query/README.md

+5 -1

@@ -8,7 +8,7 @@ timeouts and thread-safe parsers.

8 - **Strategy-pattern parsers** - MIME-type routing to dedicated parser classes

9 - **Centralized fetching** - local and HTTP(S) resources are fetched once, size-checked, then passed to parsers

10 - **LiteParse first path** - fast local parsing for PDFs and supported document/image formats, with legacy fallbacks

11 -- **Thread-safe execution** - all sync parsers offloaded to asyncio.to_thread

11 +- **Bounded parser execution** - sync parsers are offloaded to asyncio.to_thread and globally capped across chats

12 - **Configurable timeouts** - per-document and gather-level timeouts

13 - **Expanded format support** - PDF, HTML, text, YAML, XML, TOML, JS, TS, images, and catch-all Unstructured

@@ -23,10 +23,14 @@ See default_config.yaml for all options. Key settings:

23 | max_remote_bytes | 52428800 | Max remote document size |

24 | per_document_timeout | 60 | Max time for a single document parse |

25 | gather_timeout | 120 | Max time for all documents combined |

26 +| parser_concurrency | 1 | Max parser jobs running across all chats in one process |

27 +| context_intro_chunks | 2 | Leading chunks included per document for title/abstract grounding |

28 | chunk_size | 1000 | Text splitter chunk size |

29 | chunk_overlap | 100 | Text splitter overlap |

30 | search_threshold | 0.5 | Similarity search threshold |

31 | liteparse_enabled | true | Prefer LiteParse before legacy parser fallbacks |

32 +| liteparse_num_workers | 1 | Max LiteParse OCR workers per parser job |

33 +| liteparse_subprocess | true | Run LiteParse in a child process so native crashes fall back safely |

34 | thread_offload | true | Offload sync parsers to thread pool |

36 LiteParse is installed into the Agent Zero framework runtime from hooks.py during

plugins/_document_query/default_config.yaml

+4 -1

@@ -9,6 +9,8 @@ per_document_timeout: 60 # max time for a single document parse

9 gather_timeout: 120 # max time for all documents combined in one call

11 # --- Parser settings ---

12 +parser_concurrency: 1 # max parser jobs running across all chats in this process

13 +context_intro_chunks: 2 # always include leading chunks per document for title/abstract grounding

14 chunk_size: 1000

15 chunk_overlap: 100

16 search_threshold: 0.5

@@ -26,6 +28,7 @@ liteparse_target_pages:

28 liteparse_dpi: 150

29 liteparse_preserve_very_small_text: false

30 liteparse_output_format: text

29 -liteparse_num_workers:

31 +liteparse_num_workers: 1 # LiteParse defaults to CPU cores - 1; cap it for web runtime stability

32 +liteparse_subprocess: true # isolate LiteParse native runtime crashes from the Web UI process

33 pdf_ocr_fallback: true # enable legacy Tesseract fallback after PyMuPDF

34 thread_offload: true # offload sync parsers to thread pool

plugins/_document_query/helpers/document_query.py

+116 -16

@@ -7,8 +7,9 @@ a thread pool and bounded by configurable timeouts.

8 import asyncio

9 import json

10 +import threading

11 from datetime import datetime

11 -from typing import Callable, List, Optional, Sequence, Tuple

12 +from typing import Any, Callable, List, Optional, Sequence, Tuple

13 from urllib.parse import urlparse

15 from langchain.schema import SystemMessage, HumanMessage

@@ -25,6 +26,33 @@ from plugins._document_query.helpers.parsers import BaseParser, get_parsers_for_

28 DEFAULT_SEARCH_THRESHOLD = 0.5

29 +DEFAULT_PARSER_CONCURRENCY = 1

30 +SMALL_DOCUMENT_FALLBACK_MAX_CHARS = 12000

31 +_PARSER_SEMAPHORES: dict[tuple[int, int], asyncio.Semaphore] = {}

32 +_PARSER_SEMAPHORES_LOCK = threading.Lock()

33 +

34 +

35 +def _positive_int(value: Any, default: int) -> int:

36 + try:

37 + parsed = int(value)

38 + except (TypeError, ValueError):

39 + return default

40 + return parsed if parsed > 0 else default

41 +

42 +

43 +def _parser_semaphore(config: dict) -> asyncio.Semaphore:

44 + concurrency = _positive_int(

45 + config.get("parser_concurrency"),

46 + DEFAULT_PARSER_CONCURRENCY,

47 + )

48 + loop = asyncio.get_running_loop()

49 + key = (id(loop), concurrency)

50 + with _PARSER_SEMAPHORES_LOCK:

51 + semaphore = _PARSER_SEMAPHORES.get(key)

52 + if semaphore is None:

53 + semaphore = asyncio.Semaphore(concurrency)

54 + _PARSER_SEMAPHORES[key] = semaphore

55 + return semaphore

58 def _load_config(agent: Agent) -> dict:

@@ -207,7 +235,7 @@ class DocumentQueryHelper:

235

236 gather_timeout = self.config.get("gather_timeout", 120)

237 try:

210 - await asyncio.wait_for(

238 + document_contents = await asyncio.wait_for(

239 asyncio.gather(

240 *[self.document_get_content(uri, True) for uri in document_uris]

241 ),

@@ -220,6 +248,14 @@ class DocumentQueryHelper:

248 search_threshold = self.config.get("search_threshold", DEFAULT_SEARCH_THRESHOLD)

249 search_limit = self.config.get("search_limit", 100)

250 selected_chunks = {}

251 + normalized_uris = [self.store.normalize_uri(uri) for uri in document_uris]

252 + intro_chunk_count = _positive_int(

253 + self.config.get("context_intro_chunks"),

254 + 2,

255 + )

256 + for uri in normalized_uris:

257 + for chunk in await self._get_document_intro_chunks(uri, intro_chunk_count):

258 + selected_chunks[chunk.metadata["id"]] = chunk

259

260 for question in questions:

261 self.progress_callback(f"Optimizing query: {question}")

@@ -233,7 +269,6 @@ class DocumentQueryHelper:

269

270 await self.agent.handle_intervention()

271 self.progress_callback(f"Searching documents with query: {optimized_query}")

236 - normalized_uris = [self.store.normalize_uri(uri) for uri in document_uris]

272 doc_filter = " or ".join(

273 [f"document_uri == '{uri}'" for uri in normalized_uris]

274 )

@@ -246,18 +281,48 @@ class DocumentQueryHelper:

281 selected_chunks[chunk.metadata["id"]] = chunk

282

283 if not selected_chunks:

284 + fallback_content = self._small_document_fallback_content(

285 + document_uris,

286 + document_contents,

287 + )

288 + if fallback_content:

289 + self.progress_callback(

290 + "No matching chunks found; using extracted document content"

291 + )

292 + ai_response = await self._answer_questions_from_content(

293 + fallback_content,

294 + questions,

295 + "extracted document content",

296 + )

297 + self.progress_callback(f"Q&A process completed")

298 + return True, ai_response

299 +

300 self.progress_callback("No relevant content found in the documents")

301 content = f"!!! No content found for documents: {json.dumps(document_uris)} matching queries: {json.dumps(questions)}"

302 return False, content

303

304 + content = "\n\n----\n\n".join(

305 + [chunk.page_content for chunk in selected_chunks.values()]

306 + )

307 + ai_response = await self._answer_questions_from_content(

308 + content,

309 + questions,

310 + f"{len(selected_chunks)} chunks",

311 + )

312 + self.progress_callback(f"Q&A process completed")

313 + return True, ai_response

314 +

315 + async def _answer_questions_from_content(

316 + self,

317 + content: str,

318 + questions: Sequence[str],

319 + context_label: str,

320 + ) -> str:

321 self.progress_callback(

254 - f"Processing {len(questions)} questions in context of {len(selected_chunks)} chunks"

322 + f"Processing {len(questions)} questions in context of {context_label}"

323 )

324 await self.agent.handle_intervention()

325 questions_str = "\n".join([f" * {question}" for question in questions])

258 - content = "\n\n----\n\n".join(

259 - [chunk.page_content for chunk in selected_chunks.values()]

260 - )

326 qa_system_message = self.agent.parse_prompt("fw.document_query.system_prompt.md")

327 qa_user_message = f"# Document:\n{content}\n\n# Queries:\n{questions_str}"

328 ai_response, _reasoning = await self.agent.call_chat_model(

@@ -267,8 +332,41 @@ class DocumentQueryHelper:

332 ],

333 explicit_caching=False,

334 )

270 - self.progress_callback(f"Q&A process completed")

271 - return True, str(ai_response)

335 + return str(ai_response)

336 +

337 + @staticmethod

338 + def _small_document_fallback_content(

339 + document_uris: Sequence[str],

340 + document_contents: Sequence[str],

341 + max_chars: int = SMALL_DOCUMENT_FALLBACK_MAX_CHARS,

342 + ) -> str:

343 + blocks = []

344 + for document_uri, document_content in zip(document_uris, document_contents):

345 + text = (document_content or "").strip()

346 + if text:

347 + blocks.append(f"# Source: {document_uri}\n\n{text}")

348 +

349 + if not blocks:

350 + return ""

351 +

352 + content = "\n\n----\n\n".join(blocks)

353 + if len(content) > max_chars:

354 + return ""

355 + return content

356 +

357 + async def _get_document_intro_chunks(

358 + self,

359 + document_uri: str,

360 + limit: int,

361 + ) -> list[Document]:

362 + if limit <= 0:

363 + return []

364 + if not hasattr(self.store, "_get_document_chunks"):

365 + return []

366 + chunks = await self.store._get_document_chunks(document_uri)

367 + return sorted(chunks, key=lambda chunk: chunk.metadata.get("chunk_index", 0))[

368 + :limit

369 + ]

370

371 async def document_get_content(

372 self, document_uri: str, add_to_db: bool = False

@@ -328,15 +426,17 @@ class DocumentQueryHelper:

426 thread_offload: bool,

427 ) -> str:

428 errors_seen = []

429 + semaphore = _parser_semaphore(self.config)

430 for parser in parsers:

431 try:

333 - self.progress_callback("Parsing document content")

334 - content = await parser.parse(

335 - document=document,

336 - config=self.config,

337 - timeout=timeout,

338 - thread_offload=thread_offload,

339 - )

432 + async with semaphore:

433 + self.progress_callback("Parsing document content")

434 + content = await parser.parse(

435 + document=document,

436 + config=self.config,

437 + timeout=timeout,

438 + thread_offload=thread_offload,

439 + )

440 if content:

441 return content

442 errors_seen.append(f"{parser.__class__.__name__}: no content")

plugins/_document_query/helpers/parsers/liteparse.py

+72 -3

@@ -2,7 +2,10 @@

3 from __future__ import annotations

5 +import json

6 import os

7 +import subprocess

8 +import sys

9 from pathlib import Path

11 from plugins._document_query.helpers.fetch import FetchedDocument

@@ -12,6 +15,8 @@ from .base import BaseParser

15 class LiteParseParser(BaseParser):

16 """Fast parser powered by run-llama/liteparse when available."""

18 + DEFAULT_NUM_WORKERS = 1

19 +

20 mimetypes = [

21 "application/pdf",

22 "application/msword",

@@ -30,6 +35,11 @@ class LiteParseParser(BaseParser):

35 return bool(config.get("liteparse_enabled", True))

37 def _parse_sync(self, document: FetchedDocument, config: dict) -> str:

38 + if config.get("liteparse_subprocess", True):

39 + return self._parse_subprocess(document, config)

40 + return self._parse_in_process(document, config)

41 +

42 + def _parse_in_process(self, document: FetchedDocument, config: dict) -> str:

43 try:

44 from liteparse import LiteParse

45 except Exception as e:

@@ -44,6 +54,56 @@ class LiteParseParser(BaseParser):

54 raise ValueError("LiteParse returned no text")

55 return text

57 + def _parse_subprocess(self, document: FetchedDocument, config: dict) -> str:

58 + with document.local_file() as file_path:

59 + payload = {

60 + "file_path": file_path,

61 + "kwargs": self._liteparse_kwargs(config),

62 + }

63 + env = os.environ.copy()

64 + project_root = str(Path(__file__).resolve().parents[4])

65 + python_path = env.get("PYTHONPATH", "")

66 + env["PYTHONPATH"] = (

67 + f"{project_root}{os.pathsep}{python_path}"

68 + if python_path

69 + else project_root

70 + )

71 + timeout = float(config.get("per_document_timeout", 60))

72 + result = subprocess.run(

73 + [

74 + sys.executable,

75 + "-m",

76 + "plugins._document_query.helpers.parsers.liteparse_worker",

77 + ],

78 + input=json.dumps(payload),

79 + text=True,

80 + capture_output=True,

81 + cwd=project_root,

82 + env=env,

83 + timeout=timeout,

84 + check=False,

85 + )

86 +

87 + if result.returncode != 0:

88 + detail = (result.stderr or result.stdout or "").strip()

89 + raise RuntimeError(

90 + "LiteParse subprocess failed"

91 + f" with exit code {result.returncode}: {detail[-2000:]}"

92 + )

93 +

94 + try:

95 + response = json.loads(result.stdout)

96 + except json.JSONDecodeError as e:

97 + raise RuntimeError(

98 + "LiteParse subprocess returned invalid output: "

99 + f"{result.stdout[-2000:]}"

100 + ) from e

101 +

102 + text = response.get("text", "") or ""

103 + if not text.strip():

104 + raise ValueError("LiteParse returned no text")

105 + return text

106 +

107 def _liteparse_kwargs(self, config: dict) -> dict:

108 kwargs = {

109 "ocr_enabled": bool(config.get("liteparse_ocr_enabled", True)),

@@ -67,9 +127,10 @@ class LiteParseParser(BaseParser):

127 if value not in (None, ""):

128 kwargs[liteparse_key] = value

129

70 - num_workers = config.get("liteparse_num_workers")

71 - if num_workers not in (None, ""):

72 - kwargs["num_workers"] = int(num_workers)

130 + kwargs["num_workers"] = _positive_int(

131 + config.get("liteparse_num_workers"),

132 + self.DEFAULT_NUM_WORKERS,

133 + )

134

135 tessdata_path = config.get("liteparse_tessdata_path") or _detect_tessdata_path()

136 if tessdata_path:

@@ -91,3 +152,11 @@ def _detect_tessdata_path() -> str:

152 if candidate and (Path(candidate) / "eng.traineddata").is_file():

153 return candidate

154 return ""

155 +

156 +

157 +def _positive_int(value, default: int) -> int:

158 + try:

159 + parsed = int(value)

160 + except (TypeError, ValueError):

161 + return default

162 + return parsed if parsed > 0 else default

plugins/_document_query/helpers/parsers/liteparse_worker.py new

+24

@@ -0,0 +1,24 @@

1 +"""Subprocess entry point for isolating LiteParse native runtime crashes."""

2 +

3 +from __future__ import annotations

4 +

5 +import json

6 +import sys

7 +

8 +

9 +def main() -> int:

10 + payload = json.load(sys.stdin)

11 + file_path = payload["file_path"]

12 + kwargs = payload.get("kwargs") or {}

13 +

14 + from liteparse import LiteParse

15 +

16 + parser = LiteParse(**kwargs)

17 + result = parser.parse(file_path)

18 + text = getattr(result, "text", "") or ""

19 + json.dump({"text": text}, sys.stdout)

20 + return 0

21 +

22 +

23 +if __name__ == "__main__":

24 + raise SystemExit(main())

tests/test_document_query_fallback.py

@@ -44,6 +44,7 @@ def test_document_qa_uses_small_document_content_when_search_finds_no_chunks():

44 helper = object.__new__(DocumentQueryHelper)

45 helper.agent = agent

46 helper.store = FakeStore()

47 + helper.config = {}

48 helper.progress_callback = progress.append

50 async def document_get_content(uri, add_to_db=False):

tests/test_document_query_plugin.py

+84

@@ -9,6 +9,7 @@ from plugins._document_query.helpers.fetch import FetchedDocument, fetch_public_

9 from plugins._document_query.helpers.document_query import DocumentQueryHelper

10 from plugins._document_query.helpers.parsers.base import BaseParser

11 from plugins._document_query.helpers.parsers import get_parsers_for_mimetype

12 +from plugins._document_query.helpers.parsers.liteparse import LiteParseParser

13 from plugins._document_query.helpers.parsers.text import TextParser

@@ -27,6 +28,24 @@ class ParserNameShouldNotLeak(BaseParser):

28 return "parsed"

31 +class CountingAsyncParser(BaseParser):

32 + mimetypes = ["text/plain"]

33 + active = 0

34 + max_active = 0

35 +

36 + async def _parse_async(self, document: FetchedDocument, config: dict) -> str:

37 + type(self).active += 1

38 + type(self).max_active = max(type(self).max_active, type(self).active)

39 + try:

40 + await asyncio.sleep(0.02)

41 + return document.uri

42 + finally:

43 + type(self).active -= 1

44 +

45 + def _parse_sync(self, document: FetchedDocument, config: dict) -> str:

46 + return document.uri

47 +

48 +

49 def test_fetch_file_detects_mimetype_and_reads_once(tmp_path):

50 document = tmp_path / "notes.txt"

51 document.write_text("hello\nworld\n", encoding="utf-8")

@@ -92,6 +111,25 @@ def test_liteparse_is_installed_by_docker_and_plugin_hook_requirements():

111 assert plugin_requirements.strip().splitlines() == ["liteparse>=2.0.0,<3.0.0"]

112

113

114 +def test_default_config_bounds_liteparse_runtime_concurrency():

115 + default_config = (

116 + ROOT / "plugins" / "_document_query" / "default_config.yaml"

117 + ).read_text(encoding="utf-8")

118 +

119 + assert "parser_concurrency: 1" in default_config

120 + assert "context_intro_chunks: 2" in default_config

121 + assert "liteparse_num_workers: 1" in default_config

122 + assert "liteparse_subprocess: true" in default_config

123 +

124 +

125 +def test_liteparse_parser_caps_workers_by_default():

126 + parser = LiteParseParser()

127 +

128 + assert parser._liteparse_kwargs({})["num_workers"] == 1

129 + assert parser._liteparse_kwargs({"liteparse_num_workers": "3"})["num_workers"] == 3

130 + assert parser._liteparse_kwargs({"liteparse_num_workers": ""})["num_workers"] == 1

131 +

132 +

133 def test_query_optimize_prompt_filename_is_spelled_correctly():

134 prompt_dir = ROOT / "plugins" / "_document_query" / "prompts"

135 helper_source = (

@@ -129,6 +167,52 @@ def test_parser_progress_is_user_facing_and_generic():

167 assert progress == ["Parsing document content"]

168

169

170 +def test_parse_document_limits_parser_concurrency_across_helpers():

171 + CountingAsyncParser.active = 0

172 + CountingAsyncParser.max_active = 0

173 + fetched_a = FetchedDocument(

174 + uri="/tmp/a.txt",

175 + source_uri="/tmp/a.txt",

176 + scheme="file",

177 + mimetype="text/plain",

178 + content=b"a",

179 + local_path=None,

180 + )

181 + fetched_b = FetchedDocument(

182 + uri="/tmp/b.txt",

183 + source_uri="/tmp/b.txt",

184 + scheme="file",

185 + mimetype="text/plain",

186 + content=b"b",

187 + local_path=None,

188 + )

189 + helper_a = object.__new__(DocumentQueryHelper)

190 + helper_a.config = {"parser_concurrency": 1}

191 + helper_a.progress_callback = lambda _msg: None

192 + helper_b = object.__new__(DocumentQueryHelper)

193 + helper_b.config = {"parser_concurrency": 1}

194 + helper_b.progress_callback = lambda _msg: None

195 +

196 + async def parse_both():

197 + return await asyncio.gather(

198 + helper_a._parse_document(

199 + document=fetched_a,

200 + parsers=[CountingAsyncParser()],

201 + timeout=1,

202 + thread_offload=False,

203 + ),

204 + helper_b._parse_document(

205 + document=fetched_b,

206 + parsers=[CountingAsyncParser()],

207 + timeout=1,

208 + thread_offload=False,

209 + ),

210 + )

211 +

212 + assert sorted(run_async(parse_both())) == ["/tmp/a.txt", "/tmp/b.txt"]

213 + assert CountingAsyncParser.max_active == 1

214 +

215 +

216 def test_document_query_prompt_uses_progressive_skill_disclosure():

217 from helpers.skills import find_skill

218