fix(datasources): preserve governed CSV originals and fresh detail state

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

src/govoplan_datasources/backend/csv_sources.py

@@ -0,0 +1,71 @@
+"""Durable CSV source fidelity; no preview retention or catalogue content."""
+
+from collections.abc import Mapping, Sequence
+
+from govoplan_core.core.datasources import DatasourceValidationError
+from govoplan_core.core.tabular_sources import (
+    TabularCsvSource,
+    TabularSourceError,
+    csv_source_payload,
+    parse_tabular_csv,
+    verified_csv_source_text,
+    csv_projection_matches,
+)
+from govoplan_datasources.backend.tabular import (
+    MAX_STAGE_BYTES,
+    MAX_STAGE_ROWS,
+    normalize_rows,
+    parse_csv_rows,
+)
+
+
+def prepare_csv_source(
+    source: TabularCsvSource | None, rows: Sequence[Mapping[str, object]]
+) -> dict[str, object] | None:
+    if source is None:
+        return None
+    try:
+        payload = csv_source_payload(source, max_bytes=MAX_STAGE_BYTES)
+        if source.parser_profile == "core.csv.v1":
+            expected = parse_tabular_csv(
+                source.text,
+                delimiter=source.delimiter,
+                value_mode=source.value_mode,
+                max_rows=MAX_STAGE_ROWS,
+                max_bytes=MAX_STAGE_BYTES,
+            )
+        elif source.parser_profile == "datasources.csv.v1":
+            expected = parse_csv_rows(
+                source.text, delimiter=source.delimiter, value_mode=source.value_mode
+            )
+        else:
+            raise DatasourceValidationError("Unsupported original CSV parser profile.")
+        if not csv_projection_matches(normalize_rows(expected), rows):
+            raise DatasourceValidationError(
+                "Stage rows do not match their original CSV source and parsing mode."
+            )
+        return payload
+    except TabularSourceError as exc:
+        raise DatasourceValidationError(str(exc)) from exc
+
+
+def verify_stage_csv(
+    payload: Mapping[str, object] | None, rows: Sequence[Mapping[str, object]]
+) -> None:
+    if payload is None:
+        return
+    try:
+        text = verified_csv_source_text(payload)
+        prepare_csv_source(
+            TabularCsvSource(
+                text=text,
+                delimiter=payload["delimiter"],
+                value_mode=payload["value_mode"],
+                parser_profile=payload["parser_profile"],
+            ),
+            rows,
+        )
+    except (KeyError, TypeError, TabularSourceError) as exc:
+        raise DatasourceValidationError(
+            "Original CSV source evidence is invalid."
+        ) from exc

src/govoplan_datasources/backend/db/models.py

@@ -245,6 +245,7 @@ class DatasourceMaterializationRecord(Base, TimestampMixin):
     # Kept nullable-in-practice for rolling upgrades. New materializations use
     # immutable payload rows and leave this legacy field empty.
     rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
+    csv_source_: Mapped[dict[str, Any] | None] = mapped_column("csv_source", JSON, nullable=True, deferred=True)
     fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
     row_count: Mapped[int] = mapped_column(Integer, nullable=False)
     byte_count: Mapped[int] = mapped_column(Integer, nullable=False)
@@ -405,6 +406,7 @@ class DatasourceStageRecord(Base, TimestampMixin):
         nullable=False,
     )
     rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
+    csv_source_: Mapped[dict[str, Any] | None] = mapped_column("csv_source", JSON, nullable=True, deferred=True)
     fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
     row_count: Mapped[int] = mapped_column(Integer, nullable=False)
     byte_count: Mapped[int] = mapped_column(Integer, nullable=False)

src/govoplan_datasources/backend/governance.py

@@ -286,6 +286,10 @@ def stage_subject_digest(
             "fingerprint": stage.fingerprint,
             "validation_policy_hash": stage.validation_.get("policy_hash"),
             "approval_policy": dict(policy or {}),
+            **({"csv_source": {
+                key: stage.csv_source_.get(key)
+                for key in ("sha256", "byte_count", "delimiter", "value_mode", "parser_profile")
+            }} if stage.csv_source_ is not None else {}),
         }
     )
 
@@ -562,6 +566,7 @@ def apply_retention_plan(
             evidence_hashes.append(evidence.event_hash)
             materialization.payload_id = None
             materialization.rows = []
+            materialization.csv_source_ = None
             materialization.state = "disposed"
             materialization.disposed_at = plan.as_of
             materialization.disposition_ = {

src/govoplan_datasources/backend/manifest.py

@@ -68,7 +68,7 @@ from govoplan_datasources.backend.payloads import ExternalArtifactPayloadBackend
 
 MODULE_ID = "datasources"
 MODULE_NAME = "Datasources"
-MODULE_VERSION = "0.1.25"
+MODULE_VERSION = "0.1.26"
 DATASOURCE_INTERFACE_VERSION = "0.2.0"
 
 ARCHITECTURE = ModuleArchitectureDeclaration(
@@ -487,6 +487,72 @@ manifest = ModuleManifest(
     ),
     architecture=ARCHITECTURE,
     documentation=(
+        DocumentationTopic(
+            id="datasources.csv-source-fidelity",
+            title="CSV input fidelity, original export and retention",
+            summary="Keep original upload text alongside durable stages and immutable materializations without bypassing visibility or retention.",
+            body=(
+                "New CSV upload dialogs default to Preserve text; values remain strings, including whitespace, exact decimal digits, large identifiers and explicit empty records. Infer types (legacy) is an explicit compatibility option. "
+                "Existing API calls without csv_value_mode retain their legacy behavior. The API accepts text or legacy_typed; parsed headers are normalized, and malformed text-mode row shapes are rejected. "
+                "Durable CSV stages retain the original supplied UTF-8 text separately from metadata, bind it to the exact scalar types and values of the parsed rows, and include source evidence in approval. Promotion and freezing preserve the original with the corresponding immutable materialization. This does not persist preview rows. "
+                "Original text and rows each have a 5 MB limit; there are at most 10,000 rows. Each stage/materialization, including a frozen copy, can retain up to 5 MB of original text plus serialization and metadata storage overhead. "
+                "GET /api/v1/datasources/{datasource_id}/materializations/{materialization_id}/original-csv is an audited, uncached attachment export for current datasources:source:admin authority. "
+                "Current and historical visibility and policy-provider decisions still apply. Any row filter or denied field prevents raw export, even for an administrator; use governed row previews instead. No raw input is returned in catalogue, stage or history DTOs. "
+                "Freezing verifies the existing original evidence before copying and retains inherited access restrictions; the original may carry at most 32 distinct governance snapshots, with explicit rejection beyond that limit. "
+                "Original text is checksum-verified and follows the same governed retention as its stage/materialization. Stage deletion removes its original; materialization payload disposal also clears that original while retaining audit evidence. "
+                "No original is invented for historical imports. Missing or disposed originals are reported unavailable. Export preserves the API's supplied text as UTF-8, including unsanitized spreadsheet formulas, and is not a reconstruction of an earlier file encoding."
+            ),
+            layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=8,
+            translations={"de": {
+                "title": "CSV-Original, genauer Import und Aufbewahrung",
+                "summary": "Originaltext mit dauerhaften Staging-Ständen und Materialisierungen erhalten, ohne Sichtbarkeit oder Aufbewahrung zu umgehen.",
+                "body": (
+                    "Neue CSV-Importdialoge wählen standardmäßig Text erhalten: Werte einschließlich Leerzeichen, genauer Dezimalstellen, großer Kennungen und ausdrücklich leerer Datensätze bleiben Zeichenketten. Typen ableiten (bisheriges Verhalten) ist eine ausdrückliche Kompatibilitätsoption. "
+                    "Bestehende API-Aufrufe ohne csv_value_mode bleiben unverändert; unterstützt werden text und legacy_typed. Überschriften werden normalisiert, fehlerhafte Zeilenformen im Textmodus abgelehnt. "
+                    "Dauerhafte CSV-Staging-Stände speichern den gelieferten UTF-8-Originaltext getrennt von Metadaten, prüfen genaue Typen und Werte der abgeleiteten Zeilen und binden den Quellnachweis in die Freigabe ein. Übernahme und Einfrieren erhalten das Original bei der jeweiligen unveränderlichen Materialisierung. Vorschauzeilen werden dadurch nicht gespeichert. "
+                    "Original und Zeilen sind jeweils auf 5 MB und die Tabelle auf 10.000 Zeilen begrenzt. Pro Staging-Stand oder Materialisierung einschließlich eingefrorener Kopien werden bis zu 5 MB Originaltext zuzüglich Speicher für Serialisierung und Metadaten aufbewahrt. "
+                    "GET /api/v1/datasources/{datasource_id}/materializations/{materialization_id}/original-csv liefert einen protokollierten, nicht zwischengespeicherten Download mit aktuellem Recht datasources:source:admin. "
+                    "Aktuelle und historische Sichtbarkeit sowie externe Richtlinienentscheidungen gelten weiterhin. Jeder Zeilenfilter oder ein nicht freigegebenes Feld verhindert den Originalexport auch für die Administration; verwenden Sie dann die gesteuerte Zeilenvorschau. Katalog, Staging- und Verlaufsantworten enthalten keinen Originaltext. "
+                    "Einfrieren prüft den vorhandenen Originalnachweis vor dem Kopieren und erhält geerbte Zugriffsbeschränkungen. Höchstens 32 unterschiedliche Richtlinienstände dürfen ein Original begleiten; darüber hinaus wird ausdrücklich abgewiesen. "
+                    "Prüfsummen werden vor dem Export geprüft. Das Original folgt der gesteuerten Aufbewahrung: Löschen eines Staging-Stands entfernt dessen Original; Entsorgung einer Materialisierung entfernt auch deren Original, behält aber Auditnachweise. "
+                    "Für historische Importe wird kein Original erfunden. Fehlende oder entsorgte Originale werden als nicht verfügbar gemeldet. Der Download erhält den an die API gelieferten Text als UTF-8 einschließlich unveränderter Tabellenformeln, nicht eine frühere Dateikodierung."
+                ),
+            }},
+        ),
+        DocumentationTopic(
+            id="datasources.detail-freshness",
+            title="Preview and revision history after a datasource change",
+            summary="Reload the selected datasource details even when its catalogue reference stays the same.",
+            body=(
+                "After a successful catalogue reload, the selected datasource preview and materialization history are read again. "
+                "This includes manual Reload and reloads following refresh, freezing, stage promotion or governance changes, even when the selected reference or current materialization is unchanged. "
+                "While reloading, previous rows and history are not presented as the newly selected or updated datasource. "
+                "An older request cannot replace results for a newer selection, reload or authentication context. "
+                "Callbacks retained from an earlier authorization context cannot start a new reload or change success/error messages and selection. "
+                "After a real identity, credential or permission change, the old catalogue is hidden until the new context has loaded it; open dialog inputs are reset to avoid carrying drafts into another authorization context. "
+                "Cosmetic profile changes and ordinary refreshes with equivalent authority do not reset dialogs or trigger duplicate detail reads. "
+                "Preview and history are independent reads: a preview failure can still leave authorized history available. "
+                "Live datasource previews may read the provider through the existing bounded, authorized preview API; this is not background polling or automatic source refresh. "
+                "Rows are not normalized, reordered or persisted by this display update. Snapshot immutability, retention rules and all server-side permission checks remain unchanged."
+            ),
+            layer="always", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=6,
+            translations={"de": {
+                "title": "Vorschau und Revisionsverlauf nach einer Datenquellenänderung",
+                "summary": "Details der ausgewählten Datenquelle auch bei unverändertem Katalogverweis neu laden.",
+                "body": (
+                    "Nach erfolgreichem Neuladen des Katalogs werden Vorschau und Materialisierungsverlauf der ausgewählten Datenquelle erneut abgerufen. "
+                    "Dies gilt für Neu laden sowie für das Neuladen nach Aktualisierung, Einfrieren, Staging-Übernahme oder Governance-Änderungen, auch wenn Auswahlverweis oder aktuelle Materialisierung gleich bleiben. "
+                    "Während des Neuladens werden bisherige Zeilen und Verläufe nicht als neue Auswahl oder aktualisierte Datenquelle dargestellt. "
+                    "Eine ältere Anfrage kann Ergebnisse einer neueren Auswahl, eines neueren Abrufs oder eines geänderten Authentifizierungskontexts nicht ersetzen. "
+                    "Rückrufe eines früheren Berechtigungskontexts starten keinen neuen Abruf und ändern weder Erfolgs-/Fehlermeldungen noch die Auswahl. "
+                    "Nach einem tatsächlichen Identitäts-, Zugangsdaten- oder Rechtewechsel bleibt der alte Katalog verborgen, bis er im neuen Kontext geladen wurde; offene Dialogeingaben werden zurückgesetzt, damit Entwürfe nicht in einen anderen Berechtigungskontext gelangen. "
+                    "Rein optische Profiländerungen und gewöhnliche Aktualisierungen mit gleichwertigen Rechten setzen Dialoge nicht zurück und lösen keine doppelten Detailabrufe aus. "
+                    "Vorschau und Verlauf sind getrennte Lesezugriffe: Bei einem Vorschaufehler kann ein berechtigter Verlauf weiterhin verfügbar sein. "
+                    "Vorschauen lebender Datenquellen können den Anbieter über die bestehende begrenzte und berechtigungsgeprüfte Vorschau-API lesen; dies ist weder Hintergrund-Polling noch eine automatische Quellenaktualisierung. "
+                    "Die Anzeigeaktualisierung normalisiert, sortiert oder speichert keine Zeilen dauerhaft. Unveränderlichkeit von Snapshots, Aufbewahrungsregeln und sämtliche serverseitigen Rechteprüfungen bleiben erhalten."
+                ),
+            }},
+        ),
         DocumentationTopic(
             id="datasources.workspace-layout",
             title="Datasources workspace actions",

src/govoplan_datasources/backend/migrations/versions/e2b8d4a0f6c3_csv_source_evidence.py

@@ -0,0 +1,24 @@
+"""Preserve original CSV for new durable imports, separate from catalogue DTOs."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "e2b8d4a0f6c3"
+down_revision = "d1a7c3e9f5b2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "datasource_stages", sa.Column("csv_source", sa.JSON(), nullable=True)
+    )
+    op.add_column(
+        "datasource_materializations", sa.Column("csv_source", sa.JSON(), nullable=True)
+    )
+
+
+def downgrade() -> None:
+    for table in ("datasource_materializations", "datasource_stages"):
+        with op.batch_alter_table(table) as batch:
+            batch.drop_column("csv_source")

src/govoplan_datasources/backend/router.py

@@ -5,7 +5,7 @@ import json
 from collections.abc import Mapping
 from datetime import UTC, datetime
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from sqlalchemy.orm import Session
 
 from govoplan_core.audit.logging import audit_event
@@ -25,6 +25,7 @@ from govoplan_core.core.datasources import (
     datasource_origins,
 )
 from govoplan_core.db.session import get_session
+from govoplan_core.core.tabular_sources import TabularCsvSource
 from govoplan_datasources.backend.runtime import get_registry
 from govoplan_datasources.backend.schemas import (
     DatasourceFieldResponse,
@@ -203,7 +204,7 @@ def api_create_stage(
         rows = (
             tuple(payload.rows or ())
             if payload.format == "json"
-            else parse_csv_rows(payload.csv_text or "", delimiter=payload.delimiter)
+            else parse_csv_rows(payload.csv_text or "", delimiter=payload.delimiter, value_mode=payload.csv_value_mode)
         )
         stage = _provider().create_stage(
             session,
@@ -225,6 +226,12 @@ def api_create_stage(
                 },
                 metadata=payload.metadata,
                 governance=_governance(payload.governance),
+                csv_source=(TabularCsvSource(
+                    text=payload.csv_text or "",
+                    delimiter=payload.delimiter,
+                    value_mode=payload.csv_value_mode,
+                    parser_profile="datasources.csv.v1",
+                ) if payload.format == "csv" else None),
             ),
         )
     except DatasourceError as exc:
@@ -629,6 +636,27 @@ def api_list_materializations(
     )
 
 
+@router.get("/{datasource_id}/materializations/{materialization_id}/original-csv")
+def api_original_csv(
+    datasource_id: str,
+    materialization_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> Response:
+    _require_any_scope(principal, ADMIN_SCOPE)
+    try:
+        text = _provider().original_csv(
+            session, principal,
+            datasource_ref=f"datasource:{datasource_id}",
+            materialization_ref=f"materialization:{materialization_id}",
+        )
+    except DatasourceError as exc:
+        raise _http_error(exc) from exc
+    _audit(session, principal, action="datasources.original_csv.exported", object_type="datasource_materialization", object_id=f"materialization:{materialization_id}", details={"datasource_ref": f"datasource:{datasource_id}", "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(
     "/{datasource_id}/refresh",
     response_model=DatasourceStagePromoteResponse,

src/govoplan_datasources/backend/schemas.py

@@ -209,6 +209,7 @@ class DatasourceStageCreateRequest(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", "|"] = ","
     target_datasource_ref: str | None = None
     provenance: dict[str, Any] = Field(default_factory=dict)

src/govoplan_datasources/backend/service.py

@@ -39,6 +39,8 @@ from govoplan_core.core.datasources import (
     datasource_visibility_policy_provider,
 )
 from govoplan_core.db.base import utcnow
+from govoplan_core.core.tabular_sources import csv_source_summary, verified_csv_source_text, TabularSourceError
+from govoplan_datasources.backend.csv_sources import prepare_csv_source, verify_stage_csv
 from govoplan_datasources.backend.db.models import (
     DatasourceGovernanceReferenceRecord,
     DatasourceMaterializationRecord,
@@ -471,6 +473,40 @@ class SqlDatasourceProvider:
             )
         return result
 
+    def original_csv(
+        self,
+        session: object,
+        principal: object,
+        *,
+        datasource_ref: str,
+        materialization_ref: str,
+    ) -> str:
+        """Export original input only to admins with unrestricted current/history visibility."""
+        db, api_principal = _context(session, principal, ADMIN_SCOPE)
+        item = _required_datasource(db, tenant_id=api_principal.tenant_id, datasource_ref=datasource_ref)
+        request = DatasourceReadRequest(datasource_ref=datasource_ref, materialization_ref=materialization_ref)
+        materialization = _selected_materialization(db, item=item, request=request)
+        if materialization is None or materialization.disposed_at is not None:
+            raise DatasourceUnavailableError("Original CSV source is unavailable or was disposed under retention policy.")
+        if materialization.csv_source_ is None:
+            raise DatasourceNotFoundError("Original CSV was not retained for this materialization; historical typed rows cannot reconstruct it.")
+        try:
+            original = verified_csv_source_text(materialization.csv_source_, expected_summary=materialization.metadata_.get("csv_source") or {})
+        except TabularSourceError as exc:
+            raise DatasourceUnavailableError(str(exc)) from exc
+        history = materialization.csv_source_.get("governance_history", [])
+        if not isinstance(history, list) or len(history) > 32 or any(not isinstance(value, Mapping) for value in history):
+            raise DatasourceAccessError("Original CSV inherited governance is invalid.")
+        for snapshot in (None, materialization.governance_snapshot_, *history):
+            plan = self._visibility_plan(
+                db, api_principal, item=item, request=request, materialized=True,
+                schema=_fields(materialization.schema_),
+                governance_override=DatasourceGovernance.from_mapping(snapshot) if snapshot is not None else None,
+            )
+            if plan.row_filters or any(not rule.allowed for rule in plan.field_rules):
+                raise DatasourceAccessError("Original CSV export requires unrestricted row and field visibility; use the governed row preview instead.")
+        return original
+
     def _read_materialized(
         self,
         session: Session,
@@ -732,6 +768,9 @@ class SqlDatasourceProvider:
             )
         )
         rows = normalize_rows(stage.rows)
+        if stage.csv_source is not None and "csv_source" in stage.metadata:
+            raise DatasourceValidationError("csv_source metadata is reserved for verified CSV import evidence.")
+        csv_payload = prepare_csv_source(stage.csv_source, rows)
         schema = infer_schema(rows)
         governance = replace(
             governance,
@@ -768,7 +807,11 @@ class SqlDatasourceProvider:
             byte_count=encoded_size(rows),
             validation_=validation,
             provenance_=dict(stage.provenance),
-            metadata_=dict(stage.metadata),
+            metadata_={
+                **dict(stage.metadata),
+                **({"csv_source": csv_source_summary(csv_payload)} if csv_payload is not None else {}),
+            },
+            csv_source_=csv_payload,
             governance_=governance.to_dict(),
             created_by=_actor_id(api_principal),
         )
@@ -880,6 +923,7 @@ class SqlDatasourceProvider:
             )
         if stage.state != "ready":
             raise DatasourceValidationError("Only ready stages can be promoted.")
+        verify_stage_csv(stage.csv_source_, stage.rows)
         datasource = (
             db.get(DatasourceRecord, stage.target_datasource_id)
             if stage.target_datasource_id
@@ -960,6 +1004,7 @@ class SqlDatasourceProvider:
                 "stage_approval": dict(stage.approval_),
             },
             metadata=dict(stage.metadata_),
+            csv_source=stage.csv_source_,
             set_current=True,
         )
         stage.state = "promoted"
@@ -1283,6 +1328,11 @@ class SqlDatasourceProvider:
             raise DatasourceUnavailableError(
                 "The datasource has no current state to freeze."
             )
+        if current.csv_source_ is not None:
+            try:
+                verified_csv_source_text(current.csv_source_, expected_summary=current.metadata_.get("csv_source") or {})
+            except TabularSourceError as exc:
+                raise DatasourceUnavailableError(str(exc)) from exc
         current_payload = payload_for_materialization(db, current)
         materialization = _append_materialization(
             db,
@@ -1295,6 +1345,8 @@ class SqlDatasourceProvider:
             frozen=True,
             frozen_label=label,
             source_timestamp=current.source_timestamp,
+            csv_source=current.csv_source_,
+            csv_source_governance=current.governance_snapshot_,
             provenance={
                 **dict(current.provenance_),
                 "frozen_from": _materialization_ref(current.id),
@@ -1550,8 +1602,9 @@ class SqlDatasourceProvider:
         schema: Sequence[DatasourceField],
         action: str = "read",
         additional_policies: Sequence[Mapping[str, object]] = (),
+        governance_override: DatasourceGovernance | None = None,
     ) -> VisibilityPlan:
-        governance = _datasource_governance(item)
+        governance = governance_override or _datasource_governance(item)
         policies: list[Mapping[str, object]] = []
         if governance.visibility_policy:
             policies.append(governance.visibility_policy)
@@ -1766,6 +1819,8 @@ def _append_materialization(
     set_current: bool,
     reusable_payload: DatasourcePayloadRecord | None = None,
     state: str = "published",
+    csv_source: Mapping[str, object] | None = None,
+    csv_source_governance: Mapping[str, object] | None = None,
 ) -> DatasourceMaterializationRecord:
     datasource = _lock_datasource_for_materialization(session, datasource)
     revision = _allocate_materialization_revision(session, datasource)
@@ -1788,6 +1843,18 @@ def _append_materialization(
         datasource,
         schema,
     )
+    csv_payload = dict(csv_source) if csv_source is not None else None
+    if csv_payload is not None:
+        history = csv_payload.get("governance_history", [])
+        if not isinstance(history, list) or any(not isinstance(value, Mapping) for value in history):
+            raise DatasourceValidationError("Original CSV inherited governance is invalid.")
+        history = [dict(value) for value in history]
+        for snapshot in (csv_source_governance, _datasource_governance(datasource).to_dict()):
+            if snapshot is not None and dict(snapshot) not in history:
+                history.append(dict(snapshot))
+        if len(history) > 32:
+            raise DatasourceValidationError("Original CSV governance history exceeds the safe copy limit; review the source before creating a new stage.")
+        csv_payload["governance_history"] = history
     materialization = DatasourceMaterializationRecord(
         tenant_id=datasource.tenant_id,
         datasource_id=datasource.id,
@@ -1798,6 +1865,7 @@ def _append_materialization(
         payload_id=payload.id,
         payload_checksum=payload.checksum,
         rows=[],
+        csv_source_=csv_payload,
         fingerprint=fingerprint,
         row_count=payload.row_count,
         byte_count=payload.byte_count,
@@ -1805,7 +1873,7 @@ def _append_materialization(
         frozen_label=_clean_optional(frozen_label),
         source_timestamp=source_timestamp,
         provenance_=dict(provenance or {}),
-        metadata_=dict(metadata or {}),
+        metadata_={**dict(metadata or {}), **({"csv_source": csv_source_summary(csv_payload)} if csv_payload is not None else {})},
         governance_snapshot_=_datasource_governance(datasource).to_dict(),
         created_by=actor_id,
     )

src/govoplan_datasources/backend/tabular.py

@@ -10,6 +10,8 @@ from decimal import Decimal
 from typing import Any
 
 from govoplan_core.core.datasources import DatasourceField, DatasourceValidationError
+from govoplan_core.core.tabular_sources import CsvValueMode, TabularSourceError, parse_tabular_csv
+from govoplan_core.core.tabular_sources import infer_tabular_schema, tabular_type_name as _type_name
 
 
 MAX_STAGE_ROWS = 10_000
@@ -36,10 +38,22 @@ def parse_csv_rows(
     csv_text: str,
     *,
     delimiter: str = ",",
+    value_mode: CsvValueMode = "legacy_typed",
 ) -> tuple[dict[str, Any], ...]:
     if delimiter not in {",", ";", "\t", "|"}:
         raise DatasourceValidationError("Unsupported CSV delimiter.")
-    if len(csv_text.encode("utf-8")) > MAX_STAGE_BYTES:
+    if value_mode == "text":
+        try:
+            return normalize_rows(parse_tabular_csv(csv_text, delimiter=delimiter, value_mode="text", max_rows=MAX_STAGE_ROWS, max_bytes=MAX_STAGE_BYTES))
+        except TabularSourceError as exc:
+            raise DatasourceValidationError(str(exc)) from exc
+    if value_mode != "legacy_typed":
+        raise DatasourceValidationError("Unsupported CSV value mode.")
+    try:
+        encoded_text = csv_text.encode("utf-8")
+    except UnicodeError as exc:
+        raise DatasourceValidationError("CSV input must be valid Unicode encodable as UTF-8.") from exc
+    if len(encoded_text) > MAX_STAGE_BYTES:
         raise DatasourceValidationError(
             f"Staging is limited to {MAX_STAGE_BYTES // 1_000_000} MB."
         )
@@ -58,6 +72,8 @@ def parse_csv_rows(
             raise DatasourceValidationError("CSV headers must be unique.")
         rows: list[dict[str, object]] = []
         for line_number, values in enumerate(reader, start=2):
+            if len(rows) >= MAX_STAGE_ROWS:
+                raise DatasourceValidationError(f"Staging is limited to {MAX_STAGE_ROWS:,} rows.")
             if len(values) != len(header):
                 raise DatasourceValidationError(
                     f"CSV row {line_number} has {len(values)} values; expected {len(header)}."
@@ -76,26 +92,7 @@ def parse_csv_rows(
 def infer_schema(
     rows: Sequence[Mapping[str, object]],
 ) -> tuple[DatasourceField, ...]:
-    names: list[str] = []
-    for row in rows:
-        for name in row:
-            if name not in names:
-                names.append(name)
-    result: list[DatasourceField] = []
-    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(
-            DatasourceField(
-                name=name,
-                data_type=data_type,
-                nullable=len(concrete) != len(values),
-            )
-        )
-    return tuple(result)
+    return tuple(DatasourceField(name=column.name, data_type=column.data_type, nullable=column.nullable) for column in infer_tabular_schema(rows, type_name=_type_name))
 
 
 def fingerprint_rows(
@@ -168,22 +165,6 @@ def _json_value(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 _csv_value(value: str) -> object:
     cleaned = value.strip()
     if not cleaned: