From 067c8273b627140215724917e891c56e16bf9493 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 12:05:35 +0200 Subject: [PATCH] Preserve connector source runtime evidence --- README.md | 3 ++ src/govoplan_datasources/backend/manifest.py | 4 +- src/govoplan_datasources/backend/router.py | 38 ++++++++++++++++ src/govoplan_datasources/backend/schemas.py | 33 ++++++++++++++ src/govoplan_datasources/backend/service.py | 44 ++++++++++++++++++- tests/test_lifecycle.py | 39 ++++++++++++++++ webui/src/api/datasources.ts | 26 +++++++++++ .../features/datasources/DatasourcesPage.tsx | 19 +++++++- 8 files changed, 202 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index a899269..38f822a 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,9 @@ Datasource contracts rather than connector implementations. The first executable slice supports tabular static uploads, connector-backed live and cached sources, staging and promotion, refresh, immutable snapshots, explicit frozen states, previews, retirement, and atomic producer publication. +Origin discovery retains each provider's source mode, structured health, and +declared pushdown support. Live previews preserve the provider's effective row, +serialized-byte, and elapsed-time limits and its redacted diagnostics. Producer modules can append a bounded tabular result or create a new static datasource through an idempotent capability. The publication ledger retains the producer run, output materialization, provenance, and replay identity. diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 3a48f6b..5e025b1 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/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", diff --git a/src/govoplan_datasources/backend/router.py b/src/govoplan_datasources/backend/router.py index a26da21..26cf775 100644 --- a/src/govoplan_datasources/backend/router.py +++ b/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), + ), ) diff --git a/src/govoplan_datasources/backend/schemas.py b/src/govoplan_datasources/backend/schemas.py index 6e06cab..6de1ae7 100644 --- a/src/govoplan_datasources/backend/schemas.py +++ b/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): diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index 8269636..c47a0a3 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/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, *, diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index f506050..e40ec06 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -22,6 +22,11 @@ from govoplan_core.core.datasources import ( DatasourceUnavailableError, DatasourceValidationError, ) +from govoplan_core.core.tabular_sources import ( + TabularPreviewDiagnostic, + TabularPushdown, + TabularSourceHealth, +) from govoplan_core.db.base import Base, utcnow from govoplan_datasources.backend.db.models import ( DatasourceGovernanceReferenceRecord, @@ -88,6 +93,13 @@ class FakeOriginProvider: fingerprint=f"version-{len(self.rows)}-{self.rows[-1]['name']}", row_count=len(self.rows), updated_at=utcnow(), + source_mode="cached", + pushdown=TabularPushdown(projections=True, pagination=True), + health=TabularSourceHealth( + status="healthy", + code="snapshot.ready", + summary="The immutable snapshot is ready.", + ), ) def list_origins( @@ -126,6 +138,18 @@ class FakeOriginProvider: rows=tuple(dict(row) for row in rows), total_rows=len(self.rows), truncated=request.offset + len(rows) < len(self.rows), + returned_bytes=64, + elapsed_ms=4, + 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.", + ), + ), ) @@ -513,10 +537,25 @@ class DatasourceLifecycleTests(unittest.TestCase): ) self.assertEqual(2, live_read.total_rows) + self.assertEqual(64, live_read.returned_bytes) + self.assertEqual("preview.complete", live_read.diagnostics[0].code) + self.assertEqual( + "cached", + live_read.datasource.metadata["source_contract"]["source_mode"], + ) + self.assertTrue( + live_read.datasource.metadata["source_contract"]["pushdown"][ + "projections" + ] + ) self.assertEqual(1, cached_before.total_rows) self.assertEqual(2, cached_live.total_rows) self.assertEqual(2, cached_after.total_rows) self.assertEqual(materialization.ref, refreshed.current_materialization_ref) + self.assertEqual( + "healthy", + refreshed.metadata["source_contract"]["health"]["status"], + ) def test_tenant_and_scope_isolation(self) -> None: stage = self.provider.create_stage( diff --git a/webui/src/api/datasources.ts b/webui/src/api/datasources.ts index 506072e..86895fd 100644 --- a/webui/src/api/datasources.ts +++ b/webui/src/api/datasources.ts @@ -164,6 +164,21 @@ export type DatasourceOrigin = { updated_at?: string | null; capabilities: string[]; metadata: Record; + source_mode: "live" | "cached" | "file_backed" | "static"; + pushdown: { + projections: boolean; + pagination: boolean; + filters: string[]; + aggregations: string[]; + sorting: string[]; + }; + health: { + status: "healthy" | "warning" | "error" | "unknown"; + code: string; + summary: string; + checked_at?: string | null; + details: Record; + }; }; export type DatasourcePreview = { @@ -172,6 +187,17 @@ export type DatasourcePreview = { total_rows: number; truncated: boolean; materialization?: DatasourceMaterialization | null; + returned_bytes: number; + elapsed_ms: number; + effective_row_limit: number; + effective_byte_limit: number; + effective_timeout_ms: number; + diagnostics: Array<{ + severity: "info" | "warning" | "error"; + code: string; + message: string; + details: Record; + }>; }; export async function listDatasources( diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx index d9a22dd..14cd153 100644 --- a/webui/src/features/datasources/DatasourcesPage.tsx +++ b/webui/src/features/datasources/DatasourcesPage.tsx @@ -834,7 +834,8 @@ function OriginDetail({ origin }: { origin: DatasourceOrigin }) { <>
- + + @@ -849,8 +850,11 @@ function OriginDetail({ origin }: { origin: DatasourceOrigin }) {
Origin reference{origin.ref} - Shape{origin.shape} + Kind{readableToken(origin.kind)} + Shape{readableToken(origin.shape)} Fingerprint{shortFingerprint(origin.fingerprint)} + Health{origin.health.summary} + Pushdown{pushdownSummary(origin)}
@@ -864,6 +868,17 @@ function OriginDetail({ origin }: { origin: DatasourceOrigin }) { ); } +function pushdownSummary(origin: DatasourceOrigin): string { + const capabilities = [ + origin.pushdown.projections ? "projections" : "", + origin.pushdown.pagination ? "pagination" : "", + ...origin.pushdown.filters.map((item) => `filter:${item}`), + ...origin.pushdown.aggregations.map((item) => `aggregate:${item}`), + ...origin.pushdown.sorting.map((item) => `sort:${item}`) + ].filter(Boolean); + return capabilities.join(", ") || "None declared"; +} + function GovernanceDialog({ open, settings,