Define bounded tabular source contracts

docs/TABULAR_SOURCE_CONTRACT.md

@@ -0,0 +1,21 @@
+# Tabular Source Preview Contract
+
+Core defines provider-neutral DTOs for optional tabular source providers. A
+source declares whether it is live, cached, file-backed, or static; its schema
+and immutable fingerprint; structured health; and the exact projection,
+pagination, filter, aggregation, and sorting operations that the provider can
+push down. Consumers must not infer pushdown support from a provider name.
+
+Every preview request carries independent row, byte, and elapsed-time budgets.
+A provider may tighten these values but must return its effective limits,
+returned byte count, elapsed milliseconds, truncation state, and structured
+diagnostics. Equivalent fields on the Datasources read request and result
+preserve that evidence when a live source is consumed through the catalogue.
+A row that cannot fit within the byte budget fails explicitly rather than
+leaking a partial value. Timeout, stale fingerprint, unavailable source, and
+authorization failures remain distinct provider-neutral errors.
+
+Connector health and preview diagnostics must contain no credentials, endpoint
+userinfo, row values, or unbounded remote error bodies. A Datasource origin
+preserves this contract so registration and staging do not erase source mode,
+health, pushdown, or preview-limit evidence.

src/govoplan_core/core/datasources.py

@@ -9,6 +9,14 @@ from govoplan_core.core.external_references import (
     SOURCE_AUTHORITY_MODES,
     SourceAuthorityMode,
 )
+from govoplan_core.core.tabular_sources import (
+    DEFAULT_PREVIEW_BYTES,
+    DEFAULT_PREVIEW_TIMEOUT_MS,
+    TabularPreviewDiagnostic,
+    TabularPushdown,
+    TabularSourceHealth,
+    TabularSourceMode,
+)
 
 
 CAPABILITY_DATASOURCE_CATALOGUE = "datasources.catalogue"
@@ -265,6 +273,8 @@ class DatasourceReadRequest:
     offset: int = 0
     columns: tuple[str, ...] = ()
     expected_fingerprint: str | None = None
+    max_bytes: int = DEFAULT_PREVIEW_BYTES
+    timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS
 
 
 @dataclass(frozen=True, slots=True)
@@ -274,6 +284,12 @@ class DatasourceReadResult:
     total_rows: int
     truncated: bool
     materialization: DatasourceMaterialization | None = None
+    returned_bytes: int = 0
+    elapsed_ms: int = 0
+    effective_row_limit: int = 0
+    effective_byte_limit: int = 0
+    effective_timeout_ms: int = 0
+    diagnostics: tuple[TabularPreviewDiagnostic, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -339,6 +355,9 @@ class DatasourceOrigin:
     updated_at: datetime | None = None
     capabilities: tuple[str, ...] = ("read",)
     metadata: Mapping[str, object] = field(default_factory=dict)
+    source_mode: TabularSourceMode = "cached"
+    pushdown: TabularPushdown = field(default_factory=TabularPushdown)
+    health: TabularSourceHealth = field(default_factory=TabularSourceHealth)
 
 
 @dataclass(frozen=True, slots=True)
@@ -348,6 +367,8 @@ class DatasourceOriginReadRequest:
     offset: int = 0
     columns: tuple[str, ...] = ()
     expected_fingerprint: str | None = None
+    max_bytes: int = DEFAULT_PREVIEW_BYTES
+    timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS
 
 
 @dataclass(frozen=True, slots=True)
@@ -356,6 +377,12 @@ class DatasourceOriginReadResult:
     rows: tuple[Mapping[str, object], ...]
     total_rows: int
     truncated: bool
+    returned_bytes: int = 0
+    elapsed_ms: int = 0
+    effective_row_limit: int = 0
+    effective_byte_limit: int = 0
+    effective_timeout_ms: int = 0
+    diagnostics: tuple[TabularPreviewDiagnostic, ...] = ()
 
 
 @runtime_checkable

src/govoplan_core/core/tabular_sources.py

@@ -6,11 +6,17 @@ import re
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Protocol, runtime_checkable
+from typing import Literal, Protocol, runtime_checkable
 
 
 CAPABILITY_CONNECTORS_TABULAR_SOURCES = "connectors.tabularSources"
 CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER = "connectors.tabularSnapshotWriter"
+DEFAULT_PREVIEW_BYTES = 1_000_000
+DEFAULT_PREVIEW_TIMEOUT_MS = 2_000
+
+TabularSourceMode = Literal["live", "cached", "file_backed", "static"]
+TabularHealthStatus = Literal["healthy", "warning", "error", "unknown"]
+TabularDiagnosticSeverity = Literal["info", "warning", "error"]
 
 
 class TabularSourceError(ValueError):
@@ -29,6 +35,10 @@ class TabularSourceValidationError(TabularSourceError):
     pass
 
 
+class TabularSourceUnavailableError(TabularSourceError):
+    pass
+
+
 def parse_tabular_csv(
     csv_text: str,
     *,
@@ -131,6 +141,32 @@ class TabularColumn:
     nullable: bool = True
 
 
+@dataclass(frozen=True, slots=True)
+class TabularPushdown:
+    projections: bool = False
+    pagination: bool = False
+    filters: tuple[str, ...] = ()
+    aggregations: tuple[str, ...] = ()
+    sorting: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class TabularSourceHealth:
+    status: TabularHealthStatus = "unknown"
+    code: str = "source.health_unknown"
+    summary: str = "Source health has not been checked."
+    checked_at: datetime | None = None
+    details: Mapping[str, object] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class TabularPreviewDiagnostic:
+    severity: TabularDiagnosticSeverity
+    code: str
+    message: str
+    details: Mapping[str, object] = field(default_factory=dict)
+
+
 @dataclass(frozen=True, slots=True)
 class TabularSource:
     """Opaque, policy-filtered source reference exposed to consuming modules."""
@@ -148,6 +184,9 @@ class TabularSource:
     updated_at: datetime | None = None
     capabilities: tuple[str, ...] = ("read",)
     metadata: Mapping[str, object] = field(default_factory=dict)
+    source_mode: TabularSourceMode = "cached"
+    pushdown: TabularPushdown = field(default_factory=TabularPushdown)
+    health: TabularSourceHealth = field(default_factory=TabularSourceHealth)
 
 
 @dataclass(frozen=True, slots=True)
@@ -157,6 +196,8 @@ class TabularReadRequest:
     offset: int = 0
     columns: tuple[str, ...] = ()
     expected_fingerprint: str | None = None
+    max_bytes: int = DEFAULT_PREVIEW_BYTES
+    timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS
 
 
 @dataclass(frozen=True, slots=True)
@@ -165,6 +206,12 @@ class TabularReadResult:
     rows: tuple[Mapping[str, object], ...]
     total_rows: int
     truncated: bool
+    returned_bytes: int = 0
+    elapsed_ms: int = 0
+    effective_row_limit: int = 0
+    effective_byte_limit: int = 0
+    effective_timeout_ms: int = 0
+    diagnostics: tuple[TabularPreviewDiagnostic, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -247,7 +294,11 @@ def _capability(registry: object | None, name: str) -> object | None:
 __all__ = [
     "CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER",
     "CAPABILITY_CONNECTORS_TABULAR_SOURCES",
+    "DEFAULT_PREVIEW_BYTES",
+    "DEFAULT_PREVIEW_TIMEOUT_MS",
     "TabularColumn",
+    "TabularPreviewDiagnostic",
+    "TabularPushdown",
     "TabularReadRequest",
     "TabularReadResult",
     "TabularSnapshotInput",
@@ -257,6 +308,9 @@ __all__ = [
     "TabularSourceError",
     "TabularSourceNotFoundError",
     "TabularSourceProvider",
+    "TabularSourceHealth",
+    "TabularSourceMode",
+    "TabularSourceUnavailableError",
     "TabularSourceValidationError",
     "parse_tabular_csv",
     "tabular_snapshot_writer",

tests/test_tabular_source_contract.py

@@ -2,17 +2,25 @@ from __future__ import annotations
 
 import unittest
 
+from govoplan_core.core.datasources import (
+    DatasourceDescriptor,
+    DatasourceReadRequest,
+    DatasourceReadResult,
+)
 from govoplan_core.core.modules import ModuleContext, ModuleManifest
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.core.tabular_sources import (
     CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
     CAPABILITY_CONNECTORS_TABULAR_SOURCES,
     TabularColumn,
+    TabularPreviewDiagnostic,
+    TabularPushdown,
     TabularReadRequest,
     TabularReadResult,
     TabularSnapshotInput,
     TabularSnapshotWriter,
     TabularSource,
+    TabularSourceHealth,
     TabularSourceProvider,
     TabularSourceValidationError,
     parse_tabular_csv,
@@ -30,6 +38,13 @@ class _TabularProvider:
         schema=(TabularColumn(name="case_id", data_type="string", nullable=False),),
         fingerprint="abc123",
         row_count=1,
+        source_mode="cached",
+        pushdown=TabularPushdown(projections=True, pagination=True),
+        health=TabularSourceHealth(
+            status="healthy",
+            code="snapshot.ready",
+            summary="Immutable snapshot is ready.",
+        ),
     )
 
     def list_sources(self, session, principal, *, query="", limit=100):
@@ -51,6 +66,18 @@ class _TabularProvider:
             rows=selected,
             total_rows=len(rows),
             truncated=len(selected) < len(rows),
+            returned_bytes=18,
+            elapsed_ms=1,
+            effective_row_limit=request.limit,
+            effective_byte_limit=request.max_bytes,
+            effective_timeout_ms=request.timeout_ms,
+            diagnostics=(
+                TabularPreviewDiagnostic(
+                    severity="info",
+                    code="preview.bounded",
+                    message="The preview used explicit budgets.",
+                ),
+            ),
         )
 
     def create_snapshot(self, session, principal, *, snapshot):
@@ -96,8 +123,47 @@ class TabularSourceContractTests(unittest.TestCase):
         )
 
         self.assertEqual(({"case_id": "A-1"},), result.rows)
+        self.assertEqual("cached", result.source.source_mode)
+        self.assertTrue(result.source.pushdown.projections)
+        self.assertEqual("healthy", result.source.health.status)
+        self.assertEqual("preview.bounded", result.diagnostics[0].code)
+        self.assertEqual(1_000_000, request.max_bytes)
+        self.assertEqual(2_000, request.timeout_ms)
         self.assertEqual(provider.source, provider.create_snapshot(object(), object(), snapshot=snapshot))
 
+    def test_datasource_read_contract_preserves_live_preview_evidence(self) -> None:
+        request = DatasourceReadRequest(datasource_ref="datasource:monthly-cases")
+        result = DatasourceReadResult(
+            datasource=DatasourceDescriptor(
+                ref=request.datasource_ref,
+                source_name="monthly_cases",
+                name="Monthly cases",
+                kind="database",
+                mode="live",
+                shape="tabular",
+            ),
+            rows=(),
+            total_rows=0,
+            truncated=False,
+            returned_bytes=2,
+            elapsed_ms=3,
+            effective_row_limit=request.limit,
+            effective_byte_limit=request.max_bytes,
+            effective_timeout_ms=request.timeout_ms,
+            diagnostics=(
+                TabularPreviewDiagnostic(
+                    severity="info",
+                    code="preview.complete",
+                    message="The bounded preview completed.",
+                ),
+            ),
+        )
+
+        self.assertEqual(1_000_000, request.max_bytes)
+        self.assertEqual(2_000, request.timeout_ms)
+        self.assertEqual(2, result.returned_bytes)
+        self.assertEqual("preview.complete", result.diagnostics[0].code)
+
     def test_shared_csv_parser_preserves_identifier_zeroes_and_rejects_extra_values(self) -> None:
         rows = parse_tabular_csv(
             "case_id;amount;active\n0012;7.5;true\n\n",