diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0600516..e2d50ed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,15 @@
 
 本项目的发布说明遵循语义化版本，并记录用户可见的兼容性边界。
 
+## 未发布
+
+### 修复
+
+- 限制单页 PDF 视觉候选数量，避免异常重叠对象触发高复杂度区域合并。
+- 异步取消不再依赖仍在运行的事件循环清理 SDK 自有临时源文件。
+- PPTX 在完整提取媒体和幻灯片内容前执行 `max_pages` 校验。
+- 空 TXT、Markdown 和无名文本字节返回更明确的 `NoUsableContentError` 消息。
+
 ## 0.1.0 - Alpha
 
 OpenDocs 的首个公开 Alpha 将本地文档转换为 Markdown。
diff --git a/src/opendocs/parsers/office/package.py b/src/opendocs/parsers/office/package.py
index 384085c..4b0b523 100644
--- a/src/opendocs/parsers/office/package.py
+++ b/src/opendocs/parsers/office/package.py
@@ -224,6 +224,33 @@ def validate_office_package(path: Path, *, document_type: DocumentType) -> Offic
     )
 
 
+def _local_name(tag: str) -> str:
+    return tag.rsplit("}", 1)[-1]
+
+
+def enforce_pptx_page_limit(path: Path, *, max_pages: int) -> None:
+    if isinstance(max_pages, bool) or not isinstance(max_pages, int):
+        raise TypeError("max_pages must be an int")
+    if max_pages <= 0:
+        raise ValueError("max_pages must be greater than zero")
+
+    layout = validate_office_package(path, document_type=DocumentType.PPTX)
+    try:
+        with ZipFile(path) as archive:
+            root = ET.fromstring(archive.read(layout.main_part_name))
+    except (BadZipFile, KeyError, OSError, ET.ParseError) as error:
+        raise CorruptDocumentError("PPTX presentation part is corrupt") from error
+
+    page_count = sum(
+        _local_name(slide.tag) == "sldId"
+        for child in root
+        if _local_name(child.tag) == "sldIdLst"
+        for slide in child
+    )
+    if page_count > max_pages:
+        raise LimitExceededError(f"PPTX exceeds the configured {max_pages} page limit")
+
+
 def open_validated_office_document(
     path: Path,
     *,
diff --git a/src/opendocs/parsers/office/parser.py b/src/opendocs/parsers/office/parser.py
index cbcdc8d..08dacaa 100644
--- a/src/opendocs/parsers/office/parser.py
+++ b/src/opendocs/parsers/office/parser.py
@@ -30,6 +30,7 @@
     document_from_wire,
     document_to_wire,
 )
+from opendocs.parsers.office.package import enforce_pptx_page_limit
 from opendocs.source import ParseWorkspace, ResolvedSource
 from opendocs.vision.base import VisionClient, VisionRequest, VisionRequestKind, VisionResult
 from opendocs.vision.images import (
@@ -58,6 +59,7 @@ def _extract_office_to_wire(
     document_type_value: str,
     path: Path,
     workspace_path: Path,
+    max_pages: int,
 ) -> dict[str, object]:
     document_type = DocumentType(document_type_value)
     workspace = ParseWorkspace(workspace_path)
@@ -68,6 +70,7 @@ def _extract_office_to_wire(
     elif document_type is DocumentType.PPTX:
         from opendocs.parsers.office.pptx import extract_pptx
 
+        enforce_pptx_page_limit(path, max_pages=max_pages)
         document = extract_pptx(path, workspace)
     else:
         raise ValueError("native Office extraction requires DOCX or PPTX")
@@ -143,7 +146,7 @@ async def parse(self, source: ResolvedSource, *, options: ParseOptions) -> Parse
             deadline = min(deadline, self._deadline)
         try:
             async with asyncio.timeout_at(deadline):
-                document = await self._extract(source)
+                document = await self._extract(source, max_pages=options.max_pages)
                 if (
                     self._document_type is DocumentType.PPTX
                     and len(document.pages) > options.max_pages
@@ -178,12 +181,13 @@ async def parse(self, source: ResolvedSource, *, options: ParseOptions) -> Parse
             f"{self._document_type.value.upper()} produced no usable content"
         )
 
-    async def _extract(self, source: ResolvedSource) -> OfficeDocument:
+    async def _extract(self, source: ResolvedSource, *, max_pages: int) -> OfficeDocument:
         wire = await self._runtime.run_native(
             _extract_office_to_wire,
             self._document_type.value,
             source.path,
             self._runtime.workspace.path,
+            max_pages,
         )
         try:
             document = document_from_wire(wire)
diff --git a/src/opendocs/parsers/pdf/analyze.py b/src/opendocs/parsers/pdf/analyze.py
index 6506978..71e7da9 100644
--- a/src/opendocs/parsers/pdf/analyze.py
+++ b/src/opendocs/parsers/pdf/analyze.py
@@ -32,6 +32,7 @@
     page_to_wire,
 )
 from opendocs.parsers.pdf.routing import (
+    MAX_VISUAL_CANDIDATES_PER_PAGE,
     VECTOR_OBJECT_COUNT_MIN,
     build_visual_regions,
     significant_image,
@@ -165,6 +166,10 @@ def _reading_order_ambiguity(words: Sequence[PdfWord]) -> tuple[bool, list[BBox]
             left_area = (left.bbox.right - left.bbox.left) * (left.bbox.bottom - left.bbox.top)
             right_area = (right.bbox.right - right.bbox.left) * (right.bbox.bottom - right.bbox.top)
             if horizontal * vertical / min(left_area, right_area) >= 0.20:
+                if len(overlapping) >= MAX_VISUAL_CANDIDATES_PER_PAGE:
+                    raise LimitExceededError(
+                        "PDF visual region candidates exceed the resource budget"
+                    )
                 overlapping.append(
                     BBox(
                         min(left.bbox.left, right.bbox.left),
diff --git a/src/opendocs/parsers/pdf/routing.py b/src/opendocs/parsers/pdf/routing.py
index 34945b8..de90a1f 100644
--- a/src/opendocs/parsers/pdf/routing.py
+++ b/src/opendocs/parsers/pdf/routing.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from opendocs._models import BBox
+from opendocs.errors import LimitExceededError
 from opendocs.parsers.pdf.extract import bbox_area, intersection_area, union_area
 from opendocs.parsers.pdf.models import PageFacts, PageRoute, PageRouteDecision, VisualRegion
 
@@ -8,6 +9,7 @@
 FULL_PAGE_IMAGE_AREA_MIN = 0.85
 FULL_VISION_UNION_AREA_MIN = 0.60
 MAX_REGIONS_PER_PAGE = 4
+MAX_VISUAL_CANDIDATES_PER_PAGE = 1_024
 REGION_PADDING = 0.015
 REGION_MERGE_GAP = 0.02
 VECTOR_OBJECT_COUNT_MIN = 30
@@ -33,6 +35,8 @@ def _should_merge(left: BBox, right: BBox) -> bool:
 def build_visual_regions(
     candidates: list[tuple[BBox, str]],
 ) -> tuple[VisualRegion, ...]:
+    if len(candidates) > MAX_VISUAL_CANDIDATES_PER_PAGE:
+        raise LimitExceededError("PDF visual region candidates exceed the resource budget")
     merged = [(_expand(bbox), {reason}, index) for index, (bbox, reason) in enumerate(candidates)]
     while True:
         for left_index, (left, reasons, source_index) in enumerate(merged):
diff --git a/src/opendocs/parsers/text.py b/src/opendocs/parsers/text.py
index 8141efa..3a68ff6 100644
--- a/src/opendocs/parsers/text.py
+++ b/src/opendocs/parsers/text.py
@@ -4,7 +4,7 @@
 import re
 
 from opendocs._models import DocumentType, MarkdownBlock, ParsedDocument, TextBlock
-from opendocs.errors import CorruptDocumentError, LimitExceededError
+from opendocs.errors import CorruptDocumentError, LimitExceededError, NoUsableContentError
 from opendocs.options import ParseOptions
 from opendocs.source import ResolvedSource
 
@@ -47,6 +47,8 @@ async def parse(
     ) -> ParsedDocument:
         del options
         value = await asyncio.to_thread(_read_utf8, source)
+        if not value.strip():
+            raise NoUsableContentError(f"{self._document_type.value} document is empty")
 
         if self._document_type is DocumentType.MARKDOWN:
             blocks = (MarkdownBlock(markdown=value),)
diff --git a/src/opendocs/source.py b/src/opendocs/source.py
index 7366a70..d2e926c 100644
--- a/src/opendocs/source.py
+++ b/src/opendocs/source.py
@@ -5,6 +5,7 @@
 import shutil
 import sys
 import tempfile
+import threading
 import warnings
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager, suppress
@@ -145,10 +146,21 @@ def _write_temporary(path: Path, data: bytes) -> None:
         handle.write(data)
 
 
-def _cleanup_finished_write(task: asyncio.Task[None], path: Path) -> None:
+def _consume_finished_write(task: asyncio.Task[None]) -> None:
     if not task.cancelled():
         task.exception()
-    _schedule_background_cleanup(path)
+
+
+def _write_temporary_with_cancellation_cleanup(
+    path: Path,
+    data: bytes,
+    cleanup_requested: threading.Event,
+) -> None:
+    try:
+        _write_temporary(path, data)
+    finally:
+        if cleanup_requested.is_set():
+            _cleanup_owned_path_now(path)
 
 
 async def _unlink_if_exists(path: Path) -> None:
@@ -169,21 +181,16 @@ def _warn_cleanup_failure(path: Path, error: BaseException) -> None:
     )
 
 
-def _consume_task_exception(task: asyncio.Task[object], path: Path) -> None:
-    if not task.cancelled():
-        exception = task.exception()
-        if exception is not None:
-            _warn_cleanup_failure(path, exception)
-
-
-def _schedule_background_cleanup(path: Path) -> None:
-    cleanup_task = asyncio.create_task(_cleanup_owned_path(path))
-    cleanup_task.add_done_callback(lambda task: _consume_task_exception(task, path))
+def _cleanup_owned_path_now(path: Path) -> None:
+    try:
+        path.unlink(missing_ok=True)
+    except OSError as error:
+        _warn_cleanup_failure(path, error)
 
 
 async def _cleanup_after_cancellation(path: Path, *, wait: bool) -> None:
     if not wait:
-        _schedule_background_cleanup(path)
+        _cleanup_owned_path_now(path)
         return
 
     try:
@@ -196,7 +203,15 @@ async def _cleanup_after_cancellation(path: Path, *, wait: bool) -> None:
 
 async def _write_owned(data: bytes, *, wait_for_cleanup_on_cancel: bool) -> Path:
     path = _create_temporary_path(data)
-    write_task = asyncio.create_task(asyncio.to_thread(_write_temporary, path, data))
+    cleanup_requested = threading.Event()
+    write_task = asyncio.create_task(
+        asyncio.to_thread(
+            _write_temporary_with_cancellation_cleanup,
+            path,
+            data,
+            cleanup_requested,
+        )
+    )
     try:
         await asyncio.shield(write_task)
     except asyncio.CancelledError:
@@ -205,7 +220,9 @@ async def _write_owned(data: bytes, *, wait_for_cleanup_on_cancel: bool) -> Path
                 await write_task
             await _cleanup_after_cancellation(path, wait=True)
         else:
-            write_task.add_done_callback(lambda completed: _cleanup_finished_write(completed, path))
+            cleanup_requested.set()
+            await _cleanup_after_cancellation(path, wait=False)
+            write_task.add_done_callback(_consume_finished_write)
         raise
     except OSError:
         await _unlink_if_exists(path)
diff --git a/tests/test_api.py b/tests/test_api.py
index 2cdd466..3fa51c3 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import inspect
 import io
 import threading
 import time
@@ -17,6 +16,7 @@
 from opendocs import (
     CorruptDocumentError,
     DocumentTimeoutError,
+    NoUsableContentError,
     OpenDocsWarning,
     ParseOptions,
     SyncInAsyncContextError,
@@ -87,6 +87,11 @@ def test_parse_and_aparse_match_for_unnamed_text_bytes_and_streams() -> None:
     assert async_stream.closed is False
 
 
+def test_empty_bytes_report_an_explicit_empty_document_error() -> None:
+    with pytest.raises(NoUsableContentError, match="text document is empty"):
+        parse(b"")
+
+
 @pytest.mark.asyncio
 async def test_parse_rejects_a_running_event_loop() -> None:
     with pytest.raises(SyncInAsyncContextError, match="await aparse"):
@@ -138,16 +143,12 @@ def test_sync_output_truncation_warning_points_to_the_parse_caller() -> None:
 
 
 @pytest.mark.asyncio
-async def test_timeout_during_detection_returns_promptly_when_source_cleanup_blocks(
+async def test_timeout_during_detection_cleans_without_async_cleanup(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     entered = asyncio.Event()
     cleanup_started = asyncio.Event()
-    cleanup_finished = asyncio.Event()
-    release_cleanup = asyncio.Event()
     temporary_path: Path | None = None
-    cleanup_impl = source_module._cleanup_owned_path
-    real_unlink = Path.unlink
 
     async def slow_detect(source: ResolvedSource) -> DocumentType:
         nonlocal temporary_path
@@ -156,36 +157,24 @@ async def slow_detect(source: ResolvedSource) -> DocumentType:
         await asyncio.Event().wait()
         raise AssertionError("timeout should cancel detection")
 
-    async def blocked_cleanup(path: Path) -> None:
+    async def forbidden_async_cleanup(path: Path) -> None:
+        del path
         cleanup_started.set()
-        await release_cleanup.wait()
-        await cleanup_impl(path)
-        cleanup_finished.set()
-
-    def guarded_unlink(self: Path, *, missing_ok: bool = False) -> None:
-        for frame in inspect.stack():
-            if frame.function == "_cleanup_cancelled_owned_source":
-                raise AssertionError("api.py must not unlink owned temp files directly")
-        real_unlink(self, missing_ok=missing_ok)
+        raise AssertionError("cancelled async cleanup must not be scheduled")
 
     monkeypatch.setattr("opendocs.api._detect", slow_detect)
-    monkeypatch.setattr("opendocs.source._cleanup_owned_path", blocked_cleanup)
-    monkeypatch.setattr(Path, "unlink", guarded_unlink)
+    monkeypatch.setattr("opendocs.source._cleanup_owned_path", forbidden_async_cleanup)
 
     parse_task = asyncio.create_task(aparse(b"hello", options=ParseOptions(timeout=0.05)))
     await entered.wait()
     started_at = asyncio.get_running_loop().time()
 
-    try:
-        with pytest.raises(DocumentTimeoutError):
-            await asyncio.wait_for(parse_task, timeout=0.3)
-    finally:
-        release_cleanup.set()
+    with pytest.raises(DocumentTimeoutError):
+        await asyncio.wait_for(parse_task, timeout=0.3)
 
     assert asyncio.get_running_loop().time() - started_at < 0.3
     assert temporary_path is not None
-    await asyncio.wait_for(cleanup_started.wait(), timeout=1)
-    await asyncio.wait_for(cleanup_finished.wait(), timeout=1)
+    assert cleanup_started.is_set() is False
     assert not temporary_path.exists()
 
 
@@ -364,6 +353,7 @@ async def test_timeout_cleanup_failure_warns_without_replacing_document_timeout_
 ) -> None:
     entered = asyncio.Event()
     temporary_path: Path | None = None
+    real_unlink = Path.unlink
 
     async def blocked_detect(source: ResolvedSource) -> DocumentType:
         nonlocal temporary_path
@@ -372,11 +362,13 @@ async def blocked_detect(source: ResolvedSource) -> DocumentType:
         await asyncio.Event().wait()
         raise AssertionError("blocked detect should not resume")
 
-    async def fail_cleanup(path: Path) -> None:
-        raise PermissionError(f"cleanup blocked for {path.name}")
+    def fail_owned_unlink(self: Path, *, missing_ok: bool = False) -> None:
+        if temporary_path is not None and self == temporary_path:
+            raise PermissionError(f"cleanup blocked for {self.name}")
+        real_unlink(self, missing_ok=missing_ok)
 
     monkeypatch.setattr("opendocs.api._detect", blocked_detect)
-    monkeypatch.setattr(source_module, "_cleanup_owned_path", fail_cleanup)
+    monkeypatch.setattr(Path, "unlink", fail_owned_unlink)
 
     with warnings.catch_warnings(record=True) as captured:
         warnings.simplefilter("always", OpenDocsWarning)
@@ -389,7 +381,7 @@ async def fail_cleanup(path: Path) -> None:
     assert warning.code == "source_cleanup_failed"
     assert str(temporary_path) in str(warning)
     assert "cleanup blocked" in str(warning)
-    temporary_path.unlink(missing_ok=True)
+    real_unlink(temporary_path, missing_ok=True)
 
 
 @pytest.mark.asyncio
@@ -428,6 +420,7 @@ async def test_aparse_propagates_external_cancellation_when_source_cleanup_fails
 ) -> None:
     entered = asyncio.Event()
     temporary_path: Path | None = None
+    real_unlink = Path.unlink
 
     async def blocked_detect(source: ResolvedSource) -> DocumentType:
         nonlocal temporary_path
@@ -436,11 +429,13 @@ async def blocked_detect(source: ResolvedSource) -> DocumentType:
         await asyncio.Event().wait()
         raise AssertionError("blocked detect should not resume")
 
-    async def fail_cleanup(path: Path) -> None:
-        raise PermissionError(f"cleanup blocked for {path.name}")
+    def fail_owned_unlink(self: Path, *, missing_ok: bool = False) -> None:
+        if temporary_path is not None and self == temporary_path:
+            raise PermissionError(f"cleanup blocked for {self.name}")
+        real_unlink(self, missing_ok=missing_ok)
 
     monkeypatch.setattr("opendocs.api._detect", blocked_detect)
-    monkeypatch.setattr(source_module, "_cleanup_owned_path", fail_cleanup)
+    monkeypatch.setattr(Path, "unlink", fail_owned_unlink)
 
     with warnings.catch_warnings(record=True) as captured:
         warnings.simplefilter("always", OpenDocsWarning)
@@ -457,7 +452,7 @@ async def fail_cleanup(path: Path) -> None:
     assert warning.code == "source_cleanup_failed"
     assert str(temporary_path) in str(warning)
     assert "cleanup blocked" in str(warning)
-    temporary_path.unlink(missing_ok=True)
+    real_unlink(temporary_path, missing_ok=True)
 
 
 def test_public_all_exposes_only_the_documented_surface() -> None:
diff --git a/tests/test_office_package.py b/tests/test_office_package.py
index 4415dcc..1d4d892 100644
--- a/tests/test_office_package.py
+++ b/tests/test_office_package.py
@@ -17,6 +17,7 @@
     MAX_TOTAL_ARCHIVE_BYTES,
     MAX_TOTAL_MEDIA_BYTES,
     MAX_XML_PART_BYTES,
+    enforce_pptx_page_limit,
     extract_package_media,
     open_validated_office_document,
     validate_office_package,
@@ -86,6 +87,26 @@ def test_validate_office_package_accepts_minimal_docx_and_pptx(tmp_path: Path) -
     assert pptx_layout.main_part_name == "ppt/presentation.xml"
 
 
+def test_pptx_page_limit_is_checked_from_main_part_before_extraction(tmp_path: Path) -> None:
+    path = tmp_path / "sample.pptx"
+    _write_zip(
+        path,
+        [
+            ("[Content_Types].xml", b"<Types/>"),
+            ("_rels/.rels", _doc_root_rels("ppt/presentation.xml")),
+            (
+                "ppt/presentation.xml",
+                b"""<p:presentation xmlns:p='urn:test'>
+<p:sldIdLst><p:sldId id='1'/><p:sldId id='2'/></p:sldIdLst>
+</p:presentation>""",
+            ),
+        ],
+    )
+
+    with pytest.raises(LimitExceededError, match="1 page limit"):
+        enforce_pptx_page_limit(path, max_pages=1)
+
+
 @pytest.mark.parametrize(
     ("factory", "error_type", "message"),
     [
diff --git a/tests/test_office_parser.py b/tests/test_office_parser.py
index 8b9d3f2..7906656 100644
--- a/tests/test_office_parser.py
+++ b/tests/test_office_parser.py
@@ -5,6 +5,7 @@
 
 import pytest  # pyright: ignore[reportMissingImports]
 
+import opendocs.parsers.office.parser as parser_module
 from opendocs._models import BBox, DocumentType, PageBreakBlock, TextBlock
 from opendocs._runtime import ParserRuntime
 from opendocs.errors import (
@@ -22,7 +23,7 @@
     OfficePage,
     document_to_wire,
 )
-from opendocs.parsers.office.parser import OfficeParser
+from opendocs.parsers.office.parser import OfficeParser, _extract_office_to_wire
 from opendocs.source import ParseWorkspace, ResolvedSource
 from opendocs.vision.base import VisionRequest, VisionRequestKind, VisionResult, VisionTextElement
 
@@ -119,6 +120,31 @@ async def _parse(
     )
 
 
+def test_native_pptx_page_limit_runs_before_document_extraction(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    events: list[str] = []
+
+    def reject_page_limit(path: Path, *, max_pages: int) -> None:
+        del path, max_pages
+        events.append("limit")
+        raise LimitExceededError("page limit")
+
+    def forbidden_extract(path: Path, workspace: ParseWorkspace) -> OfficeDocument:
+        del path, workspace
+        events.append("extract")
+        raise AssertionError("PPTX extraction must not start after the page limit fails")
+
+    monkeypatch.setattr(parser_module, "enforce_pptx_page_limit", reject_page_limit)
+    monkeypatch.setattr("opendocs.parsers.office.pptx.extract_pptx", forbidden_extract)
+
+    with pytest.raises(LimitExceededError, match="page limit"):
+        _extract_office_to_wire("pptx", tmp_path / "source.pptx", tmp_path, 1)
+
+    assert events == ["limit"]
+
+
 @pytest.mark.asyncio
 async def test_office_parser_native_only_preserves_slots_without_model_calls(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/tests/test_pdf_analyze.py b/tests/test_pdf_analyze.py
index d9d4d90..9f4cc3c 100644
--- a/tests/test_pdf_analyze.py
+++ b/tests/test_pdf_analyze.py
@@ -5,16 +5,23 @@
 
 import pytest  # pyright: ignore[reportMissingImports]
 
+import opendocs.parsers.pdf.analyze as analyze_module
 from opendocs._models import BBox, CoordinateTransform
 from opendocs._runtime import ParserRuntime
 from opendocs.errors import CorruptDocumentError, LimitExceededError
 from opendocs.parsers.pdf.analyze import (
     _analyze_page,
     _normalized_box,
+    _reading_order_ambiguity,
     analyze_pdf,
     analyze_pdf_native,
 )
-from opendocs.parsers.pdf.models import NativeTableCandidate, NativeTextCandidate, PageRoute
+from opendocs.parsers.pdf.models import (
+    NativeTableCandidate,
+    NativeTextCandidate,
+    PageRoute,
+    PdfWord,
+)
 from opendocs.parsers.pdf.routing import route_page
 from opendocs.source import parse_workspace
 
@@ -374,6 +381,17 @@ def unexpected_reading_order(_words: object) -> object:
     assert page.closed is True
 
 
+def test_reading_order_regions_stop_at_visual_candidate_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(analyze_module, "MAX_VISUAL_CANDIDATES_PER_PAGE", 2)
+    bbox = BBox(0.1, 0.1, 0.2, 0.2)
+    words = tuple(PdfWord("x", bbox, index) for index in range(3))
+
+    with pytest.raises(LimitExceededError, match="visual region candidates"):
+        _reading_order_ambiguity(words)
+
+
 def test_each_page_cache_is_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     path = tmp_path / "input.pdf"
     path.write_bytes(b"%PDF-")
diff --git a/tests/test_pdf_routing.py b/tests/test_pdf_routing.py
index 6130dad..8981474 100644
--- a/tests/test_pdf_routing.py
+++ b/tests/test_pdf_routing.py
@@ -3,12 +3,14 @@
 import pytest
 
 from opendocs._models import BBox
+from opendocs.errors import LimitExceededError
 from opendocs.parsers.pdf.extract import measure_text_quality
 from opendocs.parsers.pdf.models import PageFacts, PageRoute, VisualRegion
 from opendocs.parsers.pdf.routing import (
     FULL_PAGE_IMAGE_AREA_MIN,
     FULL_VISION_UNION_AREA_MIN,
     MAX_REGIONS_PER_PAGE,
+    MAX_VISUAL_CANDIDATES_PER_PAGE,
     build_visual_regions,
     route_page,
 )
@@ -96,3 +98,13 @@ def test_visual_regions_merge_in_stable_source_order_with_reason_union() -> None
     assert len(regions) == 2
     assert regions[0].reasons == ("image", "table_structure_uncertain")
     assert regions[1].reasons == ("dense_drawing",)
+
+
+def test_visual_region_builder_rejects_candidates_beyond_resource_budget() -> None:
+    candidates = [
+        (BBox(0.1, 0.1, 0.2, 0.2), "reading_order_ambiguous")
+        for _ in range(MAX_VISUAL_CANDIDATES_PER_PAGE + 1)
+    ]
+
+    with pytest.raises(LimitExceededError, match="visual region candidates"):
+        build_visual_regions(candidates)
diff --git a/tests/test_source.py b/tests/test_source.py
index a731288..adc4680 100644
--- a/tests/test_source.py
+++ b/tests/test_source.py
@@ -3,6 +3,7 @@
 import asyncio
 import io
 import threading
+import time
 import warnings
 from pathlib import Path
 from typing import Any, cast
@@ -344,33 +345,73 @@ async def hold_source() -> None:
     with pytest.raises(asyncio.CancelledError):
         await task
 
+    assert temporary_path is not None
+    assert not temporary_path.exists()
+
+
+def test_write_cancellation_cleans_after_event_loop_closes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+    temporary_path: Path | None = None
+
+    def delayed_write(path: Path, data: bytes) -> None:
+        nonlocal temporary_path
+        temporary_path = path
+        started.set()
+        assert release.wait(timeout=5), "test did not release the background write"
+        path.write_bytes(data)
+        finished.set()
+
+    monkeypatch.setattr(source_module, "_write_temporary", delayed_write)
+
+    async def cancel_write() -> None:
+        task = asyncio.create_task(materialize_source(b"hello").__aenter__())
+        assert await asyncio.to_thread(started.wait, 1), "background write did not start"
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(cancel_write())
+    finally:
+        pending = asyncio.all_tasks(loop)
+        for task in pending:
+            task.cancel()
+        if pending:
+            loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+        loop.close()
+    release.set()
+
+    assert finished.wait(timeout=1), "background write did not finish"
     assert temporary_path is not None
     for _ in range(100):
         if not temporary_path.exists():
             break
-        await asyncio.sleep(0.01)
-    assert not temporary_path.exists()
+        time.sleep(0.01)
+    try:
+        assert not temporary_path.exists()
+    finally:
+        temporary_path.unlink(missing_ok=True)
 
 
 @pytest.mark.asyncio
-async def test_post_yield_cancellation_returns_promptly_and_cleans_eventually(
+async def test_post_yield_cancellation_cleans_without_async_cleanup(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     entered = asyncio.Event()
     cleanup_started = asyncio.Event()
-    cleanup_finished = asyncio.Event()
-    release_cleanup = asyncio.Event()
     temporary_path: Path | None = None
 
-    cleanup_impl = source_module._cleanup_owned_path
-
-    async def blocked_cleanup(path: Path) -> None:
+    async def forbidden_async_cleanup(path: Path) -> None:
+        del path
         cleanup_started.set()
-        await release_cleanup.wait()
-        await cleanup_impl(path)
-        cleanup_finished.set()
+        raise AssertionError("cancelled async cleanup must not be scheduled")
 
-    monkeypatch.setattr(source_module, "_cleanup_owned_path", blocked_cleanup)
+    monkeypatch.setattr(source_module, "_cleanup_owned_path", forbidden_async_cleanup)
 
     async def hold_source() -> None:
         nonlocal temporary_path
@@ -383,30 +424,28 @@ async def hold_source() -> None:
     await entered.wait()
     task.cancel()
 
-    try:
-        with pytest.raises(asyncio.CancelledError):
-            await asyncio.wait_for(task, timeout=0.2)
-    finally:
-        release_cleanup.set()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(task, timeout=0.2)
 
     assert temporary_path is not None
-    await asyncio.wait_for(cleanup_finished.wait(), timeout=1)
+    assert cleanup_started.is_set() is False
     assert not temporary_path.exists()
 
 
 @pytest.mark.asyncio
-async def test_post_yield_cancellation_warns_when_background_cleanup_fails(
+async def test_post_yield_cancellation_warns_when_immediate_cleanup_fails(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     entered = asyncio.Event()
-    release_cleanup = asyncio.Event()
     temporary_path: Path | None = None
+    real_unlink = Path.unlink
 
-    async def fail_cleanup(path: Path) -> None:
-        await release_cleanup.wait()
-        raise PermissionError(f"cleanup blocked for {path.name}")
+    def fail_owned_unlink(self: Path, *, missing_ok: bool = False) -> None:
+        if temporary_path is not None and self == temporary_path:
+            raise PermissionError(f"cleanup blocked for {self.name}")
+        real_unlink(self, missing_ok=missing_ok)
 
-    monkeypatch.setattr(source_module, "_cleanup_owned_path", fail_cleanup)
+    monkeypatch.setattr(Path, "unlink", fail_owned_unlink)
 
     with warnings.catch_warnings(record=True) as captured:
         warnings.simplefilter("always", OpenDocsWarning)
@@ -422,11 +461,8 @@ async def hold_source() -> None:
         await entered.wait()
         task.cancel()
 
-        try:
-            with pytest.raises(asyncio.CancelledError):
-                await asyncio.wait_for(task, timeout=0.2)
-        finally:
-            release_cleanup.set()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=0.2)
 
         warning = await _wait_for_open_docs_warning(captured)
 
@@ -434,7 +470,7 @@ async def hold_source() -> None:
     assert warning.code == "source_cleanup_failed"
     assert str(temporary_path) in str(warning)
     assert "cleanup blocked" in str(warning)
-    temporary_path.unlink(missing_ok=True)
+    real_unlink(temporary_path, missing_ok=True)
 
 
 @pytest.mark.asyncio
@@ -477,13 +513,15 @@ async def materialize() -> None:
 
 
 @pytest.mark.asyncio
-async def test_cancellation_during_write_warns_when_background_cleanup_fails(
+async def test_cancellation_during_write_warns_when_immediate_cleanup_fails(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     started = threading.Event()
     release_write = threading.Event()
-    release_cleanup = asyncio.Event()
+    final_cleanup_attempted = threading.Event()
     temporary_path: Path | None = None
+    real_unlink = Path.unlink
+    cleanup_attempts = 0
 
     def delayed_write(path: Path, data: bytes) -> None:
         nonlocal temporary_path
@@ -492,12 +530,17 @@ def delayed_write(path: Path, data: bytes) -> None:
         assert release_write.wait(timeout=5), "test did not release the background write"
         path.write_bytes(data)
 
-    async def fail_cleanup(path: Path) -> None:
-        await release_cleanup.wait()
-        raise PermissionError(f"cleanup blocked for {path.name}")
+    def fail_owned_unlink(self: Path, *, missing_ok: bool = False) -> None:
+        nonlocal cleanup_attempts
+        if temporary_path is not None and self == temporary_path:
+            cleanup_attempts += 1
+            if cleanup_attempts >= 2:
+                final_cleanup_attempted.set()
+            raise PermissionError(f"cleanup blocked for {self.name}")
+        real_unlink(self, missing_ok=missing_ok)
 
     monkeypatch.setattr(source_module, "_write_temporary", delayed_write)
-    monkeypatch.setattr(source_module, "_cleanup_owned_path", fail_cleanup)
+    monkeypatch.setattr(Path, "unlink", fail_owned_unlink)
 
     with warnings.catch_warnings(record=True) as captured:
         warnings.simplefilter("always", OpenDocsWarning)
@@ -515,12 +558,12 @@ async def materialize() -> None:
                 await asyncio.wait_for(task, timeout=0.2)
         finally:
             release_write.set()
-            release_cleanup.set()
 
         warning = await _wait_for_open_docs_warning(captured)
+        assert await asyncio.to_thread(final_cleanup_attempted.wait, 1)
 
     assert temporary_path is not None
     assert warning.code == "source_cleanup_failed"
     assert str(temporary_path) in str(warning)
     assert "cleanup blocked" in str(warning)
-    temporary_path.unlink(missing_ok=True)
+    real_unlink(temporary_path, missing_ok=True)
diff --git a/tests/test_text_parser.py b/tests/test_text_parser.py
index 68db40c..dd81951 100644
--- a/tests/test_text_parser.py
+++ b/tests/test_text_parser.py
@@ -8,6 +8,7 @@
 from opendocs import (
     CorruptDocumentError,
     LimitExceededError,
+    NoUsableContentError,
     ParseOptions,
     UnsupportedDocumentError,
 )
@@ -81,6 +82,31 @@ async def test_text_parser_consumes_whitespace_only_separator_lines(
     assert render_markdown(document, max_output_chars=400_000).markdown == "alpha\n\n beta\n"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("document_type", "name", "message"),
+    [
+        (DocumentType.TEXT, "empty.txt", "text document is empty"),
+        (DocumentType.MARKDOWN, "empty.md", "markdown document is empty"),
+    ],
+)
+async def test_text_parser_reports_empty_documents_explicitly(
+    tmp_path: Path,
+    document_type: DocumentType,
+    name: str,
+    message: str,
+) -> None:
+    path = tmp_path / name
+    path.write_bytes(b" \n\t")
+    parser = build_default_registry().get(document_type)
+
+    with pytest.raises(NoUsableContentError, match=message):
+        await parser.parse(
+            ResolvedSource(path=path, original_name=path.name, owned=False),
+            options=ParseOptions(),
+        )
+
+
 @pytest.mark.asyncio
 async def test_text_parser_rejects_invalid_utf8(tmp_path: Path) -> None:
     path = tmp_path / "bad.txt"