feat(connectors): add managed file and PostgreSQL origins

src/govoplan_connectors/backend/datasource_origins.py

@@ -102,12 +102,16 @@ class ConnectorDatasourceOriginProvider:
 
 
 def _origin(source: TabularSource) -> DatasourceOrigin:
+    kind = {
+        "managed_file": "file",
+        "postgresql": "database",
+    }.get(source.provider, "upload")
     return DatasourceOrigin(
         ref=source.ref,
         source_name=source.source_name,
         name=source.name,
         description=source.description,
-        kind="upload",
+        kind=kind,
         shape="tabular",
         supported_modes=("live", "cached"),
         provider=f"connectors.{source.provider}",

src/govoplan_connectors/backend/manifest.py

@@ -18,6 +18,7 @@ from govoplan_core.core.modules import (
     FrontendModule,
     MigrationSpec,
     ModuleInterfaceProvider,
+    ModuleInterfaceRequirement,
     ModuleManifest,
     PermissionDefinition,
     RoleTemplate,
@@ -36,6 +37,7 @@ from govoplan_core.core.tabular_sources import (
     CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
     CAPABILITY_CONNECTORS_TABULAR_SOURCES,
 )
+from govoplan_core.core.files import CAPABILITY_FILES_TABULAR_CONTENT
 from govoplan_core.core.sanctions import (
     CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,
 )
@@ -81,7 +83,7 @@ from govoplan_connectors.backend.provider_state import (
 
 
 MODULE_ID = "connectors"
-MODULE_VERSION = "0.1.19"
+MODULE_VERSION = "0.1.20"
 TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
 DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
 SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0"
@@ -98,6 +100,16 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
             reference="tests/test_tabular_sources.py",
             summary="Exercises tenant-safe immutable tabular snapshots and bounded reads.",
         ),
+        ModuleMaturityEvidence(
+            kind="test",
+            reference="tests/test_tabular_origin_provider.py",
+            summary="Exercises exact managed-file versions, reviewed refresh, live SQL projection, configuration drift, and tenant isolation.",
+        ),
+        ModuleMaturityEvidence(
+            kind="test",
+            reference="tests/test_tabular_adapters.py",
+            summary="Exercises bounded CSV/XLSX parsing and the credential-governed read-only PostgreSQL adapter.",
+        ),
         ModuleMaturityEvidence(
             kind="test",
             reference="tests/test_sanctions_sources.py",
@@ -120,7 +132,7 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
         ),
     ),
     known_limits=(
-        "The executable generic datasource origin is an immutable tabular snapshot; database and arbitrary REST profiles remain future providers.",
+        "Tabular origins support immutable snapshots, exact managed CSV/XLSX versions, and read-only PostgreSQL tables; arbitrary REST and other database adapters remain future providers.",
         "Feed publication renders a governed document but does not yet push it to an external publishing endpoint.",
         "The generic governed runtime simulates deterministic mapping and validation; provider-specific live writes remain owned by explicit connector adapters.",
     ),
@@ -158,7 +170,7 @@ EXTERNAL_PROVIDERS = (
     ExternalProviderDeclaration(
         id=TABULAR_PROVIDER_ID,
         module_id=MODULE_ID,
-        label="Immutable tabular snapshot provider",
+        label="Governed tabular source providers",
         maturity="read",
         operations=("discover", "search", "read", "preview", "dry_run"),
         objects=(
@@ -170,10 +182,10 @@ EXTERNAL_PROVIDERS = (
             ),
         ),
         behavior=ProviderBehaviorDeclaration(
-            revision_tokens="Source fingerprints and immutable snapshot ids are retained.",
-            concurrency="Reads may require the expected fingerprint; snapshots never mutate in place.",
-            freshness="Snapshot acquisition time and source timestamp are exposed.",
-            health="Import validation and source-read failures are explicit.",
+            revision_tokens="Source fingerprints, exact file versions, SQL configuration revisions, credential revisions, and discovery revisions are retained.",
+            concurrency="Reads may require the expected fingerprint; file versions remain pinned and SQL configuration or schema drift blocks reads until reviewed refresh.",
+            freshness="Snapshot acquisition time, exact/current file versions, and live SQL discovery health are exposed.",
+            health="Import validation, Files access/integrity, SQL authentication, configuration drift, schema drift, timeouts, and source unavailability are explicit without exposing secrets.",
             max_read_items=1000,
             idempotency="Feed imports accept a caller request key and replay the same committed immutable source without refetching.",
             retry="Read-only acquisition may be retried only as a new deliberate request after a failed atomic operation.",
@@ -187,7 +199,7 @@ EXTERNAL_PROVIDERS = (
             classifications=("internal", "confidential", "restricted"),
             purposes=("governed import", "dataflow input", "evidence reconstruction"),
             retention="Datasources or the consuming domain supplies retention and hold policy.",
-            secret_handling="Generic snapshots contain no connector credential; transport credentials stay in credential envelopes.",
+            secret_handling="Snapshots and managed-file sources contain no connector credential; PostgreSQL uses a scoped Core credential-envelope reference and never stores or returns the resolved secret.",
         ),
         capability_names=(
             CAPABILITY_CONNECTORS_TABULAR_SOURCES,
@@ -347,11 +359,13 @@ def _router(_context):
 
 
 def _provider(_context) -> SqlTabularSourceProvider:
-    return SqlTabularSourceProvider()
+    return SqlTabularSourceProvider(registry=getattr(_context, "registry", None))
 
 
 def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider:
-    return ConnectorDatasourceOriginProvider()
+    return ConnectorDatasourceOriginProvider(
+        SqlTabularSourceProvider(registry=getattr(_context, "registry", None))
+    )
 
 
 def _sanctions_snapshot_provider(
@@ -451,6 +465,14 @@ manifest = ModuleManifest(
         ),
         ModuleInterfaceProvider(name=CONNECTORS_DSAR_CAPABILITY, version="0.1.0"),
     ),
+    requires_interfaces=(
+        ModuleInterfaceRequirement(
+            name=CAPABILITY_FILES_TABULAR_CONTENT,
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
+    ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_router,
@@ -646,11 +668,16 @@ manifest = ModuleManifest(
                 "fingerprints, and bounded reads. Dataflow stores only opaque source "
                 "references and expected fingerprints. Each source declares its live, "
                 "cached, file-backed, or static mode, structured health, and supported "
-                "projection, filter, aggregation, sorting, and pagination pushdown. The "
-                "first executable provider imports immutable JSON or CSV snapshots, "
-                "supports projection and pagination, and exposes them as Datasource "
-                "origins. Database and API providers can implement the same origin "
-                "contract without changing Datasources or Dataflow."
+                "projection, filter, aggregation, sorting, and pagination pushdown. "
+                "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 "
+                "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 "
+                "schema drift until reviewed refresh. Credentials, endpoints, storage "
+                "keys, and raw file internals are never exposed through the origin."
             ),
             layer="available",
             documentation_types=("admin", "user"),

src/govoplan_connectors/backend/provider_state.py

@@ -97,6 +97,13 @@ def _tabular_state(
     observed_at: datetime,
 ) -> ExternalProviderRuntimeState:
     active = source.status == "active"
+    live_origin = source.provider in {"managed_file", "postgresql"}
+    labels = {
+        "managed_file": "Exact managed-file origin",
+        "postgresql": "Live PostgreSQL origin",
+        "snapshot": "Immutable tabular snapshot",
+    }
+    label = labels.get(source.provider, "Tabular source")
     return ExternalProviderRuntimeState(
         provider_id=TABULAR_PROVIDER_ID,
         binding_ref=f"connectors:tabular-source:{source.id}",
@@ -104,17 +111,20 @@ def _tabular_state(
         observed_at=observed_at,
         configured=True,
         active=active,
-        health="healthy" if active else "inactive",
-        freshness="not_applicable",
+        health=("unknown" if active and live_origin else "healthy" if active else "inactive"),
+        freshness="unknown" if active and live_origin else "not_applicable",
         conflict="not_applicable",
         recovery="ready" if active else "not_applicable",
         last_success_at=_aware(source.updated_at or source.created_at),
         detail=(
-            "Immutable tabular snapshot is available."
+            f"{label} is configured; live access and drift are checked on preview."
+            if active and live_origin
+            else f"{label} is available."
             if active
-            else "Immutable tabular snapshot is inactive."
+            else f"{label} is inactive."
         ),
         metrics={
+            "provider": source.provider,
             "row_count": int(source.row_count),
             "byte_count": int(source.byte_count),
             "schema_version": int(source.schema_version),

src/govoplan_connectors/backend/router.py

@@ -32,6 +32,7 @@ from govoplan_connectors.backend.schemas import (
     FeedImportRequest,
     FeedPublicationEntryPayload,
     FeedRenderPayload,
+    ManagedFileSourceCreateRequest,
     SanctionsAcquisitionRunListResponse,
     SanctionsAcquisitionRunResponse,
     SanctionsRefreshResponse,
@@ -40,6 +41,7 @@ from govoplan_connectors.backend.schemas import (
     SanctionsSourceListResponse,
     SanctionsSourceResponse,
     SnapshotCreateRequest,
+    SqlSourceCreateRequest,
     TabularColumnResponse,
     TabularHealthResponse,
     TabularPreviewDiagnosticResponse,
@@ -460,12 +462,82 @@ def api_create_tabular_snapshot(
     return _source_response(source)
 
 
+@router.post(
+    "/tabular-sources/files",
+    response_model=TabularSourceResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_managed_file_source(
+    payload: ManagedFileSourceCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        source = provider.create_file_source(
+            session,
+            principal,
+            name=payload.name,
+            source_name=payload.source_name,
+            description=payload.description,
+            file_asset_id=payload.file_asset_id,
+            file_version_id=payload.file_version_id,
+            delimiter=payload.delimiter,
+            sheet_name=payload.sheet_name,
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    _audit_tabular_source_change(
+        session,
+        principal,
+        source,
+        operation="created",
+    )
+    session.commit()
+    return _source_response(source)
+
+
+@router.post(
+    "/tabular-sources/sql",
+    response_model=TabularSourceResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_sql_source(
+    payload: SqlSourceCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        source = provider.create_sql_source(
+            session,
+            principal,
+            name=payload.name,
+            source_name=payload.source_name,
+            description=payload.description,
+            configuration_id=payload.configuration_id,
+            schema_name=payload.schema_name,
+            table_name=payload.table_name,
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    _audit_tabular_source_change(
+        session,
+        principal,
+        source,
+        operation="created",
+    )
+    session.commit()
+    return _source_response(source)
+
+
 @router.get(
     "/tabular-sources/{source_id}/preview",
     response_model=TabularSourcePreviewResponse,
 )
 def api_preview_tabular_source(
     source_id: str,
+    kind: Annotated[str, Query(pattern=r"^(snapshot|file|sql)$")] = "snapshot",
     limit: int = Query(default=100, ge=1, le=500),
     offset: int = Query(default=0, ge=0),
     max_bytes: int = Query(default=1_000_000, ge=2, le=5_000_000),
@@ -479,7 +551,7 @@ def api_preview_tabular_source(
             session,
             principal,
             request=TabularReadRequest(
-                source_ref=f"snapshot:{source_id}",
+                source_ref=f"{kind}:{source_id}",
                 limit=limit,
                 offset=offset,
                 max_bytes=max_bytes,
@@ -510,17 +582,47 @@ def api_preview_tabular_source(
     )
 
 
+@router.post(
+    "/tabular-sources/{source_id}/refresh",
+    response_model=TabularSourceResponse,
+)
+def api_refresh_tabular_source(
+    source_id: str,
+    kind: Annotated[str, Query(pattern=r"^(file|sql)$")],
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        source = provider.refresh_source(
+            session,
+            principal,
+            source_ref=f"{kind}:{source_id}",
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    _audit_tabular_source_change(
+        session,
+        principal,
+        source,
+        operation="refreshed",
+    )
+    session.commit()
+    return _source_response(source)
+
+
 @router.delete(
     "/tabular-sources/{source_id}",
     response_model=TabularSourceDeleteResponse,
 )
 def api_delete_tabular_source(
     source_id: str,
+    kind: Annotated[str, Query(pattern=r"^(snapshot|file|sql)$")] = "snapshot",
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> TabularSourceDeleteResponse:
     _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
-    source_ref = f"snapshot:{source_id}"
+    source_ref = f"{kind}:{source_id}"
     try:
         source = provider.delete_snapshot(
             session,
@@ -534,7 +636,7 @@ def api_delete_tabular_source(
         tenant_id=principal.tenant_id,
         user_id=getattr(principal.user, "id", None),
         api_key_id=principal.api_key_id,
-        action="connectors.tabular_snapshot.deleted",
+        action="connectors.tabular_source.deleted",
         object_type="connector_tabular_source",
         object_id=source_ref,
         details={"source_name": source.source_name, "fingerprint": source.fingerprint},
@@ -928,6 +1030,32 @@ def _source_response(source: TabularSource) -> TabularSourceResponse:
     )
 
 
+def _audit_tabular_source_change(
+    session: Session,
+    principal: ApiPrincipal,
+    source: TabularSource,
+    *,
+    operation: str,
+) -> None:
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action=f"connectors.tabular_source.{operation}",
+        object_type="connector_tabular_source",
+        object_id=source.ref,
+        details={
+            "provider": source.provider,
+            "source_name": source.source_name,
+            "fingerprint": source.fingerprint,
+            "schema_version": source.schema_version,
+            "row_count": source.row_count,
+            "source_mode": source.source_mode,
+        },
+    )
+
+
 def _sanctions_snapshot_response(
     snapshot: SanctionsSnapshotReference,
 ) -> SanctionsSnapshotResponse:

src/govoplan_connectors/backend/schemas.py

@@ -105,6 +105,40 @@ class SnapshotCreateRequest(BaseModel):
         return self
 
 
+class ManagedFileSourceCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=300)
+    source_name: str = Field(
+        min_length=1,
+        max_length=120,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_]*$",
+    )
+    description: str | None = Field(default=None, max_length=4000)
+    file_asset_id: str = Field(min_length=1, max_length=36)
+    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)
+
+
+class SqlSourceCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=300)
+    source_name: str = Field(
+        min_length=1,
+        max_length=120,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_]*$",
+    )
+    description: str | None = Field(default=None, max_length=4000)
+    configuration_id: str = Field(min_length=1, max_length=36)
+    schema_name: str | None = Field(
+        default=None,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$",
+    )
+    table_name: str = Field(
+        min_length=1,
+        max_length=128,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$",
+    )
+
+
 class TabularColumnResponse(BaseModel):
     name: str
     data_type: str
@@ -250,7 +284,9 @@ __all__ = [
     "FeedEntryPayload",
     "FeedImportRequest",
     "FeedRenderPayload",
+    "ManagedFileSourceCreateRequest",
     "SnapshotCreateRequest",
+    "SqlSourceCreateRequest",
     "SanctionsAcquisitionRunListResponse",
     "SanctionsAcquisitionRunResponse",
     "SanctionsRefreshResponse",

src/govoplan_connectors/backend/tabular_adapters.py

@@ -0,0 +1,849 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import re
+import zipfile
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal
+from io import BytesIO
+from typing import Any
+
+from openpyxl import load_workbook
+from sqlalchemy import (
+    JSON,
+    BigInteger,
+    Boolean,
+    Date,
+    DateTime,
+    Float,
+    Integer,
+    LargeBinary,
+    MetaData,
+    Numeric,
+    String,
+    Table,
+    Text,
+    create_engine,
+    func,
+    select,
+)
+from sqlalchemy.engine import Engine, URL, make_url
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.files import (
+    ManagedTabularFileAccessError,
+    ManagedTabularFileError,
+    ManagedTabularFileValidationError,
+    managed_tabular_file_provider,
+)
+from govoplan_core.core.tabular_sources import (
+    TabularColumn,
+    TabularPreviewDiagnostic,
+    TabularPushdown,
+    TabularSourceHealth,
+    TabularSourceUnavailableError,
+    TabularSourceValidationError,
+    parse_tabular_csv,
+)
+from govoplan_core.security.credential_envelopes import (
+    CredentialAccessContext,
+    CredentialEnvelopeError,
+    resolve_credential_envelope,
+)
+from govoplan_core.security.redaction import is_sensitive_key
+from govoplan_connectors.backend.db.models import ConnectorConfiguration
+
+
+MAX_FILE_BYTES = 5_000_000
+MAX_FILE_ROWS = 10_000
+MAX_FILE_COLUMNS = 500
+MAX_XLSX_ENTRIES = 5_000
+MAX_XLSX_EXPANDED_BYTES = 50_000_000
+MAX_XLSX_COMPRESSION_RATIO = 100
+POSTGRESQL_SCHEMES = frozenset({"postgresql", "postgresql+psycopg"})
+_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$")
+
+
+@dataclass(frozen=True, slots=True)
+class TabularOriginInspection:
+    provider: str
+    schema: tuple[TabularColumn, ...]
+    fingerprint: str
+    row_count: int
+    byte_count: int
+    metadata: Mapping[str, object]
+    health: TabularSourceHealth
+    pushdown: TabularPushdown
+    rows: tuple[Mapping[str, object], ...] = ()
+    diagnostics: tuple[TabularPreviewDiagnostic, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class TabularOriginRead:
+    inspection: TabularOriginInspection
+    rows: tuple[Mapping[str, object], ...]
+    total_rows: int
+    diagnostics: tuple[TabularPreviewDiagnostic, ...] = ()
+
+
+class ManagedFileTabularAdapter:
+    def __init__(self, registry: object | None) -> None:
+        self._registry = registry
+
+    def inspect(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        file_asset_id: str,
+        file_version_id: str | None,
+        delimiter: str = ",",
+        sheet_name: str | None = None,
+    ) -> TabularOriginInspection:
+        provider = managed_tabular_file_provider(self._registry)
+        if provider is None:
+            raise TabularSourceUnavailableError(
+                "Managed file sources require the Files module."
+            )
+        try:
+            metadata = provider.get_tabular_file(
+                session,
+                principal,
+                file_asset_id=file_asset_id,
+                file_version_id=file_version_id,
+            )
+            if metadata is None:
+                raise TabularSourceUnavailableError(
+                    "Managed tabular file or version is unavailable."
+                )
+            content = provider.read_tabular_file(
+                session,
+                principal,
+                file_asset_id=file_asset_id,
+                file_version_id=metadata.file_version_id,
+                max_bytes=MAX_FILE_BYTES,
+            )
+        except ManagedTabularFileAccessError as exc:
+            raise TabularSourceUnavailableError(
+                "Managed tabular file access is no longer authorized."
+            ) from exc
+        except ManagedTabularFileValidationError as exc:
+            raise TabularSourceValidationError(str(exc)) from exc
+        except ManagedTabularFileError as exc:
+            raise TabularSourceUnavailableError(str(exc)) from exc
+        rows, resolved_sheet = parse_managed_tabular_content(
+            content.payload,
+            filename=content.file.filename,
+            content_type=content.file.content_type,
+            delimiter=delimiter,
+            sheet_name=sheet_name,
+        )
+        schema = infer_tabular_schema(rows)
+        fingerprint = origin_fingerprint(
+            schema,
+            tokens=(
+                "managed_file",
+                content.file.file_asset_id,
+                content.file.file_version_id,
+                content.file.sha256,
+                resolved_sheet or "",
+                delimiter,
+            ),
+        )
+        try:
+            current = provider.get_tabular_file(
+                session,
+                principal,
+                file_asset_id=file_asset_id,
+            )
+        except ManagedTabularFileAccessError as exc:
+            raise TabularSourceUnavailableError(
+                "Managed tabular file access is no longer authorized."
+            ) from exc
+        except ManagedTabularFileError as exc:
+            raise TabularSourceUnavailableError(str(exc)) from exc
+        changed = bool(
+            current is not None
+            and current.file_version_id != content.file.file_version_id
+        )
+        health = TabularSourceHealth(
+            status="warning" if changed else "healthy",
+            code=(
+                "files.newer_version_available"
+                if changed
+                else "files.exact_version_ready"
+            ),
+            summary=(
+                "A newer managed file version is available for explicit review."
+                if changed
+                else "The exact managed file version passed access and integrity checks."
+            ),
+            checked_at=content.file.updated_at,
+            details={
+                "pinned_version_id": content.file.file_version_id,
+                "current_version_id": (
+                    current.file_version_id if current is not None else None
+                ),
+            },
+        )
+        return TabularOriginInspection(
+            provider="managed_file",
+            schema=schema,
+            fingerprint=fingerprint,
+            row_count=len(rows),
+            byte_count=len(content.payload),
+            metadata={
+                "origin_kind": "managed_file",
+                "file_asset_id": content.file.file_asset_id,
+                "file_version_id": content.file.file_version_id,
+                "file_sha256": content.file.sha256,
+                "filename": content.file.filename,
+                "content_type": content.file.content_type,
+                "format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv",
+                "delimiter": delimiter,
+                "sheet_name": resolved_sheet,
+            },
+            health=health,
+            pushdown=TabularPushdown(projections=True, pagination=True),
+            rows=rows,
+            diagnostics=(
+                (
+                    TabularPreviewDiagnostic(
+                        severity="warning",
+                        code="files.newer_version_available",
+                        message=(
+                            "The preview remains pinned to the reviewed file version; "
+                            "refresh the source to adopt the newer version."
+                        ),
+                        details=dict(health.details),
+                    ),
+                )
+                if changed
+                else ()
+            ),
+        )
+
+    def read(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        metadata: Mapping[str, object],
+    ) -> TabularOriginRead:
+        inspection = self.inspect(
+            session,
+            principal,
+            file_asset_id=_required_metadata(metadata, "file_asset_id"),
+            file_version_id=_required_metadata(metadata, "file_version_id"),
+            delimiter=str(metadata.get("delimiter") or ","),
+            sheet_name=_optional_text(metadata.get("sheet_name")),
+        )
+        expected_sha256 = _required_metadata(metadata, "file_sha256")
+        if inspection.metadata.get("file_sha256") != expected_sha256:
+            raise TabularSourceUnavailableError(
+                "Managed file content no longer matches its exact-version checksum."
+            )
+        return TabularOriginRead(
+            inspection=inspection,
+            rows=inspection.rows,
+            total_rows=inspection.row_count,
+            diagnostics=inspection.diagnostics,
+        )
+
+
+class PostgresqlTabularAdapter:
+    """Read-only SQLAlchemy adapter with a production PostgreSQL allow-list."""
+
+    def __init__(
+        self,
+        *,
+        engine_factory: Callable[[URL], Engine] | None = None,
+        allow_sqlite_for_tests: bool = False,
+    ) -> None:
+        self._engine_factory = engine_factory or (
+            lambda url: create_engine(url, pool_pre_ping=True)
+        )
+        self._allow_sqlite_for_tests = allow_sqlite_for_tests
+
+    def inspect(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        configuration_id: str,
+        table_name: str,
+        schema_name: str | None = None,
+        timeout_ms: int = 2_000,
+    ) -> TabularOriginInspection:
+        configuration, url, credential_revision = self._connection(
+            session,
+            principal,
+            configuration_id=configuration_id,
+        )
+        table_name = _sql_identifier(table_name, "table")
+        schema_name = (
+            _sql_identifier(schema_name, "schema") if schema_name else None
+        )
+        engine = self._engine_factory(_bounded_connection_url(url, timeout_ms))
+        try:
+            with engine.connect() as connection:
+                _configure_read_only(connection, url, timeout_ms)
+                table = Table(
+                    table_name,
+                    MetaData(),
+                    schema=schema_name,
+                    autoload_with=connection,
+                )
+                schema = tuple(_sql_column(column) for column in table.columns)
+                if not schema:
+                    raise TabularSourceValidationError(
+                        "SQL tabular sources require at least one column."
+                    )
+                row_count = int(
+                    connection.execute(select(func.count()).select_from(table)).scalar_one()
+                )
+        except TabularSourceValidationError:
+            raise
+        except SQLAlchemyError as exc:
+            raise TabularSourceUnavailableError(
+                "SQL source discovery failed; verify the active configuration, credential, table, and provider health."
+            ) from exc
+        finally:
+            engine.dispose()
+        fingerprint = origin_fingerprint(
+            schema,
+            tokens=(
+                "postgresql",
+                configuration.id,
+                configuration.effective_hash,
+                credential_revision or "",
+                schema_name or "",
+                table_name,
+            ),
+        )
+        return TabularOriginInspection(
+            provider="postgresql",
+            schema=schema,
+            fingerprint=fingerprint,
+            row_count=row_count,
+            byte_count=0,
+            metadata={
+                "origin_kind": "sql",
+                "configuration_id": configuration.id,
+                "configuration_revision": configuration.resource_revision,
+                "configuration_hash": configuration.effective_hash,
+                "credential_revision": credential_revision,
+                "schema_name": schema_name,
+                "table_name": table_name,
+            },
+            health=TabularSourceHealth(
+                status="healthy",
+                code="sql.source_ready",
+                summary="The governed PostgreSQL source is reachable and its schema was discovered.",
+                checked_at=configuration.updated_at,
+                details={
+                    "configuration_id": configuration.id,
+                    "configuration_revision": configuration.resource_revision,
+                },
+            ),
+            pushdown=TabularPushdown(projections=True, pagination=True),
+        )
+
+    def read(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        metadata: Mapping[str, object],
+        columns: Sequence[str],
+        offset: int,
+        limit: int,
+        timeout_ms: int,
+    ) -> TabularOriginRead:
+        configuration_id = _required_metadata(metadata, "configuration_id")
+        current_configuration, _current_url, current_credential_revision = (
+            self._connection(
+                session,
+                principal,
+                configuration_id=configuration_id,
+            )
+        )
+        if current_configuration.effective_hash != metadata.get(
+            "configuration_hash"
+        ):
+            raise TabularSourceValidationError(
+                "The SQL connector configuration changed; refresh the source before previewing it."
+            )
+        if current_credential_revision != metadata.get("credential_revision"):
+            raise TabularSourceValidationError(
+                "The SQL source credential changed; refresh the source before previewing it."
+            )
+        inspection = self.inspect(
+            session,
+            principal,
+            configuration_id=configuration_id,
+            table_name=_required_metadata(metadata, "table_name"),
+            schema_name=_optional_text(metadata.get("schema_name")),
+            timeout_ms=timeout_ms,
+        )
+        if inspection.fingerprint != metadata.get("discovery_fingerprint"):
+            raise TabularSourceValidationError(
+                "The SQL source schema drifted; refresh and review the source before previewing it."
+            )
+        configuration, url, _credential_revision = self._connection(
+            session,
+            principal,
+            configuration_id=configuration_id,
+        )
+        engine = self._engine_factory(_bounded_connection_url(url, timeout_ms))
+        try:
+            with engine.connect() as connection:
+                _configure_read_only(connection, url, timeout_ms)
+                table = Table(
+                    _required_metadata(metadata, "table_name"),
+                    MetaData(),
+                    schema=_optional_text(metadata.get("schema_name")),
+                    autoload_with=connection,
+                )
+                selected_names = tuple(dict.fromkeys(str(item) for item in columns))
+                selected = (
+                    [table.c[name] for name in selected_names]
+                    if selected_names
+                    else list(table.columns)
+                )
+                statement = select(*selected).offset(max(0, int(offset))).limit(
+                    max(1, int(limit))
+                )
+                rows = tuple(
+                    _json_row(dict(row._mapping))
+                    for row in connection.execute(statement)
+                )
+        except KeyError as exc:
+            raise TabularSourceValidationError(
+                f"Unknown SQL source column: {exc.args[0]}"
+            ) from exc
+        except SQLAlchemyError as exc:
+            raise TabularSourceUnavailableError(
+                "SQL source preview failed or exceeded its provider budget."
+            ) from exc
+        finally:
+            engine.dispose()
+        return TabularOriginRead(
+            inspection=inspection,
+            rows=rows,
+            total_rows=inspection.row_count,
+        )
+
+    def _connection(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        configuration_id: str,
+    ) -> tuple[ConnectorConfiguration, URL, str | None]:
+        configuration = session.scalar(
+            select(ConnectorConfiguration).where(
+                ConnectorConfiguration.id == configuration_id,
+                ConnectorConfiguration.tenant_id == principal.tenant_id,
+            )
+        )
+        if configuration is None:
+            raise TabularSourceUnavailableError(
+                "SQL connector configuration is unavailable."
+            )
+        if configuration.status != "active":
+            raise TabularSourceUnavailableError(
+                "SQL connector configuration is not active."
+            )
+        endpoint = _optional_text(configuration.endpoint_url)
+        if not endpoint:
+            raise TabularSourceUnavailableError(
+                "SQL connector configuration has no endpoint."
+            )
+        specification = dict(configuration.effective_configuration or {})
+        provider = str(specification.get("provider") or "").strip().casefold()
+        protocol = str(specification.get("protocol") or "").strip().casefold()
+        if provider not in {"postgres", "postgresql", "sql"} or protocol not in {
+            "postgres",
+            "postgresql",
+            "sql",
+        }:
+            raise TabularSourceValidationError(
+                "The selected connector configuration is not a PostgreSQL tabular reader."
+            )
+        try:
+            url = make_url(endpoint)
+        except (TypeError, ValueError) as exc:
+            raise TabularSourceValidationError(
+                "SQL connector endpoint is invalid."
+            ) from exc
+        if url.username or url.password:
+            raise TabularSourceValidationError(
+                "SQL connector endpoints must not contain credentials."
+            )
+        if any(is_sensitive_key(key) for key in url.query):
+            raise TabularSourceValidationError(
+                "SQL connector endpoint query parameters must not contain credentials."
+            )
+        allowed_schemes = set(POSTGRESQL_SCHEMES)
+        if self._allow_sqlite_for_tests:
+            allowed_schemes.update({"sqlite", "sqlite+pysqlite"})
+        if url.drivername not in allowed_schemes:
+            raise TabularSourceValidationError(
+                "Only the governed PostgreSQL tabular adapter is enabled."
+            )
+        credential_revision: str | None = None
+        credential_ref = _optional_text(configuration.credential_ref)
+        if credential_ref:
+            try:
+                credential = resolve_credential_envelope(
+                    session,
+                    credential_id=credential_ref,
+                    context=CredentialAccessContext(
+                        tenant_id=principal.tenant_id,
+                        user_id=_principal_user_id(principal),
+                        group_ids=frozenset(principal.principal.group_ids),
+                        target_scope_type="tenant",
+                        target_scope_id=principal.tenant_id,
+                        module_id="connectors",
+                        server_ref=endpoint,
+                    ),
+                )
+            except CredentialEnvelopeError as exc:
+                raise TabularSourceUnavailableError(
+                    "SQL source credential is unavailable, inactive, or outside its allowed scope."
+                ) from exc
+            public_data = dict(credential.public_data)
+            secret_data = dict(credential.secret_data)
+            username = _optional_text(
+                public_data.get("username")
+                or secret_data.get("username")
+                or secret_data.get("user")
+            )
+            password = _optional_text(secret_data.get("password"))
+            if url.drivername in POSTGRESQL_SCHEMES and (not username or not password):
+                raise TabularSourceUnavailableError(
+                    "SQL source credential does not provide a username and password."
+                )
+            url = url.set(username=username, password=password)
+            credential_revision = credential.revision
+        elif url.drivername in POSTGRESQL_SCHEMES:
+            raise TabularSourceUnavailableError(
+                "PostgreSQL tabular sources require a credential envelope reference."
+            )
+        return configuration, url, credential_revision
+
+
+def parse_managed_tabular_content(
+    payload: bytes,
+    *,
+    filename: str,
+    content_type: str | None,
+    delimiter: str,
+    sheet_name: str | None,
+) -> tuple[tuple[Mapping[str, object], ...], str | None]:
+    if len(payload) > MAX_FILE_BYTES:
+        raise TabularSourceValidationError(
+            f"Managed tabular files are limited to {MAX_FILE_BYTES:,} bytes."
+        )
+    if _is_xlsx(filename, content_type):
+        return _parse_xlsx(payload, sheet_name=sheet_name)
+    try:
+        text = payload.decode("utf-8-sig")
+    except UnicodeDecodeError as exc:
+        raise TabularSourceValidationError(
+            "Managed CSV files must use UTF-8 encoding."
+        ) from exc
+    return (
+        tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS)),
+        None,
+    )
+
+
+def _parse_xlsx(
+    payload: bytes,
+    *,
+    sheet_name: str | None,
+) -> tuple[tuple[Mapping[str, object], ...], str]:
+    _validate_xlsx_archive(payload)
+    try:
+        workbook = load_workbook(
+            BytesIO(payload),
+            read_only=True,
+            data_only=True,
+            keep_links=False,
+        )
+    except Exception as exc:
+        raise TabularSourceValidationError(
+            "Managed XLSX content could not be parsed."
+        ) from exc
+    try:
+        available = tuple(workbook.sheetnames)
+        if not available:
+            raise TabularSourceValidationError(
+                "Managed XLSX content requires at least one worksheet."
+            )
+        selected_name = _optional_text(sheet_name) or available[0]
+        if selected_name not in available:
+            raise TabularSourceValidationError(
+                f"Managed XLSX worksheet {selected_name!r} was not found."
+            )
+        worksheet = workbook[selected_name]
+        iterator = worksheet.iter_rows(values_only=True)
+        try:
+            raw_headers = next(iterator)
+        except StopIteration as exc:
+            raise TabularSourceValidationError(
+                "Managed XLSX worksheet requires a header row."
+            ) from exc
+        headers = _xlsx_headers(raw_headers)
+        rows: list[Mapping[str, object]] = []
+        for values in iterator:
+            if len(rows) >= MAX_FILE_ROWS:
+                raise TabularSourceValidationError(
+                    f"Managed XLSX worksheets are limited to {MAX_FILE_ROWS:,} data rows."
+                )
+            normalized = tuple(values[: len(headers)])
+            if all(value in (None, "") for value in normalized):
+                continue
+            rows.append(
+                {
+                    header: _json_value(
+                        normalized[index] if index < len(normalized) else None
+                    )
+                    for index, header in enumerate(headers)
+                }
+            )
+        return tuple(rows), selected_name
+    finally:
+        workbook.close()
+
+
+def _validate_xlsx_archive(payload: bytes) -> None:
+    try:
+        with zipfile.ZipFile(BytesIO(payload)) as archive:
+            entries = archive.infolist()
+            if len(entries) > MAX_XLSX_ENTRIES:
+                raise TabularSourceValidationError(
+                    "Managed XLSX content contains too many archive entries."
+                )
+            expanded = sum(max(0, item.file_size) for item in entries)
+            if expanded > MAX_XLSX_EXPANDED_BYTES:
+                raise TabularSourceValidationError(
+                    "Managed XLSX content exceeds the expanded-size limit."
+                )
+            compressed = sum(max(1, item.compress_size) for item in entries)
+            if expanded > compressed * MAX_XLSX_COMPRESSION_RATIO:
+                raise TabularSourceValidationError(
+                    "Managed XLSX content exceeds the compression-ratio limit."
+                )
+            if any(
+                item.filename.startswith(("/", "\\"))
+                or ".." in item.filename.replace("\\", "/").split("/")
+                for item in entries
+            ):
+                raise TabularSourceValidationError(
+                    "Managed XLSX content contains an unsafe archive path."
+                )
+    except zipfile.BadZipFile as exc:
+        raise TabularSourceValidationError(
+            "Managed XLSX content is not a valid workbook archive."
+        ) from exc
+
+
+def _xlsx_headers(values: Sequence[object]) -> tuple[str, ...]:
+    if len(values) > MAX_FILE_COLUMNS:
+        raise TabularSourceValidationError(
+            f"Managed XLSX worksheets are limited to {MAX_FILE_COLUMNS:,} columns."
+        )
+    headers = tuple(str(value or "").strip() for value in values)
+    while headers and not headers[-1]:
+        headers = headers[:-1]
+    if not headers or any(not header for header in headers):
+        raise TabularSourceValidationError(
+            "Managed XLSX content requires a non-empty header row."
+        )
+    if len(set(headers)) != len(headers):
+        raise TabularSourceValidationError(
+            "Managed XLSX column names must be unique."
+        )
+    return headers
+
+
+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)
+
+
+def origin_fingerprint(
+    schema: Sequence[TabularColumn],
+    *,
+    tokens: Sequence[str],
+) -> str:
+    payload = {
+        "schema": [
+            {
+                "name": column.name,
+                "data_type": column.data_type,
+                "nullable": column.nullable,
+            }
+            for column in schema
+        ],
+        "tokens": list(tokens),
+    }
+    return hashlib.sha256(
+        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+
+
+def _sql_column(column: Any) -> TabularColumn:
+    column_type = column.type
+    if isinstance(column_type, Boolean):
+        data_type = "boolean"
+    elif isinstance(column_type, (Integer, BigInteger)):
+        data_type = "integer"
+    elif isinstance(column_type, (Numeric, Float)):
+        data_type = "number"
+    elif isinstance(column_type, DateTime):
+        data_type = "datetime"
+    elif isinstance(column_type, Date):
+        data_type = "date"
+    elif isinstance(column_type, JSON):
+        data_type = "object"
+    elif isinstance(column_type, LargeBinary):
+        data_type = "binary"
+    elif isinstance(column_type, (String, Text)):
+        data_type = "string"
+    else:
+        data_type = str(column_type).casefold()
+    return TabularColumn(
+        name=str(column.name),
+        data_type=data_type,
+        nullable=bool(column.nullable),
+    )
+
+
+def _configure_read_only(connection: Any, url: URL, timeout_ms: int) -> None:
+    if url.drivername not in POSTGRESQL_SCHEMES:
+        return
+    effective_timeout = max(1, min(int(timeout_ms), 30_000))
+    connection.exec_driver_sql("SET TRANSACTION READ ONLY")
+    connection.exec_driver_sql(f"SET LOCAL statement_timeout = {effective_timeout}")
+
+
+def _bounded_connection_url(url: URL, timeout_ms: int) -> URL:
+    if url.drivername not in POSTGRESQL_SCHEMES:
+        return url
+    connect_timeout = max(1, math.ceil(min(int(timeout_ms), 30_000) / 1_000))
+    return url.update_query_dict({"connect_timeout": str(connect_timeout)})
+
+
+def _sql_identifier(value: object, label: str) -> str:
+    normalized = str(value or "").strip()
+    if not _IDENTIFIER.fullmatch(normalized):
+        raise TabularSourceValidationError(
+            f"SQL {label} names must use a simple identifier."
+        )
+    return normalized
+
+
+def _required_metadata(metadata: Mapping[str, object], key: str) -> str:
+    value = _optional_text(metadata.get(key))
+    if not value:
+        raise TabularSourceValidationError(
+            f"Tabular source metadata is missing {key}."
+        )
+    return value
+
+
+def _optional_text(value: object | None) -> str | None:
+    normalized = str(value or "").strip()
+    return normalized or None
+
+
+def _principal_user_id(principal: ApiPrincipal) -> str | None:
+    return _optional_text(getattr(principal.user, "id", None))
+
+
+def _is_xlsx(filename: str, content_type: str | None) -> bool:
+    normalized_type = str(content_type or "").split(";", 1)[0].strip().casefold()
+    return str(filename or "").strip().casefold().endswith(".xlsx") or normalized_type == (
+        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+    )
+
+
+def _json_row(row: Mapping[str, object]) -> dict[str, object]:
+    return {str(key): _json_value(value) for key, value in row.items()}
+
+
+def _json_value(value: object) -> object:
+    if value is None or isinstance(value, (str, bool, int)):
+        return value
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            return str(value)
+        return value
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, (datetime, date)):
+        return value.isoformat()
+    if isinstance(value, bytes):
+        return value.hex()
+    if isinstance(value, (list, dict)):
+        return json.loads(json.dumps(value, default=str))
+    return str(value)
+
+
+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()
+
+
+__all__ = [
+    "MAX_FILE_BYTES",
+    "MAX_FILE_ROWS",
+    "ManagedFileTabularAdapter",
+    "PostgresqlTabularAdapter",
+    "TabularOriginInspection",
+    "TabularOriginRead",
+    "infer_tabular_schema",
+    "origin_fingerprint",
+    "parse_managed_tabular_content",
+]

src/govoplan_connectors/backend/tabular_sources.py

@@ -27,8 +27,14 @@ from govoplan_core.core.tabular_sources import (
     TabularSourceValidationError,
     parse_tabular_csv,
 )
+from govoplan_core.core.runtime import get_registry
 from govoplan_core.db.base import utcnow
 from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.tabular_adapters import (
+    ManagedFileTabularAdapter,
+    PostgresqlTabularAdapter,
+    TabularOriginInspection,
+)
 
 
 READ_SCOPE = "connectors:source:read"
@@ -42,8 +48,16 @@ MAX_READ_TIMEOUT_MS = 2_000
 
 
 class SqlTabularSourceProvider:
-    def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None:
+    def __init__(
+        self,
+        *,
+        registry: object | None = None,
+        clock: Callable[[], float] = time.monotonic,
+        sql_adapter: PostgresqlTabularAdapter | None = None,
+    ) -> None:
+        self._registry = registry
         self._clock = clock
+        self._sql_adapter = sql_adapter or PostgresqlTabularAdapter()
 
     def list_sources(
         self,
@@ -109,7 +123,6 @@ class SqlTabularSourceProvider:
                 "The source fingerprint changed; refresh the source node before running it."
             )
 
-        started = self._clock()
         limit = max(1, min(int(request.limit), MAX_READ_ROWS))
         byte_limit = max(2, min(int(request.max_bytes), MAX_READ_BYTES))
         timeout_ms = max(1, min(int(request.timeout_ms), MAX_READ_TIMEOUT_MS))
@@ -149,10 +162,51 @@ class SqlTabularSourceProvider:
             raise TabularSourceValidationError(
                 f"Unknown source columns: {', '.join(unknown_columns)}"
             )
+        started = self._clock()
+        source_rows: Sequence[Mapping[str, object]]
+        total_rows = item.row_count
+        source = _source_dto(item)
+        base_offset = offset
+        if item.provider == "managed_file":
+            read = self._file_adapter().read(
+                db,
+                api_principal,
+                metadata=item.metadata_,
+            )
+            if read.inspection.fingerprint != item.fingerprint:
+                raise TabularSourceValidationError(
+                    "The managed file source changed; refresh and review it before previewing."
+                )
+            source_rows = read.rows[offset:]
+            total_rows = read.total_rows
+            source = _source_dto(item, inspection=read.inspection)
+            diagnostics.extend(read.diagnostics)
+        elif item.provider == "postgresql":
+            read = self._sql_adapter.read(
+                db,
+                api_principal,
+                metadata=item.metadata_,
+                columns=selected_columns,
+                offset=offset,
+                limit=limit + 1,
+                timeout_ms=timeout_ms,
+            )
+            source_rows = read.rows
+            total_rows = read.total_rows
+            source = _source_dto(item, inspection=read.inspection)
+            diagnostics.extend(read.diagnostics)
+        else:
+            source_rows = item.rows[offset:]
+
+        if int(max(0.0, self._clock() - started) * 1_000) >= timeout_ms:
+            raise TabularSourceUnavailableError(
+                "Tabular source preview exceeded its time budget."
+            )
+
         rows: list[dict[str, object]] = []
         returned_bytes = 2
         stopped_for = ""
-        for row in item.rows[offset:]:
+        for row in source_rows:
             if len(rows) >= limit:
                 stopped_for = "rows"
                 break
@@ -203,10 +257,10 @@ class SqlTabularSourceProvider:
                 )
             )
         return TabularReadResult(
-            source=_source_dto(item),
+            source=source,
             rows=tuple(rows),
-            total_rows=item.row_count,
-            truncated=offset + len(rows) < item.row_count,
+            total_rows=total_rows,
+            truncated=base_offset + len(rows) < total_rows,
             returned_bytes=returned_bytes,
             elapsed_ms=elapsed_ms,
             effective_row_limit=limit,
@@ -215,6 +269,173 @@ class SqlTabularSourceProvider:
             diagnostics=tuple(diagnostics),
         )
 
+    def create_file_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        name: str,
+        source_name: str,
+        file_asset_id: str,
+        file_version_id: str | None = None,
+        description: str | None = None,
+        delimiter: str = ",",
+        sheet_name: str | None = None,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        inspection = self._file_adapter().inspect(
+            db,
+            api_principal,
+            file_asset_id=file_asset_id,
+            file_version_id=file_version_id,
+            delimiter=delimiter,
+            sheet_name=sheet_name,
+        )
+        return self._create_origin(
+            db,
+            api_principal,
+            name=name,
+            source_name=source_name,
+            description=description,
+            inspection=inspection,
+        )
+
+    def create_sql_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        name: str,
+        source_name: str,
+        configuration_id: str,
+        table_name: str,
+        schema_name: str | None = None,
+        description: str | None = None,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        inspection = self._sql_adapter.inspect(
+            db,
+            api_principal,
+            configuration_id=configuration_id,
+            table_name=table_name,
+            schema_name=schema_name,
+            timeout_ms=MAX_READ_TIMEOUT_MS,
+        )
+        return self._create_origin(
+            db,
+            api_principal,
+            name=name,
+            source_name=source_name,
+            description=description,
+            inspection=inspection,
+        )
+
+    def refresh_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        source_ref: str,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=source_ref,
+        )
+        if item is None:
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        if item.provider == "managed_file":
+            inspection = self._file_adapter().inspect(
+                db,
+                api_principal,
+                file_asset_id=str(item.metadata_.get("file_asset_id") or ""),
+                file_version_id=None,
+                delimiter=str(item.metadata_.get("delimiter") or ","),
+                sheet_name=_clean_optional(item.metadata_.get("sheet_name")),
+            )
+        elif item.provider == "postgresql":
+            inspection = self._sql_adapter.inspect(
+                db,
+                api_principal,
+                configuration_id=str(
+                    item.metadata_.get("configuration_id") or ""
+                ),
+                table_name=str(item.metadata_.get("table_name") or ""),
+                schema_name=_clean_optional(item.metadata_.get("schema_name")),
+                timeout_ms=MAX_READ_TIMEOUT_MS,
+            )
+        else:
+            raise TabularSourceValidationError(
+                "Immutable snapshots cannot be refreshed; import a replacement snapshot."
+            )
+        item.schema_version += 1
+        item.schema_ = [_column_payload(column) for column in inspection.schema]
+        item.fingerprint = inspection.fingerprint
+        item.row_count = inspection.row_count
+        item.byte_count = inspection.byte_count
+        item.metadata_ = {
+            **dict(inspection.metadata),
+            "discovery_fingerprint": inspection.fingerprint,
+        }
+        item.updated_by = _actor_id(api_principal)
+        db.flush()
+        return _source_dto(item, inspection=inspection)
+
+    def _create_origin(
+        self,
+        db: Session,
+        principal: ApiPrincipal,
+        *,
+        name: str,
+        source_name: str,
+        description: str | None,
+        inspection: TabularOriginInspection,
+    ) -> TabularSource:
+        clean_name = str(name or "").strip()
+        clean_source_name = str(source_name or "").strip()
+        if not clean_name or not clean_source_name:
+            raise TabularSourceValidationError(
+                "Tabular source name and source identifier are required."
+            )
+        existing = db.scalar(
+            select(ConnectorTabularSource.id).where(
+                ConnectorTabularSource.tenant_id == principal.tenant_id,
+                ConnectorTabularSource.source_name == clean_source_name,
+            )
+        )
+        if existing is not None:
+            raise TabularSourceValidationError(
+                f"A tabular source named {clean_source_name!r} already exists."
+            )
+        actor_id = _actor_id(principal)
+        item = ConnectorTabularSource(
+            tenant_id=principal.tenant_id,
+            provider=inspection.provider,
+            source_name=clean_source_name,
+            name=clean_name,
+            description=_clean_optional(description),
+            status="active",
+            schema_version=1,
+            schema_=[_column_payload(column) for column in inspection.schema],
+            rows=[],
+            fingerprint=inspection.fingerprint,
+            row_count=inspection.row_count,
+            byte_count=inspection.byte_count,
+            metadata_={
+                **dict(inspection.metadata),
+                "discovery_fingerprint": inspection.fingerprint,
+            },
+            created_by=actor_id,
+            updated_by=actor_id,
+        )
+        db.add(item)
+        db.flush()
+        return _source_dto(item, inspection=inspection)
+
+    def _file_adapter(self) -> ManagedFileTabularAdapter:
+        return ManagedFileTabularAdapter(self._registry or get_registry())
+
     def create_snapshot(
         self,
         session: object,
@@ -347,21 +568,81 @@ def _source_record(
     tenant_id: str,
     source_ref: str,
 ) -> ConnectorTabularSource | None:
-    source_id = source_ref.removeprefix("snapshot:")
-    if not source_id or source_id == source_ref:
+    prefix, separator, source_id = str(source_ref or "").partition(":")
+    provider_by_prefix = {
+        "snapshot": "snapshot",
+        "file": "managed_file",
+        "sql": "postgresql",
+    }
+    provider = provider_by_prefix.get(prefix)
+    if not separator or not source_id or provider is None:
         return None
     return session.scalar(
         select(ConnectorTabularSource).where(
             ConnectorTabularSource.id == source_id,
             ConnectorTabularSource.tenant_id == tenant_id,
+            ConnectorTabularSource.provider == provider,
             ConnectorTabularSource.deleted_at.is_(None),
         )
     )
 
 
-def _source_dto(item: ConnectorTabularSource) -> TabularSource:
+def _source_dto(
+    item: ConnectorTabularSource,
+    *,
+    inspection: TabularOriginInspection | None = None,
+) -> TabularSource:
+    prefix_by_provider = {
+        "snapshot": "snapshot",
+        "managed_file": "file",
+        "postgresql": "sql",
+    }
+    source_mode = {
+        "managed_file": "file_backed",
+        "postgresql": "live",
+    }.get(item.provider, "cached")
+    if inspection is not None:
+        pushdown = inspection.pushdown
+        health = inspection.health
+    elif item.provider == "managed_file":
+        pushdown = TabularPushdown(projections=True, pagination=True)
+        health = TabularSourceHealth(
+            status="unknown",
+            code="files.exact_version_not_checked",
+            summary="The exact managed file version will be re-authorized and integrity-checked on preview.",
+            checked_at=item.updated_at,
+            details={
+                "file_version_id": item.metadata_.get("file_version_id"),
+            },
+        )
+    elif item.provider == "postgresql":
+        pushdown = TabularPushdown(projections=True, pagination=True)
+        health = TabularSourceHealth(
+            status="unknown",
+            code="sql.health_not_checked",
+            summary="The live SQL source will be checked against its pinned configuration, credential, and schema on preview.",
+            checked_at=item.updated_at,
+            details={
+                "configuration_id": item.metadata_.get("configuration_id"),
+                "configuration_revision": item.metadata_.get(
+                    "configuration_revision"
+                ),
+            },
+        )
+    else:
+        pushdown = TabularPushdown(
+            projections=True,
+            pagination=True,
+        )
+        health = TabularSourceHealth(
+            status="healthy",
+            code="snapshot.ready",
+            summary="The immutable connector snapshot is ready.",
+            checked_at=item.updated_at,
+            details={"immutable": True},
+        )
     return TabularSource(
-        ref=f"snapshot:{item.id}",
+        ref=f"{prefix_by_provider.get(item.provider, 'snapshot')}:{item.id}",
         provider=item.provider,
         source_name=item.source_name,
         name=item.name,
@@ -374,18 +655,9 @@ def _source_dto(item: ConnectorTabularSource) -> TabularSource:
         updated_at=item.updated_at,
         capabilities=("read", "preview"),
         metadata=dict(item.metadata_),
-        source_mode="cached",
-        pushdown=TabularPushdown(
-            projections=True,
-            pagination=True,
-        ),
-        health=TabularSourceHealth(
-            status="healthy",
-            code="snapshot.ready",
-            summary="The immutable connector snapshot is ready.",
-            checked_at=item.updated_at,
-            details={"immutable": True},
-        ),
+        source_mode=source_mode,
+        pushdown=pushdown,
+        health=health,
     )