diff --git a/docs/DATASOURCE_AND_DEFINITION_GRAPH_ARCHITECTURE.md b/docs/DATASOURCE_AND_DEFINITION_GRAPH_ARCHITECTURE.md index fc9b89b..3bf8b5e 100644 --- a/docs/DATASOURCE_AND_DEFINITION_GRAPH_ARCHITECTURE.md +++ b/docs/DATASOURCE_AND_DEFINITION_GRAPH_ARCHITECTURE.md @@ -23,6 +23,8 @@ The initial provider path is: 4. Any datasource may expose a frozen state for reproducible execution evidence. 5. Dataflow stores an opaque datasource reference, state policy, and expected fingerprint. +6. A pinned Dataflow run may publish a complete bounded result as a new + immutable materialization through an idempotent Datasources capability. Database, REST/HTTP, LDAP/directory, managed file, watched-directory, feed, and stream providers fit behind the same origin contract. Provider-specific @@ -51,21 +53,30 @@ kind of Dataflow or leaking either module into Core. ## Current Implementation - Core graph and datasource contracts are versioned at `0.1.0`. -- Workflow exposes node-library discovery and definition validation APIs. +- Workflow exposes a reusable graph editor, node-library discovery, validation, + tenant-isolated definitions, immutable revisions, and activation pinning. - Datasources exposes catalogue, origins, staging, promotion, preview, - materialization history, refresh, freeze, and retirement APIs. + materialization history, refresh, freeze, retirement, and producer + publication APIs. - Datasources WebUI exposes all current lifecycle views. - Connectors adapts existing tabular snapshots to datasource origins. - Dataflow consumes only Datasources catalogue/lifecycle capabilities and can request current, live, or latest-frozen state. +- Dataflow exposes a pinned run-lifecycle capability and a Run/Publish surface. + Its first synchronous runner records lineage and terminal state, publishes + only complete bounded results, and retains output datasource/materialization + references. +- The focused composition check proves Connector origin -> Datasource -> + pinned Dataflow run -> frozen published materialization, including replay. ## Next Slices -1. Persist and version Workflow definitions, then reuse the graph UI shell. +1. Add persisted Workflow instances, resumable transitions, human activities, + retry policy, and event subscriptions against pinned definition revisions. 2. Add SQL database and governed REST origin providers with credential-envelope references and bounded pushdown. 3. Add managed-file and directory origins. 4. Add datasource quality rules, schema compatibility policy, retention, and promotion approvals. -5. Publish Dataflow outputs as new materializations and expose run lifecycle to - Workflow. +5. Add asynchronous Dataflow workers and durable artifact-backed outputs for + runs that exceed the synchronous row/time/byte limits. diff --git a/tools/checks/check-datasource-composition.py b/tools/checks/check-datasource-composition.py index ae14377..0149de2 100644 --- a/tools/checks/check-datasource-composition.py +++ b/tools/checks/check-datasource-composition.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Exercise the connector -> datasource -> dataflow capability path.""" +"""Exercise connector -> datasource -> dataflow publication capabilities.""" from __future__ import annotations @@ -9,7 +9,17 @@ from sqlalchemy.orm import sessionmaker from govoplan_connectors.backend.db.models import ConnectorTabularSource from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef -from govoplan_core.core.datasources import datasource_catalogue, datasource_lifecycle +from govoplan_core.core.dataflows import ( + DataflowPublicationTarget, + DataflowRunRequest, + dataflow_run_lifecycle, +) +from govoplan_core.core.datasources import ( + DatasourceReadRequest, + datasource_catalogue, + datasource_lifecycle, + datasource_publication, +) from govoplan_core.core.modules import ModuleContext from govoplan_core.core.tabular_sources import ( TabularSnapshotInput, @@ -22,11 +32,18 @@ from govoplan_dataflow.backend.schemas import ( GraphNode, GraphPosition, PipelineGraph, + PipelineCreateRequest, PipelinePreviewRequest, ) -from govoplan_dataflow.backend.service import preview_pipeline +from govoplan_dataflow.backend.db.models import ( + DataflowPipeline, + DataflowPipelineRevision, + DataflowRun, +) +from govoplan_dataflow.backend.service import create_pipeline, preview_pipeline from govoplan_datasources.backend.db.models import ( DatasourceMaterializationRecord, + DatasourcePublicationRecord, DatasourceRecord, DatasourceStageRecord, ) @@ -47,6 +64,10 @@ def main() -> int: DatasourceRecord.__table__, DatasourceMaterializationRecord.__table__, DatasourceStageRecord.__table__, + DatasourcePublicationRecord.__table__, + DataflowPipeline.__table__, + DataflowPipelineRevision.__table__, + DataflowRun.__table__, ], ) session_factory = sessionmaker(bind=engine) @@ -55,7 +76,15 @@ def main() -> int: writer = tabular_snapshot_writer(registry) lifecycle = datasource_lifecycle(registry) catalogue = datasource_catalogue(registry) - if writer is None or lifecycle is None or catalogue is None: + publisher = datasource_publication(registry) + runner = dataflow_run_lifecycle(registry) + if ( + writer is None + or lifecycle is None + or catalogue is None + or publisher is None + or runner is None + ): raise RuntimeError("Datasource composition capabilities are incomplete.") origin = writer.create_snapshot( @@ -102,8 +131,73 @@ def main() -> int: raise RuntimeError(f"Unexpected Dataflow rows: {result.rows!r}") if result.source_fingerprints[0]["source_ref"] != datasource.ref: raise RuntimeError("Dataflow lineage did not retain the datasource reference.") + pipeline = create_pipeline( + session, + tenant_id="tenant-1", + actor_id="account-1", + payload=PipelineCreateRequest( + name="Monthly case output", + status="active", + graph=_graph( + datasource_ref=datasource.ref, + fingerprint=datasource.fingerprint, + ), + editor_mode="graph", + ), + ) + run_request = DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="composition-run-1", + publication=DataflowPublicationTarget( + name="Monthly case result", + source_name="monthly_case_result", + freeze=True, + frozen_label="Composition evidence", + ), + ) + published = runner.start_run( + session, + principal, + request=run_request, + ) + replayed = runner.start_run( + session, + principal, + request=run_request, + ) + if published.status != "succeeded": + raise RuntimeError(f"Dataflow publication failed: {published.error}") + if replayed.ref != published.ref or not replayed.replayed: + raise RuntimeError("Dataflow run idempotency did not replay the prior run.") + if ( + not published.output_datasource_ref + or not published.output_materialization_ref + ): + raise RuntimeError("Dataflow publication did not retain output references.") + output = catalogue.read_datasource( + session, + principal, + request=DatasourceReadRequest( + datasource_ref=published.output_datasource_ref, + ), + ) + if list(output.rows) != expected_rows: + raise RuntimeError( + f"Unexpected published Dataflow rows: {list(output.rows)!r}" + ) + if ( + output.materialization is None + or output.materialization.ref != published.output_materialization_ref + or output.materialization.frozen_at is None + ): + raise RuntimeError( + "Published Datasource materialization is not pinned and frozen." + ) engine.dispose() - print("Connector -> Datasources -> Dataflow composition passed.") + print( + "Connector -> Datasources -> pinned Dataflow publication composition passed." + ) return 0 diff --git a/tools/checks/check-focused.sh b/tools/checks/check-focused.sh index cfe0201..7960b4b 100644 --- a/tools/checks/check-focused.sh +++ b/tools/checks/check-focused.sh @@ -88,6 +88,9 @@ cd /mnt/DATA/git/govoplan-dataflow/webui cd /mnt/DATA/git/govoplan-datasources/webui "$NPM" run typecheck +cd /mnt/DATA/git/govoplan-workflow/webui +"$NPM" run typecheck + cd /mnt/DATA/git/govoplan-mail/webui "$NPM" run test:mail-ui