fix: prevent document ingestion finalization errors

apps/api/app/services/document_ingestion/handoff_service.py

@@ -29,6 +29,7 @@
 class _UploadedFileJob(Protocol):
     job_id: str
     job_type: str
+    status: str
 
 
 class DocumentIngestionHandoffService:
@@ -66,6 +67,17 @@ async def start_uploaded_file_workflow(
                 ],
             )
 
+        # Upload completion can arrive through both the S3 notification and the
+        # confirm-upload endpoint.  Once either path has moved the job out of
+        # waiting-file, the other path must be a no-op instead of dispatching a
+        # second worker task for the same logical job.
+        if job.status != JobStatus.WAITING_FILE.value:
+            logger.info(
+                "Upload handoff already completed: "
+                f"job_id={job.job_id}, status={job.status}"
+            )
+            return
+
         outcome = await self._state_machine.transition_outcome(
             db,
             job.job_id,
@@ -75,6 +87,18 @@ async def start_uploaded_file_workflow(
             "system",
         )
         if not outcome.succeeded:
+            # A concurrent handoff may have won the CAS transition after this
+            # caller loaded the waiting-file snapshot.  The state machine
+            # reports the winner's state as ``from_state``; treat that result
+            # as an idempotent no-op and do not enqueue another task.
+            if outcome.reason == "invalid_transition" and outcome.from_state != (
+                JobStatus.WAITING_FILE.value
+            ):
+                logger.info(
+                    "Upload handoff won by another trigger: "
+                    f"job_id={job.job_id}, status={outcome.from_state}"
+                )
+                return
             logger.warning(
                 "Upload handoff transition rejected: "
                 f"job_id={job.job_id}, reason={outcome.reason}"

apps/api/tests/contract/test_s3_event_contract.py

@@ -119,6 +119,93 @@ async def start_uploaded_file_parse(
     ]
 
 
+@pytest.mark.asyncio
+async def test_should_not_dispatch_a_second_task_for_a_replayed_upload_event(
+    api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]],
+    monkeypatch: MonkeyPatch,
+) -> None:
+    workflow_calls: list[dict[str, str]] = []
+
+    class FakeDocumentIngestionWorkerDispatcher:
+        async def start_uploaded_file_parse(
+            self,
+            *,
+            job_id: str,
+            user_id: str,
+        ) -> str:
+            workflow_calls.append({"job_id": job_id, "user_id": user_id})
+            return "contract-task-id"
+
+    async with api_client_factory() as api_client:
+        user_id, job_id = await _insert_waiting_file_job()
+        handoff_service = importlib.import_module(
+            "app.services.document_ingestion.handoff_service"
+        )
+        monkeypatch.setattr(
+            handoff_service,
+            "DocumentIngestionWorkerDispatcher",
+            FakeDocumentIngestionWorkerDispatcher,
+        )
+
+        first_response = await api_client.post(
+            "/api/v1/internal/s3-events",
+            json=_build_s3_event_payload(job_id),
+        )
+        replay_response = await api_client.post(
+            "/api/v1/internal/s3-events",
+            json=_build_s3_event_payload(job_id),
+        )
+
+    assert first_response.status_code == 200
+    assert replay_response.status_code == 200
+    assert workflow_calls == [{"job_id": job_id, "user_id": user_id}]
+
+
+@pytest.mark.asyncio
+async def test_should_treat_a_concurrent_upload_handoff_cas_winner_as_a_no_op() -> None:
+    from app.services.document_ingestion.handoff_service import (
+        DocumentIngestionHandoffService,
+    )
+    from shared.core.state_machine.transition_outcome import JobTransitionOutcome
+
+    class FakeStateMachine:
+        async def transition_outcome(self, *args: object, **kwargs: object) -> object:
+            del args, kwargs
+            return JobTransitionOutcome.rejected(
+                job_id="job-race",
+                to_state="pending",
+                reason="invalid_transition",
+                attempts=1,
+                from_state="pending",
+            )
+
+    class FakeDispatcher:
+        async def start_uploaded_file_parse(
+            self,
+            *,
+            job_id: str,
+            user_id: str,
+        ) -> str:
+            del job_id, user_id
+            raise AssertionError("CAS loser must not dispatch a duplicate task")
+
+    service = DocumentIngestionHandoffService(
+        state_machine=FakeStateMachine(),
+        worker_dispatcher=FakeDispatcher(),
+    )
+
+    await service.start_uploaded_file_workflow(
+        db=cast(object, None),
+        job=SimpleNamespace(
+            job_id="job-race",
+            job_type="document_ingestion",
+            status="waiting-file",
+        ),
+        user_id="contract-user",
+        trigger="s3_upload_completed",
+    )
+
+
 @pytest.mark.asyncio
 async def test_should_accept_a_pre_rename_waiting_file_job_type_during_upload_handoff(
     api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]],

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

@@ -32,7 +32,10 @@
     cleanup_stage_tracker,
     init_stage_tracker,
 )
-from shared.core.exceptions.domain_exceptions import ValidationException
+from shared.core.exceptions.domain_exceptions import (
+    UnavailableException,
+    ValidationException,
+)
 from shared.models.schemas.job_metadata import JobMetadataHelper
 from shared.services.ai.llm_overrides import cleanup_llm_overrides, init_llm_overrides
 from shared.services.ai.token_tracking import cleanup_token_tracker, init_token_tracker
@@ -63,21 +66,39 @@ def execute(self, job_id: str, user_id: str | None) -> dict[str, object]:
                 "reason": "job_already_terminal",
             }
 
-        with RedisJobLock(job_context.redis_service, job_id):
-            task_workspace = TemporaryParseWorkspace.create(job_id)
-            try:
-                result = _run_parse_job(
-                    job_id=job_id,
-                    job_context=job_context,
-                    lifecycle_service=lifecycle_service,
-                    task_workspace=task_workspace,
-                )
-            finally:
-                task_workspace.cleanup()
+        try:
+            with RedisJobLock(job_context.redis_service, job_id):
+                task_workspace = TemporaryParseWorkspace.create(job_id)
+                try:
+                    result = _run_parse_job(
+                        job_id=job_id,
+                        job_context=job_context,
+                        lifecycle_service=lifecycle_service,
+                        task_workspace=task_workspace,
+                    )
+                finally:
+                    task_workspace.cleanup()
+        except UnavailableException as exc:
+            if not _is_processing_lock_contention(exc):
+                raise
+            logger.info(
+                "Skipping duplicate parse delivery while another worker owns "
+                f"the processing lock: job_id={job_id}"
+            )
+            return {
+                "status": "skipped",
+                "job_id": job_id,
+                "reason": "job_already_processing",
+            }
 
         return result
 
 
+def _is_processing_lock_contention(error: UnavailableException) -> bool:
+    """Identify lock contention without swallowing unrelated 503 failures."""
+    return error.internal_message.startswith("Could not acquire processing lock")
+
+
 def _run_parse_job(
     *,
     job_id: str,

apps/worker/tests/contract/test_parse_task_contract.py

@@ -121,6 +121,105 @@ def test_parse_task_should_process_uploaded_file_through_real_contract_boundarie
     assert contract.find_task_workspaces(tmp_path, job["job_id"]) == []
 
 
+def test_parse_task_should_publish_each_non_null_source_path_once(
+    worker_contract_environment: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    contract = WorkerParseContract.create()
+    contract.use_workspace_root(monkeypatch, tmp_path)
+    contract.use_billing(monkeypatch, is_enabled=False)
+
+    source_file_name = "contract-duplicate-path.xlsx"
+    job = contract.create_file_job(
+        source_file_name=source_file_name,
+        job_id_prefix="job_duplicate_path",
+    )
+    contract.upload_source_file(
+        local_file_path=_SAMPLE_XLSX_PATH,
+        s3_key=job["s3_key"],
+    )
+
+    def fake_execute_document_parse(
+        *,
+        job_id: str,
+        job_context: object,
+        prepared_source: object,
+        output_dir: str,
+    ) -> ParseOutput:
+        del job_id, job_context, prepared_source
+        parsed_df = pd.DataFrame(
+            [
+                {
+                    "know_id": "first-shared-path",
+                    "type": "text",
+                    "content": "first shared-path chunk",
+                    "path": f"{source_file_name}/Root/Shared",
+                },
+                {
+                    "know_id": "duplicate-shared-path",
+                    "type": "text",
+                    "content": "duplicate shared-path chunk",
+                    "path": f"{source_file_name}/Root/Shared",
+                },
+                {
+                    "know_id": "nul-duplicate-shared-path",
+                    "type": "text",
+                    "content": "NUL duplicate shared-path chunk",
+                    "path": f"{source_file_name}/Root\x00/Shared",
+                },
+                {
+                    "know_id": "first-null-path",
+                    "type": "text",
+                    "content": "first null-path chunk",
+                    "path": "",
+                },
+                {
+                    "know_id": "second-null-path",
+                    "type": "text",
+                    "content": "second null-path chunk",
+                    "path": "",
+                },
+            ]
+        )
+        return ParseOutput(output_dir=output_dir, parsed_df=parsed_df)
+
+    monkeypatch.setattr(
+        "app.services.document_ingestion.processing_run.execute_document_parse",
+        fake_execute_document_parse,
+    )
+
+    celery_result = contract.enqueue_parse_task(
+        job_id=job["job_id"],
+        user_id=job["user_id"],
+    )
+
+    assert celery_result.successful()
+    assert celery_result.result["status"] == "success"
+
+    observed = contract.observe_successful_job(job["job_id"])
+    job_chunks = observed["job_chunks"]
+    document_chunks = observed["document_chunks"]
+
+    assert [row["chunk_id"] for row in job_chunks] == [
+        "first-shared-path",
+        "duplicate-shared-path",
+        "nul-duplicate-shared-path",
+        "first-null-path",
+        "second-null-path",
+    ]
+    assert [row["chunk_id"] for row in document_chunks] == [
+        "first-shared-path",
+        "first-null-path",
+        "second-null-path",
+    ]
+    assert [row["source_chunk_path"] for row in document_chunks] == [
+        f"{source_file_name}/Root/Shared",
+        None,
+        None,
+    ]
+
+
 def test_parse_task_result_zip_includes_page_citation_assets(
     worker_contract_environment: None,
     monkeypatch: pytest.MonkeyPatch,
@@ -223,6 +322,87 @@ def fake_execute_document_parse(
     ]
 
 
+def test_parse_task_sanitizes_nul_characters_at_database_boundary(
+    worker_contract_environment: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    contract = WorkerParseContract.create()
+    contract.use_workspace_root(monkeypatch, tmp_path)
+    contract.use_billing(monkeypatch, is_enabled=False)
+
+    source_file = tmp_path / "nul-source.txt"
+    source_file.write_text("source", encoding="utf-8")
+    job = contract.create_file_job(
+        source_file_name=source_file.name,
+        job_id_prefix="job_parse_nul",
+    )
+    contract.upload_source_file(
+        local_file_path=source_file,
+        s3_key=job["s3_key"],
+    )
+
+    def fake_execute_document_parse(
+        *,
+        job_id: str,
+        job_context: object,
+        prepared_source: object,
+        output_dir: str,
+    ) -> ParseOutput:
+        del job_id, job_context, prepared_source
+        parsed_df = pd.DataFrame(
+            [
+                {
+                    "know_id": "chunk-with-nul",
+                    "type": "text",
+                    "content": "Text\x00content",
+                    "path": f"{source_file.name}/Root\x00/Introduction",
+                    "length": 12,
+                    "keywords": "",
+                    "summary": "Summary\x00value",
+                    "tokens": "",
+                    "connectto": "",
+                    "page_nums": "1",
+                    "extra_metadata": json.dumps({"note": "Nested\x00value"}),
+                }
+            ]
+        )
+        return ParseOutput(output_dir=output_dir, parsed_df=parsed_df)
+
+    monkeypatch.setattr(
+        "app.services.document_ingestion.processing_run.execute_document_parse",
+        fake_execute_document_parse,
+    )
+
+    celery_result = contract.enqueue_parse_task(
+        job_id=job["job_id"],
+        user_id=job["user_id"],
+    )
+
+    assert celery_result.successful()
+    assert celery_result.result["status"] == "success"
+
+    observed = contract.observe_successful_job(job["job_id"])
+    job_chunk = observed["job_chunks"][0]
+    document_chunk = observed["document_chunks"][0]
+
+    assert job_chunk["text"] == "Textcontent"
+    assert job_chunk["path"] == f"{source_file.name}/Root/Introduction"
+    assert job_chunk["chunk_metadata"]["summary"] == "Summaryvalue"
+    assert job_chunk["chunk_metadata"]["note"] == "Nestedvalue"
+    assert document_chunk["content"] == "Textcontent"
+    assert document_chunk["source_chunk_path"] == (
+        f"{source_file.name}/Root/Introduction"
+    )
+
+    result_zip = contract.read_result_zip(
+        result_s3_key=observed["result"]["result_s3_key"],
+        tmp_path=tmp_path,
+    )
+    archived_chunk = result_zip["chunks"]["chunks"][0]
+    assert archived_chunk["content"] == "Text\x00content"
+
+
 def test_parse_task_should_charge_user_when_billing_is_enabled(
     worker_contract_environment: None,
     monkeypatch: pytest.MonkeyPatch,