diff --git a/package.json b/package.json
index 80c26e5..a217ede 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/files-webui",
-  "version": "0.1.26",
+  "version": "0.1.27",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",
diff --git a/pyproject.toml b/pyproject.toml
index 3e73e9b..f11e00d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,14 +4,14 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-files"
-version = "0.1.26"
+version = "0.1.27"
 description = "GovOPlaN files module with backend and WebUI integration."
 readme = "README.md"
 requires-python = ">=3.12"
 license = { file = "LICENSE" }
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.45",
+  "govoplan-core>=0.1.46",
   "defusedxml>=0.7,<1",
   "pyzipper>=0.3.6,<1",
   "python-multipart>=0.0.31,<1",
diff --git a/src/govoplan_files/backend/db/models.py b/src/govoplan_files/backend/db/models.py
index 2805cd1..e267c92 100644
--- a/src/govoplan_files/backend/db/models.py
+++ b/src/govoplan_files/backend/db/models.py
@@ -14,11 +14,13 @@ from sqlalchemy import (
     String,
     Text,
     UniqueConstraint,
+    event,
     text,
 )
 from sqlalchemy.orm import Mapped, mapped_column
 
 from govoplan_core.db.base import Base, TimestampMixin
+from govoplan_files.backend.storage.provenance import source_identity_hash, source_provenance_from_metadata
 
 
 def new_uuid() -> str:
@@ -192,6 +194,10 @@ class FileFolder(Base, TimestampMixin):
 
 class FileAsset(Base, TimestampMixin):
     __tablename__ = "file_assets"
+    __table_args__ = (
+        Index("ix_file_assets_user_source", "tenant_id", "owner_type", "owner_user_id", "source_identity_hash"),
+        Index("ix_file_assets_group_source", "tenant_id", "owner_type", "owner_group_id", "source_identity_hash"),
+    )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
     tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
@@ -227,6 +233,15 @@ class FileAsset(Base, TimestampMixin):
     metadata_: Mapped[dict[str, Any] | None] = mapped_column(
         "metadata", JSON, nullable=True
     )
+    source_identity_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
+
+@event.listens_for(FileAsset, "before_insert")
+@event.listens_for(FileAsset, "before_update")
+def _refresh_asset_source_identity(_mapper: object, _connection: object, asset: FileAsset) -> None:
+    # All owning metadata writes (including copy/restore) use ORM instances.
+    # Direct/bulk SQL metadata writers must also update this derived column.
+    asset.source_identity_hash = source_identity_hash(source_provenance_from_metadata(asset.metadata_))
 
 
 class FileVersion(Base, TimestampMixin):
diff --git a/src/govoplan_files/backend/german_documentation.py b/src/govoplan_files/backend/german_documentation.py
index f383c5b..ee4812b 100644
--- a/src/govoplan_files/backend/german_documentation.py
+++ b/src/govoplan_files/backend/german_documentation.py
@@ -1,9 +1,8 @@
 from __future__ import annotations
 
-from dataclasses import replace
 from typing import Iterable
 
-from govoplan_core.core.modules import DocumentationTopic
+from govoplan_core.core.modules import DocumentationTopic, localize_documentation_topics as _localize_topics
 
 
 _TRANSLATIONS = {
@@ -144,15 +143,4 @@ _TRANSLATIONS = {
 def localize_documentation_topics(
     topics: Iterable[DocumentationTopic],
 ) -> tuple[DocumentationTopic, ...]:
-    localized: list[DocumentationTopic] = []
-    for topic in topics:
-        german = _TRANSLATIONS.get(topic.id)
-        if german is None:
-            localized.append(topic)
-            continue
-        translations = {
-            locale: dict(value) for locale, value in topic.translations.items()
-        }
-        translations["de"] = {**translations.get("de", {}), **german}
-        localized.append(replace(topic, translations=translations))
-    return tuple(localized)
+    return _localize_topics(topics, locale="de", translations=_TRANSLATIONS)
diff --git a/src/govoplan_files/backend/manifest.py b/src/govoplan_files/backend/manifest.py
index c8a1416..d6718d9 100644
--- a/src/govoplan_files/backend/manifest.py
+++ b/src/govoplan_files/backend/manifest.py
@@ -461,7 +461,7 @@ def _dsar_provider(context: ModuleContext) -> object:
 manifest = ModuleManifest(
     id="files",
     name="Files",
-    version="0.1.26",
+    version="0.1.27",
     required_capabilities=(
         CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
         CAPABILITY_AUTH_PERMISSION_EVALUATOR,
@@ -652,6 +652,21 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=localize_documentation_topics((
+        DocumentationTopic(
+            id="files.source-identity-provenance",
+            title="Verify imported bytes and resolve source ambiguity",
+            summary="Bind provenance to acquired bytes and use tenant/owner-scoped source identities. All caller and browse metadata is preserved solely under import_annotations, never flattened into acquisition evidence.",
+            body="A supplied source_revision is an expected revision, not a label to assign to newer bytes. A stale or unavailable expected revision returns a conflict before persistence; refresh the provider listing and retry deliberately. Stored revisions come from the acquired object, and acquired_sha256 is computed from its actual bytes. Provider checksums are retained separately as provider_checksum_claims, not presented as locally verified hashes; caller and browse annotations remain in import_annotations and cannot replace acquisition evidence. S3 reads pin a version or condition on ETag; SMB denies write/delete sharing while reading and checks stable metadata; Seafile checks the source revision before and after acquisition. Remote provider revision claims are not cryptographic proof against a dishonest provider. Source lookup uses indexed canonical provenance within tenant and owner, never filename or equal content. Migration backfills the index in bounded batches without deleting, merging or selecting a winning copy. Multiple active files with one source identity cause an explicit conflict. Review the retained files and provenance before any authorized correction; copying, soft deletion and restoration preserve identity semantics. The index is non-unique because legitimate copies remain supported. It is not a new exactly-once concurrency guarantee. Owning ORM metadata writes refresh the index; maintenance bulk SQL must update the derived source hash consistently.",
+            layer="available", documentation_types=("user", "admin"), audience=("file_user", "file_admin"), order=17,
+            conditions=(DocumentationCondition(required_modules=("files",), any_scopes=("files:file:read", "files:file:upload", "files:file:admin")),),
+            links=(DocumentationLink(label="Files", href="/files", kind="runtime"), DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository")),
+            metadata={"kind": "reference", "related_topic_ids": ["files.governed-connectors-and-provenance", "files.workflow.import-managed-snapshot"]},
+            translations={"de": {
+                "title": "Importierte Bytes prüfen und Quellmehrdeutigkeit klären",
+                "summary": "Herkunft an gelesene Bytes binden und nach Mandant und Eigentümer indizierte Quellidentitäten nutzen. Alle Aufrufer- und Vorschauangaben bleiben ausschließlich unter import_annotations erhalten, niemals als Erwerbsnachweise auf gleicher Ebene.",
+                "body": "Eine übergebene source_revision ist eine erwartete Revision, kein Etikett für neuere Bytes. Eine veraltete oder nicht prüfbare Erwartung führt vor dem Speichern zum Konflikt; die Anbieterliste aktualisieren und bewusst erneut versuchen. Gespeicherte Revisionen stammen vom gelesenen Objekt; acquired_sha256 wird aus dessen tatsächlichen Bytes berechnet. Anbieterprüfsummen bleiben getrennt als provider_checksum_claims erhalten und gelten nicht als lokal verifizierte Hashes. Angaben des Aufrufers und der Vorschau bleiben in import_annotations und ersetzen keine Erwerbsnachweise. S3 bindet den Abruf an eine Version oder ETag-Bedingung; SMB verhindert Schreib-/Löschfreigabe während des Lesens und prüft stabile Metadaten; Seafile prüft die Revision vor und nach dem Abruf. Revisionsangaben eines unehrlichen Anbieters sind kein kryptografischer Beweis. Die Quellsuche verwendet indizierte kanonische Herkunft innerhalb von Mandant und Eigentümer, niemals Dateiname oder gleichen Inhalt. Die Migration ergänzt den Index stapelweise und löscht, vereinigt oder bevorzugt keine Kopie. Mehrere aktive Dateien mit derselben Quellidentität führen zum ausdrücklichen Konflikt. Vor berechtigter Korrektur Dateien und Herkunft prüfen; Kopieren, Papierkorb und Wiederherstellen erhalten die Identitätsregeln. Der Index bleibt wegen zulässiger Kopien nicht eindeutig und verspricht keine neue Exactly-once-Garantie bei Nebenläufigkeit. ORM-Metadatenänderungen aktualisieren ihn; administratives Bulk-SQL muss den abgeleiteten Quellhash konsistent mitpflegen.",
+            }},
+        ),
         DocumentationTopic(
             id="files.archive-worker-limits",
             title="Bound archive inspection and extraction work",
diff --git a/src/govoplan_files/backend/migrations/versions/a2b3c4d5e701_file_source_identity_index.py b/src/govoplan_files/backend/migrations/versions/a2b3c4d5e701_file_source_identity_index.py
new file mode 100755
index 0000000..2f71b30
--- /dev/null
+++ b/src/govoplan_files/backend/migrations/versions/a2b3c4d5e701_file_source_identity_index.py
@@ -0,0 +1,104 @@
+"""Index source identities without merging or discarding legacy copies.
+
+Revision ID: a2b3c4d5e701
+Revises: a2b3c4d5e6f8, a2b3c4d5e6f9
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "a2b3c4d5e701"
+down_revision = ("a2b3c4d5e6f8", "a2b3c4d5e6f9")
+branch_labels = None
+depends_on = None
+
+
+def _digest(metadata: object) -> str | None:
+    # Frozen v1 identity semantics: never import evolving application code.
+    provenance = (
+        metadata.get("source_provenance") if isinstance(metadata, dict) else None
+    )
+    if not isinstance(provenance, dict):
+        return None
+
+    def clean(value: object) -> str | None:
+        return (str(value).strip() or None) if value is not None else None
+
+    connector = clean(provenance.get("connector_id"))
+    provider = clean(provenance.get("provider"))
+    external = clean(provenance.get("external_id"))
+    path = clean(provenance.get("external_path"))
+    extra = (
+        provenance.get("metadata")
+        if isinstance(provenance.get("metadata"), dict)
+        else {}
+    )
+    if connector and external:
+        identity = ("external_id", connector, provider, external)
+    elif connector and path:
+        identity = (
+            "external_path",
+            connector,
+            provider,
+            clean(
+                extra.get("library_id") or extra.get("profile_id") or extra.get("share")
+            ),
+            path,
+        )
+    else:
+        return None
+    return hashlib.sha256(
+        json.dumps(identity, ensure_ascii=True, separators=(",", ":")).encode("ascii")
+    ).hexdigest()
+
+
+def upgrade() -> None:
+    op.add_column(
+        "file_assets", sa.Column("source_identity_hash", sa.String(64), nullable=True)
+    )
+    table = sa.table(
+        "file_assets",
+        sa.column("id", sa.String),
+        sa.column("metadata", sa.JSON),
+        sa.column("source_identity_hash", sa.String),
+    )
+    connection = op.get_bind()
+    cursor = None
+    while True:
+        statement = (
+            sa.select(table.c.id, table.c.metadata).order_by(table.c.id).limit(500)
+        )
+        if cursor is not None:
+            statement = statement.where(table.c.id > cursor)
+        rows = connection.execute(statement).all()
+        if not rows:
+            break
+        for row in rows:
+            digest = _digest(row.metadata)
+            if digest is not None:
+                connection.execute(
+                    table.update()
+                    .where(table.c.id == row.id)
+                    .values(source_identity_hash=digest)
+                )
+        cursor = rows[-1].id
+    # Deliberately non-unique: retained copies/legacy duplicates survive intact.
+    # Runtime sync reports ambiguity instead of selecting a newest/winning copy.
+    for owner in ("user", "group"):
+        op.create_index(
+            f"ix_file_assets_{owner}_source",
+            "file_assets",
+            ["tenant_id", "owner_type", f"owner_{owner}_id", "source_identity_hash"],
+        )
+
+
+def downgrade() -> None:
+    for owner in ("user", "group"):
+        op.drop_index(f"ix_file_assets_{owner}_source", table_name="file_assets")
+    with op.batch_alter_table("file_assets") as batch:
+        batch.drop_column("source_identity_hash")
diff --git a/src/govoplan_files/backend/route_support.py b/src/govoplan_files/backend/route_support.py
index 2583d88..6831f45 100644
--- a/src/govoplan_files/backend/route_support.py
+++ b/src/govoplan_files/backend/route_support.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import hashlib
+
 import json
 import os
 import tempfile
@@ -62,7 +64,8 @@ from govoplan_files.backend.storage.connector_credential_store import (
 from govoplan_files.backend.storage.connector_browse import (
     normalize_connector_browse_path,
 )
-from govoplan_files.backend.storage.connector_imports import read_connector_file
+from govoplan_files.backend.storage.connector_imports import ConnectorRevisionConflict, read_connector_file
+from govoplan_files.backend.storage.common import FileSourceConflict
 from govoplan_files.backend.storage.connector_deployment import (
     connector_effective_endpoint_url,
 )
@@ -406,6 +409,8 @@ def _ensure_campaign_file_access(
 
 def _http_error(exc: Exception, *, not_found: bool = False) -> HTTPException:
     code = status.HTTP_404_NOT_FOUND if not_found else status.HTTP_400_BAD_REQUEST
+    if isinstance(exc, FileSourceConflict):
+        code = status.HTTP_409_CONFLICT
     return HTTPException(status_code=code, detail=str(exc))
 
 
@@ -860,12 +865,23 @@ def _download_connector_payload(
         path=source_path,
         max_bytes=settings.file_upload_max_bytes,
     )
+    expected = str(payload.source_revision or "").strip()
+    observed = {downloaded.revision}
+    if profile.provider == "s3":
+        observed.add(downloaded.metadata.get("etag"))
+    if expected and expected not in observed:
+        raise ConnectorRevisionConflict("Connector source revision changed or is unavailable; refresh the source before importing")
+    provider_metadata = dict(downloaded.metadata)
+    checksum_claims = {key: provider_metadata.pop(key) for key in tuple(provider_metadata) if key.startswith("checksum_")}
     provenance_metadata = {
+        **provider_metadata,
         "profile_id": profile.id,
         "library_id": payload.library_id,
         "library_path": source_path,
-        **downloaded.metadata,
-        **payload.metadata,
+        "size": len(downloaded.data),
+        "acquired_sha256": hashlib.sha256(downloaded.data).hexdigest(),
+        "provider_checksum_claims": checksum_claims,
+        "import_annotations": dict(payload.metadata),
     }
     metadata = source_metadata(
         source_provenance={
@@ -878,7 +894,7 @@ def _download_connector_payload(
             "external_url": downloaded.external_url,
             "metadata": provenance_metadata,
         },
-        source_revision=payload.source_revision or downloaded.revision,
+        source_revision=downloaded.revision,
     )
     return source_path, downloaded, metadata or {}
 
diff --git a/src/govoplan_files/backend/routes/connector_io.py b/src/govoplan_files/backend/routes/connector_io.py
index 6eb8122..846bea2 100644
--- a/src/govoplan_files/backend/routes/connector_io.py
+++ b/src/govoplan_files/backend/routes/connector_io.py
@@ -28,6 +28,7 @@ from govoplan_files.backend.storage.connector_browse import (
     browse_connector_profile,
     normalize_connector_browse_path,
 )
+from govoplan_files.backend.storage.common import FileSourceConflict
 from govoplan_files.backend.storage.connector_imports import (
     ConnectorImportError,
     ConnectorImportUnsupported,
@@ -250,7 +251,7 @@ def sync_connector_space_folder(
             if detail.startswith("Skipped upload target:"):
                 action = "skipped"
                 counts["skipped"] += 1
-            elif detail.startswith("Target file already exists:"):
+            elif isinstance(exc, FileSourceConflict) or detail.startswith("Target file already exists:"):
                 action = "conflict"
                 counts["conflicts"] += 1
             else:
diff --git a/src/govoplan_files/backend/storage/common.py b/src/govoplan_files/backend/storage/common.py
index 3c096bc..7c65f2f 100644
--- a/src/govoplan_files/backend/storage/common.py
+++ b/src/govoplan_files/backend/storage/common.py
@@ -10,6 +10,10 @@ class FileStorageError(RuntimeError):
     pass
 
 
+class FileSourceConflict(FileStorageError):
+    """A source cannot be selected or revision-bound without ambiguity."""
+
+
 @dataclass(slots=True)
 class UploadedStoredFile:
     asset: FileAsset
diff --git a/src/govoplan_files/backend/storage/connector_imports.py b/src/govoplan_files/backend/storage/connector_imports.py
index 600fb44..2ab052d 100644
--- a/src/govoplan_files/backend/storage/connector_imports.py
+++ b/src/govoplan_files/backend/storage/connector_imports.py
@@ -32,6 +32,7 @@ from govoplan_files.backend.storage.connector_browse import (
 from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
 from govoplan_files.backend.storage.http_client import ConnectorHttpError, request_connector_bytes
 from govoplan_files.backend.storage.paths import filename_from_path
+from govoplan_files.backend.storage.common import FileSourceConflict
 
 
 class ConnectorImportError(RuntimeError):
@@ -42,6 +43,10 @@ class ConnectorImportUnsupported(ConnectorImportError):
     pass
 
 
+class ConnectorRevisionConflict(ConnectorImportError, FileSourceConflict):
+    pass
+
+
 @dataclass(frozen=True, slots=True)
 class ConnectorDownloadedFile:
     filename: str
@@ -120,6 +125,14 @@ def _read_seafile_file(profile: ConnectorProfile, *, library_id: str, path: str,
     if len(data) > max_bytes:
         raise ConnectorImportError(f"Seafile file exceeds limit of {max_bytes} bytes")
     detail = detail if isinstance(detail, dict) else {}
+    try:
+        after = _request_json("GET", _seafile_url(profile, f"api2/repos/{repo_id}/file/detail/"), headers=headers, params={"p": file_path})
+    except ConnectorBrowseError as exc:
+        raise ConnectorImportError(str(exc)) from exc
+    before_revision = _clean(detail.get("id") or detail.get("mtime") or detail.get("last_modified"))
+    after_revision = _clean(after.get("id") or after.get("mtime") or after.get("last_modified")) if isinstance(after, dict) else None
+    if not before_revision or before_revision != after_revision or (_int(detail.get("size")) is not None and _int(detail.get("size")) != len(data)):
+        raise ConnectorRevisionConflict("Seafile source changed or could not be revision-verified during download; refresh and retry")
     filename = filename_from_path(str(detail.get("name") or path))
     content_type = response.headers.get("content-type") or mimetypes.guess_type(filename)[0]
     external_id = f"{repo_id}:{normalize_connector_browse_path(path)}"
@@ -127,7 +140,7 @@ def _read_seafile_file(profile: ConnectorProfile, *, library_id: str, path: str,
         filename=filename,
         data=data,
         content_type=content_type,
-        revision=_clean(detail.get("id") or detail.get("mtime") or detail.get("last_modified")),
+        revision=before_revision,
         external_id=external_id,
         external_url=download_url,
         metadata={
@@ -202,12 +215,18 @@ def _read_smb_file(profile: ConnectorProfile, *, path: str, max_bytes: int) -> C
         unc_path = _smb_unc_path(location, file_path)
         smbclient = _smbclient_module()
         kwargs = _smb_client_kwargs(profile, location)
-        stat_result = smbclient.stat(unc_path, **kwargs)
-        size = _smb_stat_size(stat_result)
-        if size is not None and size > max_bytes:
-            raise ConnectorImportError(f"SMB file exceeds limit of {max_bytes} bytes")
-        with smbclient.open_file(unc_path, mode="rb", **kwargs) as handle:
-            data = handle.read(max_bytes + 1)
+        # Deny concurrent write/delete sharing while observing metadata and bytes.
+        with smbclient.open_file(unc_path, mode="rb", share_access="r", **kwargs) as handle:
+            stat_result = smbclient.stat(unc_path, **kwargs)
+            size = _smb_stat_size(stat_result)
+            if size is not None and size > max_bytes:
+                raise ConnectorImportError(f"SMB file exceeds limit of {max_bytes} bytes")
+            data = handle.read(min(size, max_bytes) + 1 if size is not None else max_bytes + 1)
+            after = smbclient.stat(unc_path, **kwargs)
+            if (_smb_stat_revision(stat_result) != _smb_stat_revision(after)
+                    or getattr(stat_result, "st_ino", None) != getattr(after, "st_ino", None)
+                    or size is None or len(data) != size):
+                raise ConnectorRevisionConflict("SMB source changed or could not be verified during download; refresh and retry")
     except ConnectorBrowseUnsupported as exc:
         raise ConnectorImportUnsupported(str(exc)) from exc
     except ConnectorBrowseError as exc:
@@ -264,16 +283,30 @@ def _download_s3_object(
     bucket: str,
     key: str,
     version_id: str | None,
+    etag: str | None,
     max_bytes: int,
 ) -> tuple[Any, bytes]:
     request: dict[str, object] = {"Bucket": bucket, "Key": key}
-    if version_id:
+    # S3's literal "null" version is replaceable when versioning is suspended.
+    if version_id and version_id != "null":
         request["VersionId"] = version_id
+    elif etag:
+        request["IfMatch"] = etag
+    else:
+        raise ConnectorRevisionConflict("S3 source has no revision or ETag for a verified download")
     try:
         response = client.get_object(**request)
         body = response.get("Body")
-        data = body.read(max_bytes + 1) if hasattr(body, "read") else bytes(response.get("Body") or b"")
+        try:
+            data = body.read(max_bytes + 1) if hasattr(body, "read") else bytes(response.get("Body") or b"")
+        finally:
+            close = getattr(body, "close", None)
+            if callable(close):
+                close()
     except Exception as exc:  # pragma: no cover - concrete exception types are dependency-version specific
+        error = getattr(exc, "response", {})
+        if isinstance(error, dict) and str(error.get("Error", {}).get("Code")) in {"PreconditionFailed", "412"}:
+            raise ConnectorRevisionConflict("S3 source changed during conditional download; refresh and retry") from exc
         raise ConnectorImportError(f"S3 object download failed: {exc}") from exc
     if len(data) > max_bytes:
         raise ConnectorImportError(f"S3 object exceeds limit of {max_bytes} bytes")
@@ -322,12 +355,16 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
             bucket=bucket,
             key=key,
             version_id=version_id,
+            etag=_clean(detail.get("ETag")),
             max_bytes=max_bytes,
         )
     finally:
         close = getattr(client, "close", None)
         if callable(close):
             close()
+    if (_clean(response.get("VersionId")) not in (None, version_id)
+            or (_clean(detail.get("ETag")) and _clean(response.get("ETag")) != _clean(detail.get("ETag")))):
+        raise ConnectorRevisionConflict("S3 returned another source revision; refresh and retry")
     content_type = _clean(response.get("ContentType") if isinstance(response, dict) else None) or _clean(detail.get("ContentType")) or mimetypes.guess_type(key)[0]
     etag = _clean(response.get("ETag") if isinstance(response, dict) else None) or _clean(detail.get("ETag"))
     filename = filename_from_path(key)
@@ -335,7 +372,7 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
         filename=filename,
         data=data,
         content_type=content_type,
-        revision=version_id or etag or _clean(detail.get("LastModified")),
+        revision=(version_id if version_id != "null" else None) or etag or _clean(detail.get("LastModified")),
         external_id=f"{bucket}:{key}",
         external_url=f"s3://{bucket}/{key}",
         metadata=_s3_download_metadata(
diff --git a/src/govoplan_files/backend/storage/files.py b/src/govoplan_files/backend/storage/files.py
index d0a44a1..83e3b0b 100644
--- a/src/govoplan_files/backend/storage/files.py
+++ b/src/govoplan_files/backend/storage/files.py
@@ -22,9 +22,9 @@ from govoplan_files.backend.storage.backends import (
     StorageObjectMissing,
     get_storage_backend,
 )
-from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, UploadedStoredFile, utcnow
+from govoplan_files.backend.storage.common import FileConflictResolution, FileSourceConflict, FileStorageError, UploadedStoredFile, utcnow
 from govoplan_files.backend.storage.paths import filename_from_path, join_folder_filename, normalize_folder, normalize_logical_path
-from govoplan_files.backend.storage.provenance import source_provenance_from_metadata
+from govoplan_files.backend.storage.provenance import source_identity as _source_identity, source_identity_hash, source_provenance_from_metadata
 from govoplan_files.backend.storage.recovery import begin_blob_write_recovery
 from govoplan_files.backend.storage.integrity import (
     QUARANTINED_BLOB_STATUSES,
@@ -414,14 +414,17 @@ def find_asset_by_source(
         return None
     assets = (
         _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id)
-        .filter(FileAsset.deleted_at.is_(None))
-        .order_by(FileAsset.updated_at.desc())
+        .filter(FileAsset.deleted_at.is_(None), FileAsset.source_identity_hash == source_identity_hash(source_provenance))
+        .limit(2)
         .all()
     )
-    for asset in assets:
-        if _source_identity(source_provenance_from_metadata(asset.metadata_ or {})) == wanted:
-            return asset
-    return None
+    if len(assets) > 1:
+        raise FileSourceConflict("Multiple active files have this source identity; review their provenance before synchronizing. No file was selected or merged.")
+    if not assets:
+        return None
+    if _source_identity(source_provenance_from_metadata(assets[0].metadata_ or {})) != wanted:
+        raise FileSourceConflict("The indexed file source identity does not match its provenance; synchronization was stopped.")
+    return assets[0]
 
 
 def update_file_asset_content(
@@ -1161,29 +1164,6 @@ def _next_version_number(session: Session, asset_id: str) -> int:
     return (int(row[0]) if row else 0) + 1
 
 
-def _source_identity(provenance: dict[str, Any] | None) -> tuple[object, ...] | None:
-    if not provenance:
-        return None
-    connector_id = _clean_identity(provenance.get("connector_id"))
-    provider = _clean_identity(provenance.get("provider"))
-    external_id = _clean_identity(provenance.get("external_id"))
-    if connector_id and external_id:
-        return ("external_id", connector_id, provider, external_id)
-    external_path = _clean_identity(provenance.get("external_path"))
-    metadata = provenance.get("metadata") if isinstance(provenance.get("metadata"), dict) else {}
-    library_id = _clean_identity(metadata.get("library_id") or metadata.get("profile_id") or metadata.get("share"))
-    if connector_id and external_path:
-        return ("external_path", connector_id, provider, library_id, external_path)
-    return None
-
-
-def _clean_identity(value: object) -> str | None:
-    if value is None:
-        return None
-    text = str(value).strip()
-    return text or None
-
-
 def _copy_asset_to_path(
     session: Session,
     asset: FileAsset,
diff --git a/src/govoplan_files/backend/storage/provenance.py b/src/govoplan_files/backend/storage/provenance.py
index 1ba5757..ad506a9 100644
--- a/src/govoplan_files/backend/storage/provenance.py
+++ b/src/govoplan_files/backend/storage/provenance.py
@@ -1,12 +1,40 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+import hashlib
+import json
 from typing import Any
 
 
 SOURCE_PROVENANCE_METADATA_KEY = "source_provenance"
 SOURCE_REVISION_METADATA_KEY = "source_revision"
 
+
+def source_identity(provenance: dict[str, Any] | None) -> tuple[object, ...] | None:
+    """The established source identity; neither filename nor content identity."""
+    if not provenance:
+        return None
+    def clean(value: object) -> str | None:
+        return str(value).strip() or None if value is not None else None
+    connector_id = clean(provenance.get("connector_id"))
+    provider = clean(provenance.get("provider"))
+    external_id = clean(provenance.get("external_id"))
+    if connector_id and external_id:
+        return ("external_id", connector_id, provider, external_id)
+    external_path = clean(provenance.get("external_path"))
+    metadata = provenance.get("metadata") if isinstance(provenance.get("metadata"), dict) else {}
+    library_id = clean(metadata.get("library_id") or metadata.get("profile_id") or metadata.get("share"))
+    if connector_id and external_path:
+        return ("external_path", connector_id, provider, library_id, external_path)
+    return None
+
+
+def source_identity_hash(provenance: dict[str, Any] | None) -> str | None:
+    identity = source_identity(provenance)
+    if identity is None:
+        return None
+    return hashlib.sha256(json.dumps(identity, ensure_ascii=True, separators=(",", ":")).encode("ascii")).hexdigest()
+
 _PROVENANCE_STRING_FIELDS = {
     "source_type",
     "connector_id",
diff --git a/tests/test_connector_provenance_binding.py b/tests/test_connector_provenance_binding.py
new file mode 100755
index 0000000..e56ab10
--- /dev/null
+++ b/tests/test_connector_provenance_binding.py
@@ -0,0 +1,244 @@
+import hashlib
+import io
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from govoplan_files.backend.route_support import (
+    _download_connector_payload,
+    _http_error,
+)
+from govoplan_files.backend.schemas import FileConnectorImportRequest
+from govoplan_files.backend.storage.connector_imports import (
+    ConnectorDownloadedFile,
+    ConnectorRevisionConflict,
+    _read_seafile_file,
+    _read_smb_file,
+    _read_s3_file,
+)
+from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
+
+
+class ConnectorProvenanceTests(unittest.TestCase):
+    def acquire(self, download, *, expected=None, annotations=None):
+        profile = ConnectorProfile(id="profile", label="Synthetic", provider="s3")
+        request = FileConnectorImportRequest(
+            library_id="bucket",
+            path="source.txt",
+            source_revision=expected,
+            metadata=annotations or {},
+        )
+        with (
+            patch(
+                "govoplan_files.backend.route_support.connector_policy_decision",
+                return_value=SimpleNamespace(allowed=True),
+            ),
+            patch(
+                "govoplan_files.backend.route_support.read_connector_file",
+                return_value=download,
+            ),
+        ):
+            return _download_connector_payload(profile, request, operation="sync")
+
+    def test_stale_or_unavailable_expected_revision_cannot_label_new_bytes(self):
+        for revision in ("B", None):
+            with (
+                self.subTest(revision=revision),
+                self.assertRaises(ConnectorRevisionConflict) as caught,
+            ):
+                self.acquire(
+                    ConnectorDownloadedFile(
+                        filename="file", data=b"B", revision=revision
+                    ),
+                    expected="A",
+                )
+            self.assertEqual(409, _http_error(caught.exception).status_code)
+
+    def test_actual_digest_is_computed_and_claims_and_annotations_are_not_authority(
+        self,
+    ):
+        annotations = {
+            "profile_id": "forged",
+            "size": 999,
+            "checksum_sha256": "caller-claim",
+            "acquired_sha256": "forged",
+            "note": "keep this",
+        }
+        download = ConnectorDownloadedFile(
+            filename="source.txt",
+            data=b"actual",
+            revision="version-B",
+            metadata={"etag": "etag-B", "checksum_sha256": "provider-claim"},
+        )
+        _, actual, metadata = self.acquire(
+            download, expected="etag-B", annotations=annotations
+        )
+        self.assertEqual("version-B", metadata["source_revision"])
+        evidence = metadata["source_provenance"]["metadata"]
+        self.assertEqual(
+            hashlib.sha256(actual.data).hexdigest(), evidence["acquired_sha256"]
+        )
+        self.assertEqual(len(actual.data), evidence["size"])
+        self.assertEqual("profile", evidence["profile_id"])
+        self.assertNotIn("checksum_sha256", evidence)
+        self.assertEqual(
+            {"checksum_sha256": "provider-claim"}, evidence["provider_checksum_claims"]
+        )
+        self.assertEqual(annotations, evidence["import_annotations"])
+        self.assertEqual(
+            metadata,
+            self.acquire(download, expected="version-B", annotations=annotations)[2],
+        )
+
+    def test_caller_fields_stay_annotations_when_provider_omits_evidence(self):
+        annotations = {
+            "sha256": "caller-hash",
+            "etag": "caller-etag",
+            "mtime": "caller-modified-time",
+            "checksum_sha256": "caller-checksum",
+            "provider_checksum_claims": {"checksum_sha256": "caller-claim"},
+            "connector_space_id": "browse-space",
+            "browse_etag": "browse-revision",
+            "folder_sync": True,
+            "note": {"values": ["001", "  keep spaces  ", None]},
+        }
+        download = ConnectorDownloadedFile(filename="source.txt", data=b"actual")
+        _, _, metadata = self.acquire(download, annotations=annotations)
+        evidence = metadata["source_provenance"]["metadata"]
+        self.assertEqual(annotations, evidence["import_annotations"])
+        self.assertEqual({}, evidence["provider_checksum_claims"])
+        self.assertEqual(
+            hashlib.sha256(download.data).hexdigest(), evidence["acquired_sha256"]
+        )
+        self.assertEqual(
+            {
+                "profile_id", "library_id", "library_path", "size",
+                "acquired_sha256", "provider_checksum_claims", "import_annotations",
+            },
+            set(evidence),
+        )
+
+    def test_seafile_rechecks_revision_after_download(self):
+        profile = ConnectorProfile(
+            id="profile",
+            label="Synthetic",
+            provider="seafile",
+            endpoint_url="https://example.invalid",
+        )
+        before = {"id": "A", "name": "source.txt", "size": 4}
+        for after in ({**before, "id": "B"}, before):
+            with (
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._seafile_headers",
+                    return_value={},
+                ),
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._request_json",
+                    side_effect=[before, "https://example.invalid/download", after],
+                ),
+                patch(
+                    "govoplan_files.backend.storage.connector_imports.request_connector_bytes",
+                    return_value=SimpleNamespace(
+                        status_code=200, content=b"DATA", headers={}
+                    ),
+                ),
+            ):
+                if after["id"] == "B":
+                    with self.assertRaises(ConnectorRevisionConflict):
+                        _read_seafile_file(
+                            profile, library_id="repo", path="source.txt", max_bytes=10
+                        )
+                else:
+                    result = _read_seafile_file(
+                        profile, library_id="repo", path="source.txt", max_bytes=10
+                    )
+                    self.assertEqual(("A", b"DATA"), (result.revision, result.data))
+
+    def test_smb_observes_stable_metadata_while_write_and_delete_sharing_are_denied(
+        self,
+    ):
+        profile = ConnectorProfile(id="profile", label="Synthetic", provider="smb")
+        before = SimpleNamespace(st_size=4, st_mtime_ns=10, st_mtime=1, st_ino=1)
+        for after in (
+            SimpleNamespace(st_size=4, st_mtime_ns=11, st_mtime=1, st_ino=1),
+            before,
+        ):
+            with (
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._smb_location",
+                    return_value=SimpleNamespace(
+                        share="share", server="example.invalid", port=445
+                    ),
+                ),
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._smb_unc_path",
+                    return_value="synthetic",
+                ),
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._smb_client_kwargs",
+                    return_value={},
+                ),
+                patch(
+                    "govoplan_files.backend.storage.connector_imports._smbclient_module"
+                ) as factory,
+            ):
+                sdk = factory.return_value
+                sdk.stat.side_effect = [before, after]
+                sdk.open_file.return_value = io.BytesIO(b"DATA")
+                if after is before:
+                    self.assertEqual(
+                        b"DATA",
+                        _read_smb_file(profile, path="source.txt", max_bytes=10).data,
+                    )
+                else:
+                    with self.assertRaises(ConnectorRevisionConflict):
+                        _read_smb_file(profile, path="source.txt", max_bytes=10)
+                self.assertEqual("r", sdk.open_file.call_args.kwargs["share_access"])
+
+    def test_s3_conditional_or_versioned_read_and_response_binding(self):
+        profile = ConnectorProfile(id="profile", label="Synthetic", provider="s3")
+        for version in (None, "version-A", "null"):
+            for returned_etag in ("etag-A", "etag-B"):
+                with (
+                    self.subTest(version=version, etag=returned_etag),
+                    patch(
+                        "govoplan_files.backend.storage.connector_imports._s3_import_client"
+                    ) as factory,
+                ):
+                    client = factory.return_value
+                    client.head_object.return_value = {
+                        "ContentLength": 4,
+                        "ETag": "etag-A",
+                        "VersionId": version,
+                    }
+                    body = io.BytesIO(b"DATA")
+                    client.get_object.return_value = {
+                        "Body": body,
+                        "ETag": returned_etag,
+                        "VersionId": version,
+                    }
+                    if returned_etag == "etag-A":
+                        result = _read_s3_file(
+                            profile,
+                            library_id="bucket",
+                            path="source.txt",
+                            max_bytes=10,
+                        )
+                        self.assertEqual(version if version and version != "null" else "etag-A", result.revision)
+                    else:
+                        with self.assertRaises(ConnectorRevisionConflict):
+                            _read_s3_file(
+                                profile,
+                                library_id="bucket",
+                                path="source.txt",
+                                max_bytes=10,
+                            )
+                    request = client.get_object.call_args.kwargs
+                    immutable = version and version != "null"
+                    self.assertEqual(version if immutable else "etag-A", request["VersionId" if immutable else "IfMatch"])
+                    self.assertTrue(body.closed)
+                    client.close.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_manifest_documentation.py b/tests/test_manifest_documentation.py
index 6721269..5e4fc73 100644
--- a/tests/test_manifest_documentation.py
+++ b/tests/test_manifest_documentation.py
@@ -4,6 +4,7 @@ import unittest
 
 
 STATIC_TOPIC_IDS = {
+    "files.source-identity-provenance",
     "files.archive-worker-limits",
     "files.configuration-package.managed-storage",
     "files.quick-access-and-product-area",
@@ -46,6 +47,15 @@ class FilesManifestDocumentationTests(unittest.TestCase):
     def topic(self, topic_id: str):
         return self.topics[topic_id]
 
+    def test_source_provenance_and_legacy_ambiguity_are_bilingual(self):
+        topic = self.topic("files.source-identity-provenance")
+        self.assertEqual({"admin", "user"}, set(topic.documentation_types))
+        for body in (topic.body, topic.translations["de"]["body"]):
+            for field in ("source_revision", "acquired_sha256", "provider_checksum_claims", "import_annotations"):
+                self.assertIn(field, body)
+        self.assertIn("non-unique", topic.body)
+        self.assertIn("nicht eindeutig", topic.translations["de"]["body"])
+
     def test_archive_resource_boundary_is_static_and_bilingual(self):
         topic = self.topic("files.archive-worker-limits")
         self.assertEqual("available", topic.layer)
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 7924d6a..6e82557 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -31,7 +31,7 @@ class FilesMigrationTests(unittest.TestCase):
                 )
                 with engine.connect() as connection:
                     self.assertIn(
-                        "a2b3c4d5e6f9",
+                        "a2b3c4d5e701",
                         set(MigrationContext.configure(connection).get_current_heads()),
                     )
             finally:
diff --git a/tests/test_source_identity_index.py b/tests/test_source_identity_index.py
new file mode 100755
index 0000000..3eeb300
--- /dev/null
+++ b/tests/test_source_identity_index.py
@@ -0,0 +1,199 @@
+from datetime import datetime, timezone
+import importlib
+import unittest
+
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from sqlalchemy import (
+    JSON,
+    Column,
+    MetaData,
+    String,
+    Table,
+    create_engine,
+    inspect,
+    select,
+    text,
+)
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db import models as _access_models  # register FK metadata
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_files.backend.db.models import FileAsset
+from govoplan_files.backend.storage.common import FileSourceConflict
+from govoplan_files.backend.storage.files import find_asset_by_source
+from govoplan_files.backend.storage.provenance import source_identity_hash
+
+
+def provenance(external="remote-1"):
+    return {"connector_id": " connector ", "provider": "s3", "external_id": external}
+
+
+class SourceIdentityIndexTests(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite:///:memory:")
+        for table in (
+            _access_models.Account.__table__,
+            _access_models.User.__table__,
+            _access_models.Group.__table__,
+            FileAsset.__table__,
+            ChangeSequenceEntry.__table__,
+        ):
+            table.create(self.engine)
+        self.session = Session(self.engine)
+
+    def tearDown(self):
+        self.session.close()
+        self.engine.dispose()
+
+    def asset(self, name, *, tenant="tenant", owner="owner", source=None, group=False):
+        row = FileAsset(
+            id=name,
+            tenant_id=tenant,
+            owner_type="group" if group else "user",
+            owner_user_id=None if group else owner,
+            owner_group_id=owner if group else None,
+            display_path=f"{name}.txt",
+            filename=f"{name}.txt",
+            metadata_={"source_provenance": source or provenance()},
+        )
+        self.session.add(row)
+        self.session.flush()
+        return row
+
+    def find(self, source=None, *, tenant="tenant", owner="owner", group=False):
+        return find_asset_by_source(
+            self.session,
+            tenant_id=tenant,
+            owner_type="group" if group else "user",
+            owner_id=owner,
+            source_provenance=source or provenance(),
+        )
+
+    def test_index_lookup_preserves_tenant_owner_provider_and_external_identity(self):
+        row = self.asset("wanted")
+        self.asset("other-tenant", tenant="another")
+        self.asset("other-owner", owner="another")
+        self.asset("group", group=True)
+        self.asset("other-provider", source={**provenance(), "provider": "webdav"})
+        for index in range(100):
+            self.asset(f"unrelated-{index}", source=provenance(f"remote-{index + 2}"))
+        self.assertIs(row, self.find())
+        self.assertIs(row, self.find())  # retry
+        self.assertEqual("group", self.find(group=True).id)
+        statement = (
+            select(FileAsset.id)
+            .where(
+                FileAsset.tenant_id == "tenant",
+                FileAsset.owner_type == "user",
+                FileAsset.owner_user_id == "owner",
+                FileAsset.source_identity_hash == source_identity_hash(provenance()),
+                FileAsset.deleted_at.is_(None),
+            )
+            .limit(2)
+        )
+        sql = str(
+            statement.compile(self.engine, compile_kwargs={"literal_binds": True})
+        )
+        plan = self.session.execute(text("EXPLAIN QUERY PLAN " + sql)).all()
+        self.assertIn("ix_file_assets_user_source", str(plan))
+        self.assertIn("LIMIT 2", sql)
+
+    def test_metadata_copy_delete_restore_and_duplicates_never_choose_a_winner(self):
+        row = self.asset("original")
+        copy = self.asset("copy", source=dict(row.metadata_["source_provenance"]))
+        self.assertEqual(row.source_identity_hash, copy.source_identity_hash)
+        with self.assertRaisesRegex(FileSourceConflict, "Multiple active"):
+            self.find()
+        copy.deleted_at = datetime.now(timezone.utc)
+        self.session.flush()
+        self.assertIs(row, self.find())
+        copy.deleted_at = None
+        self.session.flush()
+        with self.assertRaises(FileSourceConflict):
+            self.find()
+        copy.metadata_ = {"source_provenance": provenance("changed")}
+        self.session.flush()
+        self.assertIs(row, self.find())
+        self.assertIs(copy, self.find(provenance("changed")))
+        copy.metadata_ = {}
+        self.session.flush()
+        self.assertIsNone(copy.source_identity_hash)
+        self.assertIsNone(self.find(provenance("changed")))
+
+    def test_backfill_retains_all_legacy_records_and_metadata_in_bounded_batches(self):
+        migration = importlib.import_module(
+            "govoplan_files.backend.migrations.versions.a2b3c4d5e701_file_source_identity_index"
+        )
+        engine = create_engine("sqlite:///:memory:")
+        metadata = MetaData()
+        old = Table(
+            "file_assets",
+            metadata,
+            Column("id", String, primary_key=True),
+            Column("tenant_id", String),
+            Column("owner_type", String),
+            Column("owner_user_id", String),
+            Column("owner_group_id", String),
+            Column("metadata", JSON),
+        )
+        metadata.create_all(engine)
+        rows = [
+            {
+                "id": f"{index:04d}",
+                "tenant_id": "tenant",
+                "owner_type": "user",
+                "owner_user_id": "owner",
+                "metadata": {
+                    "source_provenance": provenance(
+                        "duplicate" if index < 2 else str(index)
+                    )
+                },
+            }
+            for index in range(503)
+        ]
+        rows.append(
+            {
+                "id": "no-source",
+                "tenant_id": "tenant",
+                "owner_type": "user",
+                "owner_user_id": "owner",
+                "metadata": {"unrelated": [1, 2]},
+            }
+        )
+        try:
+            with engine.begin() as connection:
+                connection.execute(old.insert(), rows)
+                with Operations.context(MigrationContext.configure(connection)):
+                    migration.upgrade()
+                new = Table("file_assets", MetaData(), autoload_with=connection)
+                actual = (
+                    connection.execute(select(new).order_by(new.c.id)).mappings().all()
+                )
+                self.assertEqual(len(rows), len(actual))
+                self.assertEqual(
+                    [row["metadata"] for row in rows],
+                    [row["metadata"] for row in actual],
+                )
+                self.assertEqual(
+                    actual[0]["source_identity_hash"], actual[1]["source_identity_hash"]
+                )
+                for row in actual:
+                    self.assertEqual(
+                        source_identity_hash(row["metadata"].get("source_provenance")),
+                        row["source_identity_hash"],
+                    )
+                indexes = inspect(connection).get_indexes("file_assets")
+                self.assertEqual(2, len(indexes))
+                self.assertFalse(any(index["unique"] for index in indexes))
+                with Operations.context(MigrationContext.configure(connection)):
+                    migration.downgrade()
+                self.assertEqual(
+                    len(rows), len(connection.execute(select(old.c.id)).all())
+                )
+        finally:
+            engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/webui/package.json b/webui/package.json
index 5da9fc2..64d139a 100644
--- a/webui/package.json
+++ b/webui/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/files-webui",
-  "version": "0.1.26",
+  "version": "0.1.27",
   "private": true,
   "type": "module",
   "main": "src/index.ts",