diff --git a/pyproject.toml b/pyproject.toml index a4f1994..f54d256 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-dataflow" -version = "0.1.18" +version = "0.1.19" description = "Governed graphical and SQL data pipelines for GovOPlaN." readme = "README.md" requires-python = ">=3.12" diff --git a/src/govoplan_dataflow/__init__.py b/src/govoplan_dataflow/__init__.py index f2ed6e2..260ef6f 100644 --- a/src/govoplan_dataflow/__init__.py +++ b/src/govoplan_dataflow/__init__.py @@ -1,3 +1,3 @@ from __future__ import annotations -__version__ = "0.1.18" +__version__ = "0.1.19" diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 7524013..2cf9eef 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -56,7 +56,7 @@ from govoplan_dataflow.backend.dsar_provider import ( MODULE_ID = "dataflow" MODULE_NAME = "Dataflow" -MODULE_VERSION = "0.1.18" +MODULE_VERSION = "0.1.19" READ_SCOPE = "dataflow:pipeline:read" WRITE_SCOPE = "dataflow:pipeline:write" @@ -262,7 +262,12 @@ DOCUMENTATION = ( "Scope determines ownership and Policy inheritance. Templates can be derived but not run; complete " "flows may be previewed, revisioned, automated, and executed when effective Policy allows it. Saving " "appends an immutable revision. A scoped copy pins its source revision and content hash. Triggers pin " - "the revision and authorization grant, then re-evaluate authority for every delivery. Runs create " + "the revision and authorization grant, then re-evaluate authority for every delivery. Allow runs is " + "the definition-level admission boundary and does not grant a caller permission. A one-time run uses " + "the configured tenant-local date and time. The missed-run policy either coalesces elapsed interval " + "occurrences into one latest delivery or skips them; it never silently replays every missed occurrence. " + "The concurrency limit bounds active deliveries for that trigger and does not increase tenant worker " + "capacity. Runs create " "durable command and recovery evidence. Publishing creates a governed Datasource materialization, and " "environment promotion changes which immutable revision is eligible for staging or production runs. " "Recording a reconciliation decision uses optimistic concurrency and appends an immutable revision; " @@ -285,6 +290,10 @@ DOCUMENTATION = ( "help_contexts": [ "dataflow.field.scope", "dataflow.field.definition-kind", + "dataflow.field.allow-runs", + "dataflow.field.trigger-run-at", + "dataflow.field.trigger-missed-runs", + "dataflow.field.trigger-concurrency", "dataflow.action.save", "dataflow.action.derive", "dataflow.action.trigger", @@ -309,7 +318,11 @@ DOCUMENTATION = ( "environment, progress, cancellation, output, and recovery state. Database-only runs commit atomically. " "Publication to a governed Datasource uses forward recovery: an unknown provider outcome is reconciled " "before retry so output is not duplicated. Staging and production promotion is explicit and does not " - "rewrite a revision. Cancellation is best effort once external work has started; the final evidence " + "rewrite a revision. Freezing a published state assigns a durable label to the exact immutable output; " + "it does not copy or detach the data from Datasources retention, hold, and access rules. Artifact-backed " + "outputs return the same stable publication, datasource, and materialization references as inline " + "outputs. Datasource warnings and review-required states remain visible to Workflow instead of being " + "collapsed into success. Cancellation is best effort once external work has started; the final evidence " "states whether work stopped, completed, failed, or requires operator reconciliation. Scheduled, event, " "and queued execution is partitioned by tenant module entitlement before a run is claimed. Disabling " "Dataflow stops new admission and leaves accepted runs available for an explicit operator decision." @@ -325,6 +338,7 @@ DOCUMENTATION = ( "dataflow.runs", "dataflow.action.queue-run", "dataflow.action.publish", + "dataflow.field.freeze-publication", "dataflow.action.promote-staging", "dataflow.action.promote-production", "dataflow.state.recovery-attention", diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index b54b716..a344277 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -1584,6 +1584,26 @@ def _publish_pipeline_result( run.output_publication_ref = publication.ref run.output_datasource_ref = publication.datasource.ref run.output_materialization_ref = publication.materialization.ref + if publication.status in {"published_with_warnings", "review_required"}: + diagnostics = list(run.diagnostics) + diagnostics.append( + DataflowDiagnostic( + severity="warning", + code=( + "publication.review_required" + if publication.status == "review_required" + else "publication.warning" + ), + message=( + "The output materialization requires review before it can " + "become the datasource's current state." + if publication.status == "review_required" + else "The output was published with datasource validation " + "warnings." + ), + ).model_dump(mode="json") + ) + run.diagnostics = diagnostics def _mark_pipeline_run_failed( diff --git a/tests/test_service.py b/tests/test_service.py index 31fbc7c..fe19b3a 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -123,8 +123,9 @@ def runtime_identity() -> RuntimeIdentity: class FakePublicationProvider: - def __init__(self) -> None: + def __init__(self, status: str = "published") -> None: self.requests = [] + self.status = status def publish_rows(self, _session, _principal, *, request): self.requests.append(request) @@ -139,7 +140,7 @@ class FakePublicationProvider: ) return DatasourcePublicationResult( ref="publication:publication-1", - status="published", + status=self.status, datasource=descriptor, materialization=DatasourceMaterialization( ref="materialization:materialization-1", @@ -539,6 +540,36 @@ class DataflowServiceTests(unittest.TestCase): ), ) + def test_publication_review_state_is_exposed_to_workflow_as_a_warning( + self, + ) -> None: + pipeline = self._create() + provider = FakePublicationProvider("review_required") + + run, _ = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(provider), + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="review-publication", + publication=DataflowPublicationTarget( + name="Review output", + source_name="review_output", + ), + ), + ) + + self.assertEqual("succeeded", run.status) + self.assertEqual( + "publication.review_required", + run.diagnostics[-1]["code"], + ) + self.assertEqual("warning", run.diagnostics[-1]["severity"]) + def test_publication_without_datasources_finishes_as_failed_run(self) -> None: pipeline = self._create() run, _ = start_pipeline_run( diff --git a/webui/package.json b/webui/package.json index ad58da9..0a3b511 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/dataflow-webui", - "version": "0.1.18", + "version": "0.1.19", "private": true, "type": "module", "main": "src/index.ts", diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 699866b..0d82129 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -1201,6 +1201,10 @@ function DefinitionSettingsDialog({ /> onChange({ allowRun: value })} @@ -1737,7 +1741,7 @@ function DataflowTriggersDialog({ {kind === "once" ? ( - + setRunAt(event.target.value)} /> ) : null} @@ -1752,7 +1756,7 @@ function DataflowTriggersDialog({ onChange={(event) => setIntervalMinutes(Math.max(1, Number(event.target.value)))} /> - +