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.25"
+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

@@ -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 = {
@@ -39,7 +38,9 @@ _TRANSLATIONS = {
         "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. 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 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": {
@@ -62,15 +63,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_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

@@ -126,7 +126,7 @@ from govoplan_connectors.backend.german_documentation import (
 
 
 MODULE_ID = "connectors"
-MODULE_VERSION = "0.1.25"
+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"
@@ -967,6 +967,38 @@ manifest = ModuleManifest(
         ),
     ),
     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",
@@ -1115,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 "
@@ -1128,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",

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_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.25",
+  "version": "0.1.27",
   "private": true,
   "type": "module",
   "main": "src/index.ts",