fix: reject duplicate document source bytes

CONTEXT.md

@@ -69,6 +69,18 @@ state, and starts parsing work.
 The API-side handoff that advances an uploaded file Job to pending state and
 enqueues the worker parse task with the user-aware Celery queue policy.
 
+### Source Content Admission
+
+The check that rejects identical original file bytes within one User and
+Namespace before parsing or billing, including unchanged Document updates.
+Filenames do not determine duplicates. Only current active Document revisions
+and in-progress Jobs participate; failed Jobs and archived or superseded
+revisions do not block retries. The check happens after source upload or URL
+download, so a duplicate is reported through the normal Job failure contract.
+Demo Corpus publication retains its separate admission policy.
+Existing sources without recorded fingerprints do not participate, and their
+original files are not fetched for comparison.
+
 ### Worker Document Parsing
 
 The worker-side workflow that turns a source file into parsed DataFrame rows,

apps/api/alembic/versions/5e6f7a8b9c0d_add_source_content_hash_index.py

@@ -0,0 +1,71 @@
+"""Index future source content fingerprints without backfilling existing Jobs."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+from sqlalchemy import text
+
+revision: str = "5e6f7a8b9c0d"
+down_revision: str = "4d5e6f7a8b9c"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def _create_index(*, concurrently: bool) -> None:
+    existing_index: tuple[bool, str, str] | None = (
+        op.get_bind()
+        .execute(
+            text(
+                "SELECT indexes.indisvalid AND indexes.indisready, "
+                "tables.relname, pg_get_indexdef(indexes.indexrelid) "
+                "FROM pg_index AS indexes "
+                "JOIN pg_class AS classes ON classes.oid = indexes.indexrelid "
+                "JOIN pg_class AS tables ON tables.oid = indexes.indrelid "
+                "JOIN pg_namespace AS namespaces ON namespaces.oid = classes.relnamespace "
+                "WHERE namespaces.nspname = current_schema() "
+                "AND classes.relname = 'idx_jobs_user_source_content_hash'"
+            )
+        )
+        .tuples()
+        .first()
+    )
+    if existing_index is not None:
+        is_ready, table_name, definition = existing_index
+        normalized_definition: str = " ".join(definition.split())
+        intended_definition: str = (
+            "USING btree (user_id, ((job_metadata ->> 'source_content_sha256'::text))) "
+            "WHERE ((job_metadata ->> 'source_content_sha256'::text) IS NOT NULL)"
+        )
+        if (
+            is_ready
+            and table_name == "jobs"
+            and normalized_definition.startswith("CREATE INDEX ")
+            and normalized_definition.endswith(intended_definition)
+        ):
+            return
+    concurrent_clause: str = "CONCURRENTLY " if concurrently else ""
+    op.execute(
+        f"DROP INDEX {concurrent_clause}IF EXISTS idx_jobs_user_source_content_hash"
+    )
+    op.execute(
+        f"CREATE INDEX {concurrent_clause}IF NOT EXISTS idx_jobs_user_source_content_hash "
+        "ON jobs (user_id, (job_metadata ->> 'source_content_sha256')) "
+        "WHERE job_metadata ->> 'source_content_sha256' IS NOT NULL"
+    )
+
+
+def upgrade() -> None:
+    uses_external_transaction: bool = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    if uses_external_transaction:
+        _create_index(concurrently=False)
+        return
+    with op.get_context().autocommit_block():
+        _create_index(concurrently=True)
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS idx_jobs_user_source_content_hash")

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

@@ -10,10 +10,8 @@
 from app.repositories.job_repository import JobRepository
 from app.services.document_ingestion.command import DocumentIngestionCommand
 from app.services.document_ingestion.scope_service import (
-    find_active_document_by_source_file_name,
     is_active_document_job_unique_violation,
     raise_document_ingestion_conflict,
-    raise_duplicate_filename_conflict,
 )
 from app.services.jobs.result_projection import to_job_status_value
 from app.services.rate_limit.data_structures import CurrentUser
@@ -147,29 +145,6 @@ async def _cache_job_creation_state(
         }
         await job_info_service.save_job_info(job_id, job_info)
 
-    async def _reject_duplicate_source_file_name(
-        self,
-        db: AsyncSession,
-        *,
-        payload: JobCreateBase,
-        current_user: CurrentUser,
-        scope: ResolvedDocumentIngestionScope,
-        source_file_name: str,
-    ) -> None:
-        if payload.document_id or scope.job_metadata.get("corpus_target") == "DEMO":
-            return
-        existing = await find_active_document_by_source_file_name(
-            db,
-            user_id=current_user.user_id,
-            namespace=scope.namespace,
-            source_file_name=source_file_name,
-        )
-        if existing is not None:
-            raise_duplicate_filename_conflict(
-                existing_document_id=existing.document_id,
-                source_file_name=source_file_name,
-            )
-
     async def _create_file_job(
         self,
         db: AsyncSession,
@@ -180,13 +155,6 @@ async def _create_file_job(
         scope: ResolvedDocumentIngestionScope,
     ) -> JobResponse:
         assert payload.file_name is not None
-        await self._reject_duplicate_source_file_name(
-            db,
-            payload=payload,
-            current_user=current_user,
-            scope=scope,
-            source_file_name=payload.file_name,
-        )
         file_extension = os.path.splitext(payload.file_name)[1]
         s3_key = f"uploads/{job_id}{file_extension}"
         JobMetadataHelper.set_file_source(
@@ -264,13 +232,6 @@ async def _create_url_job(
             source_url=payload.source_url,
             file_extension=file_extension,
         )
-        await self._reject_duplicate_source_file_name(
-            db,
-            payload=payload,
-            current_user=current_user,
-            scope=scope,
-            source_file_name=source_file_name,
-        )
         s3_key = f"uploads/{job_id}{file_extension}"
         JobMetadataHelper.set_url_source(
             scope.job_metadata,

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

@@ -16,7 +16,6 @@
     NotFoundException,
     ValidationException,
 )
-from shared.models.database.document import Document
 from shared.models.database.job import Job
 from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace
 
@@ -68,45 +67,6 @@ def raise_document_ingestion_conflict(
     )
 
 
-async def find_active_document_by_source_file_name(
-    db: AsyncSession,
-    *,
-    user_id: str,
-    namespace: str,
-    source_file_name: str,
-) -> Document | None:
-    result = await db.execute(
-        select(Document)
-        .where(Document.user_id == user_id)
-        .where(Document.namespace == namespace)
-        .where(Document.status == "active")
-        .where(Document.source_file_name == source_file_name)
-        .order_by(Document.created_at.asc(), Document.document_id.asc())
-        .limit(1)
-    )
-    return result.scalars().first()
-
-
-def raise_duplicate_filename_conflict(
-    *,
-    existing_document_id: str,
-    source_file_name: str,
-) -> None:
-    raise ConflictException(
-        user_message=(
-            f"A document named {source_file_name!r} already exists. "
-            "To replace it, retry with document_id set to details.id."
-        ),
-        reason="ALREADY_EXISTS",
-        resource="Document",
-        resource_id=existing_document_id,
-        internal_message=(
-            "Duplicate source_file_name="
-            f"{source_file_name!r} document_id={existing_document_id}"
-        ),
-    )
-
-
 async def resolve_effective_document_scope(
     db: AsyncSession,
     *,

apps/api/tests/contract/test_job_creation_contract.py

@@ -874,7 +874,7 @@ async def test_should_inherit_existing_document_namespace_when_update_namespace_
 
 
 @pytest.mark.asyncio
-async def test_should_return_conflict_when_creating_a_job_for_a_duplicate_file_name(
+async def test_should_accept_same_file_name_before_source_content_is_available(
     developer_api_client_factory: Callable[
         [], AbstractAsyncContextManager[AsyncClient]
     ],
@@ -894,21 +894,12 @@ async def test_should_return_conflict_when_creating_a_job_for_a_duplicate_file_n
         response = await api_client.post("/api/v1/jobs", json=payload)
         jobs_after = await _count_jobs()
 
-    assert response.status_code == 409
+    assert response.status_code == 200
     response_json: dict[str, object] = response.json()
-    error = cast(dict[str, object], response_json["error"])
-    assert response_json["success"] is False
-    assert error["code"] == "ALREADY_EXISTS"
-    assert error["message"] == (
-        f"A document named {file_name!r} already exists. "
-        "To replace it, retry with document_id set to details.id."
-    )
-    assert error["details"] == {
-        "reason": "ALREADY_EXISTS",
-        "resource": "Document",
-        "id": document_id,
-    }
-    assert jobs_after == jobs_before
+    assert response_json["status"] == "waiting-file"
+    assert response_json["document_id"] != document_id
+    assert response_json["upload_url"]
+    assert jobs_after == jobs_before + 1
 
 
 @pytest.mark.asyncio
@@ -967,7 +958,7 @@ async def test_should_allow_same_file_name_in_a_different_namespace(
 
 
 @pytest.mark.asyncio
-async def test_should_return_conflict_when_url_source_resolves_to_a_duplicate_file_name(
+async def test_should_accept_url_source_with_same_resolved_file_name(
     monkeypatch: MonkeyPatch,
     developer_api_client_factory: Callable[
         [], AbstractAsyncContextManager[AsyncClient]
@@ -981,6 +972,7 @@ async def test_should_return_conflict_when_url_source_resolves_to_a_duplicate_fi
         "source_url": f"https://example.com/contracts/{file_name}",
         "data_id": "contract-job-duplicate-filename-url",
     }
+    scheduled_tasks: list[dict[str, object]] = []
 
     class _FakeHeadResponse:
         def __init__(self) -> None:
@@ -998,6 +990,25 @@ async def head(
             assert follow_redirects is False
             return _FakeHeadResponse()
 
+    class _FakeCeleryTask:
+        def apply_async(
+            self,
+            *,
+            args: list[object],
+            kwargs: dict[str, object],
+        ) -> None:
+            scheduled_tasks.append({"args": args, "kwargs": kwargs})
+
+    class _FakeCeleryApp:
+        def __init__(self) -> None:
+            from types import SimpleNamespace
+
+            self.conf = SimpleNamespace(task_routes={})
+
+        def signature(self, task_name: str) -> _FakeCeleryTask:
+            assert task_name == "app.core.tasks.document_ingestion_tasks.upload_url_file_task"
+            return _FakeCeleryTask()
+
     def resolve_public_address(
         host: str,
         port: int | None,
@@ -1007,6 +1018,7 @@ def resolve_public_address(
         del host, port, args, kwargs
         return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
 
+    import shared.core.celery_app as celery_app_module
     import shared.services.http.client_pool as client_pool_module
 
     monkeypatch.setattr(socket, "getaddrinfo", resolve_public_address)
@@ -1015,26 +1027,38 @@ def resolve_public_address(
         "get_async_client",
         lambda: _FakeAsyncHttpClient(),
     )
+    monkeypatch.setattr(
+        celery_app_module,
+        "get_celery_app",
+        lambda: _FakeCeleryApp(),
+    )
 
     async with developer_api_client_factory() as api_client:
         await _insert_document(document_id=document_id, source_file_name=file_name)
         jobs_before = await _count_jobs()
         response = await api_client.post("/api/v1/jobs", json=payload)
         jobs_after = await _count_jobs()
+        response_json: dict[str, object] = response.json()
+        job_id = cast(str, response_json["job_id"])
+        job_metadata = cast(
+            dict[str, object], (await _load_job_record(job_id))["job_metadata"]
+        )
 
-    assert response.status_code == 409
-    error = cast(dict[str, object], response.json()["error"])
-    assert error["code"] == "ALREADY_EXISTS"
-    assert error["details"] == {
-        "reason": "ALREADY_EXISTS",
-        "resource": "Document",
-        "id": document_id,
-    }
-    assert jobs_after == jobs_before
+    assert response.status_code == 200
+    assert response_json["source_type"] == "url"
+    assert response_json["document_id"] != document_id
+    assert job_metadata["source_file_name"] == file_name
+    assert jobs_after == jobs_before + 1
+    assert scheduled_tasks == [
+        {
+            "args": [job_id, payload["source_url"], "local-dev-user"],
+            "kwargs": {"job_type": "document_ingestion"},
+        }
+    ]
 
 
 @pytest.mark.asyncio
-async def test_should_return_conflict_when_creating_a_v2_job_for_a_duplicate_file_name(
+async def test_should_accept_v2_job_with_same_file_name_before_upload(
     developer_api_client_factory: Callable[
         [], AbstractAsyncContextManager[AsyncClient]
     ],
@@ -1054,15 +1078,11 @@ async def test_should_return_conflict_when_creating_a_v2_job_for_a_duplicate_fil
         response = await api_client.post("/api/v2/jobs", json=payload)
         jobs_after = await _count_jobs()
 
-    assert response.status_code == 409
-    error = cast(dict[str, object], response.json()["error"])
-    assert error["code"] == "ALREADY_EXISTS"
-    assert error["details"] == {
-        "reason": "ALREADY_EXISTS",
-        "resource": "Document",
-        "id": document_id,
-    }
-    assert jobs_after == jobs_before
+    assert response.status_code == 200
+    response_json: dict[str, object] = response.json()
+    assert response_json["status"] == "waiting-file"
+    assert response_json["document_id"] != document_id
+    assert jobs_after == jobs_before + 1
 
 
 @pytest.mark.asyncio