Bound connector source previews

README.md

@@ -17,7 +17,9 @@ or reporting behavior.
 The first executable connector capability provides tenant-isolated tabular
 origins. Operators can import bounded JSON or CSV snapshots, inspect inferred
 schemas, and expose immutable source references and content fingerprints
-through `connectors.datasource_origins@0.1.0`.
+through `connectors.datasource_origins@0.1.0`. Preview reads enforce provider
+ceilings for rows, serialized bytes, and elapsed time and report the effective
+limits and any truncation as structured diagnostics.
 
 Connectors owns acquisition, connection profiles, credentials, discovery, and
 provider health. `govoplan-datasources` registers an origin as a governed live
@@ -25,6 +27,12 @@ or cached datasource and owns staging, materializations, frozen states, and
 consumer access. Dataflow consumes that Datasources contract and never imports
 connector implementations or stores connector credentials.
 
+Each origin declares whether it is live, cached, file-backed, or static, its
+structured health state, and which projection, filter, aggregation, sorting,
+and pagination operations it can push down. The immutable snapshot provider
+currently supports projection and pagination only; consumers must keep other
+operations in Dataflow rather than assuming transport-side execution.
+
 Database, REST/HTTP, directory, managed-file, and warehouse providers can
 implement the same origin contract without changing Datasources or Dataflow.
 

src/govoplan_connectors/backend/datasource_origins.py

@@ -7,6 +7,7 @@ from govoplan_core.core.datasources import (
     DatasourceOrigin,
     DatasourceOriginReadRequest,
     DatasourceOriginReadResult,
+    DatasourceUnavailableError,
     DatasourceValidationError,
 )
 from govoplan_core.core.tabular_sources import (
@@ -15,6 +16,7 @@ from govoplan_core.core.tabular_sources import (
     TabularSourceAccessError,
     TabularSourceError,
     TabularSourceNotFoundError,
+    TabularSourceUnavailableError,
     TabularSourceValidationError,
 )
 from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider
@@ -79,6 +81,8 @@ class ConnectorDatasourceOriginProvider:
                     offset=request.offset,
                     columns=request.columns,
                     expected_fingerprint=request.expected_fingerprint,
+                    max_bytes=request.max_bytes,
+                    timeout_ms=request.timeout_ms,
                 ),
             )
         except TabularSourceError as exc:
@@ -88,6 +92,12 @@ class ConnectorDatasourceOriginProvider:
             rows=result.rows,
             total_rows=result.total_rows,
             truncated=result.truncated,
+            returned_bytes=result.returned_bytes,
+            elapsed_ms=result.elapsed_ms,
+            effective_row_limit=result.effective_row_limit,
+            effective_byte_limit=result.effective_byte_limit,
+            effective_timeout_ms=result.effective_timeout_ms,
+            diagnostics=result.diagnostics,
         )
 
 
@@ -116,6 +126,9 @@ def _origin(source: TabularSource) -> DatasourceOrigin:
         updated_at=source.updated_at,
         capabilities=source.capabilities,
         metadata=dict(source.metadata),
+        source_mode=source.source_mode,
+        pushdown=source.pushdown,
+        health=source.health,
     )
 
 
@@ -124,6 +137,8 @@ def _datasource_error(exc: TabularSourceError):
         return DatasourceAccessError(str(exc))
     if isinstance(exc, TabularSourceNotFoundError):
         return DatasourceNotFoundError(str(exc))
+    if isinstance(exc, TabularSourceUnavailableError):
+        return DatasourceUnavailableError(str(exc))
     if isinstance(exc, TabularSourceValidationError):
         return DatasourceValidationError(str(exc))
     return DatasourceValidationError(str(exc))

src/govoplan_connectors/backend/manifest.py

@@ -453,7 +453,9 @@ manifest = ModuleManifest(
                 "Connectors owns endpoint discovery, authentication hand-off, transport limits, retries, and protocol health. "
                 "Domain modules own field mapping, validation, reconciliation, and record mutation. The shared Core runtime "
                 "contract reports redacted effects and diagnostics with source revisions, fingerprints, and immutable input hashes. "
-                "A commit must reject stale, truncated, conflicting, or error-bearing previews, and credentials never appear in URLs or samples."
+                "Tabular previews enforce effective row, serialized-byte, and elapsed-time ceilings and report limit truncation "
+                "as structured diagnostics. A commit must reject stale, truncated, conflicting, or error-bearing previews, and "
+                "credentials never appear in URLs or samples."
             ),
             layer="available",
             documentation_types=("admin", "user"),
@@ -468,8 +470,11 @@ manifest = ModuleManifest(
             body=(
                 "Connectors owns source configuration, access checks, schema discovery, "
                 "fingerprints, and bounded reads. Dataflow stores only opaque source "
-                "references and expected fingerprints. The first executable provider "
+                "references and expected fingerprints. Each source declares its live, "
-                "imports immutable JSON or CSV snapshots and exposes them as Datasource "
+                "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."
             ),

src/govoplan_connectors/backend/router.py

@@ -16,6 +16,7 @@ from govoplan_core.core.tabular_sources import (
     TabularSourceAccessError,
     TabularSourceError,
     TabularSourceNotFoundError,
+    TabularSourceUnavailableError,
 )
 from govoplan_core.core.feeds import (
     FeedCapabilityError,
@@ -38,6 +39,9 @@ from govoplan_connectors.backend.schemas import (
     SanctionsSourceResponse,
     SnapshotCreateRequest,
     TabularColumnResponse,
+    TabularHealthResponse,
+    TabularPreviewDiagnosticResponse,
+    TabularPushdownResponse,
     TabularSourceDeleteResponse,
     TabularSourceListResponse,
     TabularSourcePreviewResponse,
@@ -85,6 +89,11 @@ def _http_error(exc: TabularSourceError) -> HTTPException:
         return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
     if isinstance(exc, TabularSourceAccessError):
         return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
+    if isinstance(exc, TabularSourceUnavailableError):
+        return HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=str(exc),
+        )
     return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
 
 
@@ -377,6 +386,8 @@ def api_preview_tabular_source(
     source_id: str,
     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),
+    timeout_ms: int = Query(default=2_000, ge=1, le=10_000),
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> TabularSourcePreviewResponse:
@@ -389,6 +400,8 @@ def api_preview_tabular_source(
                 source_ref=f"snapshot:{source_id}",
                 limit=limit,
                 offset=offset,
+                max_bytes=max_bytes,
+                timeout_ms=timeout_ms,
             ),
         )
     except TabularSourceError as exc:
@@ -398,6 +411,20 @@ def api_preview_tabular_source(
         rows=[dict(row) for row in result.rows],
         total_rows=result.total_rows,
         truncated=result.truncated,
+        returned_bytes=result.returned_bytes,
+        elapsed_ms=result.elapsed_ms,
+        effective_row_limit=result.effective_row_limit,
+        effective_byte_limit=result.effective_byte_limit,
+        effective_timeout_ms=result.effective_timeout_ms,
+        diagnostics=[
+            TabularPreviewDiagnosticResponse(
+                severity=item.severity,
+                code=item.code,
+                message=item.message,
+                details=dict(item.details),
+            )
+            for item in result.diagnostics
+        ],
     )
 
 
@@ -604,6 +631,25 @@ def _source_response(source: TabularSource) -> TabularSourceResponse:
         updated_at=source.updated_at.isoformat() if source.updated_at else None,
         capabilities=list(source.capabilities),
         metadata=dict(source.metadata),
+        source_mode=source.source_mode,
+        pushdown=TabularPushdownResponse(
+            projections=source.pushdown.projections,
+            pagination=source.pushdown.pagination,
+            filters=list(source.pushdown.filters),
+            aggregations=list(source.pushdown.aggregations),
+            sorting=list(source.pushdown.sorting),
+        ),
+        health=TabularHealthResponse(
+            status=source.health.status,
+            code=source.health.code,
+            summary=source.health.summary,
+            checked_at=(
+                source.health.checked_at.isoformat()
+                if source.health.checked_at
+                else None
+            ),
+            details=dict(source.health.details),
+        ),
     )
 
 

src/govoplan_connectors/backend/schemas.py

@@ -100,6 +100,29 @@ class TabularColumnResponse(BaseModel):
     nullable: bool
 
 
+class TabularPushdownResponse(BaseModel):
+    projections: bool
+    pagination: bool
+    filters: list[str]
+    aggregations: list[str]
+    sorting: list[str]
+
+
+class TabularHealthResponse(BaseModel):
+    status: Literal["healthy", "warning", "error", "unknown"]
+    code: str
+    summary: str
+    checked_at: str | None
+    details: dict[str, Any]
+
+
+class TabularPreviewDiagnosticResponse(BaseModel):
+    severity: Literal["info", "warning", "error"]
+    code: str
+    message: str
+    details: dict[str, Any]
+
+
 class TabularSourceResponse(BaseModel):
     ref: str
     provider: str
@@ -114,6 +137,9 @@ class TabularSourceResponse(BaseModel):
     updated_at: str | None
     capabilities: list[str]
     metadata: dict[str, Any]
+    source_mode: Literal["live", "cached", "file_backed", "static"]
+    pushdown: TabularPushdownResponse
+    health: TabularHealthResponse
 
 
 class TabularSourceListResponse(BaseModel):
@@ -125,6 +151,12 @@ class TabularSourcePreviewResponse(BaseModel):
     rows: list[dict[str, Any]]
     total_rows: int
     truncated: bool
+    returned_bytes: int
+    elapsed_ms: int
+    effective_row_limit: int
+    effective_byte_limit: int
+    effective_timeout_ms: int
+    diagnostics: list[TabularPreviewDiagnosticResponse]
 
 
 class TabularSourceDeleteResponse(BaseModel):
@@ -216,6 +248,9 @@ __all__ = [
     "SanctionsSourceListResponse",
     "SanctionsSourceResponse",
     "TabularColumnResponse",
+    "TabularHealthResponse",
+    "TabularPreviewDiagnosticResponse",
+    "TabularPushdownResponse",
     "TabularSourceDeleteResponse",
     "TabularSourceListResponse",
     "TabularSourcePreviewResponse",

src/govoplan_connectors/backend/tabular_sources.py

@@ -2,10 +2,11 @@ from __future__ import annotations
 
 import hashlib
 import json
+import time
 from collections.abc import Mapping, Sequence
 from datetime import datetime
 from decimal import Decimal
-from typing import Any
+from typing import Any, Callable
 
 from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
@@ -13,12 +14,16 @@ from sqlalchemy.orm import Session
 from govoplan_core.auth import ApiPrincipal, has_scope
 from govoplan_core.core.tabular_sources import (
     TabularColumn,
+    TabularPreviewDiagnostic,
+    TabularPushdown,
     TabularReadRequest,
     TabularReadResult,
     TabularSnapshotInput,
     TabularSource,
     TabularSourceAccessError,
     TabularSourceNotFoundError,
+    TabularSourceHealth,
+    TabularSourceUnavailableError,
     TabularSourceValidationError,
     parse_tabular_csv,
 )
@@ -32,9 +37,14 @@ ADMIN_SCOPE = "connectors:source:admin"
 MAX_SNAPSHOT_ROWS = 10_000
 MAX_SNAPSHOT_BYTES = 5_000_000
 MAX_READ_ROWS = 500
+MAX_READ_BYTES = 1_000_000
+MAX_READ_TIMEOUT_MS = 2_000
 
 
 class SqlTabularSourceProvider:
+    def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None:
+        self._clock = clock
+
     def list_sources(
         self,
         session: object,
@@ -99,8 +109,39 @@ 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))
         offset = max(0, int(request.offset))
+        diagnostics: list[TabularPreviewDiagnostic] = []
+        if limit != request.limit:
+            diagnostics.append(
+                _preview_diagnostic(
+                    "preview.row_limit_tightened",
+                    "The provider tightened the requested row limit.",
+                    requested=request.limit,
+                    effective=limit,
+                )
+            )
+        if byte_limit != request.max_bytes:
+            diagnostics.append(
+                _preview_diagnostic(
+                    "preview.byte_limit_tightened",
+                    "The provider tightened the requested byte limit.",
+                    requested=request.max_bytes,
+                    effective=byte_limit,
+                )
+            )
+        if timeout_ms != request.timeout_ms:
+            diagnostics.append(
+                _preview_diagnostic(
+                    "preview.timeout_tightened",
+                    "The provider tightened the requested time limit.",
+                    requested=request.timeout_ms,
+                    effective=timeout_ms,
+                )
+            )
         selected_columns = tuple(dict.fromkeys(request.columns))
         known_columns = {column["name"] for column in item.schema_}
         unknown_columns = [column for column in selected_columns if column not in known_columns]
@@ -108,20 +149,70 @@ class SqlTabularSourceProvider:
             raise TabularSourceValidationError(
                 f"Unknown source columns: {', '.join(unknown_columns)}"
             )
-        window = item.rows[offset : offset + limit]
+        rows: list[dict[str, object]] = []
-        rows = tuple(
+        returned_bytes = 2
-            {
+        stopped_for = ""
+        for row in item.rows[offset:]:
+            if len(rows) >= limit:
+                stopped_for = "rows"
+                break
+            elapsed_ms = int(max(0.0, self._clock() - started) * 1_000)
+            if elapsed_ms >= timeout_ms:
+                if not rows:
+                    raise TabularSourceUnavailableError(
+                        "Tabular source preview exceeded its time budget."
+                    )
+                stopped_for = "time"
+                break
+            selected = {
                 key: value
                 for key, value in row.items()
                 if not selected_columns or key in selected_columns
             }
-            for row in window
+            row_bytes = len(
+                json.dumps(
+                    selected,
+                    sort_keys=True,
+                    separators=(",", ":"),
+                    default=str,
+                ).encode("utf-8")
+            )
+            additional_bytes = row_bytes + (1 if rows else 0)
+            if returned_bytes + additional_bytes > byte_limit:
+                if not rows:
+                    raise TabularSourceValidationError(
+                        "A single source row exceeds the preview byte limit."
+                    )
+                stopped_for = "bytes"
+                break
+            rows.append(selected)
+            returned_bytes += additional_bytes
+        elapsed_ms = int(max(0.0, self._clock() - started) * 1_000)
+        if stopped_for:
+            labels = {
+                "rows": ("preview.row_limit_reached", "row"),
+                "bytes": ("preview.byte_limit_reached", "byte"),
+                "time": ("preview.timeout_reached", "time"),
+            }
+            code, label = labels[stopped_for]
+            diagnostics.append(
+                TabularPreviewDiagnostic(
+                    severity="warning",
+                    code=code,
+                    message=f"The preview stopped at its effective {label} limit.",
+                )
             )
         return TabularReadResult(
             source=_source_dto(item),
-            rows=rows,
+            rows=tuple(rows),
             total_rows=item.row_count,
             truncated=offset + len(rows) < item.row_count,
+            returned_bytes=returned_bytes,
+            elapsed_ms=elapsed_ms,
+            effective_row_limit=limit,
+            effective_byte_limit=byte_limit,
+            effective_timeout_ms=timeout_ms,
+            diagnostics=tuple(diagnostics),
         )
 
     def create_snapshot(
@@ -283,6 +374,33 @@ 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},
+        ),
+    )
+
+
+def _preview_diagnostic(
+    code: str,
+    message: str,
+    *,
+    requested: int,
+    effective: int,
+) -> TabularPreviewDiagnostic:
+    return TabularPreviewDiagnostic(
+        severity="info",
+        code=code,
+        message=message,
+        details={"requested": requested, "effective": effective},
     )
 
 
@@ -367,6 +485,8 @@ def _escape_like(value: str) -> str:
 __all__ = [
     "ADMIN_SCOPE",
     "MAX_READ_ROWS",
+    "MAX_READ_BYTES",
+    "MAX_READ_TIMEOUT_MS",
     "MAX_SNAPSHOT_BYTES",
     "MAX_SNAPSHOT_ROWS",
     "READ_SCOPE",

tests/test_datasource_origins.py

@@ -81,6 +81,11 @@ class ConnectorDatasourceOriginTests(unittest.TestCase):
         self.assertEqual((self.source.ref,), tuple(item.ref for item in origins))
         self.assertEqual(("live", "cached"), origins[0].supported_modes)
         self.assertEqual(({"id": 1, "name": "Ada"},), result.rows)
+        self.assertEqual("cached", origins[0].source_mode)
+        self.assertTrue(origins[0].pushdown.projections)
+        self.assertEqual("healthy", origins[0].health.status)
+        self.assertGreater(result.returned_bytes, 2)
+        self.assertEqual(1_000_000, result.effective_byte_limit)
 
 
 if __name__ == "__main__":

tests/test_tabular_sources.py

@@ -12,6 +12,7 @@ from govoplan_core.core.tabular_sources import (
     TabularReadRequest,
     TabularSnapshotInput,
     TabularSourceAccessError,
+    TabularSourceUnavailableError,
     TabularSourceValidationError,
 )
 from govoplan_core.db.base import Base
@@ -91,6 +92,94 @@ class ConnectorsTabularSourceTests(unittest.TestCase):
         self.assertEqual(1, len(preview.rows))
         self.assertTrue(preview.truncated)
         self.assertEqual(created.fingerprint, preview.source.fingerprint)
+        self.assertEqual("cached", preview.source.source_mode)
+        self.assertTrue(preview.source.pushdown.projections)
+        self.assertTrue(preview.source.pushdown.pagination)
+        self.assertEqual("healthy", preview.source.health.status)
+        self.assertGreater(preview.returned_bytes, 2)
+        self.assertEqual(1, preview.effective_row_limit)
+        self.assertEqual("preview.row_limit_reached", preview.diagnostics[0].code)
+
+    def test_preview_enforces_byte_time_and_provider_ceiling_budgets(self) -> None:
+        created = self.provider.create_snapshot(
+            self.session,
+            principal(),
+            snapshot=TabularSnapshotInput(
+                name="Bounded",
+                source_name="bounded",
+                rows=(
+                    {"id": 1, "value": "first"},
+                    {"id": 2, "value": "second"},
+                ),
+            ),
+        )
+        self.session.commit()
+
+        bounded = self.provider.read_source(
+            self.session,
+            principal(),
+            request=TabularReadRequest(
+                source_ref=created.ref,
+                limit=500,
+                max_bytes=35,
+                timeout_ms=2_000,
+            ),
+        )
+        self.assertEqual(1, len(bounded.rows))
+        self.assertTrue(bounded.truncated)
+        self.assertEqual(
+            "preview.byte_limit_reached",
+            bounded.diagnostics[-1].code,
+        )
+        with self.assertRaisesRegex(
+            TabularSourceValidationError,
+            "single source row exceeds",
+        ):
+            self.provider.read_source(
+                self.session,
+                principal(),
+                request=TabularReadRequest(
+                    source_ref=created.ref,
+                    max_bytes=2,
+                ),
+            )
+
+        tightened = self.provider.read_source(
+            self.session,
+            principal(),
+            request=TabularReadRequest(
+                source_ref=created.ref,
+                limit=5_000,
+                max_bytes=5_000_000,
+                timeout_ms=10_000,
+            ),
+        )
+        self.assertEqual(500, tightened.effective_row_limit)
+        self.assertEqual(1_000_000, tightened.effective_byte_limit)
+        self.assertEqual(2_000, tightened.effective_timeout_ms)
+        self.assertEqual(
+            {
+                "preview.row_limit_tightened",
+                "preview.byte_limit_tightened",
+                "preview.timeout_tightened",
+            },
+            {item.code for item in tightened.diagnostics},
+        )
+
+        times = iter((0.0, 0.01))
+        timeout_provider = SqlTabularSourceProvider(clock=lambda: next(times))
+        with self.assertRaisesRegex(
+            TabularSourceUnavailableError,
+            "time budget",
+        ):
+            timeout_provider.read_source(
+                self.session,
+                principal(),
+                request=TabularReadRequest(
+                    source_ref=created.ref,
+                    timeout_ms=1,
+                ),
+            )
 
     def test_tenant_and_scope_isolation_are_enforced(self) -> None:
         created = self.provider.create_snapshot(