Update README performance benchmark

README.md

@@ -89,7 +89,7 @@ A: Yes. Knowhere extracts them, runs them through VLMs for summarization and fea
 
 ## Performance Benchmark
 
-Agents using Knowhere outperform those working from raw documents or MinerU-parsed output on real-world tasks: searching, modifying, and answering questions.
+Agents using Knowhere outperform those working from raw documents, MinerU-parsed output, or Unstructured output on real-world tasks: searching, modifying, and answering questions.
 
 <p align="center">
   <img alt="Benchmark Performance: Agent + Knowhere vs Others" src="docs/assets/benchmark.png" width="900">
@@ -99,11 +99,11 @@ Agents using Knowhere outperform those working from raw documents or MinerU-pars
 
 ### Key Advantages
 
-- **Accuracy**: +36% first-try accuracy and +10% recall over raw documents.
+- **Accuracy**: +36% first-try accuracy and +11% recall over raw documents.
 - **Reliability**: 79% accuracy with feedback, vs. a ~53% ceiling on raw docs.
 - **Efficiency**: Fewer loops, fewer tokens, less time. Agents navigate a structured graph instead of reading monolithic text.
 
-*(Internal evaluation across identical agentic RAG tasks. Baseline: MinerU output fed directly to agents.)*
+*(Internal evaluation across identical agentic RAG tasks. Baselines: raw documents, MinerU output, and Unstructured output fed directly to agents.)*
 
 > [!NOTE]
 > **📊 Benchmarks are actively expanding.** More parsers and retrieval baselines coming soon.

apps/worker/app/services/document_ingestion/processing_billing.py

@@ -129,8 +129,9 @@ def record_processing_start(
     billing_snapshot: ParseJobBillingSnapshot,
     processing_started_at: datetime,
     workload_estimate: WorkloadEstimate,
+    extra_metadata: dict[str, object] | None = None,
 ) -> None:
-    metadata_updates = {
+    metadata_updates: dict[str, object] = {
         "page_count": workload_estimate.page_count,
         "billing_status": billing_snapshot.billing_status,
         "billing_amount_micro_dollars": billing_snapshot.billing_amount_micro_dollars,
@@ -142,6 +143,8 @@ def record_processing_start(
         metadata_updates["workload_estimate_fallback_reason"] = (
             workload_estimate.fallback_reason
         )
+    if extra_metadata is not None:
+        metadata_updates.update(extra_metadata)
     with get_sync_db_context() as db:
         job_result = db.execute(select(Job).where(Job.job_id == job_id).with_for_update())
         job = job_result.scalar_one_or_none()

apps/worker/app/services/document_ingestion/processing_run.py

@@ -25,7 +25,7 @@
 )
 from loguru import logger
 
-from shared.core.exceptions.domain_exceptions import ValidationException
+from shared.core.exceptions.domain_exceptions import ValidationException, Violation
 from shared.services.jobs.lifecycle.service import get_sync_job_lifecycle_service
 from shared.services.redis.distributed_lock import RedisJobLock
 from shared.services.redis.redis_sync_service import (
@@ -97,6 +97,7 @@ def _run_parse_job(
         file_extension=prepared_source.file_extension,
         page_count=page_count,
     ):
+        violations = _build_pdf_page_limit_violations(page_count)
         billing_snapshot = record_skipped_parse_job_billing(
             job_id=job_id,
             workload_estimate=workload_estimate,
@@ -107,8 +108,9 @@ def _run_parse_job(
             billing_snapshot=billing_snapshot,
             processing_started_at=processing_started_at,
             workload_estimate=workload_estimate,
+            extra_metadata={"error_details": {"violations": violations}},
         )
-        _raise_pdf_page_limit_exceeded(page_count)
+        _raise_pdf_page_limit_exceeded(page_count, violations)
 
     billing_snapshot = charge_parse_job_pages(
         job_id=job_id,
@@ -168,7 +170,21 @@ def _is_pdf_page_limit_exceeded(*, file_extension: str, page_count: int) -> bool
     return file_extension == ".pdf" and page_count > settings.MAX_PDF_PAGE_LIMIT
 
 
-def _raise_pdf_page_limit_exceeded(page_count: int) -> None:
+def _build_pdf_page_limit_violations(page_count: int) -> list[Violation]:
+    from shared.core.config import settings
+
+    pdf_page_limit = settings.MAX_PDF_PAGE_LIMIT
+    return [
+        {
+            "field": "page_count",
+            "description": f"PDF has {page_count} pages, limit is {pdf_page_limit}",
+        }
+    ]
+
+
+def _raise_pdf_page_limit_exceeded(
+    page_count: int, violations: list[Violation]
+) -> None:
     from shared.core.config import settings
 
     pdf_page_limit = settings.MAX_PDF_PAGE_LIMIT
@@ -177,10 +193,5 @@ def _raise_pdf_page_limit_exceeded(page_count: int) -> None:
             f"Document too large: {page_count} pages exceeds the {pdf_page_limit}-page limit. "
             "Please split the document and upload in smaller batches."
         ),
-        violations=[
-            {
-                "field": "page_count",
-                "description": f"PDF has {page_count} pages, limit is {pdf_page_limit}",
-            }
-        ],
+        violations=violations,
     )

apps/worker/tests/contract/conftest.py

@@ -9,13 +9,20 @@
 
 import pytest
 from pytest_postgresql import factories
+from celery import Celery
 from pytest import MonkeyPatch
 from shared.testing import contract_runtime
 from shared.testing.contract_runtime import PostgreSQLProcess
 from shared.testing.postgresql_environment import find_executable
 
 _REPO_ROOT: Path = Path(__file__).resolve().parents[4]
 _WORKER_ROOT: Path = _REPO_ROOT / "apps" / "worker"
+_DOCUMENT_INGESTION_TASK_NAMES: tuple[str, ...] = (
+    "app.core.tasks.document_ingestion_tasks.upload_url_file_task",
+    "app.core.tasks.kb_tasks.upload_url_file_task",
+    "app.core.tasks.document_ingestion_tasks.parse_task",
+    "app.core.tasks.kb_tasks.parse_task",
+)
 
 
 def _resolve_postgresql_executable() -> str | None:
@@ -28,6 +35,11 @@ def _resolve_postgresql_executable() -> str | None:
     return str(executable_path) if executable_path is not None else None
 
 
+def _clear_document_ingestion_task_registrations(celery_app: Celery) -> None:
+    for task_name in _DOCUMENT_INGESTION_TASK_NAMES:
+        celery_app.tasks.pop(task_name, None)
+
+
 _contract_postgresql_proc = factories.postgresql_proc(
     executable=_resolve_postgresql_executable(),
     port=contract_runtime.CONTRACT_POSTGRESQL_PORT_RANGE,
@@ -64,6 +76,7 @@ def worker_contract_environment(
     celery_app = get_celery_app()
     monkeypatch.setattr(celery_app.conf, "task_always_eager", True)
     monkeypatch.setattr(celery_app.conf, "task_eager_propagates", False)
+    _clear_document_ingestion_task_registrations(celery_app)
     importlib.import_module("app.core.tasks.document_ingestion_tasks")
 
     try:

packages/shared-python/shared/services/redis/redis_sync_service.py

@@ -261,7 +261,6 @@ def close(self):
             self._client = None
             logger.info("Sync Redis client closed")
 
-
 class SyncRedisServiceFactory:
     """Factory for sync Redis service instances."""