Preserve connector source runtime evidence

src/govoplan_datasources/backend/manifest.py

@@ -347,7 +347,9 @@ manifest = ModuleManifest(
                 "Datasources owns data identity, provenance, lifecycle, read semantics, "
                 "and the typed governance catalogue for authority, purpose, quality, "
                 "freshness, classification, correction, and dependent services, flows, "
-                "reports, controls, and decisions. Governance metadata visibility does not "
+                "reports, controls, and decisions. It preserves origin source mode, "
+                "structured health, declared pushdown, and effective row, byte, and time "
+                "limits for live previews. Governance metadata visibility does not "
                 "grant access to protected rows."
             ),
             layer="available",

src/govoplan_datasources/backend/router.py

@@ -31,9 +31,12 @@ from govoplan_datasources.backend.schemas import (
     DatasourceListResponse,
     DatasourceMaterializationListResponse,
     DatasourceMaterializationResponse,
+    DatasourceOriginHealthResponse,
     DatasourceOriginListResponse,
+    DatasourceOriginPushdownResponse,
     DatasourceOriginRegisterRequest,
     DatasourceOriginResponse,
+    DatasourcePreviewDiagnosticResponse,
     DatasourcePreviewResponse,
     DatasourceResponse,
     DatasourceRetireResponse,
@@ -383,6 +386,8 @@ def api_preview_datasource(
     offset: int = Query(default=0, ge=0),
     consistency: str = Query(default="current", pattern="^(current|live|frozen)$"),
     materialization_ref: str | None = Query(default=None, max_length=120),
+    max_bytes: int = Query(default=1_000_000, ge=1_024, le=20_000_000),
+    timeout_ms: int = Query(default=2_000, ge=100, le=10_000),
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> DatasourcePreviewResponse:
@@ -397,6 +402,8 @@ def api_preview_datasource(
                 consistency=consistency,  # type: ignore[arg-type]
                 limit=limit,
                 offset=offset,
+                max_bytes=max_bytes,
+                timeout_ms=timeout_ms,
             ),
         )
     except DatasourceError as exc:
@@ -411,6 +418,20 @@ def api_preview_datasource(
             if result.materialization
             else None
         ),
+        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=[
+            DatasourcePreviewDiagnosticResponse(
+                severity=item.severity,
+                code=item.code,
+                message=item.message,
+                details=dict(item.details),
+            )
+            for item in result.diagnostics
+        ],
     )
 
 
@@ -662,6 +683,23 @@ def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse:
         updated_at=item.updated_at.isoformat() if item.updated_at else None,
         capabilities=list(item.capabilities),
         metadata=dict(item.metadata),
+        source_mode=item.source_mode,
+        pushdown=DatasourceOriginPushdownResponse(
+            projections=item.pushdown.projections,
+            pagination=item.pushdown.pagination,
+            filters=list(item.pushdown.filters),
+            aggregations=list(item.pushdown.aggregations),
+            sorting=list(item.pushdown.sorting),
+        ),
+        health=DatasourceOriginHealthResponse(
+            status=item.health.status,
+            code=item.health.code,
+            summary=item.health.summary,
+            checked_at=(
+                item.health.checked_at.isoformat() if item.health.checked_at else None
+            ),
+            details=dict(item.health.details),
+        ),
     )
 
 

src/govoplan_datasources/backend/schemas.py

@@ -17,6 +17,7 @@ DatasourceKindValue = Literal[
     "custom",
 ]
 DatasourceShapeValue = Literal["tabular", "document", "binary", "directory", "stream"]
+TabularSourceModeValue = Literal["live", "cached", "file_backed", "static"]
 SourceAuthorityModeValue = Literal[
     "native_authoritative",
     "external_authoritative",
@@ -224,6 +225,29 @@ class DatasourceStagePromoteResponse(BaseModel):
     materialization: DatasourceMaterializationResponse
 
 
+class DatasourceOriginPushdownResponse(BaseModel):
+    projections: bool
+    pagination: bool
+    filters: list[str]
+    aggregations: list[str]
+    sorting: list[str]
+
+
+class DatasourceOriginHealthResponse(BaseModel):
+    status: Literal["healthy", "warning", "error", "unknown"]
+    code: str
+    summary: str
+    checked_at: str | None
+    details: dict[str, Any]
+
+
+class DatasourcePreviewDiagnosticResponse(BaseModel):
+    severity: Literal["info", "warning", "error"]
+    code: str
+    message: str
+    details: dict[str, Any]
+
+
 class DatasourceOriginResponse(BaseModel):
     ref: str
     source_name: str
@@ -241,6 +265,9 @@ class DatasourceOriginResponse(BaseModel):
     updated_at: str | None
     capabilities: list[str]
     metadata: dict[str, Any]
+    source_mode: TabularSourceModeValue
+    pushdown: DatasourceOriginPushdownResponse
+    health: DatasourceOriginHealthResponse
 
 
 class DatasourceOriginListResponse(BaseModel):
@@ -267,6 +294,12 @@ class DatasourcePreviewResponse(BaseModel):
     total_rows: int
     truncated: bool
     materialization: DatasourceMaterializationResponse | None
+    returned_bytes: int
+    elapsed_ms: int
+    effective_row_limit: int
+    effective_byte_limit: int
+    effective_timeout_ms: int
+    diagnostics: list[DatasourcePreviewDiagnosticResponse]
 
 
 class DatasourceFreezeRequest(BaseModel):

src/govoplan_datasources/backend/service.py

@@ -665,7 +665,7 @@ class SqlDatasourceProvider:
                 "origin_provider": origin.provider,
                 "registered_at": utcnow().isoformat(),
             },
-            metadata_=dict(origin.metadata),
+            metadata_=_origin_metadata(origin),
             created_by=_actor_id(api_principal),
             updated_by=_actor_id(api_principal),
         )
@@ -856,6 +856,8 @@ class SqlDatasourceProvider:
                     offset=offset,
                     columns=columns,
                     expected_fingerprint=request.expected_fingerprint,
+                    max_bytes=request.max_bytes,
+                    timeout_ms=request.timeout_ms,
                 ),
             )
         except DatasourceError:
@@ -872,12 +874,19 @@ class SqlDatasourceProvider:
             row_count=result.origin.row_count,
             byte_count=result.origin.byte_count,
             updated_at=result.origin.updated_at,
+            metadata=_origin_metadata(result.origin, current=item.metadata_),
         )
         return DatasourceReadResult(
             datasource=descriptor,
             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,
         )
 
     def _required_origin(
@@ -964,6 +973,7 @@ class SqlDatasourceProvider:
         frozen_label: str | None = None,
         set_current: bool,
     ) -> DatasourceMaterializationRecord:
+        item.metadata_ = _origin_metadata(origin, current=item.metadata_)
         normalized = normalize_rows(rows)
         schema = infer_schema(normalized) or origin.schema
         fingerprint = fingerprint_rows(normalized, schema)
@@ -992,6 +1002,38 @@ class SqlDatasourceProvider:
         )
 
 
+def _origin_metadata(
+    origin: DatasourceOrigin,
+    *,
+    current: Mapping[str, object] | None = None,
+) -> dict[str, object]:
+    return {
+        **dict(current or {}),
+        **dict(origin.metadata),
+        "source_contract": {
+            "source_mode": origin.source_mode,
+            "pushdown": {
+                "projections": origin.pushdown.projections,
+                "pagination": origin.pushdown.pagination,
+                "filters": list(origin.pushdown.filters),
+                "aggregations": list(origin.pushdown.aggregations),
+                "sorting": list(origin.pushdown.sorting),
+            },
+            "health": {
+                "status": origin.health.status,
+                "code": origin.health.code,
+                "summary": origin.health.summary,
+                "checked_at": (
+                    origin.health.checked_at.isoformat()
+                    if origin.health.checked_at
+                    else None
+                ),
+                "details": dict(origin.health.details),
+            },
+        },
+    }
+
+
 def _append_materialization(
     session: Session,
     *,