diff --git a/apps/api/app/services/documents/lifecycle_service.py b/apps/api/app/services/documents/lifecycle_service.py
index 1bd5ae3a9..a0328d507 100644
--- a/apps/api/app/services/documents/lifecycle_service.py
+++ b/apps/api/app/services/documents/lifecycle_service.py
@@ -47,6 +47,15 @@ def _document_chunk_asset_url(
         return None
 
     try:
+        if not result_storage.verify_raw_exists(
+            job_id=job_id,
+            relative_path=file_path,
+        ):
+            logger.warning(
+                f"Skipping asset URL for missing chunk artifact: "
+                f"job_id={job_id}, file_path={file_path}"
+            )
+            return None
         return result_storage.generate_artifact_url(
             job_id=job_id,
             artifact_ref=file_path,
diff --git a/apps/api/tests/contract/test_demo_documents_contract.py b/apps/api/tests/contract/test_demo_documents_contract.py
index a1e704747..c00bee07c 100644
--- a/apps/api/tests/contract/test_demo_documents_contract.py
+++ b/apps/api/tests/contract/test_demo_documents_contract.py
@@ -62,6 +62,9 @@ def normalize_artifact_ref(self, artifact_ref: str | None) -> str | None:
             return normalized
         return None
 
+    def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
+        return relative_path in self.raw_files_by_job_id.get(job_id, set())
+
     def generate_artifact_url(
         self,
         *,
diff --git a/apps/api/tests/contract/test_documents_contract.py b/apps/api/tests/contract/test_documents_contract.py
index 6b8e33b04..ebe27e64f 100644
--- a/apps/api/tests/contract/test_documents_contract.py
+++ b/apps/api/tests/contract/test_documents_contract.py
@@ -533,6 +533,22 @@ def _upload_page_citation_asset(*, job_id: str, artifact_ref: str) -> None:
         asset_path.unlink(missing_ok=True)
 
 
+def _upload_chunk_asset(*, job_id: str, artifact_ref: str) -> None:
+    from shared.services.storage.result_storage import JobResultStorage
+
+    suffix = Path(artifact_ref).suffix or ".bin"
+    asset_path = Path("/tmp") / f"knowhere-contract-chunk-asset-{uuid4().hex}{suffix}"
+    asset_path.write_bytes(b"<table><tr><td>contract chunk asset</td></tr></table>")
+    try:
+        JobResultStorage().upload_raw_file(
+            job_id=job_id,
+            relative_path=artifact_ref,
+            local_file_path=str(asset_path),
+        )
+    finally:
+        asset_path.unlink(missing_ok=True)
+
+
 @pytest.mark.asyncio
 async def test_should_list_only_the_authenticated_users_documents_for_the_effective_namespace(
     developer_api_client_factory: Callable[
@@ -992,6 +1008,10 @@ async def test_should_include_media_asset_urls_in_document_chunk_list_when_reque
                 },
             ],
         )
+        _upload_chunk_asset(
+            job_id=revision["job_id"],
+            artifact_ref="tables/table-1.html",
+        )
         response = await api_client.get(
             f"/api/v1/documents/{document_id}/chunks",
             params={
@@ -1167,6 +1187,10 @@ async def test_should_return_one_document_chunk_by_document_chunk_id(
                 }
             ],
         )
+        _upload_chunk_asset(
+            job_id=revision["job_id"],
+            artifact_ref="images/figure-1.png",
+        )
         response = await api_client.get(
             f"/api/v1/documents/{document_id}/chunks/{chunk_id}",
             params={"include_asset_urls": "true"},
diff --git a/apps/api/tests/contract/test_retrieval_contract.py b/apps/api/tests/contract/test_retrieval_contract.py
index 58de969c7..4debea613 100644
--- a/apps/api/tests/contract/test_retrieval_contract.py
+++ b/apps/api/tests/contract/test_retrieval_contract.py
@@ -555,6 +555,9 @@ def normalize_artifact_ref(self, artifact_ref: str | None) -> str | None:
                 return None
             return normalized
 
+        def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
+            return True
+
     def fake_get_result_storage() -> FakeResultStorage:
         return FakeResultStorage()
 
diff --git a/apps/api/tests/unit/test_document_chunk_asset_url.py b/apps/api/tests/unit/test_document_chunk_asset_url.py
new file mode 100644
index 000000000..84d2739b6
--- /dev/null
+++ b/apps/api/tests/unit/test_document_chunk_asset_url.py
@@ -0,0 +1,177 @@
+"""Unit tests for document chunk asset URL generation."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from tests.support.import_environment import (
+    configure_import_environment,
+    ensure_import_paths,
+)
+
+# Defer importing apps/api `app` until test bodies run. Module-level imports
+# would cache API's `app` package and break apps/worker contract collection in
+# the same pytest process (both packages are named `app`).
+configure_import_environment()
+ensure_import_paths()
+
+_API_ROOT = str(Path(__file__).resolve().parents[2])
+
+
+def _is_api_app_module(module: ModuleType | None) -> bool:
+    if module is None:
+        return False
+    module_file = getattr(module, "__file__", None)
+    if isinstance(module_file, str) and module_file.startswith(_API_ROOT):
+        return True
+    module_paths = getattr(module, "__path__", ())
+    try:
+        return any(str(path).startswith(_API_ROOT) for path in module_paths)
+    except KeyError:
+        return False
+
+
+def _drop_api_app_modules() -> None:
+    for module_name in sorted(sys.modules, key=len, reverse=True):
+        if module_name != "app" and not module_name.startswith("app."):
+            continue
+        if _is_api_app_module(sys.modules.get(module_name)):
+            sys.modules.pop(module_name, None)
+
+
+def _prioritize_api_import_root() -> None:
+    """Keep apps/api ahead of apps/worker for the shared `app` package name."""
+    ensure_import_paths()
+    if _API_ROOT in sys.path:
+        sys.path.remove(_API_ROOT)
+    sys.path.insert(0, _API_ROOT)
+
+
+@pytest.fixture(autouse=True)
+def _clear_api_app_modules_after_unit_test():
+    """Avoid leaving API's `app` package cached for later worker contract tests."""
+    yield
+    _drop_api_app_modules()
+
+
+class FakeResultStorage:
+    def __init__(self, *, exists: bool = True) -> None:
+        self.exists = exists
+        self.checked: list[str] = []
+
+    def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
+        self.checked.append(f"{job_id}/{relative_path}")
+        return self.exists
+
+    def generate_artifact_url(
+        self,
+        *,
+        job_id: str,
+        artifact_ref: str,
+        expires_in: int = 3600,
+    ) -> str | None:
+        return f"https://assets.example.test/{job_id}/{artifact_ref}"
+
+
+def _document_chunk_asset_url(**kwargs: Any) -> str | None:
+    _prioritize_api_import_root()
+    _drop_api_app_modules()
+    from app.services.documents.lifecycle_service import (
+        _document_chunk_asset_url,
+    )
+
+    return _document_chunk_asset_url(**kwargs)
+
+
+def test_asset_url_generated_when_raw_file_exists() -> None:
+    storage = FakeResultStorage(exists=True)
+
+    url = _document_chunk_asset_url(
+        chunk_type="table",
+        job_id="job-1",
+        file_path="tables/table-1 Test.html",
+        include_asset_urls=True,
+        result_storage=storage,
+    )
+
+    assert url == "https://assets.example.test/job-1/tables/table-1 Test.html"
+    assert storage.checked == ["job-1/tables/table-1 Test.html"]
+
+
+def test_asset_url_skipped_when_raw_file_missing() -> None:
+    storage = FakeResultStorage(exists=False)
+
+    url = _document_chunk_asset_url(
+        chunk_type="table",
+        job_id="job-1",
+        file_path="tables/table-1 Test.html",
+        include_asset_urls=True,
+        result_storage=storage,
+    )
+
+    assert url is None
+    assert storage.checked == ["job-1/tables/table-1 Test.html"]
+
+
+def test_asset_url_falls_back_on_verification_error() -> None:
+    class ExplodingStorage:
+        def verify_raw_exists(self, *, job_id: str, relative_path: str) -> bool:
+            raise RuntimeError("storage unavailable")
+
+        def generate_artifact_url(
+            self,
+            *,
+            job_id: str,
+            artifact_ref: str,
+            expires_in: int = 3600,
+        ) -> str | None:
+            return f"https://assets.example.test/{job_id}/{artifact_ref}"
+
+    with patch(
+        "app.services.documents.lifecycle_service.logger.warning"
+    ) as mock_warning:
+        url = _document_chunk_asset_url(
+            chunk_type="image",
+            job_id="job-1",
+            file_path="images/page-1.png",
+            include_asset_urls=True,
+            result_storage=ExplodingStorage(),
+        )
+
+    assert url is None
+    assert mock_warning.call_count == 1
+
+
+def test_asset_url_skipped_for_non_media_chunk_types() -> None:
+    storage = FakeResultStorage(exists=True)
+
+    url = _document_chunk_asset_url(
+        chunk_type="text",
+        job_id="job-1",
+        file_path="tables/table-1 Test.html",
+        include_asset_urls=True,
+        result_storage=storage,
+    )
+
+    assert url is None
+    assert storage.checked == []
+
+
+def test_asset_url_skipped_when_include_asset_urls_false() -> None:
+    storage = FakeResultStorage(exists=True)
+
+    url = _document_chunk_asset_url(
+        chunk_type="table",
+        job_id="job-1",
+        file_path="tables/table-1 Test.html",
+        include_asset_urls=False,
+        result_storage=storage,
+    )
+
+    assert url is None
+    assert storage.checked == []