fix(connectors): retain exact CSV source evidence

Release v0.1.27. Coordinated integrity review: GovOPlaN/govoplan-core#298.

README.md

@@ -117,3 +117,21 @@ See:
 - [Governed connector configuration](docs/GOVERNED_CONNECTOR_CONFIGURATION.md)
 - [MediaWiki and BlueSpice connector](docs/MEDIAWIKI_BLUESPICE_CONNECTOR.md)
 - [Znuny and OTRS-compatible service-desk connector](docs/ZNUNY_OTRS_CONNECTOR.md)
+
+## Git-source WebUI package
+
+The repository root exposes `@govoplan/connectors-webui` for Git-tagged release
+dependencies. It mirrors the owning `webui/package.json` version, public
+TypeScript/CSS exports and peer requirements, with entry paths under
+`webui/src`. Consumers provide the shared Core/React peers; the facade runs no
+development or install scripts. The source archive contains `webui/src`, this
+README and any repository license file. Run module development checks from `webui/`; Python
+installation remains governed by `pyproject.toml`.
+
+Das Repository stellt `@govoplan/connectors-webui` am Wurzelpfad für versionierte
+Git-Abhängigkeiten bereit. Version, öffentliche TypeScript-/CSS-Exporte und
+Peer-Anforderungen entsprechen `webui/package.json`; die Einstiegspfade liegen
+unter `webui/src`. Gemeinsame Core-/React-Peers stellt die einbindende Anwendung
+bereit. Die Fassade führt keine Entwicklungs- oder Installationsskripte aus.
+Entwicklungsprüfungen bleiben in `webui/`, die Python-Installation weiterhin in
+`pyproject.toml` definiert.

docs/CONNECTOR_SOURCE_LIFECYCLE.md

@@ -132,6 +132,25 @@ explicit refresh. Missing Files capability, revoked file access, quarantine,
 oversized or malformed content, inactive/stale credentials, unreachable SQL,
 and timeout failures produce sanitized unavailable/validation diagnostics.
 
+XLSX archive checks and workbook parsing run in a fresh disposable Core worker.
+Each invocation allows 15 seconds wall time, 10 CPU seconds and 512 MiB virtual
+address space, with no file output. The typed transport allows 8 MiB input and
+64 MiB result bytes, at most 64 nesting levels and 1,000,000 value nodes. These
+transport bounds include serialization overhead. The existing workbook bounds
+remain 5,000,000 raw bytes, 50,000,000 expanded bytes, 5,000 archive entries,
+100:1 compression ratio, 500 columns and 10,000 row positions after the header.
+Authorization and exact-version file reads happen in the parent; credentials,
+SQL sessions and durable changes are never passed to the parser.
+
+Users receive an explicit failure rather than a partial source when resource
+or transport limits are exceeded; reduce workbook size or complexity before
+retrying. The Core `GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY` setting limits active work per
+API/worker process without queuing. Busy capacity may be retried later. Missing
+POSIX resource controls, cancellation and worker failure produce sanitized
+unavailable diagnostics; there is no in-process fallback. Operators must keep
+the Core worker API available and account for the aggregate memory of all
+active worker slots across API/worker replicas.
+
 All three current providers declare projection and pagination pushdown only.
 Filters, aggregations, and sorting remain in Dataflow until an adapter explicitly
 declares and tests those operations.

package.json

@@ -0,0 +1,31 @@
+{
+  "name": "@govoplan/connectors-webui",
+  "version": "0.1.27",
+  "private": true,
+  "type": "module",
+  "main": "webui/src/index.ts",
+  "module": "webui/src/index.ts",
+  "types": "webui/src/index.ts",
+  "exports": {
+    ".": {
+      "types": "./webui/src/index.ts",
+      "import": "./webui/src/index.ts"
+    },
+    "./styles/connectors.css": "./webui/src/styles/connectors.css"
+  },
+  "peerDependencies": {
+    "@govoplan/core-webui": "^0.1.18",
+    "react": ">=19.2.7 <20",
+    "react-dom": ">=19.2.7 <20"
+  },
+  "peerDependenciesMeta": {
+    "@govoplan/core-webui": {
+      "optional": true
+    }
+  },
+  "files": [
+    "webui/src",
+    "README.md",
+    "LICENSE"
+  ]
+}

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-connectors"
-version = "0.1.22"
+version = "0.1.27"
 description = "Governed connector catalogue and tabular source capabilities for GovOPlaN."
 readme = "README.md"
 requires-python = ">=3.12"
@@ -12,7 +12,7 @@ license = "AGPL-3.0-or-later"
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
   "defusedxml>=0.7,<1",
-  "govoplan-core>=0.1.33",
+  "govoplan-core>=0.1.46",
   "openpyxl>=3.1.5,<4",
 ]
 

src/govoplan_connectors/backend/db/models.py

@@ -47,6 +47,8 @@ class ConnectorTabularSource(Base, TimestampMixin):
     row_count: Mapped[int] = mapped_column(Integer, nullable=False)
     byte_count: Mapped[int] = mapped_column(Integer, nullable=False)
     metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict, nullable=False)
+    # Original upload content is deliberately excluded from ordinary catalogue loads/DTOs.
+    csv_source_: Mapped[dict[str, Any] | None] = mapped_column("csv_source", JSON, nullable=True, deferred=True)
     created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
     updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
     deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)

src/govoplan_connectors/backend/german_documentation.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+from govoplan_core.core.modules import DocumentationTopic, localize_documentation_topics as _localize_topics
+
+
+_TRANSLATIONS = {
+    "connectors.data-subject-requests": {
+        "title": "Datenschutzanfragen zu Connector-Daten",
+        "summary": "Nachvollziehbare Connector-Aktivitäten exportieren, ohne Zugangsdaten oder externe Inhalte offenzulegen.",
+        "body": (
+            "Connectors gleicht ausschließlich eine eindeutige, mandantenbezogene Konto-ID ab und kann eine bereits verifizierte Suche auf eine Quelle, einen Abruf, eine Definition, Konfiguration, Simulation, ein externes Wissens- oder Service-Desk-Profil oder einen Connector-Vorgang begrenzen. Der Export weist Konfigurations-, Abruf-, Simulations-, externe Vorgangs- und Prüfaktivitäten anhand begrenzter Lebenszyklusmetadaten aus. Zugangsdaten, Endpunktreferenzen, Quelldatensätze, externe Antworten, Anfrageinhalte, Mapping- oder Konfigurationsdokumente, Diagnosen, Provenienz, Hashwerte und Transportnachweise werden nie ausgegeben. Die Zuordnung zu handelnden Personen bleibt unveränderlicher Governance- und Vorgangsnachweis und wird nicht automatisch gelöscht. E-Mail- oder Objektkennungen ohne verifizierte Konto-ID begründen keinen Treffer."
+        ),
+    },
+    "connectors.governed-configuration": {
+        "title": "Connector-Definitionen und Simulationen steuern",
+        "summary": "Connector-Schemata und Mappings versionieren und dabei lokale Überschreibungen sowie Prüfnachweise erhalten.",
+        "body": (
+            "Connector-Administrationen erstellen paketverwaltete oder lokale Definitionen, die Anbieter, Protokoll, Fähigkeiten, Schemata, Mapping-Regeln, Validierung, Vorschau, Audit-Anforderungen, Datenschutz, Aufbewahrung, Grenzen und Wiederholungsverhalten ausdrücklich beschreiben. Jede Änderung erzeugt eine unveränderliche Revision. Eine Mandantenkonfiguration bindet genau eine Revision und speichert nur eine Zugangsdatenreferenz; Paketaktualisierungen ändern die wirksame Konfiguration erst nach bewusster Übernahme. Lokale Überschreibungen werden als geschützt angezeigt und bei der Übernahme erneut angewendet. Testläufe und Simulationen sind begrenzt, schwärzen konfigurierte Felder, sind über den Aufrufschlüssel idempotent und bewahren Konfigurations-, Mapping-, Eingabe- und externe Revisionsprovenienz. Mehrdeutige Ergebnisse folgen der Richtlinie für manuelle Prüfung, Quarantäne oder Ablehnung. Offene und quarantänisierte Nachweise müssen mit Begründung freigegeben oder abgelehnt werden. Eine erfolgreiche generische Simulation verspricht keinen anbieterspezifischen Live-Schreibzugriff."
+        ),
+    },
+    "connectors.authority-and-effects": {
+        "title": "Autorität und Wirkungen eines Connectors verstehen",
+        "summary": "Richtung, technische Reife und konfigurierte Quellenautorität getrennt und sichtbar behandeln.",
+        "body": (
+            "Ein Connector kann Daten beziehen, veröffentlichen oder bidirektional arbeiten und von reiner Erkennung bis zum vollständigen Ersatz reifen. Jede Bindung legt getrennt fest, ob GovOPlaN führend ist, einer externen Autorität folgt, einen Spiegel führt, nach Konfliktregeln synchronisiert, eine Governance-Schicht ergänzt oder nur einen Verweis bewahrt. Schreibende Anbieter müssen Revisionen, Grenzen, Idempotenz, unbekannte Ergebnisse, Nachweise, Abgleich, Korrektur, Ausfallverhalten und Anforderungen an Geheimnisse erläutern."
+        ),
+    },
+    "connectors.runtime-preview-contract": {
+        "title": "Connector-Vorschauen und Diagnosen",
+        "summary": "Für externe Transporte ein einheitliches, begrenztes und geschwärztes Testlaufformat verwenden.",
+        "body": (
+            "Connectors verantwortet Endpunkterkennung, Authentifizierungsübergabe, Transportgrenzen, Wiederholungen und Protokollzustand. Fachmodule verantworten Feldzuordnung, Validierung, Abgleich und Datensatzänderungen. Der gemeinsame Core-Laufzeitvertrag meldet geschwärzte Wirkungen und Diagnosen zusammen mit Quellrevisionen, Fingerabdrücken und unveränderlichen Eingabe-Hashes. Tabellarische Vorschauen begrenzen Zeilen, serialisierte Bytes und Laufzeit und melden Abschneidungen strukturiert. Eine Übernahme muss veraltete, abgeschnittene, widersprüchliche oder fehlerhafte Vorschauen ablehnen; Zugangsdaten erscheinen weder in URLs noch in Beispielen."
+        ),
+    },
+    "connectors.tabular-sources": {
+        "title": "Gesteuerte tabellarische Quellen",
+        "summary": "Anbieterneutrale Quellenerkennung und begrenzte Lesezugriffe für Dataflow bereitstellen.",
+        "body": (
+            "Connectors verantwortet Quellkonfiguration, Zugriffsprüfung, Schemaerkennung, Fingerabdrücke und begrenzte Lesevorgänge. Dataflow speichert nur undurchsichtige Quellreferenzen und erwartete Fingerabdrücke. Jede Quelle weist ihren Live-, Cache-, Datei- oder statischen Modus, einen strukturierten Zustand und unterstützte Projektion, Filterung, Aggregation, Sortierung und Seitennavigation aus. Unveränderliche JSON- und CSV-Snapshots bleiben verfügbar. Verwaltete CSV- und XLSX-Quellen nutzen optional Files, binden eine exakt autorisierte Version, erzwingen Archiv- und Entpackgrenzen und übernehmen neuere Versionen erst nach ausdrücklicher Aktualisierung. "
+            "XLSX-Lesevorgänge prüfen die tatsächlichen Koordinaten des ausgewählten Arbeitsblatts vor dem Aufbau des Zellrasters: höchstens 500 Spalten und 10.000 Zeilenpositionen nach der Kopfzeile einschließlich leerer Zwischenräume. Unzuverlässige Dimensionsangaben vergrößern weder das Raster noch verbergen sie Zellen; übergroße oder widersprüchliche Koordinaten führen zu einem Validierungsfehler statt zu still abgeschnittenen Daten. "
+            "Der PostgreSQL-Adapter nutzt eine aktive gesteuerte Konfiguration und eine eingegrenzte Core-Zugangsdatenhülle, liest nur einfache Schema- und Tabellenkennungen und blockiert bei Konfigurations-, Zugangsdaten- oder Schemadrift bis zur geprüften Aktualisierung. Zugangsdaten, Endpunkte, Speicherschlüssel und interne Dateiinhalte werden nie über die Quelle offengelegt."
+        ),
+    },
+    "connectors.sanctions-snapshots": {
+        "title": "Snapshots von Sanktionsquellen",
+        "summary": "Unveränderliche, per Prüfsumme verifizierbare Sanktionslistennachweise abrufen, ohne Prüfsachverhalte zu übertragen.",
+        "body": (
+            "Connectors stellt eine deterministische synthetische Testquelle und die offizielle konsolidierte XML-Liste des Sicherheitsrats der Vereinten Nationen bereit. Jeder Abruf zeichnet bedingte Transportnachweise, begrenzte Wiederholungen, Zustand, Quellmetadaten, Rohbeleg und SHA-256-Prüfsumme auf. Vor Anbieterzugriffen erwirbt die Aktualisierung eine verteilte Recovery-Sperre; unveränderlicher Snapshot und abschließender Recovery-Prüfpunkt werden anschließend in einer Transaktion gespeichert. Derselbe Anfrageschlüssel liefert dasselbe Ergebnis. Risk Compliance verantwortet Normalisierung, Abgleich, rechtliche Prüfung und Entscheidungen."
+        ),
+    },
+    "connectors.rss-atom": {
+        "title": "RSS- und Atom-Feeds",
+        "summary": "Gesteuerte Feed-Snapshots importieren und nach Sichtbarkeit gefilterte Feeds ausgeben.",
+        "body": (
+            "Connectors verantwortet begrenzten, gegen SSRF geschützten RSS-/Atom-Transport und XML-Verarbeitung. Importierte Einträge werden zu unveränderlichen tabellarischen Snapshots in Datasources und enthalten Abruf, Aktualität, ETag, Inhaltsdigest und Quellprovenienz. Ausgaben akzeptieren nur provenienzbelegte Ereignis-, Veröffentlichungs-, Fall- oder Berichtsauswahlen einer verantwortlichen Oberfläche. Die Zielgruppe setzt die Sichtbarkeitsobergrenze: Öffentliche Feeds enthalten nur öffentliche Einträge; mandantenbezogene oder private Feeds benötigen eine eigene Berechtigung. Aufrufende Stellen dürfen keine eigene Sichtbarkeitsliste vorgeben. Portal oder Reporting verantwortet dauerhafte Veröffentlichungsrouten und autorisiert jeden Zugriff auf eingeschränkte Feeds neu. Ein eigenes RSS-Modul ist erst erforderlich, wenn GovOPlaN später eine eigenständige Feed-Reader-Oberfläche benötigt."
+        ),
+    },
+}
+
+
+def localize_documentation_topics(
+    topics: Iterable[DocumentationTopic],
+) -> tuple[DocumentationTopic, ...]:
+    return _localize_topics(topics, locale="de", translations=_TRANSLATIONS)

src/govoplan_connectors/backend/german_structured_documentation.py

@@ -0,0 +1,77 @@
+"""German translations for public structured documentation metadata."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+GERMAN_STRUCTURED_TRANSLATIONS: dict[str, dict[str, Any]] = {'connectors.data-subject-requests': {'consequence_classes': {'exclude_connector_secrets': 'Gibt '
+                                                                                           'niemals '
+                                                                                           'Anmeldeinformationen, '
+                                                                                           'Endpunkte, '
+                                                                                           'externe '
+                                                                                           'Zeilen '
+                                                                                           'oder '
+                                                                                           'Beweisnutzlasten '
+                                                                                           'zurück.',
+                                                              'export_operator_attribution': 'Retouren '
+                                                                                             'minimierte '
+                                                                                             'Connector-Aktivität '
+                                                                                             'für '
+                                                                                             'das '
+                                                                                             'genaue '
+                                                                                             'Konto.',
+                                                              'retain_connector_evidence': 'Bewahrt '
+                                                                                           'die '
+                                                                                           'Konfiguration '
+                                                                                           'und '
+                                                                                           'die '
+                                                                                           'Rechenschaftspflicht '
+                                                                                           'für '
+                                                                                           'externe '
+                                                                                           'Operationen '
+                                                                                           'bei.'}},
+ 'connectors.governed-configuration': {'outcome': 'Das aktive Konnektorverhalten ist inspizierbar, '
+                                                  'versionengebunden, testbar und überprüfbar, '
+                                                  'bevor ein anbieterspezifisches Schreiben '
+                                                  'erfolgt.',
+                                       'prerequisites': ['Eine Konnektordefinition wurde '
+                                                         'installiert oder erstellt.',
+                                                         'Anmeldematerial wird außerhalb der '
+                                                         'Connector-URL gespeichert und durch eine '
+                                                         'genehmigte geheime Kennung '
+                                                         'referenziert.'],
+                                       'verification': 'Laden Sie die Konfiguration neu, prüfen '
+                                                       'Sie geschützte Pfade und effektiven Hash, '
+                                                       'führen Sie eine Simulation mit einem neuen '
+                                                       'idempotency-Schlüssel aus und lösen Sie '
+                                                       'alle ausstehenden Überprüfungsergebnisse.'},
+ 'connectors.mediawiki-bluespice': {'outcome': 'Externes Wissen bleibt identitätsstabil, '
+                                               'verlustsichtbar, ACL-sicher und migrationsfähig.',
+                                    'prerequisites': ['Eine aktive verwaltete MediaWiki Action '
+                                                      'API-Konfiguration existiert.',
+                                                      'Namespace-Ziele und Fallback-ACLs wurden '
+                                                      'überprüft.',
+                                                      'Die Such- und Wiki-Module sind optional und '
+                                                      'bleiben fähigkeitsgetrennt.'],
+                                    'verification': 'Entdecken Sie das Profil neu, führen Sie ein '
+                                                    'Delta mit Schlüsseln aus, inspizieren Sie '
+                                                    'Gesundheit und Diagnose, überprüfen Sie einen '
+                                                    'erlaubten und verweigerten Suchprinzipal und '
+                                                    'führen Sie einen Migrations-Dry-Run aus, '
+                                                    'bevor Sie zielseitig arbeiten.'},
+ 'connectors.znuny-otrs': {'outcome': 'Externe Tickets bleiben identitätsstabil, verlustsichtbar, '
+                                      'ACL-sicher, wiederherstellbar und semantisch getrennt von '
+                                      'GovOPlaN-Domäneneinträgen.',
+                           'prerequisites': ['Eine aktiv gesteuerte Znuny/OTRS GenericInterface '
+                                             'REST-Konfiguration existiert.',
+                                             'Bereitstellungsdefinierte Routen, '
+                                             'Warteschlangenpartitionen, Autorität und '
+                                             'Fallback-ACLs wurden überprüft.',
+                                             'Tickets, Helpdesk, Cases und Search bleiben '
+                                             'optionale funktionsgetrennte Verbraucher.'],
+                           'verification': 'Entdecke das Profil neu, beende einen Keyed-Full-Run, '
+                                           'führe ein Keyed-Delta aus, inspiziere die '
+                                           'Mapping-Diagnose, überprüfe einen erlaubten und '
+                                           'verweigerten Suchprinzipal und versöhne jedes '
+                                           'ergebnisunbekannte Update vor dem erneuten Versuch.'}}

src/govoplan_connectors/backend/knowledge_search.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from govoplan_connectors.backend.search_principal import principal_acl_tokens as _principal_acl_tokens
+
 from collections.abc import Mapping, Sequence
 from urllib.parse import quote
 
@@ -217,30 +219,6 @@ def search_document(
     )
 
 
-def _principal_acl_tokens(principal: object) -> tuple[str, ...]:
-    values: list[str] = []
-    for prefix, attribute in (
-        ("account", "account_id"),
-        ("membership", "membership_id"),
-        ("identity", "identity_id"),
-    ):
-        value = getattr(principal, attribute, None)
-        if value:
-            values.append(f"{prefix}:{value}")
-    for prefix, attribute in (
-        ("group", "group_ids"),
-        ("role", "role_ids"),
-        ("function", "function_assignment_ids"),
-        ("scope", "scopes"),
-    ):
-        values.extend(
-            f"{prefix}:{value}"
-            for value in getattr(principal, attribute, ())
-            if value
-        )
-    return tuple(dict.fromkeys(values))[:500]
-
-
 def _has_scope(principal: object, required: str) -> bool:
     check = getattr(principal, "has", None)
     if callable(check):

src/govoplan_connectors/backend/manifest.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+from govoplan_core.core.modules import with_documentation_structured_translations
+from govoplan_connectors.backend.german_structured_documentation import GERMAN_STRUCTURED_TRANSLATIONS
+
 from pathlib import Path
 
 from govoplan_core.core.access import (
@@ -14,6 +17,7 @@ from govoplan_core.core.datasources import CAPABILITY_DATASOURCE_ORIGINS
 from govoplan_core.core.feeds import CAPABILITY_CONNECTORS_FEEDS
 from govoplan_core.core.modules import (
     CapabilityDocumentation,
+    DocumentationCondition,
     DocumentationTopic,
     FrontendModule,
     MigrationSpec,
@@ -116,10 +120,13 @@ from govoplan_connectors.backend.provider_state import (
     service_desk_provider_states,
     tabular_provider_states,
 )
+from govoplan_connectors.backend.german_documentation import (
+    localize_documentation_topics,
+)
 
 
 MODULE_ID = "connectors"
-MODULE_VERSION = "0.1.22"
+MODULE_VERSION = "0.1.27"
 TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
 DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
 SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0"
@@ -959,7 +966,39 @@ manifest = ModuleManifest(
             label="Connectors",
         ),
     ),
-    documentation=(
+    documentation=localize_documentation_topics((
+        DocumentationTopic(
+            id="connectors.csv-source-fidelity",
+            title="Preserving original CSV input",
+            summary="Choose explicit text or legacy type inference, and retrieve the verified original of new CSV snapshots.",
+            body=(
+                "CSV snapshot and managed-file creation accept csv_value_mode=text to preserve field whitespace, decimal digits, large identifiers, boolean-looking text and explicit empty records as strings. "
+                "Text mode rejects missing/extra fields and malformed quoting rather than dropping values. Headers remain normalized and blank physical lines are not table rows; the original upload preserves these lexical details. "
+                "Omitting csv_value_mode keeps legacy_typed API behavior for existing clients. Legacy inference can trim values, infer numbers/booleans and omit empty rows; choose text when exact values matter. "
+                "New CSV snapshots retain the original supplied text, delimiter, parser profile and UTF-8 SHA-256 separately from catalogue responses; csv_source metadata is reserved verified evidence, not a place to supply content. "
+                "GET /api/v1/connectors/tabular-sources/{source_ref}/original-csv requires current connectors:source:read or connectors:source:admin authority in the same tenant, verifies the checksum, audits the export without source content, and returns an uncached attachment. "
+                "The export preserves the text submitted to the API as UTF-8, not an earlier file encoding. It does not sanitize spreadsheet formulas: treat imported originals as untrusted data. "
+                "Original text and parsed rows each remain limited to 5 MB, with at most 10,000 parsed rows. Each snapshot can retain up to 5 MB of original text, plus serialization and metadata storage overhead. Source text follows the snapshot lifecycle; retired/deleted sources are not downloadable. "
+                "Managed files retain their original through the exact Files version and its access/lifecycle policy; text mode is pinned in source metadata and preserved on refresh. Existing snapshot fingerprints and historical rows are never rewritten. "
+                "Original text cannot be reconstructed for older snapshots: the endpoint reports it unavailable. Schema inference now shares one ordered, bounded-state mechanism with Datasources while preserving legacy provider type classifications."
+            ),
+            layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=8,
+            translations={"de": {
+                "title": "Ursprüngliche CSV-Eingaben erhalten",
+                "summary": "Textwerte oder bisherige Typableitung ausdrücklich wählen und das geprüfte Original neuer CSV-Snapshots abrufen.",
+                "body": (
+                    "CSV-Snapshots und verwaltete Dateiquellen unterstützen csv_value_mode=text. Dieser Modus erhält Leerzeichen in Feldwerten, Dezimalstellen, große Kennungen, boolesch wirkenden Text und ausdrücklich leere Datensätze als Zeichenketten. "
+                    "Fehlende oder zusätzliche Felder und fehlerhafte Anführungszeichen werden abgelehnt. Überschriften werden weiterhin normalisiert; vollständig leere physische Zeilen sind keine Tabellenzeilen. Das Original erhält auch diese Texteigenschaften. "
+                    "Ohne csv_value_mode bleibt für bestehende API-Aufrufe legacy_typed aktiv. Dabei können Werte gekürzt, Zahlen/Wahrheitswerte abgeleitet und leere Zeilen ausgelassen werden. Für genaue Werte wählen Sie Text. "
+                    "Neue CSV-Snapshots speichern den gelieferten Originaltext, Trennzeichen, Parserprofil und UTF-8-SHA-256 getrennt vom Katalog. csv_source-Metadaten sind reservierter geprüfter Nachweis. "
+                    "GET /api/v1/connectors/tabular-sources/{source_ref}/original-csv benötigt aktuelle Rechte connectors:source:read oder connectors:source:admin im selben Mandanten, prüft die Prüfsumme, protokolliert den Export ohne Quellinhalt und liefert einen nicht zwischengespeicherten Download. "
+                    "Exportiert wird der an die API übermittelte Text als UTF-8, nicht eine frühere Dateikodierung. Tabellenformeln werden nicht verändert; behandeln Sie Originaldateien als nicht vertrauenswürdige Daten. "
+                    "Originaltext und verarbeitete Zeilen sind jeweils auf 5 MB begrenzt; höchstens 10.000 Zeilen werden angenommen. Pro Snapshot werden bis zu 5 MB Originaltext zuzüglich Speicher für Serialisierung und Metadaten aufbewahrt. Der Originaltext folgt dem Snapshot-Lebenszyklus; stillgelegte oder gelöschte Quellen sind nicht abrufbar. "
+                    "Verwaltete Dateien behalten ihr Original in der genauen Files-Version mit deren Zugriffs- und Lebenszyklusregeln; der Textmodus bleibt bei Aktualisierungen erhalten. Bestehende Fingerprints und historische Zeilen werden nicht umgeschrieben. "
+                    "Für ältere Snapshots kann das Original nicht rekonstruiert werden; der Abruf meldet es als nicht verfügbar. Die gemeinsame Schemaableitung erhält die bisherigen Typregeln und Spaltenreihenfolge."
+                ),
+            }},
+        ),
         DocumentationTopic(
             id="connectors.data-subject-requests",
             title="Connector data-subject requests",
@@ -1014,7 +1053,7 @@ manifest = ModuleManifest(
             related_modules=("policy", "audit", "dataflow", "ops"),
             order=38,
             metadata={
-                "kind": "guide",
+                "kind": "workflow",
                 "help_contexts": ["connectors.admin.governed-configurations"],
                 "prerequisites": [
                     "A connector definition has been installed or authored.",
@@ -1023,6 +1062,12 @@ manifest = ModuleManifest(
                 "outcome": "The active connector behavior is inspectable, version-pinned, testable, and reviewable before any provider-specific write.",
                 "verification": "Reload the configuration, inspect protected paths and effective hash, run a simulation with a new idempotency key, and resolve any pending review result.",
             },
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("connectors",),
+                    required_scopes=(ADMIN_SCOPE,),
+                ),
+            ),
         ),
         DocumentationTopic(
             id="connectors.authority-and-effects",
@@ -1038,6 +1083,38 @@ manifest = ModuleManifest(
             audience=("operator", "module_admin", "power_user", "product_owner"),
             related_modules=("datasources", "dataflow", "ops", "policy", "audit"),
             order=39,
+            metadata={
+                "kind": "reference",
+                "help_contexts": [
+                    "connectors.action.reject-ambiguous-result",
+                    "connectors.action.approve-ambiguous-result",
+                ],
+                "fields": [
+                    "Direction and technical maturity",
+                    "Configured source authority",
+                    "Revision and idempotency behavior",
+                    "Reconciliation, outage, and secret requirements",
+                ],
+                "consequences": [
+                    "The authority mode determines which side may change business state.",
+                    "A writable provider must preserve evidence and reconcile unknown outcomes before retry.",
+                ],
+            },
+            structured_translation_version="1",
+            structured_translations={
+                "de": {
+                    "fields": [
+                        "Richtung und technische Reife",
+                        "Konfigurierte Quellenautorität",
+                        "Revisions- und Idempotenzverhalten",
+                        "Anforderungen an Abgleich, Ausfallverhalten und Geheimnisse",
+                    ],
+                    "consequences": [
+                        "Der Autoritätsmodus bestimmt, welche Seite den Fachzustand ändern darf.",
+                        "Ein schreibender Anbieter muss Nachweise bewahren und unbekannte Ergebnisse vor einer Wiederholung abgleichen.",
+                    ],
+                }
+            },
         ),
         DocumentationTopic(
             id="connectors.runtime-preview-contract",
@@ -1070,7 +1147,9 @@ manifest = ModuleManifest(
                 "Immutable JSON/CSV snapshots remain available. Managed CSV/XLSX "
                 "sources use the optional Files capability, pin an exact authorized "
                 "version, apply archive and expansion limits, and require explicit "
-                "refresh before adopting a newer version. The PostgreSQL adapter uses "
+                "refresh before adopting a newer version. XLSX reads validate actual selected-worksheet coordinates before grid allocation: "
+                "at most 500 columns and 10,000 row positions after the header, including blank gaps. Unreliable declared dimensions "
+                "neither expand the grid nor conceal cells; oversized or inconsistent coordinates fail validation instead of truncating data. The PostgreSQL adapter uses "
                 "an active governed configuration and scoped Core credential envelope, "
                 "reflects simple schema/table identifiers, runs read-only bounded "
                 "projection and pagination, and blocks configuration, credential, or "
@@ -1083,6 +1162,44 @@ manifest = ModuleManifest(
             related_modules=("dataflow", "files", "reporting", "risk_compliance"),
             order=40,
         ),
+        DocumentationTopic(
+            id="connectors.xlsx-worker-limits",
+            title="XLSX processing resource limits",
+            summary="Understand isolated workbook parsing and explicit capacity failures.",
+            body=(
+                "Authorized XLSX bytes are parsed in a fresh disposable process with a 15-second wall limit, 10 CPU seconds, "
+                "512 MiB address-space limit, no file output, and 8 MiB input/64 MiB result transport ceilings. Existing limits remain "
+                "5,000,000 input bytes, 50,000,000 expanded archive bytes, 5,000 archive entries, 100:1 expansion ratio, "
+                "500 columns and 10,000 row positions after the header. Typed transport additionally limits nesting to 64 levels "
+                "and 1,000,000 value nodes. Limit failures reject the whole parse; reduce the workbook before retrying. "
+                "The shared Core GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY capacity is per API/worker process and does not queue: busy work "
+                "fails explicitly and may be retried later. POSIX resource controls are required; missing controls, cancellation "
+                "or worker failure produce sanitized unavailable diagnostics. No in-process fallback occurs. File authorization, "
+                "credentials, SQL sessions and durable source changes remain in the parent. Operators must budget aggregate "
+                "memory across process slots and keep the required Core worker API available."
+            ),
+            layer="static",
+            documentation_types=("user", "admin"),
+            audience=("user", "module_admin", "operator"),
+            translations={"de": {
+                "title": "Ressourcengrenzen der XLSX-Verarbeitung",
+                "summary": "Isoliertes Einlesen von Arbeitsmappen und ausdrückliche Kapazitätsfehler verstehen.",
+                "body": (
+                    "Autorisierte XLSX-Bytes werden in einem neuen kurzlebigen Prozess verarbeitet: höchstens 15 Sekunden Gesamtdauer, "
+                    "10 CPU-Sekunden, 512 MiB Adressraum, keine Dateiausgabe und 8 MiB Eingabe-/64 MiB Ergebnistransport. Weiterhin gelten "
+                    "5.000.000 Eingabebytes, 50.000.000 entpackte Archivbytes, 5.000 Archiveinträge, ein Entpackverhältnis von 100:1, "
+                    "500 Spalten und 10.000 Zeilenpositionen nach der Kopfzeile. Der typisierte Transport begrenzt zusätzlich die "
+                    "Verschachtelung auf 64 Ebenen und 1.000.000 Wertknoten. Grenzverletzungen lehnen den gesamten Lesevorgang ab; "
+                    "verkleinern Sie die Arbeitsmappe vor einem erneuten Versuch. Die gemeinsame Core-Einstellung GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY "
+                    "gilt pro API-/Worker-Prozess und bildet keine Warteschlange: Bei belegter Kapazität erfolgt ein ausdrücklicher Fehler; "
+                    "versuchen Sie es später erneut. POSIX-Ressourcenbegrenzungen sind erforderlich. Fehlende Kontrollen, Abbruch oder "
+                    "Worker-Fehler melden eine bereinigte Nichtverfügbarkeit. Es gibt keinen Rückfall auf Verarbeitung im Elternprozess. "
+                    "Dateiberechtigungen, Zugangsdaten, SQL-Sitzungen und dauerhafte Quelländerungen bleiben im Elternprozess. Betreiber "
+                    "müssen den Gesamtspeicher aller Prozessplätze berücksichtigen und die benötigte Core-Worker-API bereitstellen."
+                ),
+            }},
+            order=41,
+        ),
         DocumentationTopic(
             id="connectors.rss-atom",
             title="RSS and Atom feeds",
@@ -1218,7 +1335,12 @@ manifest = ModuleManifest(
                 "verification": "Rediscover the profile, finish a keyed full run, run a keyed delta, inspect mapping diagnostics, verify one allowed and denied Search principal, and reconcile every outcome-unknown update before retry.",
             },
         ),
-    ),
+    )),
+)
+
+
+manifest = with_documentation_structured_translations(
+    manifest, locale="de", translations=GERMAN_STRUCTURED_TRANSLATIONS
 )
 
 

src/govoplan_connectors/backend/migrations/versions/d2a4c6e8f0b1_csv_source_evidence.py

@@ -0,0 +1,20 @@
+"""Retain original CSV source evidence for new durable snapshots."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "d2a4c6e8f0b1"
+down_revision = "c0f1a2b3c4d5"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "connector_tabular_sources", sa.Column("csv_source", sa.JSON(), nullable=True)
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("connector_tabular_sources") as batch:
+        batch.drop_column("csv_source")

src/govoplan_connectors/backend/router.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
 from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
 from govoplan_core.core.tabular_sources import (
+    TabularCsvSource,
     TabularReadRequest,
     TabularSnapshotInput,
     TabularSource,
@@ -539,7 +540,7 @@ def api_create_tabular_snapshot(
         rows = (
             tuple(payload.rows or ())
             if payload.format == "json"
-            else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter)
+            else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter, value_mode=payload.csv_value_mode)
         )
         source = provider.create_snapshot(
             session,
@@ -550,6 +551,11 @@ def api_create_tabular_snapshot(
                 description=payload.description,
                 rows=rows,
                 metadata={"import_format": payload.format},
+                csv_source=(TabularCsvSource(
+                    text=payload.csv_text or "",
+                    delimiter=payload.delimiter,
+                    value_mode=payload.csv_value_mode,
+                ) if payload.format == "csv" else None),
             ),
         )
     except TabularSourceError as exc:
@@ -573,6 +579,35 @@ def api_create_tabular_snapshot(
     return _source_response(source)
 
 
+@router.get("/tabular-sources/{source_ref}/original-csv")
+def api_original_tabular_csv(
+    source_ref: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> Response:
+    _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
+    try:
+        text = provider.original_csv(session, principal, source_ref=source_ref)
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="connectors.original_csv.exported",
+        object_type="connector_tabular_source",
+        object_id=source_ref,
+        details={"sha256": hashlib.sha256(text.encode("utf-8")).hexdigest()},
+    )
+    session.commit()
+    return Response(
+        content=text.encode("utf-8"),
+        media_type="text/csv; charset=utf-8",
+        headers={"Content-Disposition": 'attachment; filename="original.csv"', "Cache-Control": "no-store"},
+    )
+
+
 @router.post(
     "/tabular-sources/files",
     response_model=TabularSourceResponse,
@@ -595,6 +630,7 @@ def api_create_managed_file_source(
             file_version_id=payload.file_version_id,
             delimiter=payload.delimiter,
             sheet_name=payload.sheet_name,
+            csv_value_mode=payload.csv_value_mode,
         )
     except TabularSourceError as exc:
         raise _http_error(exc) from exc

src/govoplan_connectors/backend/schemas.py

@@ -90,6 +90,7 @@ class SnapshotCreateRequest(BaseModel):
     format: Literal["json", "csv"] = "json"
     rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000)
     csv_text: str | None = Field(default=None, max_length=5_000_000)
+    csv_value_mode: Literal["legacy_typed", "text"] = "legacy_typed"
     delimiter: Literal[",", ";", "\t", "|"] = ","
 
     @model_validator(mode="after")
@@ -117,6 +118,7 @@ class ManagedFileSourceCreateRequest(BaseModel):
     file_version_id: str | None = Field(default=None, min_length=1, max_length=36)
     delimiter: Literal[",", ";", "\t", "|"] = ","
     sheet_name: str | None = Field(default=None, min_length=1, max_length=255)
+    csv_value_mode: Literal["legacy_typed", "text"] = "legacy_typed"
 
 
 class SqlSourceCreateRequest(BaseModel):

src/govoplan_connectors/backend/search_principal.py

@@ -0,0 +1,26 @@
+"""Shared connector-search ACL token projection; owner authorization stays local."""
+
+
+def principal_acl_tokens(principal: object) -> tuple[str, ...]:
+    # Keep legacy first-seen ordering and the exact 500-token authorization cap.
+    values: dict[str, None] = {}
+    for prefix, attribute in (
+        ("account", "account_id"),
+        ("membership", "membership_id"),
+        ("identity", "identity_id"),
+    ):
+        value = getattr(principal, attribute, None)
+        if value:
+            values[f"{prefix}:{value}"] = None
+    for prefix, attribute in (
+        ("group", "group_ids"),
+        ("role", "role_ids"),
+        ("function", "function_assignment_ids"),
+        ("scope", "scopes"),
+    ):
+        for value in getattr(principal, attribute, ()):
+            if value:
+                values[f"{prefix}:{value}"] = None
+                if len(values) >= 500:
+                    return tuple(values)
+    return tuple(values)

src/govoplan_connectors/backend/service_desk_search.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from govoplan_connectors.backend.search_principal import principal_acl_tokens as _principal_acl_tokens
+
 from collections.abc import Mapping, Sequence
 from urllib.parse import quote
 
@@ -288,30 +290,6 @@ def search_document(
     )
 
 
-def _principal_acl_tokens(principal: object) -> tuple[str, ...]:
-    values: list[str] = []
-    for prefix, attribute in (
-        ("account", "account_id"),
-        ("membership", "membership_id"),
-        ("identity", "identity_id"),
-    ):
-        value = getattr(principal, attribute, None)
-        if value:
-            values.append(f"{prefix}:{value}")
-    for prefix, attribute in (
-        ("group", "group_ids"),
-        ("role", "role_ids"),
-        ("function", "function_assignment_ids"),
-        ("scope", "scopes"),
-    ):
-        values.extend(
-            f"{prefix}:{value}"
-            for value in getattr(principal, attribute, ())
-            if value
-        )
-    return tuple(dict.fromkeys(values))[:500]
-
-
 def _has_scope(principal: object, required: str) -> bool:
     check = getattr(principal, "has", None)
     if callable(check):

src/govoplan_connectors/backend/tabular_adapters.py

@@ -12,7 +12,10 @@ from decimal import Decimal
 from io import BytesIO
 from typing import Any
 
+from defusedxml import ElementTree as SafeET
+from defusedxml.common import DefusedXmlException
 from openpyxl import load_workbook
+from openpyxl.utils.cell import column_index_from_string
 from sqlalchemy import (
     JSON,
     BigInteger,
@@ -50,13 +53,26 @@ from govoplan_core.core.tabular_sources import (
     TabularSourceUnavailableError,
     TabularSourceValidationError,
     parse_tabular_csv,
+    CsvValueMode,
+    infer_tabular_schema as _infer_schema,
+    tabular_type_name,
 )
 from govoplan_core.security.credential_envelopes import (
     CredentialAccessContext,
     CredentialEnvelopeError,
     resolve_credential_envelope,
 )
+from govoplan_core.security.bounded_process import (
+    ProcessBudgetError,
+    ProcessLimits,
+    run_bounded_operation,
+)
 from govoplan_core.security.redaction import is_sensitive_key
+from govoplan_core.security.worker_payload import (
+    WorkerPayloadError,
+    decode_worker_payload,
+    encode_worker_payload,
+)
 from govoplan_connectors.backend.db.models import ConnectorConfiguration
 
 
@@ -66,6 +82,13 @@ MAX_FILE_COLUMNS = 500
 MAX_XLSX_ENTRIES = 5_000
 MAX_XLSX_EXPANDED_BYTES = 50_000_000
 MAX_XLSX_COMPRESSION_RATIO = 100
+XLSX_PROCESS_LIMITS = ProcessLimits(
+    wall_seconds=15,
+    cpu_seconds=10,
+    memory_bytes=512 * 1024 * 1024,
+    input_bytes=8 * 1024 * 1024,
+    output_bytes=64 * 1024 * 1024,
+)
 POSTGRESQL_SCHEMES = frozenset({"postgresql", "postgresql+psycopg"})
 _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$")
 
@@ -105,6 +128,7 @@ class ManagedFileTabularAdapter:
         file_version_id: str | None,
         delimiter: str = ",",
         sheet_name: str | None = None,
+        csv_value_mode: CsvValueMode = "legacy_typed",
     ) -> TabularOriginInspection:
         provider = managed_tabular_file_provider(self._registry)
         if provider is None:
@@ -143,6 +167,7 @@ class ManagedFileTabularAdapter:
             content_type=content.file.content_type,
             delimiter=delimiter,
             sheet_name=sheet_name,
+            csv_value_mode=csv_value_mode,
         )
         schema = infer_tabular_schema(rows)
         fingerprint = origin_fingerprint(
@@ -154,6 +179,7 @@ class ManagedFileTabularAdapter:
                 content.file.sha256,
                 resolved_sheet or "",
                 delimiter,
+                *(("csv_text_values",) if csv_value_mode == "text" else ()),
             ),
         )
         try:
@@ -207,6 +233,7 @@ class ManagedFileTabularAdapter:
                 "content_type": content.file.content_type,
                 "format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv",
                 "delimiter": delimiter,
+                "csv_value_mode": csv_value_mode,
                 "sheet_name": resolved_sheet,
             },
             health=health,
@@ -243,6 +270,7 @@ class ManagedFileTabularAdapter:
             file_version_id=_required_metadata(metadata, "file_version_id"),
             delimiter=str(metadata.get("delimiter") or ","),
             sheet_name=_optional_text(metadata.get("sheet_name")),
+            csv_value_mode=str(metadata.get("csv_value_mode") or "legacy_typed"),
         )
         expected_sha256 = _required_metadata(metadata, "file_sha256")
         if inspection.metadata.get("file_sha256") != expected_sha256:
@@ -547,6 +575,7 @@ def parse_managed_tabular_content(
     content_type: str | None,
     delimiter: str,
     sheet_name: str | None,
+    csv_value_mode: CsvValueMode = "legacy_typed",
 ) -> tuple[tuple[Mapping[str, object], ...], str | None]:
     if len(payload) > MAX_FILE_BYTES:
         raise TabularSourceValidationError(
@@ -561,7 +590,7 @@ def parse_managed_tabular_content(
             "Managed CSV files must use UTF-8 encoding."
         ) from exc
     return (
-        tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS)),
+        tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS, max_bytes=MAX_FILE_BYTES, value_mode=csv_value_mode)),
         None,
     )
 
@@ -570,6 +599,55 @@ def _parse_xlsx(
     payload: bytes,
     *,
     sheet_name: str | None,
+) -> tuple[tuple[Mapping[str, object], ...], str]:
+    # Only already-authorized workbook bytes and a worksheet selector leave
+    # the parent; Files access, credentials and SQL sessions never cross.
+    try:
+        encoded = encode_worker_payload(
+            {"payload": payload, "sheet_name": sheet_name},
+            max_bytes=XLSX_PROCESS_LIMITS.input_bytes,
+        )
+        result = decode_worker_payload(
+            run_bounded_operation(_parse_xlsx_worker, encoded, limits=XLSX_PROCESS_LIMITS),
+            max_bytes=XLSX_PROCESS_LIMITS.output_bytes,
+        )
+    except ProcessBudgetError as exc:
+        error_type = (
+            TabularSourceUnavailableError
+            if exc.code in {"busy", "cancelled", "unavailable", "worker_failed"}
+            else TabularSourceValidationError
+        )
+        raise error_type(f"Managed XLSX processing failed ({exc.code}): {exc}") from exc
+    except (TypeError, ValueError, RecursionError) as exc:
+        raise TabularSourceValidationError("Managed XLSX data could not be safely transferred.") from exc
+    if not isinstance(result, dict):
+        raise TabularSourceUnavailableError("Managed XLSX worker returned an invalid result.")
+    if "validation_error" in result:
+        raise TabularSourceValidationError(str(result["validation_error"]))
+    rows, selected_name = result.get("rows"), result.get("sheet_name")
+    if not isinstance(rows, tuple) or not isinstance(selected_name, str):
+        raise TabularSourceUnavailableError("Managed XLSX worker returned an invalid result.")
+    return rows, selected_name
+
+
+def _parse_xlsx_worker(payload: bytes) -> bytes:
+    data = decode_worker_payload(payload, max_bytes=XLSX_PROCESS_LIMITS.input_bytes)
+    try:
+        rows, sheet_name = _parse_xlsx_content(data["payload"], sheet_name=data["sheet_name"])
+        return encode_worker_payload(
+            {"rows": rows, "sheet_name": sheet_name},
+            max_bytes=XLSX_PROCESS_LIMITS.output_bytes,
+        )
+    except TabularSourceValidationError as exc:
+        return encode_worker_payload({"validation_error": str(exc)})
+    except WorkerPayloadError:
+        return encode_worker_payload({"validation_error": "Managed XLSX parsed data exceeds the worker transport limit."})
+
+
+def _parse_xlsx_content(
+    payload: bytes,
+    *,
+    sheet_name: str | None,
 ) -> tuple[tuple[Mapping[str, object], ...], str]:
     _validate_xlsx_archive(payload)
     try:
@@ -579,6 +657,8 @@ def _parse_xlsx(
             data_only=True,
             keep_links=False,
         )
+    except MemoryError:
+        raise
     except Exception as exc:
         raise TabularSourceValidationError(
             "Managed XLSX content could not be parsed."
@@ -595,7 +675,12 @@ def _parse_xlsx(
                 f"Managed XLSX worksheet {selected_name!r} was not found."
             )
         worksheet = workbook[selected_name]
-        iterator = worksheet.iter_rows(values_only=True)
+        maximum_column = _validate_xlsx_worksheet(worksheet)
+        # Declared dimensions are not authoritative: they can inflate sparse
+        # rows/columns or conceal actual cells. Validate the XML coordinates
+        # before openpyxl synthesizes any missing cells, then ignore dimensions.
+        worksheet.reset_dimensions()
+        iterator = worksheet.iter_rows(max_col=maximum_column, values_only=True)
         try:
             raw_headers = next(iterator)
         except StopIteration as exc:
@@ -604,10 +689,10 @@ def _parse_xlsx(
             ) from exc
         headers = _xlsx_headers(raw_headers)
         rows: list[Mapping[str, object]] = []
-        for values in iterator:
-            if len(rows) >= MAX_FILE_ROWS:
+        for row_number, values in enumerate(iterator, start=1):
+            if row_number > MAX_FILE_ROWS:
                 raise TabularSourceValidationError(
-                    f"Managed XLSX worksheets are limited to {MAX_FILE_ROWS:,} data rows."
+                    f"Managed XLSX worksheets are limited to {MAX_FILE_ROWS:,} row positions after the header, including blank gaps."
                 )
             normalized = tuple(values[: len(headers)])
             if all(value in (None, "") for value in normalized):
@@ -625,6 +710,60 @@ def _parse_xlsx(
         workbook.close()
 
 
+def _validate_xlsx_worksheet(worksheet: Any) -> int:
+    """Bound actual coordinates before read-only openpyxl allocates row tuples.
+
+    The pinned openpyxl 3.x read-only source handle is streamed and closed here;
+    no untrusted dimensions or filesystem paths are used to allocate a grid.
+    """
+    maximum_column = 1
+    row_position = 0
+    rows_seen = 0
+    column_position = 0
+    cells_seen = 0
+    try:
+        with worksheet._get_source() as source:
+            for event, element in SafeET.iterparse(source, events=("start", "end"), forbid_dtd=True):
+                tag = element.tag.rsplit("}", 1)[-1]
+                if event == "end":
+                    element.clear()
+                    continue
+                if tag == "row":
+                    rows_seen += 1
+                    raw_row = element.get("r", str(row_position + 1))
+                    if len(raw_row) > 7 or not raw_row.isascii() or not raw_row.isdigit():
+                        raise TabularSourceValidationError("Managed XLSX content has an invalid row coordinate.")
+                    row_position = int(raw_row)
+                    if not 1 <= row_position <= MAX_FILE_ROWS + 1 or rows_seen > MAX_FILE_ROWS + 1:
+                        raise TabularSourceValidationError(
+                            f"Managed XLSX worksheets are limited to {MAX_FILE_ROWS:,} row positions after the header, including blank gaps."
+                        )
+                    column_position = 0
+                    cells_seen = 0
+                elif tag == "c":
+                    cells_seen += 1
+                    reference = element.get("r")
+                    if reference is not None:
+                        match = re.fullmatch(r"([A-Za-z]{1,3})([1-9][0-9]{0,6})", reference)
+                        if match is None:
+                            raise TabularSourceValidationError("Managed XLSX content has an invalid cell coordinate.")
+                        column_position = column_index_from_string(match.group(1))
+                        if int(match.group(2)) != row_position:
+                            raise TabularSourceValidationError("Managed XLSX cell and row coordinates do not agree.")
+                    else:
+                        column_position += 1
+                    if column_position > MAX_FILE_COLUMNS or cells_seen > MAX_FILE_COLUMNS:
+                        raise TabularSourceValidationError(
+                            f"Managed XLSX worksheets are limited to {MAX_FILE_COLUMNS:,} columns."
+                        )
+                    maximum_column = max(maximum_column, column_position)
+        return maximum_column
+    except (DefusedXmlException, SafeET.ParseError, ValueError) as exc:
+        if isinstance(exc, TabularSourceValidationError):
+            raise
+        raise TabularSourceValidationError("Managed XLSX worksheet XML could not be safely parsed.") from exc
+
+
 def _validate_xlsx_archive(payload: bytes) -> None:
     try:
         with zipfile.ZipFile(BytesIO(payload)) as archive:
@@ -679,26 +818,7 @@ def _xlsx_headers(values: Sequence[object]) -> tuple[str, ...]:
 def infer_tabular_schema(
     rows: Sequence[Mapping[str, object]],
 ) -> tuple[TabularColumn, ...]:
-    names: list[str] = []
-    for row in rows:
-        for name in row:
-            if name not in names:
-                names.append(name)
-    result: list[TabularColumn] = []
-    for name in names:
-        values = [row.get(name) for row in rows]
-        concrete = [value for value in values if value is not None]
-        data_type = _type_name(concrete[0]) if concrete else "unknown"
-        if any(_type_name(value) != data_type for value in concrete[1:]):
-            data_type = "mixed"
-        result.append(
-            TabularColumn(
-                name=name,
-                data_type=data_type,
-                nullable=len(concrete) != len(values),
-            )
-        )
-    return tuple(result)
+    return _infer_schema(rows, type_name=_type_name)
 
 
 def origin_fingerprint(
@@ -821,19 +941,7 @@ def _json_value(value: object) -> object:
 
 
 def _type_name(value: object) -> str:
-    if isinstance(value, bool):
-        return "boolean"
-    if isinstance(value, int):
-        return "integer"
-    if isinstance(value, (float, Decimal)):
-        return "number"
-    if isinstance(value, str):
-        return "string"
-    if isinstance(value, list):
-        return "array"
-    if isinstance(value, dict):
-        return "object"
-    return type(value).__name__.casefold()
+    return tabular_type_name(value, casefold_unknown=True)
 
 
 __all__ = [

src/govoplan_connectors/backend/tabular_sources.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, has_scope
 from govoplan_core.core.tabular_sources import (
+    CsvValueMode,
     TabularColumn,
     TabularPreviewDiagnostic,
     TabularPushdown,
@@ -26,6 +27,12 @@ from govoplan_core.core.tabular_sources import (
     TabularSourceUnavailableError,
     TabularSourceValidationError,
     parse_tabular_csv,
+    csv_source_payload,
+    csv_source_summary,
+    csv_projection_matches,
+    verified_csv_source_text,
+    infer_tabular_schema,
+    tabular_type_name as _type_name,
 )
 from govoplan_core.core.runtime import get_registry
 from govoplan_core.db.base import utcnow
@@ -281,6 +288,7 @@ class SqlTabularSourceProvider:
         description: str | None = None,
         delimiter: str = ",",
         sheet_name: str | None = None,
+        csv_value_mode: CsvValueMode = "legacy_typed",
     ) -> TabularSource:
         db, api_principal = _context(session, principal, WRITE_SCOPE)
         inspection = self._file_adapter().inspect(
@@ -290,6 +298,7 @@ class SqlTabularSourceProvider:
             file_version_id=file_version_id,
             delimiter=delimiter,
             sheet_name=sheet_name,
+            csv_value_mode=csv_value_mode,
         )
         return self._create_origin(
             db,
@@ -353,6 +362,7 @@ class SqlTabularSourceProvider:
                 file_version_id=None,
                 delimiter=str(item.metadata_.get("delimiter") or ","),
                 sheet_name=_clean_optional(item.metadata_.get("sheet_name")),
+                csv_value_mode=str(item.metadata_.get("csv_value_mode") or "legacy_typed"),
             )
         elif item.provider == "postgresql":
             inspection = self._sql_adapter.inspect(
@@ -456,6 +466,20 @@ class SqlTabularSourceProvider:
                 f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows."
             )
         rows = [_json_row(row) for row in snapshot.rows]
+        csv_payload = None
+        if snapshot.csv_source is not None:
+            if "csv_source" in snapshot.metadata:
+                raise TabularSourceValidationError("csv_source metadata is reserved for verified CSV import evidence.")
+            if snapshot.csv_source.parser_profile != "core.csv.v1":
+                raise TabularSourceValidationError("Unsupported original CSV parser profile.")
+            csv_payload = csv_source_payload(snapshot.csv_source, max_bytes=MAX_SNAPSHOT_BYTES)
+            expected = parse_csv_snapshot(
+                snapshot.csv_source.text,
+                delimiter=snapshot.csv_source.delimiter,
+                value_mode=snapshot.csv_source.value_mode,
+            )
+            if not csv_projection_matches(expected, rows):
+                raise TabularSourceValidationError("Snapshot rows do not match their original CSV source and parsing mode.")
         encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
         if len(encoded) > MAX_SNAPSHOT_BYTES:
             raise TabularSourceValidationError(
@@ -487,7 +511,11 @@ class SqlTabularSourceProvider:
             fingerprint=fingerprint,
             row_count=len(rows),
             byte_count=len(encoded),
-            metadata_=dict(snapshot.metadata),
+            metadata_={
+                **dict(snapshot.metadata),
+                **({"csv_source": csv_source_summary(csv_payload)} if csv_payload is not None else {}),
+            },
+            csv_source_=csv_payload,
             created_by=actor_id,
             updated_by=actor_id,
         )
@@ -497,6 +525,15 @@ class SqlTabularSourceProvider:
         db.flush()
         return _source_dto(item)
 
+    def original_csv(self, session: object, principal: object, *, source_ref: str) -> str:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        item = _source_record(db, tenant_id=api_principal.tenant_id, source_ref=source_ref)
+        if item is None or item.status != "active":
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        if not item.csv_source_:
+            raise TabularSourceNotFoundError("Original CSV was not retained for this source; historical typed snapshots cannot reconstruct it.")
+        return verified_csv_source_text(item.csv_source_, expected_summary=item.metadata_.get("csv_source") or {})
+
     def delete_snapshot(
         self,
         session: object,
@@ -519,35 +556,18 @@ class SqlTabularSourceProvider:
         return _source_dto(item)
 
 
-def parse_csv_snapshot(csv_text: str, *, delimiter: str) -> tuple[Mapping[str, object], ...]:
+def parse_csv_snapshot(csv_text: str, *, delimiter: str, value_mode: CsvValueMode = "legacy_typed") -> tuple[Mapping[str, object], ...]:
     return parse_tabular_csv(
         csv_text,
         delimiter=delimiter,
         max_rows=MAX_SNAPSHOT_ROWS,
+        max_bytes=MAX_SNAPSHOT_BYTES,
+        value_mode=value_mode,
     )
 
 
 def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]:
-    names: list[str] = []
-    for row in rows:
-        for name in row:
-            if name not in names:
-                names.append(name)
-    result: list[TabularColumn] = []
-    for name in names:
-        values = [row.get(name) for row in rows]
-        concrete = [value for value in values if value is not None]
-        data_type = _type_name(concrete[0]) if concrete else "unknown"
-        if any(_type_name(value) != data_type for value in concrete[1:]):
-            data_type = "mixed"
-        result.append(
-            TabularColumn(
-                name=name,
-                data_type=data_type,
-                nullable=len(concrete) != len(values),
-            )
-        )
-    return tuple(result)
+    return infer_tabular_schema(rows, type_name=_type_name)
 
 
 def snapshot_fingerprint(
@@ -717,22 +737,6 @@ def _unsupported_json(value: object) -> object:
     raise TypeError(f"{type(value).__name__} is not JSON serializable")
 
 
-def _type_name(value: object) -> str:
-    if isinstance(value, bool):
-        return "boolean"
-    if isinstance(value, int):
-        return "integer"
-    if isinstance(value, (float, Decimal)):
-        return "number"
-    if isinstance(value, str):
-        return "string"
-    if isinstance(value, list):
-        return "array"
-    if isinstance(value, dict):
-        return "object"
-    return type(value).__name__.lower()
-
-
 def _column_payload(column: TabularColumn) -> dict[str, object]:
     return {
         "name": column.name,

tests/test_documentation_contract.py

@@ -0,0 +1,20 @@
+from govoplan_connectors.backend.manifest import get_manifest
+
+
+def test_static_documentation_has_complete_german_reference_copy() -> None:
+    for topic in get_manifest().documentation:
+        german = topic.translations.get("de", {})
+        assert all(german.get(field, "").strip() for field in ("title", "summary", "body")), topic.id
+
+
+def test_documentation_exposes_conditioned_workflow_and_reference() -> None:
+    topics = {topic.id: topic for topic in get_manifest().documentation}
+    workflow = topics["connectors.governed-configuration"]
+    assert workflow.metadata.get("kind") == "workflow"
+    assert any(condition.required_scopes for condition in workflow.conditions)
+
+    reference = topics["connectors.authority-and-effects"]
+    assert reference.metadata.get("kind") == "reference"
+    assert reference.metadata.get("fields")
+    assert reference.metadata.get("consequences")
+    assert reference.structured_translations.get("de")

tests/test_migrations.py

@@ -2,16 +2,45 @@ from __future__ import annotations
 
 import tempfile
 import unittest
+from datetime import UTC, datetime
 from pathlib import Path
 
+from alembic import command
 from alembic.runtime.migration import MigrationContext
-from sqlalchemy import create_engine, inspect
+from sqlalchemy import MetaData, Table, create_engine, inspect, select
 
 from govoplan_connectors.backend.manifest import get_manifest
-from govoplan_core.db.migrations import migrate_database
+from govoplan_core.db.migrations import alembic_config, migrate_database
 
 
 class ConnectorsMigrationTests(unittest.TestCase):
+    def test_csv_evidence_upgrade_preserves_legacy_snapshot_without_inventing_source(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-connectors-csv-migration-") as directory:
+            url = f"sqlite:///{Path(directory) / 'connectors.db'}"
+            config = alembic_config(database_url=url, enabled_modules=("connectors",), manifest_factories=(get_manifest,))
+            command.upgrade(config, "c0f1a2b3c4d5")
+            engine = create_engine(url)
+            try:
+                table = Table("connector_tabular_sources", MetaData(), autoload_with=engine)
+                now = datetime.now(UTC)
+                with engine.begin() as connection:
+                    connection.execute(table.insert().values(
+                        id="legacy-csv", tenant_id="tenant-1", provider="snapshot",
+                        source_name="legacy", name="Legacy", status="active", schema_version=1,
+                        schema=[{"name": "id", "data_type": "integer", "nullable": False}],
+                        rows=[{"id": 1}], fingerprint="a" * 64, row_count=1, byte_count=10,
+                        metadata={"original_label": "CSV"}, created_at=now, updated_at=now,
+                    ))
+                    before = dict(connection.execute(select(table)).mappings().one())
+                command.upgrade(config, "d2a4c6e8f0b1")
+                upgraded = Table("connector_tabular_sources", MetaData(), autoload_with=engine)
+                with engine.connect() as connection:
+                    after = dict(connection.execute(select(upgraded)).mappings().one())
+                self.assertIsNone(after.pop("csv_source"))
+                self.assertEqual(before, after)
+            finally:
+                engine.dispose()
+
     def test_baseline_creates_connector_tables_and_head(self) -> None:
         with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory:
             url = f"sqlite:///{Path(directory) / 'connectors.db'}"
@@ -24,7 +53,7 @@ class ConnectorsMigrationTests(unittest.TestCase):
             try:
                 with engine.connect() as connection:
                     self.assertIn(
-                        "c0f1a2b3c4d5",
+                        "d2a4c6e8f0b1",
                         set(MigrationContext.configure(connection).get_current_heads()),
                     )
                     self.assertTrue(

tests/test_search_principal.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from govoplan_connectors.backend.search_principal import principal_acl_tokens
+
+
+class SearchPrincipalTests(unittest.TestCase):
+    def test_legacy_first_seen_projection_and_cap_are_unchanged(self) -> None:
+        principal = SimpleNamespace(
+            account_id=" actor ",
+            membership_id="m",
+            identity_id="i",
+            group_ids=("g", "", "g", "2"),
+            role_ids=("r", "r"),
+            function_assignment_ids=("f",),
+            scopes=tuple(f"scope-{i}" for i in range(700)),
+        )
+        legacy = []
+        for prefix, attribute in (
+            ("account", "account_id"),
+            ("membership", "membership_id"),
+            ("identity", "identity_id"),
+        ):
+            value = getattr(principal, attribute, None)
+            if value:
+                legacy.append(f"{prefix}:{value}")
+        for prefix, attribute in (
+            ("group", "group_ids"),
+            ("role", "role_ids"),
+            ("function", "function_assignment_ids"),
+            ("scope", "scopes"),
+        ):
+            legacy.extend(
+                f"{prefix}:{value}"
+                for value in getattr(principal, attribute, ())
+                if value
+            )
+        expected = tuple(dict.fromkeys(legacy))[:500]
+        self.assertEqual(expected, principal_acl_tokens(principal))
+        self.assertEqual(500, len(expected))
+        self.assertEqual((), principal_acl_tokens(object()))
+
+    def test_projection_stops_consuming_at_authorization_cap(self) -> None:
+        def bounded_scopes():
+            yield from (str(i) for i in range(500))
+            raise AssertionError("ACL projection scanned beyond its effective cap")
+
+        self.assertEqual(
+            500, len(principal_acl_tokens(SimpleNamespace(scopes=bounded_scopes())))
+        )

tests/test_tabular_sources.py

@@ -1,23 +1,27 @@
 from __future__ import annotations
 
 import unittest
+from unittest.mock import patch
+from dataclasses import replace
 
 from fastapi import HTTPException
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import sessionmaker
 
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.tabular_sources import (
+    TabularCsvSource,
     TabularReadRequest,
     TabularSnapshotInput,
     TabularSourceAccessError,
     TabularSourceUnavailableError,
     TabularSourceValidationError,
+    TabularSourceNotFoundError,
 )
 from govoplan_core.db.base import Base
 from govoplan_connectors.backend.db.models import ConnectorTabularSource
-from govoplan_connectors.backend.router import api_create_tabular_snapshot
+from govoplan_connectors.backend.router import api_create_tabular_snapshot, api_original_tabular_csv
 from govoplan_connectors.backend.schemas import SnapshotCreateRequest
 from govoplan_connectors.backend.tabular_sources import (
     READ_SCOPE,
@@ -263,6 +267,54 @@ class ConnectorsTabularSourceTests(unittest.TestCase):
 
         self.assertEqual(422, raised.exception.status_code)
 
+    def test_original_csv_round_trip_is_private_bounded_and_not_catalogue_content(self) -> None:
+        text = '\ufeffid,value\r\n9007199254740993,"  keep me  "\r\ntrue,0.123456789012345678901234567890\r\n"   ",""\r\n'
+        payload = SnapshotCreateRequest(name="CSV", source_name="csv", format="csv", csv_text=text)
+        with patch("govoplan_connectors.backend.router.audit_event"):
+            created = api_create_tabular_snapshot(payload, session=self.session, principal=principal())
+        self.session.expunge_all()
+        listed = self.provider.list_sources(self.session, principal())
+        self.assertNotIn("text", listed[0].metadata["csv_source"])
+        self.assertEqual("legacy_typed", listed[0].metadata["csv_source"]["value_mode"])
+        record = self.session.get(ConnectorTabularSource, created.ref.split(":", 1)[1])
+        self.assertIn("csv_source_", inspect(record).unloaded)
+        with patch("govoplan_connectors.backend.router.audit_event") as audit:
+            response = api_original_tabular_csv(created.ref, session=self.session, principal=principal())
+        self.assertEqual("connectors.original_csv.exported", audit.call_args.kwargs["action"])
+        self.assertEqual({"sha256"}, set(audit.call_args.kwargs["details"]))
+        with patch("govoplan_connectors.backend.router.provider.original_csv") as read, self.assertRaises(HTTPException) as denied:
+            api_original_tabular_csv(created.ref, session=self.session, principal=principal(scopes=()))
+        self.assertEqual(403, denied.exception.status_code)
+        read.assert_not_called()
+        self.assertEqual(text.encode("utf-8"), response.body)
+        self.assertEqual("no-store", response.headers["cache-control"])
+        self.assertIn("attachment", response.headers["content-disposition"])
+        with self.assertRaises(TabularSourceNotFoundError):
+            self.provider.original_csv(self.session, principal("tenant-2"), source_ref=created.ref)
+        with self.assertRaises(TabularSourceAccessError):
+            self.provider.original_csv(self.session, principal(scopes=()), source_ref=created.ref)
+        record.csv_source_ = {**record.csv_source_, "text": "tampered"}
+        self.session.flush()
+        with self.assertRaises(TabularSourceUnavailableError):
+            self.provider.original_csv(self.session, principal(), source_ref=created.ref)
+
+    def test_text_snapshot_matches_exact_projection_and_rejects_inconsistent_evidence(self) -> None:
+        source = TabularCsvSource(text='value\n"   "\n0.123456789012345678901234567890\n', value_mode="text")
+        snapshot = TabularSnapshotInput(name="Text", source_name="text", rows=parse_csv_snapshot(source.text, delimiter=",", value_mode="text"), csv_source=source)
+        with self.assertRaises(TabularSourceValidationError):
+            self.provider.create_snapshot(self.session, principal(), snapshot=replace(snapshot, rows=({"value": "changed"},)))
+        created = self.provider.create_snapshot(self.session, principal(), snapshot=snapshot)
+        self.assertEqual(2, created.row_count)
+        self.assertEqual(source.text, self.provider.original_csv(self.session, principal(), source_ref=created.ref))
+        legacy = self.provider.create_snapshot(self.session, principal(), snapshot=TabularSnapshotInput(name="Old", source_name="old", rows=({"id": 1},)))
+        with self.assertRaises(TabularSourceNotFoundError):
+            self.provider.original_csv(self.session, principal(), source_ref=legacy.ref)
+
+    def test_original_csv_projection_binding_rejects_equal_but_different_types(self) -> None:
+        for text, value in (("value\ntrue\n", 1), ("value\n1\n", True), ("value\n1\n", 1.0)):
+            with self.subTest(text=text, value=value), self.assertRaises(TabularSourceValidationError):
+                self.provider.create_snapshot(self.session, principal(), snapshot=TabularSnapshotInput(name="Invalid", source_name="invalid", rows=({"value": value},), csv_source=TabularCsvSource(text=text)))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_xlsx_safety_bounds.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import re
+import unittest
+from dataclasses import replace
+from io import BytesIO
+from unittest.mock import patch
+from zipfile import ZipFile
+
+from openpyxl import Workbook
+
+from govoplan_core.core.tabular_sources import TabularSourceValidationError
+from govoplan_connectors.backend import tabular_adapters as adapters
+
+
+def workbook_bytes(*, dimension: str | None = None, last_cell: str = "A2") -> bytes:
+    workbook = Workbook()
+    workbook.active.append(["name"])
+    workbook.active.append(["Ada"])
+    output = BytesIO()
+    workbook.save(output)
+    workbook.close()
+    result = BytesIO()
+    with ZipFile(BytesIO(output.getvalue())) as original, ZipFile(result, "w") as modified:
+        for entry in original.infolist():
+            content = original.read(entry)
+            if entry.filename == "xl/worksheets/sheet1.xml":
+                xml = content.decode()
+                if dimension is not None:
+                    xml = re.sub(r'<dimension ref="[^"]+"\s*/>', f'<dimension ref="{dimension}"/>', xml)
+                xml = xml.replace('r="A2"', f'r="{last_cell}"')
+                xml = xml.replace('<row r="2">', f'<row r="{re.search(r"[0-9]+$", last_cell).group()}">')
+                content = xml.encode()
+            modified.writestr(entry, content)
+    return result.getvalue()
+
+
+def parse(payload: bytes):
+    return adapters.parse_managed_tabular_content(payload, filename="fixture.xlsx", content_type=None, delimiter=",", sheet_name=None)
+
+
+class XlsxSafetyBoundsTests(unittest.TestCase):
+    def test_sparse_rows_cannot_bypass_limit_by_not_counting_as_data(self):
+        with self.assertRaisesRegex(TabularSourceValidationError, "row"):
+            parse(workbook_bytes(last_cell=f"A{adapters.MAX_FILE_ROWS + 2}"))
+
+    def test_forged_small_dimensions_cannot_hide_far_away_cells(self):
+        with self.assertRaisesRegex(TabularSourceValidationError, "row"):
+            parse(workbook_bytes(dimension="A1:A2", last_cell="A1000000"))
+
+    def test_forged_small_dimensions_cannot_hide_out_of_range_columns(self):
+        with self.assertRaisesRegex(TabularSourceValidationError, "column"):
+            parse(workbook_bytes(dimension="A1:A2", last_cell="XFD2"))
+
+    def test_declared_dimensions_do_not_expand_or_truncate_real_rows(self):
+        for dimension in ("A1:XFD1048576", "A1:A1"):
+            with self.subTest(dimension=dimension):
+                rows, sheet = parse(workbook_bytes(dimension=dimension))
+                self.assertEqual(({"name": "Ada"},), rows)
+                self.assertEqual("Sheet", sheet)
+
+    def test_small_blank_gaps_and_exact_limit_remain_usable(self):
+        for cell in ("A4", f"A{adapters.MAX_FILE_ROWS + 1}"):
+            rows, _sheet = parse(workbook_bytes(last_cell=cell))
+            self.assertEqual(({"name": "Ada"},), rows)
+
+    def test_workbook_is_parsed_in_a_fresh_child_not_the_parent(self):
+        with patch.object(adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran")):
+            rows, sheet = parse(workbook_bytes())
+        self.assertEqual(({"name": "Ada"},), rows)
+        self.assertEqual("Sheet", sheet)
+
+    def test_real_worker_timeout_fails_without_parent_fallback(self):
+        limits = replace(adapters.XLSX_PROCESS_LIMITS, wall_seconds=0.001)
+        with patch.object(adapters, "XLSX_PROCESS_LIMITS", limits), patch.object(
+            adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran")
+        ):
+            with self.assertRaisesRegex(TabularSourceValidationError, "timeout"):
+                parse(workbook_bytes())
+
+    def test_child_validation_keeps_missing_sheet_message(self):
+        with self.assertRaisesRegex(TabularSourceValidationError, "worksheet 'Missing' was not found"):
+            adapters.parse_managed_tabular_content(
+                workbook_bytes(), filename="fixture.xlsx", content_type=None,
+                delimiter=",", sheet_name="Missing",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/connectors-webui",
-  "version": "0.1.22",
+  "version": "0.1.27",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

webui/src/features/ConnectorGovernancePage.tsx

@@ -483,7 +483,7 @@ export default function ConnectorGovernancePage({ settings, auth }: Props) {
             <FormField label="Endpoint URL" hint="Credentials are rejected in URLs.">
               <input value={draft.endpoint_url} disabled={!canAdmin || busy} placeholder="https://provider.example/api" onChange={(event) => setDraft({ ...draft, endpoint_url: event.target.value })} />
             </FormField>
-            <FormField label="Credential reference" hint="Reference an approved secret; do not paste a secret.">
+            <FormField label="Credential reference" hint="Reference an approved secret; do not paste a secret." helpContextId="connectors.admin.governed-configurations" helpModuleId="connectors">
               <input value={draft.credential_ref} disabled={!canAdmin || busy} placeholder="vault://connectors/provider" onChange={(event) => setDraft({ ...draft, credential_ref: event.target.value })} />
             </FormField>
             <FormField label="Ambiguous-result policy">
@@ -574,7 +574,7 @@ export default function ConnectorGovernancePage({ settings, auth }: Props) {
         <FormField label="Endpoint URL">
           <input value={newDraft.endpoint_url} disabled={busy} onChange={(event) => setNewDraft({ ...newDraft, endpoint_url: event.target.value })} />
         </FormField>
-        <FormField label="Credential reference">
+        <FormField label="Credential reference" helpContextId="connectors.admin.governed-configurations" helpModuleId="connectors">
           <input value={newDraft.credential_ref} disabled={busy} onChange={(event) => setNewDraft({ ...newDraft, credential_ref: event.target.value })} />
         </FormField>
         <FormField label="Ambiguous-result policy">
@@ -597,8 +597,8 @@ export default function ConnectorGovernancePage({ settings, auth }: Props) {
       closeDisabled={busy}
       footer={<>
         <Button onClick={() => setReviewRun(null)} disabled={busy}>Cancel</Button>
-        <Button variant="danger" onClick={() => void decideReview("rejected")} disabled={busy || reviewReason.trim().length < 5}>Reject</Button>
-        <Button variant="primary" onClick={() => void decideReview("approved")} disabled={busy || reviewReason.trim().length < 5}>Approve</Button>
+        <Button variant="danger" helpContextId="connectors.action.reject-ambiguous-result" helpModuleId="connectors" onClick={() => void decideReview("rejected")} disabled={busy || reviewReason.trim().length < 5}>Reject</Button>
+        <Button variant="primary" helpContextId="connectors.action.approve-ambiguous-result" helpModuleId="connectors" onClick={() => void decideReview("approved")} disabled={busy || reviewReason.trim().length < 5}>Approve</Button>
       </>}
     >
       <p>Review {reviewRun?.summary.ambiguous ?? 0} ambiguous effects against the retained input and configuration hashes before deciding.</p>

webui/src/features/ExternalKnowledgePage.tsx

@@ -326,7 +326,7 @@ export default function ExternalKnowledgePage({ settings, auth }: Props) {
       <Button variant="secondary" onClick={() => void sync(false)} disabled={!selected || !canSync || busy || dirty}>Run delta</Button>
       <Button variant="secondary" onClick={() => void sync(true)} disabled={!selected || !canSync || busy || dirty}>Run full backfill</Button>
       <Button variant="secondary" onClick={() => setMigrationOpen(true)} disabled={!selected || !canMigrate || busy || dirty}>Preview migration</Button>
-      <Button variant="primary" onClick={() => setPublishOpen(true)} disabled={!selected || !canPublish || busy || dirty}>Publish page</Button>
+      <Button variant="primary" helpContextId="connectors.admin.external-knowledge" helpModuleId="connectors" onClick={() => setPublishOpen(true)} disabled={!selected || !canPublish || busy || dirty}>Publish page</Button>
     </>}
     discardAction={{
       label: "Discard changes",
@@ -501,7 +501,7 @@ export default function ExternalKnowledgePage({ settings, auth }: Props) {
 
     <Dialog open={publishOpen} title="Publish provider page revision" onClose={() => !busy && setPublishOpen(false)} closeDisabled={busy} footer={<>
       <Button onClick={() => setPublishOpen(false)} disabled={busy}>Cancel</Button>
-      <Button variant="primary" onClick={() => void publish()} disabled={busy || !externalPageId.trim() || !publishTitle.trim()}>Publish revision</Button>
+      <Button variant="primary" helpContextId="connectors.admin.external-knowledge" helpModuleId="connectors" onClick={() => void publish()} disabled={busy || !externalPageId.trim() || !publishTitle.trim()}>Publish revision</Button>
     </>}>
       <p className="muted">Publication is an external effect. Supply the current provider revision where possible; an unknown outcome blocks blind retry.</p>
       <FormGrid columns={2} collapseAt="standard" className="">

webui/src/module.ts

@@ -72,7 +72,7 @@ const adminSections: AdminSectionsUiCapability = {
 export const connectorsModule: PlatformWebModule = {
   id: "connectors",
   label: "Connectors",
-  version: "0.1.22",
+  version: "0.1.23",
   dependencies: [],
   optionalDependencies: ["access", "audit", "policy", "ops", "search", "wiki", "tickets", "helpdesk", "cases"],
   viewSurfaces: [