feat(dataflow): preserve publication review outcomes

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(