fix(ui): align contextual documentation with headings

Verified with the coordinated workspace changes by devkit full run
2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed).
This shared UI pass does not mark the individual module reviews complete.

docs/FILES_HANDBOOK.md

@@ -682,6 +682,54 @@ and restart workers after installation. Both paths retain complete header and
 path validation, selection and actual output limits; the native path also
 independently checks size and CRC and never extracts to filesystem paths.
 
+All ZIP/TAR metadata parsing (including ZIP central directories and TAR PAX
+headers) and decoding now run in fresh, credential-stripped child processes.
+Preview/password checks have 120 seconds wall time, 90 CPU seconds and 512 MiB
+address space. Confirmation uses one child for the entire archive, with 600
+seconds wall time and 300 CPU seconds. Its address-space limit is the larger of
+512 MiB or `3 × configured member limit + 128 MiB`. A member limit must be positive
+and at most 2 GiB; the default 50 MiB member limit and the existing 250 MiB request,
+10,000-entry, 2 GiB expanded and 100:1 limits are unchanged. The kernel file-size
+ceiling is `max(member limit, 16 KiB)`; actual member/cumulative bytes are checked
+independently. Metadata transport has a 64 MiB ceiling, final extraction receipts
+64 KiB, and status records 16 KiB. Raw archive/member bytes are not pipe DTOs.
+
+`GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY` is shared with other isolated module work:
+one admitted operation per API/worker process by default, configurable from one
+to 16. Admission covers source snapshot preparation before starting the child,
+so a busy request does not first copy its source. Capacity is not global across
+replicas. Include each replica's address-space and temporary-disk budgets when
+sizing the deployment. Resource controls are required; there is no inline parser
+fallback. These are process resource limits, not a filesystem/network sandbox or
+an aggregate cgroup memory guarantee.
+
+Each admitted operation creates a separate local OS-temporary `0700` directory
+and `0600` source snapshot, beyond the upload-once work-root quota described
+above. Plan additional capacity for one compressed source copy and at most one
+extracted member per admitted archive. Snapshot copying is bounded to the initial
+regular-file size and rejects changed sources. The staging wrapper writes only
+numbered internal filenames, never archive names. The parent rechecks selected
+logical paths, regular/no-symlink file identity, size, digest and monotonic
+progress, persists the member with the existing Files authority/transaction, and
+deletes it before acknowledging the child to decode the next member. A private
+`0600` FIFO carries one eight-byte sequence acknowledgement; completed members
+wake the parent immediately, with no fixed per-member sleep. Wake notifications
+are capped at 32 KiB; configurations allowing more than 32,768 members use Core's
+ordinary progress polling after that budget, still within the wall-time limit.
+Passwords remain request-only and are never written to these files. Handled success/error
+paths reap the child and remove the private directory. Abrupt host/parent loss
+can leave private temporary files; use the deployment's OS-temporary cleanup
+policy without deleting active work.
+
+The confirmation wall budget includes time waiting for parent storage; a slow
+destination may exhaust it after earlier members were staged for persistence.
+Existing rollback and blob recovery remain authoritative. Busy capacity, missing
+controls, CPU/memory/time/output limits, or invalid staging records produce a
+controlled failure; split the archive or explicitly retry after resolving the
+cause. There is no resumable worker or automatic confirmation retry. An optional
+UI progress-write failure remains non-fatal, but a failed private worker status
+or acknowledgement channel must abort safely.
+
 Verified members are read and stored one at a time, avoiding a whole expanded
 archive in memory. Storage-client reuse is confined to one archive operation;
 authorization, tenant isolation and policy are not cached. Batched response

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",

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",

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):

src/govoplan_files/backend/documentation.py

@@ -184,6 +184,7 @@ def _archive_topic(
             "A requested preview may safely reupload the selected file if its stage expired; confirmation is never retried automatically. "
             "Unsafe paths and special filesystem entries are rejected. Actual extracted bytes are counted instead of trusting archive headers. "
             "Archive paths are limited to 4096 UTF-8 bytes and 128 components; derived parent directories count toward the entry limit. TAR inspection checks each header's size and ratio before advancing across that member's payload. "
+            "Metadata parsing and decoding run in disposable resource-limited children, including native ZIP decoding. Preview allows 120 seconds and confirmation 600 seconds; busy workers and CPU, memory, time or transport limits fail closed. Split a rejected archive or explicitly retry later; confirmation is not automatically retried. "
             "The shared blurred dialog overlay separates upload transfer, inspection, extraction/storage and final commit, without an envelope animation. "
             "Selected file/byte totals appear immediately and extraction progress uses actual server counters. Unknown progress is indeterminate; completion is shown only after the transaction succeeds. "
             "Keep the dialog open while processing; errors release the overlay and preserve the selection for review."
@@ -200,6 +201,7 @@ def _archive_topic(
                 "Bei einer angeforderten Vorschau darf die Oberfläche die ausgewählte Datei nach Ablauf erneut übertragen; die Bestätigung wird niemals automatisch wiederholt. "
                 "Unsichere Pfade und besondere Dateisystemeinträge werden abgelehnt; tatsächlich gelesene Bytes werden gezählt. "
                 "Archivpfade sind auf 4096 UTF-8-Bytes und 128 Komponenten begrenzt; abgeleitete übergeordnete Ordner zählen zur Eintragsgrenze. Die TAR-Prüfung kontrolliert Größe und Expansionsverhältnis bei jedem Header, bevor dessen Nutzdaten übersprungen werden. "
+                "Metadatenprüfung und Dekodieren einschließlich nativem ZIP laufen in kurzlebigen ressourcenbegrenzten Kindprozessen. Für die Vorschau gelten 120 Sekunden, für die Bestätigung 600 Sekunden; ausgelastete Worker sowie CPU-, Speicher-, Zeit- oder Transportgrenzen führen zum sicheren Abbruch. Ein abgelehntes Archiv aufteilen oder später ausdrücklich erneut versuchen; die Bestätigung wird nicht automatisch wiederholt. "
                 "Die gemeinsame Überlagerung des weichgezeichneten Dialogs unterscheidet Übertragung, Prüfung, Entpacken/Speichern und abschließende Transaktionsbestätigung, ohne Briefumschlaganimation. "
                 "Ausgewählte Datei- und Bytezahlen erscheinen sofort; der Entpackfortschritt nutzt tatsächliche Serverzähler. Unbekannter Fortschritt bleibt unbestimmt, der Abschluss wird erst nach erfolgreicher Transaktion angezeigt. "
                 "Den Dialog während der Verarbeitung geöffnet lassen; Fehler entfernen die Überlagerung und erhalten die Auswahl zur Prüfung."

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 = {
@@ -57,6 +56,9 @@ _TRANSLATIONS = {
         "title": "Verwaltete Dateien und Ordner organisieren",
         "summary": "Ordner anlegen und zugängliche Inhalte mit ausdrücklicher Konfliktbehandlung umbenennen, verschieben oder kopieren.",
         "body": (
+            "Dokumentationsbücher stehen neben den Überschriften für Dateien, Verbindungen, Richtlinien "
+            "oder Integrität und neben dem Titel des Dialogs zum Entpacken verwalteter Archive. "
+            "Feldhilfe bleibt bei der Feldbezeichnung. "
             "Die Organisation bleibt in gesteuerten persönlichen oder Gruppenbereichen. Verschieben erhält die Asset-Identität; Kopieren erzeugt neue Assets und Versionen, die unveränderliche Blob-Bytes wiederverwenden. Jeder Zielkonflikt muss ausdrücklich abgelehnt, durch Umbenennen gelöst, überschrieben oder übersprungen werden. "
             "Ordner erstellen und Hochladen bleiben neben Neu laden im Kopf des Arbeitsbereichs. Nach der Auswahl von Dateien oder Ordnern öffnet Auswahl verwalten die Aktionen Verschieben, Kopieren, Umbenennen, Freigaben und Zugriffserklärungen. "
             "Löschen ist von der Organisation getrennt und benötigt weiterhin die vorhandene Bestätigung. Herunterladen und Archiv entpacken bleiben für passende Auswahlen direkt erreichbar. "
@@ -127,7 +129,7 @@ _TRANSLATIONS = {
             "Core stellt das gemeinsame lokale/S3-Objektspeicher-Backend bereit; Files verantwortet Dateimetadaten und Objektschlüssel. Lokaler Speicher eignet sich für einen Laufzeitprozess, ein gemeinsames Host-Volume für Replikate auf demselben Host. Unabhängige Hosts benötigen einen ausdrücklich vertrauenswürdigen HTTPS-S3-kompatiblen Endpunkt. PostgreSQL, Objekte und Hauptschlüssel müssen auf denselben abgestimmten Recovery-Zeitpunkt wiederhergestellt werden. "
             "Die Archivoberfläche fordert ausdrücklich einmalige Übertragung mit Zwischenspeicherung an: Erneute Passwortprüfung und Bestätigung verwenden eine an Mandant und Person gebundene Kopie, speichern jedoch niemals Passwörter. Temporäre Archiv- und Fortschrittsdateien haben Modus 0600 unter einem privaten Verzeichnis mit Modus 0700. FILE_ARCHIVE_WORK_ROOT ersetzt den Standard im temporären Betriebssystemverzeichnis, dessen Name von Dienst-UID und Hash des konfigurierten Speicherwurzelpfads abgeleitet wird. Alle Worker desselben Hosts müssen darauf zugreifen; unabhängige Hosts benötigen gemeinsamen POSIX-Speicher mit funktionierenden flock-Sperren oder feste Host-Zuordnung für Archiv- und Fortschrittsanfragen, auch bei dauerhaftem S3-Blob-Speicher. "
             "Standardgrenzen sind 2 GiB zwischengespeicherte Archive pro Arbeitsverzeichnis, vier Kopien pro Mandant/Person und 1.800 Sekunden ab der ersten Speicherung ohne Verlängerung. Abbruch- und Wechselanfragen sowie erfolgreicher Abschluss geben Kopien frei; verlassene abgelaufene Kopien werden bei späterer Archivverarbeitung bereinigt, nicht garantiert zu einem exakten Zeitpunkt ohne weitere Aktivität. Aktive POSIX-Leases schützen benutzte Kopien auch nach Ablauf vor Verdrängung. Kopien und Fortschrittsdaten sind keine fortsetzbaren Hintergrundaufträge. "
-            "Eine optionale, aktuell gehaltene Systembibliothek libarchive beschleunigt klassisches ZIPCrypto mit gespeicherten oder Deflate-komprimierten Einträgen und ASCII- oder ausdrücklich als UTF-8 markierten Namen. Fehlende Bibliotheken, AES und nicht unterstützte Kodierungen/Formate wählen vor dem Dekodieren den bisherigen Python-Leser; native Fehler führen zum sicheren Abbruch. Vollständige Header-, Pfad- und Grenzprüfungen bleiben erhalten; der native Pfad prüft CRC und Größe zusätzlich unabhängig und entpackt niemals direkt ins Dateisystem. "
+            "Eine optionale, aktuell gehaltene Systembibliothek libarchive beschleunigt klassisches ZIPCrypto mit gespeicherten oder Deflate-komprimierten Einträgen und ASCII- oder ausdrücklich als UTF-8 markierten Namen. Fehlende Bibliotheken, AES und nicht unterstützte Kodierungen/Formate wählen vor dem Dekodieren den bisherigen Python-Leser; native Fehler führen zum sicheren Abbruch. Beide Leser laufen in einem begrenzten Kindprozess mit vollständigen Header-, Pfad- und Grenzprüfungen; der native Pfad prüft CRC und Größe zusätzlich unabhängig. Nur die Zwischenspeicherschicht schreibt Dateien unter privaten servergenerierten numerischen Namen, niemals unter Archivpfaden. "
             "Die Verarbeitung speichert jeweils einen geprüften Eintrag. Wiederverwendung des Speicher-Clients innerhalb eines Archivs und gebündelte Antwortmetadaten vermeiden Zusatzarbeit, ohne Autorisierung zwischenzuspeichern oder dauerhafte Recovery-, Integritäts- und Commit-Prüfungen pro Datei zu entfernen. Fortschritt zeigt tatsächliche Datei-/Bytezähler und eine eigene Abschlussphase statt Zeitschätzungen oder vorzeitiger Erfolgsmeldung. Ein schnellerer Decoder verspricht keine entsprechende Beschleunigung des gesamten Imports oder von S3."
         ),
     },
@@ -144,15 +146,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)

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,71 @@ 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",
+            summary="ZIP and TAR parsers run in disposable resource-limited processes while Files retains authorization and storage.",
+            body=(
+                "Archive preview and password verification run in a fresh child with 120 seconds wall time, 90 CPU seconds and 512 MiB address space. "
+                "Confirmation uses one child for the entire archive, including metadata validation and native or Python decoding, with 600 seconds wall time and 300 CPU seconds. "
+                "Its address-space ceiling is the larger of 512 MiB or three times the configured member limit plus 128 MiB; the supported member limit is positive and at most 2 GiB. "
+                "The ordinary defaults remain 50 MiB per member, 250 MiB compressed request, 10,000 entries, 2 GiB expanded and 100:1 expansion. "
+                "Core's GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY admits archive snapshot creation and processing together with other isolated module work; its default is one slot per API/worker process, configurable from one to 16. "
+                "Budget every replica separately. A busy request fails before copying its source and must be retried explicitly; missing process controls, CPU/memory/time/output limits and invalid staging records fail closed without inline parsing. "
+                "Provide OS-temporary disk capacity for one additional compressed source snapshot and one extracted member per admitted archive, beyond the upload-once staging quota. "
+                "Private 0700 directories contain only server-generated 0600 filenames. The child waits for parent storage and deletion of each verified member before decoding the next. A private FIFO carries one eight-byte sequence acknowledgement; completed members wake the parent without a fixed per-file sleep. Wake bytes are capped at 32 KiB; larger non-default entry counts use ordinary progress polling thereafter. "
+                "Member paths, selection, regular-file identity, size, digest and monotonic file/byte progress are rechecked by the parent. Passwords are request-only and not written to staging. "
+                "Metadata transport is bounded to 64 MiB and status records to 16 KiB; the child file-size limit is the greater of the member limit or 16 KiB, with actual member and cumulative limits checked independently. "
+                "Slow destination storage consumes the confirmation wall-time budget. Failures use existing transaction rollback/recovery; successful decoding alone is not a committed import, and confirmation is never retried automatically. "
+                "The child has no inherited application credentials or SQL session. These are process resource controls, not a filesystem/network sandbox or an aggregate cgroup memory guarantee."
+            ),
+            translations={"de": {
+                "title": "Archivprüfung und Entpacken begrenzen",
+                "summary": "ZIP- und TAR-Parser laufen in kurzlebigen ressourcenbegrenzten Prozessen; Files behält Autorisierung und Speicherung.",
+                "body": (
+                    "Archivvorschau und Passwortprüfung laufen in einem frischen Kindprozess mit 120 Sekunden Laufzeit, 90 CPU-Sekunden und 512 MiB Adressraum. "
+                    "Die Bestätigung verwendet einen Kindprozess für das gesamte Archiv einschließlich Metadatenprüfung und nativem oder Python-Dekodieren, mit 600 Sekunden Laufzeit und 300 CPU-Sekunden. "
+                    "Seine Adressraumgrenze ist das Maximum aus 512 MiB und dem Dreifachen der konfigurierten Dateigrenze plus 128 MiB; die unterstützte Dateigrenze ist positiv und höchstens 2 GiB. "
+                    "Die üblichen Standardgrenzen bleiben 50 MiB pro Datei, 250 MiB komprimierte Anfrage, 10.000 Einträge, 2 GiB entpackte Daten und 100:1 Expansion. "
+                    "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY von Core begrenzt Quellkopie und Verarbeitung gemeinsam mit isolierter Arbeit anderer Module: standardmäßig ein Platz pro API-/Worker-Prozess, konfigurierbar von eins bis 16. "
+                    "Jedes Replikat separat budgetieren. Bei Auslastung wird vor der Quellkopie abgelehnt; ein neuer Versuch muss ausdrücklich erfolgen. Fehlende Prozesskontrollen, CPU-/Speicher-/Zeit-/Ausgabegrenzen und ungültige Zwischenspeicherdaten führen zum sicheren Abbruch ohne Parsing im Elternprozess. "
+                    "Zusätzlich zur Quote für einmalig hochgeladene Archive benötigt jedes zugelassene Archiv im temporären Betriebssystemverzeichnis Platz für eine weitere komprimierte Quellkopie und eine entpackte Datei. "
+                    "Private Verzeichnisse mit Modus 0700 enthalten ausschließlich servergenerierte Dateinamen mit Modus 0600. Der Kindprozess wartet nach jeder geprüften Datei auf Speicherung und Löschung durch den Elternprozess, bevor er die nächste dekodiert. Ein privater FIFO überträgt eine acht Byte lange Sequenzbestätigung; fertige Dateien wecken den Elternprozess ohne feste Pause pro Datei. Wecksignale sind auf 32 KiB begrenzt; größere abweichend konfigurierte Eintragszahlen nutzen danach die gewöhnliche Fortschrittsabfrage. "
+                    "Dieser prüft Pfade, Auswahl, reguläre Dateiidentität, Größe, Prüfsumme und monotonen Datei-/Bytefortschritt erneut. Passwörter bleiben auf die Anfrage beschränkt und werden nicht zwischengespeichert. "
+                    "Der Metadatentransport ist auf 64 MiB, Statusdatensätze auf 16 KiB begrenzt; die Dateigrößengrenze des Kindprozesses ist das Maximum aus Dateilimit und 16 KiB. Tatsächliche Einzel- und Gesamtgrößen werden unabhängig kontrolliert. "
+                    "Langsamer Zielspeicher zählt zur Laufzeit der Bestätigung. Fehler nutzen bestehendes Transaktions-Rollback und Recovery; erfolgreiches Dekodieren ist kein bestätigter Import und die Bestätigung wird nie automatisch wiederholt. "
+                    "Der Kindprozess erbt weder Anwendungszugangsdaten noch SQL-Sitzungen. Die Kontrollen begrenzen Prozessressourcen, sind aber keine Dateisystem-/Netzwerk-Sandbox und keine aggregierte cgroup-Speichergarantie."
+                ),
+            }},
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("file_user", "file_admin", "operator"),
+            configuration_keys=("GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY",),
+            conditions=(DocumentationCondition(
+                required_modules=("files",),
+                any_scopes=("files:file:upload", "files:file:admin", "system:settings:read"),
+            ),),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            order=28,
+        ),
         DocumentationTopic(
             id="files.tabular-content",
             title="Use managed CSV and XLSX versions as governed data sources",
@@ -993,6 +1058,8 @@ manifest = ModuleManifest(
             title="Organize managed files and folders",
             summary="Create folders and rename, move, or copy accessible managed content with explicit conflict handling.",
             body=(
+                "Documentation books sit beside the Files, connection, policy, or integrity heading and "
+                "beside the managed-archive unpack dialog title. Field help stays with its label. "
                 "Organization stays inside governed personal or group spaces. Moves preserve the asset identity, while copies create new assets and versions that reuse immutable blob bytes. "
                 "Every target conflict must be rejected, renamed, overwritten, or skipped explicitly. "
                 "Create folder and Upload remain beside Reload in the workspace header. Select files or folders and open Manage selection for Move, Copy, Rename, sharing, or access explanations. "
@@ -1393,6 +1460,8 @@ manifest = ModuleManifest(
             title="Govern file connections, folder sync, and credential deletion",
             summary="Keep endpoints, credentials, inherited policy, and bounded manual synchronization separate, with reviewable outcomes and provenance.",
             body=(
+                "Documentation books sit beside the Files, connection, policy, or integrity heading and "
+                "beside the managed-archive unpack dialog title. Field help stays with its label. "
                 "System, tenant, and one user/group/campaign leaf form the effective policy chain: deny rules win and every configured allow rule must match. "
                 "Responses redact secret values and deployment references. Deleting a database-managed credential or profile immediately scrubs Files-owned encrypted material and private metadata in the same transaction as a non-secret audit event; dependent profiles are disabled, while legacy non-owned references are only detached and audited. "
                 "Removing a connector space is a separate owner-authorized operation: it retires only the local virtual-space link and leaves provider content, imported managed files and shares, profiles, credentials, and remote references untouched. Intrinsic user and group managed spaces cannot be removed. "
@@ -1452,6 +1521,9 @@ manifest = ModuleManifest(
                     "title": "Dateiverbindungen, Ordnersynchronisierung und das Löschen von Zugangsdaten steuern",
                     "summary": "Endpunkte, Zugangsdaten, vererbte Richtlinien und begrenzte manuelle Synchronisierung getrennt und mit prüfbaren Ergebnissen sowie Herkunftsnachweisen verwalten.",
                     "body": (
+                        "Dokumentationsbücher stehen neben den Überschriften für Dateien, Verbindungen, Richtlinien "
+                        "oder Integrität und neben dem Titel des Dialogs zum Entpacken verwalteter Archive. Feldhilfe "
+                        "bleibt bei der Feldbezeichnung. "
                         "System, Mandant und genau eine Benutzer-, Gruppen- oder Kampagnenebene bilden die wirksame Richtlinienkette: Ablehnungsregeln haben Vorrang und jede konfigurierte Erlaubnisregel muss zutreffen. "
                         "Antworten blenden Geheimwerte und Bereitstellungsverweise aus. Beim Löschen datenbankverwalteter Zugangsdaten oder Profile entfernt Files eigenes verschlüsseltes Material und private Metadaten in derselben Transaktion wie das nicht geheime Audit-Ereignis. Abhängige Profile werden deaktiviert; ältere, nicht Files gehörende Verweise werden nur getrennt und auditiert. "
                         "Das Entfernen eines Connector-Bereichs ist ein eigener, eigentümerberechtigter Vorgang: Nur die lokale Verknüpfung des virtuellen Bereichs wird außer Kraft gesetzt. Inhalte beim Anbieter, importierte verwaltete Dateien und Freigaben, Profile, Zugangsdaten und Remote-Verweise bleiben erhalten. Intrinsische Benutzer- und Gruppenbereiche können nicht entfernt werden. "
@@ -1775,7 +1847,7 @@ manifest = ModuleManifest(
                 "Core supplies the common local/S3 object-storage backend while Files owns file metadata and object-key semantics. Local storage is valid for one runtime process; a shared host volume supports same-host replicas; independent hosts require an explicitly trusted HTTPS S3-compatible endpoint. Restore PostgreSQL, objects, and the master key to one coordinated recovery point. "
                 "The archive UI opts into upload-once staging: password repreview and confirmation reuse a tenant/user-bound copy, but never persist passwords. Temporary archive and progress files use mode 0600 below a private 0700 directory. FILE_ARCHIVE_WORK_ROOT overrides the default OS-temporary directory derived from the service uid and configured storage-root hash. All same-host workers must share it; independent hosts need shared POSIX storage with working flock locks or sticky routing for archive and progress requests, even when durable blobs use S3. "
                 "Defaults are 2 GiB of staged archives per work root, four stages per tenant/user, and 1,800 seconds from initial staging without renewal. Cancel/replacement requests and successful confirmation release copies; abandoned expired copies are cleaned opportunistically by later archive work, not on an exact idle deadline. POSIX active leases prevent eviction of an in-use copy even after its TTL; stages and progress are not resumable background jobs. "
-                "A maintained optional system libarchive accelerates classic stored/deflated ZIPCrypto decoding for ASCII or UTF-8-flagged names. Missing libraries, AES and unsupported encodings/formats select the original Python reader before decoding; native errors fail closed. Both paths retain full header/path/limit checks, and the native path independently verifies CRC and size without filesystem extraction. "
+                "A maintained optional system libarchive accelerates classic stored/deflated ZIPCrypto decoding for ASCII or UTF-8-flagged names. Missing libraries, AES and unsupported encodings/formats select the original Python reader before decoding; native errors fail closed. Both readers run in a bounded child and retain full header/path/limit checks; the native path independently verifies CRC and size. Only the staging wrapper writes files, using private server-generated numeric names, never archive paths. "
                 "Extraction stores one verified member at a time. Archive-local backend-client reuse and batched response metadata reduce avoidable work without caching authorization or removing durable per-file recovery, integrity or commit verification. Progress reports actual file/byte counters and a separate finalization phase, never a timed estimate or an early success. Native decoder speed does not promise an equivalent whole-import or S3 improvement."
             ),
             layer="configured",

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")

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 {}
 

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:

src/govoplan_files/backend/storage/archive_workers.py

@@ -0,0 +1,418 @@
+"""Disposable archive parsers with bounded, acknowledged member staging.
+
+This is a resource boundary for server-owned parsers, not a filesystem/network
+sandbox. Only the parent owns database state, credentials and durable storage.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager, closing
+from dataclasses import asdict, replace
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import tempfile
+
+from govoplan_core.security.bounded_process import (
+    ProcessBudgetError,
+    ProcessLimits,
+    bounded_operation_admission,
+    run_bounded_operation,
+)
+from govoplan_core.security.worker_payload import (
+    WorkerPayloadError, decode_worker_payload, encode_worker_payload,
+)
+from govoplan_files.backend.storage.common import FileStorageError
+
+
+INSPECTION_LIMITS = ProcessLimits(
+    wall_seconds=120, cpu_seconds=90, memory_bytes=512 * 1024 * 1024,
+    input_bytes=64 * 1024 * 1024, output_bytes=64 * 1024 * 1024,
+)
+EXTRACTION_LIMITS = ProcessLimits(
+    wall_seconds=600, cpu_seconds=300, memory_bytes=512 * 1024 * 1024,
+    input_bytes=64 * 1024 * 1024, output_bytes=64 * 1024,
+    file_bytes=50 * 1024 * 1024,
+)
+_STATUS_BYTES = 16 * 1024
+_MAX_INTEGER = 2**63 - 1
+_MAX_WAKE_BYTES = 32 * 1024
+
+
+def _failure(message="Archive worker returned an invalid staging record"):
+    return FileStorageError(message)
+
+
+def _read_regular(path: Path, maximum: int, *, atomic_record: bool = False) -> bytes:
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    try:
+        info = os.fstat(descriptor)
+        allowed_links = {0, 1} if atomic_record else {1}
+        if not stat.S_ISREG(info.st_mode) or info.st_nlink not in allowed_links or info.st_size > maximum:
+            raise _failure()
+    except BaseException:
+        os.close(descriptor)
+        raise
+    with os.fdopen(descriptor, "rb") as source:
+        # BufferedReader.read(n) can reserve n bytes even for a tiny file. Use
+        # the validated actual size, never the potentially multi-GiB policy cap.
+        value = source.read(info.st_size + 1)
+        after = os.fstat(source.fileno())
+        if (len(value) != info.st_size or after.st_nlink not in allowed_links
+                or _fingerprint(info, atomic_record=atomic_record) != _fingerprint(after, atomic_record=atomic_record)):
+            raise _failure()
+        return value
+
+
+def _fingerprint(info, *, atomic_record=False):
+    # Replacing a status/ack atomically may unlink an already-open old inode and
+    # update its ctime; its content, size and mtime must nevertheless be stable.
+    return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns,
+            None if atomic_record else (info.st_nlink, info.st_ctime_ns))
+
+
+def _write_record(directory: Path, name: str, value: object) -> None:
+    # name is a server constant: neither member names nor worker DTO paths.
+    payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
+    if len(payload) > _STATUS_BYTES:
+        raise _failure()
+    temporary = directory / f".{name}.tmp"
+    try:
+        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(descriptor, "wb") as output:
+            output.write(payload)
+        os.replace(temporary, directory / name)
+    except OSError as exc:
+        raise _failure("Archive worker staging channel is unavailable") from exc
+    finally:
+        try:
+            temporary.unlink(missing_ok=True)
+        except OSError:
+            pass  # The owning private-directory context retries final cleanup.
+
+
+@contextmanager
+def _ack_channel(directory: Path, *, create: bool = False):
+    """One fixed private FIFO; archive names never select a channel or file."""
+    path = directory / "ack"
+    descriptor = None
+    try:
+        try:
+            if create:
+                os.mkfifo(path, 0o600)
+            flags = os.O_RDWR if create else os.O_RDONLY
+            descriptor = os.open(path, flags | os.O_NONBLOCK | os.O_NOFOLLOW)
+            info = os.fstat(descriptor)
+            if not stat.S_ISFIFO(info.st_mode) or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600:
+                raise _failure()
+            if not create:
+                # Validate before blocking: a substituted regular file or FIFO
+                # must not hang the open. The parent holds a RDWR handle.
+                os.set_blocking(descriptor, True)
+        except OSError as exc:
+            raise _failure("Archive worker staging channel is unavailable") from exc
+        yield descriptor
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+
+
+def _send_ack(descriptor: int, sequence: int) -> None:
+    try:
+        # Eight bytes are within PIPE_BUF: one nonblocking write is atomic.
+        if os.write(descriptor, sequence.to_bytes(8, "big")) != 8:
+            raise _failure()
+    except OSError as exc:
+        raise _failure("Archive worker staging channel is unavailable") from exc
+
+
+def _receive_ack(descriptor: int, sequence: int) -> None:
+    value = b""
+    while len(value) < 8:
+        chunk = os.read(descriptor, 8 - len(value))
+        if not chunk:
+            raise _failure()
+        value += chunk
+    if int.from_bytes(value, "big") != sequence:
+        raise _failure()
+
+
+@contextmanager
+def _source_stage(archive_data):
+    if os.name != "posix" or not hasattr(os, "O_NOFOLLOW"):
+        raise FileStorageError("Required isolated-worker resource controls are unavailable")
+    with tempfile.TemporaryDirectory(prefix="govoplan-archive-worker-") as temporary:
+        directory = Path(temporary)
+        os.chmod(directory, 0o700)
+        source_path = directory / "source"
+        descriptor = os.open(source_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        try:
+            with os.fdopen(descriptor, "wb") as target:
+                if isinstance(archive_data, bytes):
+                    target.write(archive_data)
+                else:
+                    source_descriptor = os.open(os.fspath(archive_data), os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+                    with os.fdopen(source_descriptor, "rb") as source:
+                        before = os.fstat(source.fileno())
+                        if not stat.S_ISREG(before.st_mode):
+                            raise FileStorageError("Archive source must be a regular file")
+                        # Never follow a growing producer to EOF. Callers apply
+                        # the compressed request cap before supplying this path.
+                        remaining = before.st_size
+                        while remaining:
+                            chunk = source.read(min(remaining, 1024 * 1024))
+                            if not chunk:
+                                raise FileStorageError("Archive source changed during inspection")
+                            target.write(chunk)
+                            remaining -= len(chunk)
+                        extra = source.read(1)
+                        after = os.fstat(source.fileno())
+                        if extra or _fingerprint(before) != _fingerprint(after):
+                            raise FileStorageError("Archive source changed during inspection")
+        except OSError as exc:
+            raise FileStorageError("Archive source could not be staged safely") from exc
+        yield directory, source_path
+
+
+@contextmanager
+def _admitted_stage(archive_data):
+    # Share Core capacity with other expensive module work before copying any
+    # source bytes, not just after preparation has already consumed resources.
+    try:
+        with bounded_operation_admission() as admission, _source_stage(archive_data) as (directory, source):
+            yield directory, source, admission
+    except ProcessBudgetError as exc:
+        raise FileStorageError(f"Archive processing failed ({exc.code}): {exc}") from exc
+
+
+def _run(operation, data, *, limits, tick=None, admission=None):
+    try:
+        payload = encode_worker_payload(data, max_bytes=limits.input_bytes)
+        wire = run_bounded_operation(operation, payload, limits=limits, cancelled=tick, admission=admission)
+        result = decode_worker_payload(wire, max_bytes=limits.output_bytes)
+    except ProcessBudgetError as exc:
+        raise FileStorageError(f"Archive processing failed ({exc.code}): {exc}") from exc
+    except WorkerPayloadError as exc:
+        raise FileStorageError("Archive worker data exceeded safe transport limits") from exc
+    if type(result) is not dict:
+        raise _failure()
+    if "error" in result:
+        from govoplan_files.backend.storage.archives import ArchivePasswordError
+        if type(result.get("error")) is not str or type(result.get("password_error")) is not bool:
+            raise _failure()
+        error_type = ArchivePasswordError if result["password_error"] else FileStorageError
+        raise error_type(result["error"])
+    return result
+
+
+def inspect_archive_isolated(archive_data, **options):
+    from govoplan_files.backend.storage.archives import ArchiveEntry, ArchiveInspection, archive_format_for_filename
+    archive_format_for_filename(options["filename"])
+    with _admitted_stage(archive_data) as (_directory, source, admission):
+        result = _run(_inspect_worker, {"source": str(source), "options": options},
+                      limits=INSPECTION_LIMITS, admission=admission)
+    inspection = result.get("inspection")
+    if type(inspection) is not dict or type(inspection.get("entries")) is not tuple:
+        raise _failure()
+    if len(inspection["entries"]) > options["max_entries"]:
+        raise _failure()
+    try:
+        inspection["entries"] = tuple(ArchiveEntry(**entry) for entry in inspection["entries"])
+        return ArchiveInspection(**inspection)
+    except (TypeError, ValueError) as exc:
+        raise _failure() from exc
+
+
+def _inspect_worker(payload: bytes) -> bytes:
+    from govoplan_files.backend.storage.archives import ArchivePasswordError, _inspect_archive_content
+    data = decode_worker_payload(payload, max_bytes=INSPECTION_LIMITS.input_bytes)
+    try:
+        inspection = _inspect_archive_content(data["source"], **data["options"])
+        result = {"inspection": asdict(inspection)}
+    except FileStorageError as exc:
+        result = {"error": str(exc), "password_error": isinstance(exc, ArchivePasswordError)}
+    return encode_worker_payload(result, max_bytes=INSPECTION_LIMITS.output_bytes)
+
+
+def extract_archive_isolated(session, *, archive_data, filename, password, selected_paths,
+                             max_entries, max_file_bytes, max_expanded_bytes,
+                             max_expansion_ratio, progress, store_options):
+    from govoplan_files.backend.storage.archives import (
+        _safe_member_path, _store_archive_members, archive_format_for_filename,
+    )
+    from govoplan_files.backend.storage.files import archive_storage_backend_scope
+    archive_format_for_filename(filename)
+    if type(max_file_bytes) is not int or not 0 < max_file_bytes <= 2 * 1024 * 1024 * 1024:
+        raise FileStorageError("Isolated archive members require a positive limit of at most 2 GiB")
+    limits = replace(
+        EXTRACTION_LIMITS,
+        memory_bytes=max(EXTRACTION_LIMITS.memory_bytes, 3 * max_file_bytes + 128 * 1024 * 1024),
+        file_bytes=max(max_file_bytes, _STATUS_BYTES),
+    )
+    uploaded = []
+    stored_bytes = 0
+    sequence = 0
+    seen_paths = set()
+    last_extracted_bytes = 0
+    last_extracted_files = 0
+    expected_totals = None
+    selected = None if selected_paths is None else {_safe_member_path(path) for path in selected_paths}
+    if progress:
+        progress("inspecting", 0, 0, 0, 0)
+    with (
+        _admitted_stage(archive_data) as (directory, source, admission),
+        _ack_channel(directory, create=True) as acknowledgement,
+        archive_storage_backend_scope(),
+    ):
+        actual_total_limit = min(max_expanded_bytes, source.stat().st_size * max_expansion_ratio)
+
+        def poll():
+            nonlocal stored_bytes, sequence, last_extracted_bytes, last_extracted_files, expected_totals
+            try:
+                record = json.loads(_read_regular(directory / "status", _STATUS_BYTES, atomic_record=True))
+            except FileNotFoundError:
+                return False
+            except (OSError, ValueError, RecursionError) as exc:
+                raise _failure() from exc
+            if type(record) is not dict or type(record.get("sequence")) is not int:
+                raise _failure()
+            next_sequence = record["sequence"]
+            if not 0 < next_sequence <= _MAX_INTEGER or next_sequence < sequence:
+                raise _failure()
+            if next_sequence == sequence:
+                return False
+            kind = record.get("kind")
+            counters = record.get("progress")
+            if kind not in {"progress", "member"} or type(counters) is not list or len(counters) != 5:
+                raise _failure()
+            keys = {"sequence", "kind", "progress"}
+            if kind == "member":
+                keys |= {"path", "size", "sha256"}
+            if set(record) != keys:
+                raise _failure()
+            phase, completed, total, completed_bytes, total_bytes = counters
+            if phase != "extracting" or any(type(value) is not int for value in counters[1:]):
+                raise _failure()
+            if not (last_extracted_files <= completed <= total <= max_entries
+                    and 0 <= total_bytes <= actual_total_limit
+                    and last_extracted_bytes <= completed_bytes <= total_bytes):
+                raise _failure()
+            if expected_totals is not None and expected_totals != (total, total_bytes):
+                raise _failure()
+            if completed not in {len(uploaded), len(uploaded) + 1}:
+                raise _failure()
+            sequence = next_sequence
+            last_extracted_bytes = completed_bytes
+            last_extracted_files = completed
+            expected_totals = total, total_bytes
+            if progress:
+                progress(*counters)
+            if kind == "progress":
+                return False
+            if completed != len(uploaded) + 1 or type(record.get("path")) is not str:
+                raise _failure()
+            inner_path = _safe_member_path(record["path"])
+            if inner_path != record["path"] or inner_path in seen_paths:
+                raise _failure()
+            if selected is not None and not any(inner_path == path or inner_path.startswith(path + "/") for path in selected):
+                raise _failure()
+            size = record.get("size")
+            if type(size) is not int or not 0 <= size <= max_file_bytes or stored_bytes + size != completed_bytes:
+                raise _failure()
+            member_path = directory / f"member-{completed:08d}"
+            try:
+                data = _read_regular(member_path, max_file_bytes)
+            except OSError as exc:
+                raise _failure() from exc
+            if len(data) != size or hashlib.sha256(data).hexdigest() != record.get("sha256"):
+                raise _failure()
+            # The established persistence API accepts bytes, not a stream/path.
+            # Keep one member resident, with the existing transaction and quotas.
+            uploaded.extend(_store_archive_members(session, members=((inner_path, data),), **store_options))
+            stored_bytes += len(data)
+            seen_paths.add(inner_path)
+            del data
+            try:
+                member_path.unlink()
+            except OSError as exc:
+                raise _failure("Archive worker staging channel is unavailable") from exc
+            if progress:
+                progress("storing", len(uploaded), total, stored_bytes, total_bytes)
+            _send_ack(acknowledgement, sequence)
+            return False
+
+        result = _run(_extract_worker, {
+            "source": str(source), "directory": str(directory),
+            "options": {"filename": filename, "password": password, "max_entries": max_entries,
+                        "max_expanded_bytes": max_expanded_bytes, "max_expansion_ratio": max_expansion_ratio},
+            "selected_paths": None if selected is None else tuple(selected),
+            "max_file_bytes": max_file_bytes,
+        }, limits=limits, tick=poll, admission=admission)
+        if type(result.get("completed")) is not int or result["completed"] != len(uploaded):
+            raise _failure()
+        return uploaded
+
+
+def _extract_worker(payload: bytes) -> bytes:
+    from govoplan_files.backend.storage.archives import (
+        ArchivePasswordError, _inspect_archive_content, _read_selected_tar_members,
+        _read_selected_zip_members, _selected_file_paths,
+    )
+    data = decode_worker_payload(payload, max_bytes=EXTRACTION_LIMITS.input_bytes)
+    directory = Path(data["directory"])
+    sequence = 0
+    completed = 0
+
+    def report(phase, files, total, count, total_bytes, **member):
+        nonlocal sequence
+        sequence += 1
+        _write_record(directory, "status", {
+            "sequence": sequence, "kind": "member" if member else "progress",
+            "progress": [phase, files, total, count, total_bytes], **member,
+        })
+
+    try:
+        inspection = _inspect_archive_content(data["source"], **data["options"])
+        if inspection.requires_password and not inspection.password_verified:
+            raise ArchivePasswordError("Archive password is required")
+        selected = _selected_file_paths(inspection.entries, data["selected_paths"])
+        if not selected:
+            raise FileStorageError("Select at least one archive file to import")
+        total_bytes = sum(entry.size_bytes for entry in inspection.entries if entry.path in selected)
+        total_limit = min(data["options"]["max_expanded_bytes"],
+                          inspection.compressed_size_bytes * data["options"]["max_expansion_ratio"])
+        arguments = {"selected_files": selected, "max_file_bytes": data["max_file_bytes"],
+                     "max_total_bytes": total_limit, "progress": report, "total_bytes": total_bytes}
+        if inspection.archive_format == "zip":
+            members = _read_selected_zip_members(data["source"], password=data["options"]["password"], **arguments)
+        else:
+            members = _read_selected_tar_members(data["source"], **arguments)
+        actual = 0
+        with closing(members), _ack_channel(directory) as acknowledgement:
+            for inner_path, content in members:
+                completed += 1
+                actual += len(content)
+                member_path = directory / f"member-{completed:08d}"
+                descriptor = os.open(member_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+                with os.fdopen(descriptor, "wb") as output:
+                    output.write(content)
+                report("extracting", completed, len(selected), actual, total_bytes,
+                       path=inner_path, size=len(content), sha256=hashlib.sha256(content).hexdigest())
+                del content
+                # Wake Core's selector after publishing a completed member. One
+                # discarded stderr byte per member fits its 64 KiB ceiling for
+                # the default 10,000-entry cap; intermediate progress never
+                # wakes. Non-default >32,768-member workloads retain ordinary
+                # Core progress polling after the fixed wake budget is spent.
+                if completed <= _MAX_WAKE_BYTES:
+                    os.write(2, b".")
+                # Acknowledgement applies backpressure: the parent stores and
+                # removes this member before we allocate/decode the next one.
+                # Blocking on a bounded FIFO read avoids a fixed delay for each
+                # tiny file. Core still enforces the overall wall/CPU budgets.
+                _receive_ack(acknowledgement, sequence)
+        result = {"completed": completed}
+    except FileStorageError as exc:
+        result = {"error": str(exc), "password_error": isinstance(exc, ArchivePasswordError)}
+    return encode_worker_payload(result, max_bytes=EXTRACTION_LIMITS.output_bytes)

src/govoplan_files/backend/storage/archives.py

@@ -6,7 +6,6 @@ import stat
 import tarfile
 import zipfile
 from dataclasses import dataclass
-from contextlib import closing
 from io import BytesIO
 from os import PathLike
 from pathlib import Path, PurePosixPath
@@ -26,7 +25,6 @@ from govoplan_files.backend.storage.common import (
     UploadedStoredFile,
 )
 from govoplan_files.backend.storage.files import (
-    archive_storage_backend_scope,
     create_file_asset,
     current_versions_and_blobs,
 )
@@ -142,6 +140,22 @@ def inspect_archive(
     max_entries: int = ARCHIVE_UPLOAD_MAX_ENTRIES,
     max_expanded_bytes: int = 2 * 1024 * 1024 * 1024,
     max_expansion_ratio: int = 100,
+) -> ArchiveInspection:
+    from govoplan_files.backend.storage.archive_workers import inspect_archive_isolated
+    return inspect_archive_isolated(
+        archive_data, filename=filename, password=password, max_entries=max_entries,
+        max_expanded_bytes=max_expanded_bytes, max_expansion_ratio=max_expansion_ratio,
+    )
+
+
+def _inspect_archive_content(
+    archive_data: bytes | str | PathLike[str],
+    *,
+    filename: str,
+    password: str | None = None,
+    max_entries: int = ARCHIVE_UPLOAD_MAX_ENTRIES,
+    max_expanded_bytes: int = 2 * 1024 * 1024 * 1024,
+    max_expansion_ratio: int = 100,
 ) -> ArchiveInspection:
     archive_format = archive_format_for_filename(filename)
     compressed_size = _archive_size(archive_data)
@@ -206,68 +220,18 @@ def extract_archive_upload(
     max_expansion_ratio: int = 100,
     progress: ArchiveProgress | None = None,
 ) -> list[UploadedStoredFile]:
-    if progress:
-        progress("inspecting", 0, 0, 0, 0)
-    inspection = inspect_archive(
-        archive_data,
-        filename=filename,
-        password=password,
-        max_entries=max_entries,
-        max_expanded_bytes=max_expanded_bytes,
-        max_expansion_ratio=max_expansion_ratio,
+    from govoplan_files.backend.storage.archive_workers import extract_archive_isolated
+    return extract_archive_isolated(
+        session, archive_data=archive_data, filename=filename, password=password,
+        selected_paths=selected_paths, max_entries=max_entries, max_file_bytes=max_file_bytes,
+        max_expanded_bytes=max_expanded_bytes, max_expansion_ratio=max_expansion_ratio,
+        progress=progress, store_options={
+            "tenant_id": tenant_id, "owner_type": owner_type, "owner_id": owner_id,
+            "user_id": user_id, "folder": folder, "campaign_id": campaign_id,
+            "conflict_strategy": conflict_strategy, "conflict_resolutions": conflict_resolutions,
+            "metadata": metadata, "is_admin": is_admin, "encryption_vault_id": encryption_vault_id,
+        },
     )
-    if inspection.requires_password and not inspection.password_verified:
-        raise ArchivePasswordError("Archive password is required")
-    selected_files = _selected_file_paths(inspection.entries, selected_paths)
-    if not selected_files:
-        raise FileStorageError("Select at least one archive file to import")
-    selected_total_bytes = sum(
-        entry.size_bytes for entry in inspection.entries if entry.path in selected_files
-    )
-    actual_total_limit = min(
-        max_expanded_bytes,
-        inspection.compressed_size_bytes * max_expansion_ratio,
-    )
-    if inspection.archive_format == "zip":
-        members = _read_selected_zip_members(
-            archive_data,
-            selected_files=selected_files,
-            password=password,
-            max_file_bytes=max_file_bytes,
-            max_total_bytes=actual_total_limit,
-            progress=progress,
-            total_bytes=selected_total_bytes,
-        )
-    else:
-        members = _read_selected_tar_members(
-            archive_data,
-            selected_files=selected_files,
-            max_file_bytes=max_file_bytes,
-            max_total_bytes=actual_total_limit,
-            progress=progress,
-            total_bytes=selected_total_bytes,
-        )
-    # Release Python/native archive handles immediately if storage or a
-    # callback fails while the member iterator is suspended at a yield.
-    with closing(members), archive_storage_backend_scope():
-        return _store_archive_members(
-            session,
-            members=members,
-            tenant_id=tenant_id,
-            owner_type=owner_type,
-            owner_id=owner_id,
-            user_id=user_id,
-            folder=folder,
-            campaign_id=campaign_id,
-            conflict_strategy=conflict_strategy,
-            conflict_resolutions=conflict_resolutions,
-            metadata=metadata,
-            is_admin=is_admin,
-            encryption_vault_id=encryption_vault_id,
-            progress=progress,
-            total_files=len(selected_files),
-            total_bytes=selected_total_bytes,
-        )
 
 
 def extract_zip_upload(

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

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(

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,

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",

tests/test_archive_inspection_bounds.py

@@ -4,7 +4,7 @@ import tarfile
 import unittest
 from unittest.mock import patch
 
-from govoplan_files.backend.storage.archives import _safe_member_path, inspect_archive
+from govoplan_files.backend.storage.archives import _inspect_archive_content, _safe_member_path, inspect_archive
 from govoplan_files.backend.storage.common import FileStorageError
 from test_archives import _tar_bytes, _zip_bytes
 
@@ -38,21 +38,21 @@ class ArchiveInspectionBoundTests(unittest.TestCase):
         headers = _HeadersOnly(4)
         with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
             with self.assertRaisesRegex(FileStorageError, "too many entries"):
-                inspect_archive(b"fixture", filename="fixture.tar.gz", max_entries=2)
+                _inspect_archive_content(b"fixture", filename="fixture.tar.gz", max_entries=2)
         self.assertEqual(3, headers.seen)
 
     def test_tar_expanded_size_rejected_before_payload_decompression(self):
         headers = _HeadersOnly(1, size=100)
         with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
             with self.assertRaisesRegex(FileStorageError, "too large after extraction"):
-                inspect_archive(b"fixture", filename="fixture.tar.gz", max_expanded_bytes=10)
+                _inspect_archive_content(b"fixture", filename="fixture.tar.gz", max_expanded_bytes=10)
         self.assertEqual(1, headers.seen)
 
     def test_tar_ratio_rejected_before_payload_decompression(self):
         headers = _HeadersOnly(1, size=100)
         with patch("govoplan_files.backend.storage.archives._open_tar", return_value=headers):
             with self.assertRaisesRegex(FileStorageError, "expansion ratio"):
-                inspect_archive(b"fixture", filename="fixture.tar.gz", max_expansion_ratio=2)
+                _inspect_archive_content(b"fixture", filename="fixture.tar.gz", max_expansion_ratio=2)
 
     def test_derived_directories_are_included_in_entry_limit(self):
         for filename, payload in (("fixture.zip", _zip_bytes({"a/b/c/file.txt": b"x"})), ("fixture.tar.gz", _tar_bytes({"a/b/c/file.txt": b"x"}))):

tests/test_archive_performance.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from contextlib import closing
 from io import BytesIO
 from pathlib import Path
 import random
@@ -204,24 +205,16 @@ class NativeArchivePerformanceTests(unittest.TestCase):
                 "govoplan_files.backend.storage.archives.native_zip_library",
                 return_value=proxy,
             ),
-            patch(
-                "govoplan_files.backend.storage.archives.create_file_asset",
-                side_effect=FileStorageError("Destination failure"),
-            ),
         ):
             with self.assertRaisesRegex(FileStorageError, "Destination failure"):
-                extract_archive_upload(
-                    object(),
-                    tenant_id="tenant",
-                    owner_type="user",
-                    owner_id="user",
-                    user_id="user",
-                    archive_data=self.archive,
-                    filename="archive.zip",
-                    folder="",
-                    campaign_id=None,
-                    password="fixture-only",
-                )
+                # This unit test owns the native handle locally; public imports
+                # now own it in a fresh child (covered by worker cleanup tests).
+                with closing(_read_selected_zip_members(
+                    self.archive, selected_files=set(self.contents), password="fixture-only",
+                    max_file_bytes=100_000, max_total_bytes=100_000,
+                )) as members:
+                    next(members)
+                    raise FileStorageError("Destination failure")
         close.assert_called_once()
 
     def test_native_read_enforces_actual_member_and_total_limits(self):

tests/test_archive_staging.py

@@ -221,7 +221,17 @@ class ArchiveStagingTests(unittest.TestCase):
     def test_progress_persistence_failure_does_not_turn_a_committed_import_into_an_error(self):
         preview = self.preview().json()
         operation_id = str(uuid4())
-        with patch.object(archive_work.os, "replace", side_effect=OSError("Progress write unavailable")):
+        original_replace = archive_work.os.replace
+
+        def replace(source, destination):
+            # Only the optional UI progress receipt is unavailable. The private
+            # worker ack is a required integrity/backpressure channel, not UI
+            # progress, and must not be disabled by this module-global mock.
+            if Path(source).name.startswith(".progress-"):
+                raise OSError("Progress write unavailable")
+            return original_replace(source, destination)
+
+        with patch.object(archive_work.os, "replace", side_effect=replace):
             result = self.confirm(preview, operation_id=operation_id)
         self.assertEqual(result.status_code, 200, result.text)
         self.assertEqual(self.session.query(FileAsset).count(), 2)
@@ -229,6 +239,21 @@ class ArchiveStagingTests(unittest.TestCase):
         # the confirmation response remains the authoritative outcome.
         self.assertEqual(self.progress(operation_id).json()["status"], "running")
 
+    def test_required_worker_ack_failure_rolls_back_and_preserves_upload_for_explicit_retry(self):
+        preview = self.preview().json()
+        stage = self.staged_path(preview)
+        operation_id = str(uuid4())
+        with patch("govoplan_files.backend.storage.archive_workers._send_ack",
+                   side_effect=FileStorageError("Archive worker staging channel is unavailable")):
+            result = self.confirm(preview, operation_id=operation_id)
+        self.assertEqual(400, result.status_code, result.text)
+        self.assertIn("staging channel is unavailable", result.text)
+        self.assert_no_extraction()
+        self.assertEqual("failed", self.progress(operation_id).json()["status"])
+        self.assertTrue(stage.exists())
+        result = self.confirm(preview, operation_id=str(uuid4()))
+        self.assertEqual(200, result.status_code, result.text)
+
     def test_lease_cleanup_failure_cannot_mask_a_committed_import(self):
         preview = self.preview().json()
         operation_id = str(uuid4())

tests/test_archive_workers.py

@@ -0,0 +1,346 @@
+from __future__ import annotations
+
+from dataclasses import replace
+import json
+import inspect
+import os
+from pathlib import Path
+import subprocess
+import tarfile
+import tempfile
+import time
+import tracemalloc
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+import zlib
+
+from govoplan_core.security import bounded_process
+from govoplan_core.settings import settings
+from govoplan_files.backend.storage import archive_workers as workers
+from govoplan_files.backend.storage.archives import extract_archive_upload, inspect_archive
+from govoplan_files.backend.storage.common import FileStorageError
+from test_archives import _zip_bytes
+
+
+def _cpu_exhaustion_worker(payload: bytes) -> bytes:
+    # A real imported child operation proves the Files wrapper's CPU error path;
+    # the metadata tests below execute the actual archive parser, not this probe.
+    while True:
+        pass
+
+
+def _extract(payload, **options):
+    return extract_archive_upload(
+        object(), tenant_id="tenant", owner_type="user", owner_id="owner",
+        user_id="owner", filename="fixture.zip", archive_data=payload,
+        folder="destination", campaign_id=None, **options,
+    )
+
+
+class ArchiveWorkerTests(unittest.TestCase):
+    def setUp(self):
+        self.directories = []
+        self.children = []
+        real_directory = tempfile.TemporaryDirectory
+        real_popen = subprocess.Popen
+
+        def directory(*args, **kwargs):
+            context = real_directory(*args, **kwargs)
+            self.directories.append(Path(context.name))
+            return context
+
+        def popen(*args, **kwargs):
+            child = real_popen(*args, **kwargs)
+            self.children.append(child)
+            return child
+
+        self.enterContext(patch.object(workers.tempfile, "TemporaryDirectory", side_effect=directory))
+        self.enterContext(patch.object(bounded_process.subprocess, "Popen", side_effect=popen))
+        self.store = self.enterContext(patch(
+            "govoplan_files.backend.storage.archives.create_file_asset",
+            return_value=SimpleNamespace(asset=SimpleNamespace(id="asset")),
+        ))
+
+    def tearDown(self):
+        self.assertTrue(all(not directory.exists() for directory in self.directories))
+        self.assertTrue(all(child.returncode is not None for child in self.children))
+
+    def test_parent_never_parses_metadata_or_extracts_and_one_child_handles_all_members(self):
+        payload = _zip_bytes({"one.txt": b"one", "two.txt": b"two"})
+        with (
+            patch("govoplan_files.backend.storage.archives._inspect_archive_content", side_effect=AssertionError("parent parser")),
+            patch("govoplan_files.backend.storage.archives._read_selected_zip_members", side_effect=AssertionError("parent decoder")),
+        ):
+            self.assertEqual(2, inspect_archive(payload, filename="fixture.zip").file_count)
+            self.assertEqual(2, len(_extract(payload)))
+        self.assertEqual(2, len(self.children))  # one preview, one whole extraction
+        self.assertEqual([b"one", b"two"], [call.kwargs["data"] for call in self.store.call_args_list])
+
+    def test_hostile_pax_metadata_allocation_is_confined_before_any_store(self):
+        header = tarfile.TarInfo("pax")
+        header.type = tarfile.XHDTYPE
+        header.size = 1024 * 1024 * 1024
+        payload = header.tobuf() + b"\0" * 1024
+        with patch("govoplan_files.backend.storage.archives._open_tar", side_effect=AssertionError("parent TAR parser")):
+            with self.assertRaisesRegex(FileStorageError, "memory_limit|Invalid TAR|could not complete safely"):
+                inspect_archive(payload, filename="hostile.tar")
+        self.store.assert_not_called()
+        self.assertEqual(1, len(self.children))
+        # A rejected archive neither consumes the admission slot permanently nor
+        # prevents a later ordinary parse in the same parent process.
+        self.assertEqual(1, inspect_archive(_zip_bytes({"ok": b"ok"}), filename="ok.zip").file_count)
+
+    def test_compressed_pax_metadata_hits_real_child_memory_limit(self):
+        header = tarfile.TarInfo("pax")
+        header.type = tarfile.XHDTYPE
+        header.size = 1024 * 1024 * 1024
+        compressor = zlib.compressobj(1, wbits=31)
+        chunks = [compressor.compress(header.tobuf())]
+        block = b"\0" * (1024 * 1024)
+        # Build the hostile fixture incrementally: ~2.3 MiB compressed, never a
+        # 512 MiB parent allocation. TAR consumes PAX metadata before ordinary
+        # returned-member limits can inspect it; the child AS limit must win.
+        for _ in range(512):
+            chunks.append(compressor.compress(block))
+        chunks.append(compressor.flush())
+        payload = b"".join(chunks)
+        self.assertLess(len(payload), 3 * 1024 * 1024)
+        with patch("govoplan_files.backend.storage.archives._open_tar", side_effect=AssertionError("parent TAR parser")):
+            with self.assertRaisesRegex(FileStorageError, "memory_limit"):
+                inspect_archive(payload, filename="hostile.tar.gz")
+        self.store.assert_not_called()
+        self.assertEqual(1, len(self.children))
+
+    def test_real_child_timeout_reaps_and_removes_snapshot(self):
+        with patch.object(workers, "INSPECTION_LIMITS", replace(workers.INSPECTION_LIMITS, wall_seconds=0.001)):
+            with self.assertRaisesRegex(FileStorageError, "timeout"):
+                inspect_archive(_zip_bytes({"one": b"one"}), filename="fixture.zip")
+        self.assertEqual(1, len(self.children))
+        self.store.assert_not_called()
+
+    def test_real_child_cpu_limit_has_sanitized_files_error(self):
+        with (
+            patch.object(workers, "_inspect_worker", _cpu_exhaustion_worker),
+            patch.object(workers, "INSPECTION_LIMITS", replace(workers.INSPECTION_LIMITS, cpu_seconds=1, wall_seconds=10)),
+        ):
+            with self.assertRaisesRegex(FileStorageError, "cpu_limit"):
+                inspect_archive(_zip_bytes({"one": b"one"}), filename="fixture.zip")
+
+    def test_real_preview_output_is_bounded(self):
+        with patch.object(workers, "INSPECTION_LIMITS", replace(workers.INSPECTION_LIMITS, output_bytes=256)):
+            with self.assertRaisesRegex(FileStorageError, "output_limit"):
+                inspect_archive(_zip_bytes({f"file-{index}": b"x" for index in range(20)}), filename="fixture.zip")
+
+    def test_destination_failure_kills_waiting_child_and_preserves_error(self):
+        self.store.side_effect = FileStorageError("Destination failure")
+        with self.assertRaisesRegex(FileStorageError, "Destination failure"):
+            _extract(_zip_bytes({"one": b"one", "two": b"two"}))
+        self.assertEqual(1, self.store.call_count)
+        self.assertEqual(1, len(self.children))
+
+    def test_missing_ack_times_out_and_wrong_ack_fails_in_real_child(self):
+        real_send = workers._send_ack
+        for send, error in ((lambda descriptor, sequence: None, "timeout"),
+                            (lambda descriptor, sequence: real_send(descriptor, sequence + 1), "invalid staging record")):
+            with self.subTest(error=error), patch.object(workers, "_send_ack", side_effect=send):
+                with patch.object(workers, "EXTRACTION_LIMITS", replace(workers.EXTRACTION_LIMITS, wall_seconds=2)):
+                    with self.assertRaisesRegex(FileStorageError, error):
+                        _extract(_zip_bytes({"one": b"one", "two": b"two"}))
+
+    def test_parent_progress_exception_is_not_reclassified_as_transport_error(self):
+        def progress(stage, *counters):
+            if stage == "extracting":
+                raise ValueError("parent progress failure")
+
+        with self.assertRaisesRegex(ValueError, "parent progress failure"):
+            _extract(_zip_bytes({"one": b"one"}), progress=progress)
+        self.store.assert_not_called()
+
+    def test_source_path_snapshot_rejects_growth_and_shrink_without_spawning(self):
+        # A source path is server-owned; it must still not turn into an unbounded
+        # copy when another writer changes it during snapshot creation.
+        for replacement in (b"fixture plus unexpected growth", b"x"):
+            with self.subTest(replacement=replacement), tempfile.TemporaryDirectory() as temporary:
+                source = Path(temporary) / "archive"
+                source.write_bytes(b"fixture")
+                real_fstat = os.fstat
+                changed = False
+
+                def fstat(descriptor):
+                    nonlocal changed
+                    result = real_fstat(descriptor)
+                    if not changed:
+                        changed = True
+                        source.write_bytes(replacement)
+                    return result
+
+                with patch.object(workers.os, "fstat", side_effect=fstat):
+                    with self.assertRaisesRegex(FileStorageError, "source changed"):
+                        inspect_archive(source, filename="fixture.zip")
+        self.assertEqual([], self.children)
+
+    def test_busy_rejects_before_source_snapshot(self):
+        with patch.object(settings, "isolated_process_concurrency", 1), bounded_process.bounded_operation_admission():
+            with patch.object(workers, "_source_stage", side_effect=AssertionError("must admit before copying")):
+                with self.assertRaisesRegex(FileStorageError, "busy"):
+                    inspect_archive(b"fixture", filename="fixture.zip")
+        self.assertEqual([], self.children)
+        self.assertEqual([], self.directories)
+
+    def test_unsupported_format_rejects_before_snapshot(self):
+        with self.assertRaisesRegex(FileStorageError, "Unsupported archive format"):
+            inspect_archive(b"fixture", filename="fixture.exe")
+        self.assertEqual([], self.directories)
+
+    def test_symlinked_staged_member_is_never_read_or_stored(self):
+        real_read = workers._read_regular
+        attacked = False
+
+        def read(path, maximum, **options):
+            nonlocal attacked
+            if path.name.startswith("member-") and not attacked:
+                attacked = True
+                path.unlink()
+                path.symlink_to(path.parent / "source")
+            return real_read(path, maximum, **options)
+
+        with patch.object(workers, "_read_regular", side_effect=read):
+            with self.assertRaisesRegex(FileStorageError, "invalid staging record"):
+                _extract(_zip_bytes({"one": b"one"}))
+        self.assertTrue(attacked)
+        self.store.assert_not_called()
+
+    def test_forged_member_records_reject_before_store(self):
+        real_read = workers._read_regular
+        for mutation in (
+            {"path": "outside/one"}, {"path": "../escape"}, {"size": -1},
+            {"sha256": "incorrect"}, {"sequence": True}, {"unexpected": True},
+            {"progress": ["extracting", 1, 1, 4, 3]},
+        ):
+            with self.subTest(mutation=mutation):
+                self.store.reset_mock()
+
+                def read(path, maximum, **options):
+                    value = real_read(path, maximum, **options)
+                    if path.name == "status":
+                        record = json.loads(value)
+                        if record.get("kind") == "member":
+                            record.update(mutation)
+                            return json.dumps(record).encode()
+                    return value
+
+                with patch.object(workers, "_read_regular", side_effect=read):
+                    with self.assertRaises(FileStorageError):
+                        _extract(_zip_bytes({"selected/one": b"one"}), selected_paths=("selected",))
+                self.store.assert_not_called()
+
+    def test_regressing_sequence_or_changing_totals_fail_before_second_store(self):
+        real_read = workers._read_regular
+        for mutation in ({"sequence": 1}, {"progress": ["extracting", 2, 3, 6, 6]}):
+            with self.subTest(mutation=mutation):
+                self.store.reset_mock()
+
+                def read(path, maximum, **options):
+                    value = real_read(path, maximum, **options)
+                    if path.name == "status":
+                        record = json.loads(value)
+                        if record.get("kind") == "member" and record["progress"][1] == 2:
+                            record.update(mutation)
+                            return json.dumps(record).encode()
+                    return value
+
+                with patch.object(workers, "_read_regular", side_effect=read):
+                    with self.assertRaisesRegex(FileStorageError, "invalid staging record"):
+                        _extract(_zip_bytes({"one": b"one", "two": b"two"}))
+                self.assertEqual(1, self.store.call_count)
+
+    def test_record_reads_are_bounded_and_reject_non_regular_files(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            directory = Path(temporary)
+            status = directory / "status"
+            status.write_bytes(b"x" * (workers._STATUS_BYTES + 1))
+            with self.assertRaises(FileStorageError):
+                workers._read_regular(status, workers._STATUS_BYTES)
+            with self.assertRaises(FileStorageError):
+                workers._read_regular(directory, workers._STATUS_BYTES)
+
+    def test_tiny_member_read_does_not_allocate_the_configured_gibibyte_cap(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            member = Path(temporary) / "member"
+            member.write_bytes(b"x")
+            tracemalloc.start()
+            try:
+                self.assertEqual(b"x", workers._read_regular(member, 2 * 1024 * 1024 * 1024))
+                _current, peak = tracemalloc.get_traced_memory()
+            finally:
+                tracemalloc.stop()
+            self.assertLess(peak, 1024 * 1024)
+
+    def test_atomic_status_replacement_keeps_open_snapshot_valid_but_member_identity_is_strict(self):
+        for atomic_record in (True, False):
+            with self.subTest(atomic_record=atomic_record), tempfile.TemporaryDirectory() as temporary:
+                status = Path(temporary) / "status"
+                status.write_bytes(b"old")
+                replacement = Path(temporary) / "replacement"
+                replacement.write_bytes(b"new")
+                real_fstat = os.fstat
+                replaced = False
+
+                def fstat(descriptor):
+                    nonlocal replaced
+                    info = real_fstat(descriptor)
+                    if not replaced:
+                        replaced = True
+                        os.replace(replacement, status)
+                    return info
+
+                with patch.object(workers.os, "fstat", side_effect=fstat):
+                    if atomic_record:
+                        self.assertEqual(b"old", workers._read_regular(status, 16, atomic_record=True))
+                    else:
+                        with self.assertRaises(FileStorageError):
+                            workers._read_regular(status, 16)
+                self.assertEqual(b"new", status.read_bytes())
+
+    def test_private_ack_channel_is_bounded_and_validates_type_permissions_and_sequence(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            directory = Path(temporary)
+            with workers._ack_channel(directory, create=True) as writer:
+                with workers._ack_channel(directory) as reader:
+                    workers._send_ack(writer, 7)
+                    workers._receive_ack(reader, 7)
+                    workers._send_ack(writer, 8)
+                    with self.assertRaises(FileStorageError):
+                        workers._receive_ack(reader, 9)
+                    with patch.object(workers.os, "write", side_effect=BlockingIOError()):
+                        with self.assertRaisesRegex(FileStorageError, "staging channel is unavailable"):
+                            workers._send_ack(writer, 10)
+            ack = directory / "ack"
+            os.chmod(ack, 0o644)
+            with self.assertRaises(FileStorageError), workers._ack_channel(directory):
+                pass
+            ack.unlink()
+            ack.write_bytes(b"not a FIFO")
+            with self.assertRaises(FileStorageError), workers._ack_channel(directory):
+                pass
+            ack.unlink()
+            ack.symlink_to(directory / "missing")
+            with self.assertRaises(FileStorageError), workers._ack_channel(directory):
+                pass
+
+    def test_tiny_member_import_has_no_per_member_sleep_or_polling_delay(self):
+        self.assertNotIn("sleep(", inspect.getsource(workers._extract_worker))
+        for count in (100, 1000):
+            with self.subTest(count=count):
+                payload = _zip_bytes({f"member-{index}.txt": b"x" for index in range(count)})
+                started = time.monotonic()
+                self.assertEqual(count, len(_extract(payload)))
+                elapsed = time.monotonic() - started
+                # Broad synthetic regression ceiling, not a storage-throughput
+                # promise: 50 ms per member would take >=50 s for 1,000 files.
+                self.assertLess(elapsed, 15)
+
+
+if __name__ == "__main__":
+    unittest.main()

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()

tests/test_manifest_documentation.py

@@ -4,6 +4,8 @@ import unittest
 
 
 STATIC_TOPIC_IDS = {
+    "files.source-identity-provenance",
+    "files.archive-worker-limits",
     "files.configuration-package.managed-storage",
     "files.quick-access-and-product-area",
     "files.search.managed-content",
@@ -45,6 +47,27 @@ 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)
+        self.assertEqual({"admin", "user"}, set(topic.documentation_types))
+        self.assertIn("GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY", topic.configuration_keys)
+        for body in (topic.body, topic.translations["de"]["body"]):
+            for limit in ("120", "90", "512 MiB", "600", "300", "128 MiB", "64 MiB", "16 KiB", "2 GiB", "250 MiB"):
+                self.assertIn(limit, body)
+        self.assertIn("before copying", topic.body)
+        self.assertIn("not a filesystem/network sandbox", topic.body)
+        self.assertIn("keine Dateisystem-/Netzwerk-Sandbox", topic.translations["de"]["body"])
+
     def test_workspace_action_locations_and_read_only_reload_are_documented_in_both_languages(self) -> None:
         topic = self.topic("files.quick-access-and-product-area")
         self.assertEqual({"user", "admin"}, set(topic.documentation_types))

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:

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()

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",

webui/scripts/test-interface-pattern-language.mjs

@@ -39,7 +39,7 @@ assert.match(filesPage, /disabledReason=\{uploadBlocker\}/);
 assert.match(filesPage, /disabledReason=\{deleteBlocker\}/);
 assert.match(filesPage, /<ConfirmDialog[\s\S]*tone="danger"/);
 assert.match(filesPage, /<WorkspaceFrame as="main" height="viewport" surface="plain"/);
-assert.match(filesPage, /const toolbar = <WorkspaceActionBar\s+scope="workspace"\s+variant="collection"\s+refreshable/);
+assert.match(filesPage, /const toolbar = <WorkspaceActionBar\s+title="i18n:govoplan-files\.files\.6ce6c512"\s+titleHelp=\{<DocumentationHelpLink reference=\{FILES_WORKFLOW_DOCUMENTATION\} \/>\}\s+scope="workspace"\s+variant="collection"\s+refreshable/);
 assert.match(filesPage, /\{toolbar\}[\s\S]*className=\{`file-manager-shell/);
 assert.match(filesPage, /const selectionToolbar = <WorkspaceActionBar\s+scope="detail-pane"/);
 assert.match(filesPage, /<Dialog\s+open=\{toolsPanel !== null\}/);

webui/src/features/files/FileConnectorSettingsPanel.tsx

@@ -931,14 +931,14 @@ export default function FileConnectorSettingsPanel({
       <>
       <Card
           title={panelTitle}
+          titleHelp={<DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />}
           actions={
           canWrite ?
           <div className="button-row compact-actions">
-              <DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />
               <Button onClick={() => void loadProfiles()} disabled={loading || saving} disabledReason={loading ? "File connections are already loading." : saving ? "Wait for the current connector change to finish." : undefined}>{loading ? "i18n:govoplan-files.loading.b04ba49f" : "i18n:govoplan-files.reload.cce71553"}</Button>
               <Button variant="primary" onClick={startCreate} disabled={saving} disabledReason={saving ? "Wait for the current connector change to finish." : undefined}><Plus size={16} aria-hidden="true" /> i18n:govoplan-files.new_connection.ac979fe4</Button>
             </div> :
-          <DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />
+          undefined
           }>
 
         <LoadingFrame loading={loading} label="i18n:govoplan-files.loading_file_connections.bd68f224">
@@ -954,14 +954,14 @@ export default function FileConnectorSettingsPanel({
 
       <Card
           title={i18nMessage("i18n:govoplan-files.value_connector_policy.0bf7f53b", { value0: scopeLabel(scopeType) })}
+          titleHelp={<DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />}
           actions={canWrite ?
           <div className="button-row compact-actions">
-            <DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />
             <Button variant="primary" onClick={() => void savePolicyDraft()} disabled={saving || loading} disabledReason={saving ? "Wait for the current connector change to finish." : loading ? "Wait until the effective connector policy has loaded." : undefined}>
               <ShieldCheck size={16} aria-hidden="true" /> {saving ? "i18n:govoplan-files.saving.ae7e8875" : "i18n:govoplan-files.save_policy.77d67ce3"}
             </Button>
           </div> :
-          <DocumentationHelpLink reference={CONNECTOR_DOCUMENTATION} />}>
+          undefined}>
 
         <LoadingFrame loading={loading} label="i18n:govoplan-files.loading_connector_policy.f12ab285">
           <>

webui/src/features/files/FileIntegrityPanel.tsx

@@ -384,6 +384,7 @@ export default function FileIntegrityPanel({ settings, canWrite }: Props) {
     <>
       <AdminPageLayout
         title="File integrity"
+        titleHelp={<DocumentationHelpLink reference={DOCUMENTATION} label="Open Files integrity documentation" />}
         description="Run bounded storage reconciliation and resolve quarantined or unreferenced objects without acting on stale operator state."
         loading={loading}
         error={error}
@@ -401,7 +402,6 @@ export default function FileIntegrityPanel({ settings, canWrite }: Props) {
             <Button variant="primary" onClick={() => setCreateOpen(true)} disabled={!canWrite || busy}>
               <Plus size={16} /> New scan
             </Button>
-            <DocumentationHelpLink reference={DOCUMENTATION} label="Open Files integrity documentation" />
           </>
         )}
       >

webui/src/features/files/FilesPage.tsx

@@ -2421,6 +2421,8 @@ export default function FilesPage({ settings, auth }: {settings: ApiSettings;aut
   }
 
   const toolbar = <WorkspaceActionBar
+    title="i18n:govoplan-files.files.6ce6c512"
+    titleHelp={<DocumentationHelpLink reference={FILES_WORKFLOW_DOCUMENTATION} />}
     scope="workspace"
     variant="collection"
     refreshable
@@ -2430,7 +2432,6 @@ export default function FilesPage({ settings, auth }: {settings: ApiSettings;aut
     helpModuleId="files"
     reloadAction={{ onReload: () => void reloadCurrentView(), loading: reloadingView, state: viewReloadFailed ? "reload-failed" : "current", disabled: busy || connectorSpaceLoading, disabledReason: workingBlocker || (connectorSpaceLoading ? "This connector space is already refreshing." : undefined) }}
     contextActions={<Button onClick={() => setToolsPanel("connections")} disabled={busy} disabledReason={workingBlocker}><Settings2 size={16} aria-hidden="true" /> i18n:govoplan-files.tools.connections</Button>}
-    helpAction={<DocumentationHelpLink reference={FILES_WORKFLOW_DOCUMENTATION} />}
     createAction={<>
       <Button onClick={() => openDialog("create-folder", toolbarTarget())} disabled={Boolean(organizeBlocker)} disabledReason={organizeBlocker}><Plus size={16} aria-hidden="true" /> i18n:govoplan-files.create_folder.97bafaba</Button>
       <Button variant="primary" onClick={() => openDialog("upload", toolbarTarget())} disabled={Boolean(uploadBlocker)} disabledReason={uploadBlocker}><UploadCloud size={16} aria-hidden="true" /> i18n:govoplan-files.upload.8bdf057f</Button>
@@ -2928,12 +2929,11 @@ export default function FilesPage({ settings, auth }: {settings: ApiSettings;aut
       }
 
       {dialog === "upload" &&
-      <FileDialog title={managedArchiveFile ? "i18n:govoplan-files.managed_archive.unpack" : i18nMessage("i18n:govoplan-files.upload_to_value_value.b83a34b4", { value0: activeDialogSpace?.label || "i18n:govoplan-files.files.6ce6c512", value1: activeDialogTarget?.folderPath || "i18n:govoplan-files.root.e96857c5" })} busy={busy || uploadActive} onClose={() => { if (!busy) closeDialog(); }}>
+      <FileDialog title={managedArchiveFile ? "i18n:govoplan-files.managed_archive.unpack" : i18nMessage("i18n:govoplan-files.upload_to_value_value.b83a34b4", { value0: activeDialogSpace?.label || "i18n:govoplan-files.files.6ce6c512", value1: activeDialogTarget?.folderPath || "i18n:govoplan-files.root.e96857c5" })} titleHelp={managedArchiveFile ? <DocumentationHelpLink reference={{ topicId: "files.workflow.unpack-managed-archive", documentationType: "user" }} /> : undefined} busy={busy || uploadActive} onClose={() => { if (!busy) closeDialog(); }}>
         <LoadingFrame loading={uploadActive} label={operationBusyLabel} indicator="none" progress={operationProgressValue} progressLabel={operationProgressLabel}>
           <div inert={uploadActive}>
           {managedArchiveFile && <p className="form-help">{i18nMessage("i18n:govoplan-files.managed_archive.source", { value0: managedArchiveFile.filename })}</p>}
           {managedArchiveFile && <p className="form-help">i18n:govoplan-files.managed_archive.protection</p>}
-          {managedArchiveFile && <DocumentationHelpLink reference={{ topicId: "files.workflow.unpack-managed-archive", documentationType: "user" }} />}
           {error && <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert>}
           {!archivePreview &&
         <>

webui/src/features/files/components/FileManagerComponents.tsx

@@ -195,11 +195,12 @@ export function RenamePreviewList({
 
 }
 
-export function FileDialog({ title, onClose, children, busy = false }: {title: string;onClose: () => void;children: ReactNode;busy?: boolean;}) {
+export function FileDialog({ title, titleHelp, onClose, children, busy = false }: {title: string;titleHelp?: ReactNode;onClose: () => void;children: ReactNode;busy?: boolean;}) {
   return (
     <Dialog
       open
       title={title}
+      titleHelp={titleHelp}
       onClose={onClose}
       closeDisabled={busy}
       closeOnBackdrop={!busy}