Add governed Dataflow audience outputs

fixtures/golden/adrema-audience-selection/expected-output.json

@@ -0,0 +1,70 @@
+[
+  {
+    "recipient_key": "identity:alice",
+    "display_name": "Alice Example",
+    "identity_id": "identity-alice",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": "alice@example.test",
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-email-alice",
+    "selected_channel": "email",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "preference.explicit_email",
+    "_reconciliation_status": "match"
+  },
+  {
+    "recipient_key": "identity:bob",
+    "display_name": "Bob Example",
+    "identity_id": "identity-bob",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": "Example Street 2, 10000 Exampletown",
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-postal-bob",
+    "selected_channel": "postal",
+    "distribution_status": "suppressed",
+    "exclusion_reason": "preference.suppressed",
+    "policy_decision": "preference.suppressed",
+    "_reconciliation_status": "match"
+  },
+  {
+    "recipient_key": "identity:grace",
+    "display_name": "Grace Example",
+    "identity_id": "identity-grace",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": "Example Street 3, 10000 Exampletown",
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-postal-grace",
+    "selected_channel": "postal",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "preference.default_postal",
+    "_reconciliation_status": "match"
+  },
+  {
+    "recipient_key": "identity:frank",
+    "display_name": "Frank Example",
+    "identity_id": "identity-frank",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": "portal:frank",
+    "contact_point_id": "contact-portal-frank",
+    "selected_channel": "portal",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "manual.include_with_reason",
+    "_reconciliation_status": "match"
+  }
+]

fixtures/golden/adrema-audience-selection/graph.json

@@ -0,0 +1,299 @@
+{
+  "schema_version": 1,
+  "nodes": [
+    {
+      "id": "directory",
+      "type": "source.reference",
+      "label": "Frozen directory and contact snapshot",
+      "position": {"x": 40, "y": 60},
+      "config": {
+        "source_ref": "snapshot:directory-2026-01",
+        "source_name": "directory_snapshot",
+        "expected_fingerprint": "sha256:synthetic-directory-2026-01",
+        "consistency": "frozen",
+        "source_columns": [
+          "recipient_key", "display_name", "identity_id", "group_key",
+          "organization_unit_id", "function_id", "region", "purpose",
+          "effective", "effective_from", "effective_until",
+          "preferred_target", "preferred_contact_point_id", "preferred_channel",
+          "fallback_target", "fallback_contact_point_id", "fallback_channel",
+          "preference_decision", "sort_order"
+        ]
+      }
+    },
+    {
+      "id": "group-filter",
+      "type": "filter",
+      "label": "Typed group",
+      "position": {"x": 260, "y": 60},
+      "config": {"column": "group_key", "operator": "eq", "value": {"$parameter": "group_key"}}
+    },
+    {
+      "id": "organization-filter",
+      "type": "filter",
+      "label": "Organization unit",
+      "position": {"x": 480, "y": 60},
+      "config": {"column": "organization_unit_id", "operator": "eq", "value": "district-north"}
+    },
+    {
+      "id": "function-filter",
+      "type": "filter",
+      "label": "Effective function",
+      "position": {"x": 700, "y": 60},
+      "config": {"column": "function_id", "operator": "eq", "value": "resident-service"}
+    },
+    {
+      "id": "region-filter",
+      "type": "filter",
+      "label": "Geography",
+      "position": {"x": 920, "y": 60},
+      "config": {"column": "region", "operator": "eq", "value": {"$parameter": "region"}}
+    },
+    {
+      "id": "purpose-filter",
+      "type": "filter",
+      "label": "Distribution purpose",
+      "position": {"x": 1140, "y": 60},
+      "config": {"column": "purpose", "operator": "eq", "value": {"$parameter": "purpose"}}
+    },
+    {
+      "id": "effective-filter",
+      "type": "filter",
+      "label": "Effective-dated assignment",
+      "position": {"x": 1360, "y": 60},
+      "config": {"column": "effective", "operator": "eq", "value": true}
+    },
+    {
+      "id": "contact-fallback",
+      "type": "calculate",
+      "label": "Resolve preferred or fallback contact",
+      "position": {"x": 1580, "y": 60},
+      "config": {
+        "calculations": [
+          {"target_column": "selected_target", "expression": "coalesce(preferred_target, fallback_target)", "result_type": "string"},
+          {"target_column": "selected_contact_point_id", "expression": "coalesce(preferred_contact_point_id, fallback_contact_point_id)", "result_type": "string"},
+          {"target_column": "selected_channel", "expression": "coalesce(preferred_channel, fallback_channel)", "result_type": "string"},
+          {"target_column": "email", "expression": "CASE WHEN selected_channel = 'email' THEN selected_target ELSE NULL END", "result_type": "string"},
+          {"target_column": "postal_address", "expression": "CASE WHEN selected_channel = 'postal' THEN selected_target ELSE NULL END", "result_type": "string"},
+          {"target_column": "account_id", "expression": "CASE WHEN selected_channel = 'internal_mail' THEN selected_target ELSE NULL END", "result_type": "string"},
+          {"target_column": "portal_target", "expression": "CASE WHEN selected_channel = 'portal' THEN selected_target ELSE NULL END", "result_type": "string"}
+        ]
+      }
+    },
+    {
+      "id": "directory-fields",
+      "type": "select",
+      "label": "Canonical recipient candidate",
+      "position": {"x": 1800, "y": 60},
+      "config": {
+        "fields": [
+          "recipient_key", "display_name", "identity_id", "group_key",
+          "organization_unit_id", "function_id", "selected_target",
+          "selected_contact_point_id", "selected_channel", "email",
+          "postal_address", "account_id", "portal_target",
+          "preference_decision", "sort_order"
+        ]
+      }
+    },
+    {
+      "id": "candidate-quality",
+      "type": "quality.rules",
+      "label": "Recipient quality gate",
+      "position": {"x": 2020, "y": 60},
+      "config": {
+        "rules": [
+          {"id": "recipient-key", "column": "recipient_key", "operator": "not_null"},
+          {"id": "delivery-target", "column": "selected_target", "operator": "not_null"},
+          {"id": "contact-point", "column": "selected_contact_point_id", "operator": "not_null"}
+        ],
+        "action": "drop"
+      }
+    },
+    {
+      "id": "manual",
+      "type": "source.reference",
+      "label": "Frozen manual includes",
+      "position": {"x": 1580, "y": 300},
+      "config": {
+        "source_ref": "snapshot:manual-includes-2026-01",
+        "source_name": "manual_includes",
+        "expected_fingerprint": "sha256:synthetic-manual-includes-2026-01",
+        "consistency": "frozen",
+        "source_columns": [
+          "recipient_key", "display_name", "identity_id", "group_key",
+          "organization_unit_id", "function_id", "selected_target",
+          "selected_contact_point_id", "selected_channel", "email",
+          "postal_address", "account_id", "portal_target",
+          "preference_decision", "sort_order"
+        ]
+      }
+    },
+    {
+      "id": "append-includes",
+      "type": "combine.union",
+      "label": "Append manual includes",
+      "position": {"x": 2240, "y": 170},
+      "config": {"mode": "all"}
+    },
+    {
+      "id": "deduplicate",
+      "type": "distinct",
+      "label": "One candidate per identity",
+      "position": {"x": 2460, "y": 170},
+      "config": {"columns": ["recipient_key"]}
+    },
+    {
+      "id": "suppressions",
+      "type": "source.reference",
+      "label": "Frozen channel suppressions",
+      "position": {"x": 2240, "y": 410},
+      "config": {
+        "source_ref": "snapshot:suppressions-2026-01",
+        "source_name": "channel_suppressions",
+        "expected_fingerprint": "sha256:synthetic-suppressions-2026-01",
+        "consistency": "frozen",
+        "source_columns": ["recipient_key", "suppression_reason"]
+      }
+    },
+    {
+      "id": "apply-suppressions",
+      "type": "combine.join",
+      "label": "Attach suppression decisions",
+      "position": {"x": 2680, "y": 250},
+      "config": {
+        "join_type": "left",
+        "left_keys": ["recipient_key"],
+        "right_keys": ["recipient_key"],
+        "right_prefix": "suppression_"
+      }
+    },
+    {
+      "id": "decision-columns",
+      "type": "calculate",
+      "label": "Classify channel decision",
+      "position": {"x": 2900, "y": 250},
+      "config": {
+        "calculations": [
+          {"target_column": "distribution_status", "expression": "CASE WHEN suppression_suppression_reason IS NOT NULL THEN 'suppressed' ELSE 'usable' END", "result_type": "string"},
+          {"target_column": "exclusion_reason", "expression": "suppression_suppression_reason", "result_type": "string"},
+          {"target_column": "policy_decision", "expression": "coalesce(suppression_suppression_reason, preference_decision)", "result_type": "string"}
+        ]
+      }
+    },
+    {
+      "id": "decision-fields",
+      "type": "select",
+      "label": "Canonical audience decision",
+      "position": {"x": 3120, "y": 250},
+      "config": {
+        "fields": [
+          "recipient_key", "display_name", "identity_id",
+          "organization_unit_id", "function_id", "email", "postal_address",
+          "account_id", "portal_target",
+          {"column": "selected_contact_point_id", "alias": "contact_point_id"},
+          "selected_channel", "distribution_status", "exclusion_reason",
+          "policy_decision", "sort_order"
+        ]
+      }
+    },
+    {
+      "id": "stable-order",
+      "type": "sort",
+      "label": "Stable recipient order",
+      "position": {"x": 3340, "y": 250},
+      "config": {
+        "fields": [
+          {"column": "sort_order", "direction": "asc"},
+          {"column": "recipient_key", "direction": "asc"}
+        ]
+      }
+    },
+    {
+      "id": "expected",
+      "type": "source.reference",
+      "label": "Frozen expected audience",
+      "position": {"x": 3340, "y": 500},
+      "config": {
+        "source_ref": "snapshot:expected-audience-2026-01",
+        "source_name": "expected_audience",
+        "expected_fingerprint": "sha256:synthetic-expected-audience-2026-01",
+        "consistency": "frozen",
+        "source_columns": [
+          "recipient_key", "display_name", "identity_id",
+          "organization_unit_id", "function_id", "email", "postal_address",
+          "account_id", "portal_target", "contact_point_id",
+          "selected_channel", "distribution_status", "exclusion_reason",
+          "policy_decision"
+        ]
+      }
+    },
+    {
+      "id": "reconcile",
+      "type": "reconcile.compare",
+      "label": "Reconcile expected audience",
+      "position": {"x": 3560, "y": 350},
+      "config": {
+        "left_keys": ["recipient_key"],
+        "right_keys": ["recipient_key"],
+        "compare_columns": [
+          "selected_channel", "contact_point_id", "distribution_status",
+          "exclusion_reason", "policy_decision"
+        ],
+        "right_prefix": "actual_"
+      }
+    },
+    {
+      "id": "reconciled-only",
+      "type": "filter",
+      "label": "Require exact reconciliation",
+      "position": {"x": 3780, "y": 350},
+      "config": {"column": "_reconciliation_status", "operator": "eq", "value": "match"}
+    },
+    {
+      "id": "output-fields",
+      "type": "select",
+      "label": "Distribution-list dataset",
+      "position": {"x": 4000, "y": 350},
+      "config": {
+        "fields": [
+          "recipient_key", "display_name", "identity_id",
+          "organization_unit_id", "function_id", "email", "postal_address",
+          "account_id", "portal_target", "contact_point_id",
+          "selected_channel", "distribution_status", "exclusion_reason",
+          "policy_decision", "_reconciliation_status"
+        ]
+      }
+    },
+    {
+      "id": "output",
+      "type": "output",
+      "label": "Pinned audience output",
+      "position": {"x": 4220, "y": 350},
+      "config": {}
+    }
+  ],
+  "edges": [
+    {"id": "e01", "source": "directory", "target": "group-filter"},
+    {"id": "e02", "source": "group-filter", "target": "organization-filter"},
+    {"id": "e03", "source": "organization-filter", "target": "function-filter"},
+    {"id": "e04", "source": "function-filter", "target": "region-filter"},
+    {"id": "e05", "source": "region-filter", "target": "purpose-filter"},
+    {"id": "e06", "source": "purpose-filter", "target": "effective-filter"},
+    {"id": "e07", "source": "effective-filter", "target": "contact-fallback"},
+    {"id": "e08", "source": "contact-fallback", "target": "directory-fields"},
+    {"id": "e09", "source": "directory-fields", "target": "candidate-quality"},
+    {"id": "e10", "source": "candidate-quality", "target": "append-includes"},
+    {"id": "e11", "source": "manual", "target": "append-includes"},
+    {"id": "e12", "source": "append-includes", "target": "deduplicate"},
+    {"id": "e13", "source": "deduplicate", "target": "apply-suppressions", "target_port": "left"},
+    {"id": "e14", "source": "suppressions", "target": "apply-suppressions", "target_port": "right"},
+    {"id": "e15", "source": "apply-suppressions", "target": "decision-columns"},
+    {"id": "e16", "source": "decision-columns", "target": "decision-fields"},
+    {"id": "e17", "source": "decision-fields", "target": "stable-order"},
+    {"id": "e18", "source": "expected", "target": "reconcile", "target_port": "left"},
+    {"id": "e19", "source": "stable-order", "target": "reconcile", "target_port": "right"},
+    {"id": "e20", "source": "reconcile", "target": "reconciled-only"},
+    {"id": "e21", "source": "reconciled-only", "target": "output-fields"},
+    {"id": "e22", "source": "output-fields", "target": "output"}
+  ]
+}

fixtures/golden/adrema-audience-selection/inputs/directory-snapshot.json

@@ -0,0 +1,149 @@
+[
+  {
+    "recipient_key": "identity:alice",
+    "display_name": "Alice Example",
+    "identity_id": "identity-alice",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "monthly-notice",
+    "effective": true,
+    "effective_from": "2025-01-01",
+    "effective_until": null,
+    "preferred_target": "alice@example.test",
+    "preferred_contact_point_id": "contact-email-alice",
+    "preferred_channel": "email",
+    "fallback_target": "Example Street 1, 10000 Exampletown",
+    "fallback_contact_point_id": "contact-postal-alice",
+    "fallback_channel": "postal",
+    "preference_decision": "preference.explicit_email",
+    "sort_order": 10
+  },
+  {
+    "recipient_key": "identity:bob",
+    "display_name": "Bob Example",
+    "identity_id": "identity-bob",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "monthly-notice",
+    "effective": true,
+    "effective_from": "2025-06-01",
+    "effective_until": "2027-01-01",
+    "preferred_target": null,
+    "preferred_contact_point_id": null,
+    "preferred_channel": null,
+    "fallback_target": "Example Street 2, 10000 Exampletown",
+    "fallback_contact_point_id": "contact-postal-bob",
+    "fallback_channel": "postal",
+    "preference_decision": "preference.default_postal",
+    "sort_order": 20
+  },
+  {
+    "recipient_key": "identity:grace",
+    "display_name": "Grace Example",
+    "identity_id": "identity-grace",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "monthly-notice",
+    "effective": true,
+    "effective_from": "2024-01-01",
+    "effective_until": null,
+    "preferred_target": null,
+    "preferred_contact_point_id": null,
+    "preferred_channel": null,
+    "fallback_target": "Example Street 3, 10000 Exampletown",
+    "fallback_contact_point_id": "contact-postal-grace",
+    "fallback_channel": "postal",
+    "preference_decision": "preference.default_postal",
+    "sort_order": 30
+  },
+  {
+    "recipient_key": "identity:carol",
+    "display_name": "Carol Example",
+    "identity_id": "identity-carol",
+    "group_key": "employees",
+    "organization_unit_id": "district-south",
+    "function_id": "finance-officer",
+    "region": "south",
+    "purpose": "monthly-notice",
+    "effective": true,
+    "effective_from": "2025-01-01",
+    "effective_until": null,
+    "preferred_target": "carol@example.test",
+    "preferred_contact_point_id": "contact-email-carol",
+    "preferred_channel": "email",
+    "fallback_target": null,
+    "fallback_contact_point_id": null,
+    "fallback_channel": null,
+    "preference_decision": "preference.explicit_email",
+    "sort_order": 40
+  },
+  {
+    "recipient_key": "identity:dana",
+    "display_name": "Dana Example",
+    "identity_id": "identity-dana",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "newsletter",
+    "effective": true,
+    "effective_from": "2025-01-01",
+    "effective_until": null,
+    "preferred_target": "dana@example.test",
+    "preferred_contact_point_id": "contact-email-dana",
+    "preferred_channel": "email",
+    "fallback_target": null,
+    "fallback_contact_point_id": null,
+    "fallback_channel": null,
+    "preference_decision": "preference.explicit_email",
+    "sort_order": 50
+  },
+  {
+    "recipient_key": "identity:erin",
+    "display_name": "Erin Example",
+    "identity_id": "identity-erin",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "monthly-notice",
+    "effective": false,
+    "effective_from": "2024-01-01",
+    "effective_until": "2025-01-01",
+    "preferred_target": "erin@example.test",
+    "preferred_contact_point_id": "contact-email-erin",
+    "preferred_channel": "email",
+    "fallback_target": null,
+    "fallback_contact_point_id": null,
+    "fallback_channel": null,
+    "preference_decision": "preference.explicit_email",
+    "sort_order": 60
+  },
+  {
+    "recipient_key": null,
+    "display_name": "Malformed Synthetic Row",
+    "identity_id": null,
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "region": "north",
+    "purpose": "monthly-notice",
+    "effective": true,
+    "effective_from": "2025-01-01",
+    "effective_until": null,
+    "preferred_target": null,
+    "preferred_contact_point_id": null,
+    "preferred_channel": null,
+    "fallback_target": null,
+    "fallback_contact_point_id": null,
+    "fallback_channel": null,
+    "preference_decision": "preference.unresolved",
+    "sort_order": 70
+  }
+]

fixtures/golden/adrema-audience-selection/inputs/expected-audience.json

@@ -0,0 +1,66 @@
+[
+  {
+    "recipient_key": "identity:alice",
+    "display_name": "Alice Example",
+    "identity_id": "identity-alice",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": "alice@example.test",
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-email-alice",
+    "selected_channel": "email",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "preference.explicit_email"
+  },
+  {
+    "recipient_key": "identity:bob",
+    "display_name": "Bob Example",
+    "identity_id": "identity-bob",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": "Example Street 2, 10000 Exampletown",
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-postal-bob",
+    "selected_channel": "postal",
+    "distribution_status": "suppressed",
+    "exclusion_reason": "preference.suppressed",
+    "policy_decision": "preference.suppressed"
+  },
+  {
+    "recipient_key": "identity:grace",
+    "display_name": "Grace Example",
+    "identity_id": "identity-grace",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": "Example Street 3, 10000 Exampletown",
+    "account_id": null,
+    "portal_target": null,
+    "contact_point_id": "contact-postal-grace",
+    "selected_channel": "postal",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "preference.default_postal"
+  },
+  {
+    "recipient_key": "identity:frank",
+    "display_name": "Frank Example",
+    "identity_id": "identity-frank",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "email": null,
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": "portal:frank",
+    "contact_point_id": "contact-portal-frank",
+    "selected_channel": "portal",
+    "distribution_status": "usable",
+    "exclusion_reason": null,
+    "policy_decision": "manual.include_with_reason"
+  }
+]

fixtures/golden/adrema-audience-selection/inputs/manual-includes.json

@@ -0,0 +1,36 @@
+[
+  {
+    "recipient_key": "identity:alice",
+    "display_name": "Alice Example (duplicate manual include)",
+    "identity_id": "identity-alice",
+    "group_key": "residents",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "selected_target": "alice@example.test",
+    "selected_contact_point_id": "contact-email-alice",
+    "selected_channel": "email",
+    "email": "alice@example.test",
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": null,
+    "preference_decision": "manual.include_duplicate",
+    "sort_order": 11
+  },
+  {
+    "recipient_key": "identity:frank",
+    "display_name": "Frank Example",
+    "identity_id": "identity-frank",
+    "group_key": "manual",
+    "organization_unit_id": "district-north",
+    "function_id": "resident-service",
+    "selected_target": "portal:frank",
+    "selected_contact_point_id": "contact-portal-frank",
+    "selected_channel": "portal",
+    "email": null,
+    "postal_address": null,
+    "account_id": null,
+    "portal_target": "portal:frank",
+    "preference_decision": "manual.include_with_reason",
+    "sort_order": 40
+  }
+]

fixtures/golden/adrema-audience-selection/inputs/suppressions.json

@@ -0,0 +1,6 @@
+[
+  {
+    "recipient_key": "identity:bob",
+    "suppression_reason": "preference.suppressed"
+  }
+]

fixtures/golden/adrema-audience-selection/manifest.json

@@ -0,0 +1,92 @@
+{
+  "id": "adrema-audience-selection",
+  "title": "Synthetic governed AdreMa audience selection",
+  "issue": "GovOPlaN/govoplan-dataflow#18",
+  "tenant_id": "tenant-synthetic-adrema",
+  "graph": "graph.json",
+  "expected_output": "expected-output.json",
+  "parameters": {
+    "group_key": "residents",
+    "region": "north",
+    "purpose": "monthly-notice"
+  },
+  "sources": {
+    "snapshot:directory-2026-01": {
+      "fixture": "inputs/directory-snapshot.json",
+      "tenant_id": "tenant-synthetic-adrema",
+      "provider": "connectors.synthetic-directory",
+      "fingerprint": "sha256:synthetic-directory-2026-01"
+    },
+    "snapshot:manual-includes-2026-01": {
+      "fixture": "inputs/manual-includes.json",
+      "tenant_id": "tenant-synthetic-adrema",
+      "provider": "datasources.static",
+      "fingerprint": "sha256:synthetic-manual-includes-2026-01"
+    },
+    "snapshot:suppressions-2026-01": {
+      "fixture": "inputs/suppressions.json",
+      "tenant_id": "tenant-synthetic-adrema",
+      "provider": "addresses.synthetic-preferences",
+      "fingerprint": "sha256:synthetic-suppressions-2026-01"
+    },
+    "snapshot:expected-audience-2026-01": {
+      "fixture": "inputs/expected-audience.json",
+      "tenant_id": "tenant-synthetic-adrema",
+      "provider": "reporting.synthetic-baseline",
+      "fingerprint": "sha256:synthetic-expected-audience-2026-01"
+    }
+  },
+  "expected_definition_hash": "2fec104dfda522dbc708ce61b237c062919703d62d2912ba621d5873ee5b975e",
+  "expected_execution_graph_hash": "d1cbb51c837810300044ed5f660e295ace49411265108b102731174d5587a85d",
+  "expected_output_hash": "f2d744354279f80393c6f894f7854569aa686f8292e6256fe429569d4410a72e",
+  "expected_source_fingerprints": [
+    {
+      "node_id": "directory",
+      "source_ref": "snapshot:directory-2026-01",
+      "source_name": "directory_snapshot",
+      "kind": "datasource",
+      "provider": "connectors.synthetic-directory",
+      "fingerprint": "sha256:synthetic-directory-2026-01",
+      "row_count": 7,
+      "preview_rows": 7,
+      "truncated": false
+    },
+    {
+      "node_id": "manual",
+      "source_ref": "snapshot:manual-includes-2026-01",
+      "source_name": "manual_includes",
+      "kind": "datasource",
+      "provider": "datasources.static",
+      "fingerprint": "sha256:synthetic-manual-includes-2026-01",
+      "row_count": 2,
+      "preview_rows": 2,
+      "truncated": false
+    },
+    {
+      "node_id": "suppressions",
+      "source_ref": "snapshot:suppressions-2026-01",
+      "source_name": "channel_suppressions",
+      "kind": "datasource",
+      "provider": "addresses.synthetic-preferences",
+      "fingerprint": "sha256:synthetic-suppressions-2026-01",
+      "row_count": 1,
+      "preview_rows": 1,
+      "truncated": false
+    },
+    {
+      "node_id": "expected",
+      "source_ref": "snapshot:expected-audience-2026-01",
+      "source_name": "expected_audience",
+      "kind": "datasource",
+      "provider": "reporting.synthetic-baseline",
+      "fingerprint": "sha256:synthetic-expected-audience-2026-01",
+      "row_count": 4,
+      "preview_rows": 4,
+      "truncated": false
+    }
+  ],
+  "expected_recipients": ["identity:alice", "identity:grace", "identity:frank"],
+  "expected_exclusions": {
+    "identity:bob": "preference.suppressed"
+  }
+}

src/govoplan_dataflow/backend/dataset_output.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+import hashlib
+import json
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.dataflows import (
+    DataflowDatasetDescriptor,
+    DataflowDatasetRequest,
+    DataflowDatasetResult,
+    DataflowRunConflictError,
+    DataflowRunUnavailableError,
+)
+from govoplan_core.security.time import utc_now
+from govoplan_dataflow.backend.backends.base import ExecutionBudget
+from govoplan_dataflow.backend.executor import PipelineExecutionError
+from govoplan_dataflow.backend.governance import require_definition_action
+from govoplan_dataflow.backend.schemas import PipelineGraph
+from govoplan_dataflow.backend.service import (
+    _execute_pipeline_preview,
+    get_pipeline,
+    get_pipeline_revision,
+    list_pipelines,
+)
+from govoplan_dataflow.backend.subflows import substitute_parameters
+
+
+class SqlDataflowDatasetOutputProvider:
+    def __init__(self, registry: object | None = None) -> None:
+        self.registry = registry
+
+    def list_outputs(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str = "",
+        limit: int = 100,
+    ) -> tuple[DataflowDatasetDescriptor, ...]:
+        typed_session, typed_principal = _contracts(session, principal)
+        normalized_query = query.strip().casefold()
+        result: list[DataflowDatasetDescriptor] = []
+        for pipeline in list_pipelines(
+            typed_session,
+            tenant_id=typed_principal.tenant_id,
+        ):
+            if normalized_query and normalized_query not in pipeline.name.casefold():
+                continue
+            try:
+                decision = require_definition_action(
+                    pipeline,
+                    principal=typed_principal,
+                    registry=self.registry,
+                    action="run",
+                )
+            except PermissionError:
+                continue
+            revision = get_pipeline_revision(typed_session, pipeline=pipeline)
+            result.append(
+                DataflowDatasetDescriptor(
+                    pipeline_ref=pipeline.id,
+                    name=pipeline.name,
+                    description=pipeline.description,
+                    revision=revision.revision,
+                    definition_hash=revision.content_hash,
+                    status=pipeline.status,
+                    updated_at=pipeline.updated_at,
+                    parameters=dict(pipeline.metadata_.get("parameters") or {}),
+                    provenance={
+                        "module": "dataflow",
+                        "scope_type": pipeline.scope_type,
+                        "scope_id": pipeline.scope_id,
+                        "policy_decision": decision.to_dict(),
+                    },
+                )
+            )
+            if len(result) >= max(1, min(limit, 200)):
+                break
+        return tuple(result)
+
+    def read_output(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: DataflowDatasetRequest,
+    ) -> DataflowDatasetResult:
+        typed_session, typed_principal = _contracts(session, principal)
+        row_limit = max(1, min(request.row_limit, 2_000))
+        pipeline = get_pipeline(
+            typed_session,
+            tenant_id=typed_principal.tenant_id,
+            pipeline_id=request.pipeline_ref,
+        )
+        decision = require_definition_action(
+            pipeline,
+            principal=typed_principal,
+            registry=self.registry,
+            action="run",
+        )
+        revision = get_pipeline_revision(
+            typed_session,
+            pipeline=pipeline,
+            revision=request.revision,
+        )
+        if (
+            request.expected_definition_hash
+            and request.expected_definition_hash != revision.content_hash
+        ):
+            raise DataflowRunConflictError(
+                "The pinned Dataflow definition hash no longer matches the requested revision."
+            )
+        graph = PipelineGraph.model_validate(
+            substitute_parameters(revision.graph, dict(request.parameters))
+        )
+        try:
+            result, executor_version = _execute_pipeline_preview(
+                graph,
+                session=typed_session,
+                principal=typed_principal,
+                registry=self.registry,
+                backend="auto",
+                row_limit=row_limit,
+                preview_node_id=None,
+                budget=ExecutionBudget(
+                    max_output_rows=row_limit,
+                    max_batch_bytes=4_000_000,
+                    max_wall_seconds=5.0,
+                ),
+            )
+        except PipelineExecutionError as exc:
+            raise DataflowRunUnavailableError(
+                f"The pinned Dataflow output could not be evaluated: {exc}"
+            ) from exc
+        source_fingerprints = tuple(
+            dict(item) for item in result.source_fingerprints
+        )
+        if request.expected_source_fingerprints and not _fingerprints_match(
+            request.expected_source_fingerprints,
+            source_fingerprints,
+        ):
+            raise DataflowRunConflictError(
+                "Dataflow source fingerprints differ from the pinned source state."
+            )
+        output_hash = hashlib.sha256(
+            json.dumps(
+                result.rows,
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=True,
+                default=str,
+            ).encode("utf-8")
+        ).hexdigest()
+        return DataflowDatasetResult(
+            pipeline_ref=pipeline.id,
+            revision=revision.revision,
+            definition_hash=revision.content_hash,
+            rows=tuple(dict(item) for item in result.rows),
+            total_rows=result.total_rows,
+            truncated=result.truncated,
+            output_hash=output_hash,
+            executor_version=executor_version,
+            source_fingerprints=source_fingerprints,
+            diagnostics=tuple(
+                item.model_dump(mode="json") for item in result.diagnostics
+            ),
+            generated_at=utc_now(),
+            provenance={
+                "module": "dataflow",
+                "scope_type": pipeline.scope_type,
+                "scope_id": pipeline.scope_id,
+                "policy_decision": decision.to_dict(),
+                "parameters": dict(request.parameters),
+            },
+        )
+
+
+def dataset_output_provider(context: object | None = None):
+    return SqlDataflowDatasetOutputProvider(getattr(context, "registry", None))
+
+
+def _fingerprints_match(expected, actual) -> bool:
+    def normalized(values):
+        return sorted(
+            json.dumps(dict(item), sort_keys=True, separators=(",", ":"), default=str)
+            for item in values
+        )
+
+    return normalized(expected) == normalized(actual)
+
+
+def _contracts(session: object, principal: object):
+    if not hasattr(session, "scalar") or not hasattr(session, "scalars"):
+        raise TypeError("Dataflow dataset output requires a SQLAlchemy session.")
+    if not isinstance(principal, ApiPrincipal):
+        raise TypeError("Dataflow dataset output requires an API principal.")
+    return session, principal
+
+
+__all__ = ["SqlDataflowDatasetOutputProvider", "dataset_output_provider"]

src/govoplan_dataflow/backend/manifest.py

@@ -11,6 +11,7 @@ from govoplan_core.core.access import (
     CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
 )
 from govoplan_core.core.dataflows import (
+    CAPABILITY_DATAFLOW_DATASET_OUTPUT,
     CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
     CAPABILITY_DATAFLOW_RUN_WORKER,
     CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
@@ -279,7 +280,7 @@ manifest = ModuleManifest(
         ModuleInterfaceProvider(name="dataflow.pipeline_preview", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="dataflow.run_lifecycle", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="dataflow.run_worker", version=MODULE_VERSION),
-        ModuleInterfaceProvider(name="dataflow.dataset_output", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION),
         ModuleInterfaceProvider(
             name="dataflow.trigger_dispatcher",
             version=MODULE_VERSION,
@@ -378,6 +379,10 @@ manifest = ModuleManifest(
     ),
     route_factory=_dataflow_router,
     capability_factories={
+        CAPABILITY_DATAFLOW_DATASET_OUTPUT: lambda context: __import__(
+            "govoplan_dataflow.backend.dataset_output",
+            fromlist=["dataset_output_provider"],
+        ).dataset_output_provider(context),
         CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
         CAPABILITY_DATAFLOW_RUN_WORKER: _run_worker,
         CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,

tests/test_dataset_output.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.dataflows import DataflowDatasetRequest, DataflowRunConflictError
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import configure_database, reset_database
+from govoplan_dataflow.backend.dataset_output import SqlDataflowDatasetOutputProvider
+from govoplan_dataflow.backend.db.models import DataflowPipeline, DataflowPipelineRevision
+from govoplan_dataflow.backend.schemas import PipelineGraph
+from govoplan_dataflow.backend.service import definition_hash
+
+
+def principal(tenant_id: str = "tenant-1") -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id=tenant_id,
+            scopes=frozenset({"dataflow:pipeline:read", "dataflow:pipeline:run"}),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class DataflowDatasetOutputTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.database = configure_database("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.database.engine,
+            tables=[DataflowPipeline.__table__, DataflowPipelineRevision.__table__],
+        )
+
+    def tearDown(self) -> None:
+        reset_database(dispose=True)
+
+    def test_pinned_revision_executes_through_typed_registry(self) -> None:
+        graph = PipelineGraph.model_validate(
+            {
+                "nodes": [
+                    {
+                        "id": "source",
+                        "type": "source.inline",
+                        "label": "Source",
+                        "position": {"x": 0, "y": 0},
+                        "config": {
+                            "source_name": "audience",
+                            "rows": [
+                                {"recipient_key": "one", "email": "one@example.test"},
+                                {"recipient_key": "two", "email": "two@example.test"},
+                            ]
+                        },
+                    },
+                    {
+                        "id": "output",
+                        "type": "output",
+                        "label": "Output",
+                        "position": {"x": 200, "y": 0},
+                        "config": {},
+                    },
+                ],
+                "edges": [{"id": "edge-1", "source": "source", "target": "output"}],
+            }
+        )
+        content_hash = definition_hash(graph, None)
+        with self.database.session() as session:
+            pipeline = DataflowPipeline(
+                id="pipeline-1",
+                tenant_id="tenant-1",
+                scope_type="tenant",
+                scope_id="tenant-1",
+                definition_kind="flow",
+                name="Audience",
+                status="active",
+                current_revision=1,
+                allow_run=True,
+                metadata_={},
+            )
+            revision = DataflowPipelineRevision(
+                id="revision-1",
+                tenant_id="tenant-1",
+                pipeline_id=pipeline.id,
+                revision=1,
+                graph=graph.model_dump(mode="json"),
+                sql_text=None,
+                editor_mode="graph",
+                content_hash=content_hash,
+            )
+            session.add_all((pipeline, revision))
+            session.flush()
+
+            provider = SqlDataflowDatasetOutputProvider()
+            descriptors = provider.list_outputs(session, principal())
+            result = provider.read_output(
+                session,
+                principal(),
+                request=DataflowDatasetRequest(
+                    pipeline_ref=pipeline.id,
+                    revision=1,
+                    expected_definition_hash=content_hash,
+                ),
+            )
+
+            self.assertEqual(1, len(descriptors))
+            self.assertEqual(2, result.total_rows)
+            self.assertEqual("one", result.rows[0]["recipient_key"])
+            self.assertEqual(64, len(result.output_hash))
+
+            replay = provider.read_output(
+                session,
+                principal(),
+                request=DataflowDatasetRequest(
+                    pipeline_ref=pipeline.id,
+                    revision=1,
+                    expected_definition_hash=content_hash,
+                    expected_source_fingerprints=result.source_fingerprints,
+                ),
+            )
+            self.assertEqual(result.output_hash, replay.output_hash)
+
+            with self.assertRaises(DataflowRunConflictError):
+                provider.read_output(
+                    session,
+                    principal(),
+                    request=DataflowDatasetRequest(
+                        pipeline_ref=pipeline.id,
+                        revision=1,
+                        expected_definition_hash="wrong",
+                    ),
+                )
+            with self.assertRaises(DataflowRunConflictError):
+                provider.read_output(
+                    session,
+                    principal(),
+                    request=DataflowDatasetRequest(
+                        pipeline_ref=pipeline.id,
+                        revision=1,
+                        expected_definition_hash=content_hash,
+                        expected_source_fingerprints=(
+                            {"node_id": "source", "fingerprint": "changed"},
+                        ),
+                    ),
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_golden_flows.py

@@ -1,12 +1,19 @@
 from __future__ import annotations
 
 import json
+import hashlib
 import unittest
 from pathlib import Path
 
-from govoplan_dataflow.backend.executor import execute_preview
+from govoplan_dataflow.backend.executor import (
-from govoplan_dataflow.backend.graph import validate_graph
+    PipelineExecutionError,
+    ResolvedSource,
+    execute_preview,
+)
+from govoplan_dataflow.backend.expressions import ExpressionError, parse_expression
+from govoplan_dataflow.backend.graph import definition_hash, validate_graph
 from govoplan_dataflow.backend.schemas import PipelineGraph
+from govoplan_dataflow.backend.subflows import substitute_parameters
 
 
 FIXTURES = Path(__file__).parents[1] / "fixtures" / "golden"
@@ -21,13 +28,18 @@ class GoldenFlowTests(unittest.TestCase):
         for directory in directories:
             with self.subTest(flow=directory.name):
                 manifest = _json(directory / "manifest.json")
-                graph_payload = _json(directory / str(manifest["graph"]))
+                raw_graph_payload = _json(directory / str(manifest["graph"]))
+                graph_payload = json.loads(json.dumps(raw_graph_payload))
                 for item in graph_payload["nodes"]:
                     fixture = item.get("config", {}).get("fixture")
-                    if fixture:
+                    if fixture and item["type"] == "source.inline":
                         item["config"]["rows"] = _json(
                             directory / "inputs" / fixture
                         )
+                graph_payload = substitute_parameters(
+                    graph_payload,
+                    dict(manifest.get("parameters") or {}),
+                )
                 graph = PipelineGraph.model_validate(graph_payload)
                 diagnostics = validate_graph(graph)
                 self.assertEqual(
@@ -39,17 +51,147 @@ class GoldenFlowTests(unittest.TestCase):
                     ],
                 )
 
-                result = execute_preview(graph, row_limit=500)
+                resolver = _source_resolver(directory, manifest)
+                result = execute_preview(
+                    graph,
+                    row_limit=500,
+                    source_resolver=resolver,
+                )
 
                 self.assertEqual(
                     _json(directory / str(manifest["expected_output"])),
                     result.rows,
                 )
+                if "expected_definition_hash" in manifest:
+                    self.assertEqual(
+                        manifest["expected_definition_hash"],
+                        definition_hash(PipelineGraph.model_validate(raw_graph_payload)),
+                    )
+                if "expected_execution_graph_hash" in manifest:
+                    self.assertEqual(
+                        manifest["expected_execution_graph_hash"],
+                        definition_hash(graph),
+                    )
+                if "expected_output_hash" in manifest:
+                    self.assertEqual(
+                        manifest["expected_output_hash"],
+                        _rows_hash(result.rows),
+                    )
+                if "expected_source_fingerprints" in manifest:
+                    self.assertEqual(
+                        manifest["expected_source_fingerprints"],
+                        result.source_fingerprints,
+                    )
+                full_result = execute_preview(
+                    graph,
+                    row_limit=5_000,
+                    source_resolver=resolver,
+                )
+                self.assertEqual(result.rows, full_result.rows)
+                self.assertFalse(full_result.truncated)
+
+    def test_adrema_fixture_rejects_untrusted_sources_and_unsafe_expression(self) -> None:
+        directory = FIXTURES / "adrema-audience-selection"
+        manifest = _json(directory / "manifest.json")
+        graph = PipelineGraph.model_validate(
+            substitute_parameters(
+                _json(directory / "graph.json"),
+                dict(manifest["parameters"]),
+            )
+        )
+        drifted_manifest = {
+            **manifest,
+            "sources": {
+                **manifest["sources"],
+                "snapshot:directory-2026-01": {
+                    **manifest["sources"]["snapshot:directory-2026-01"],
+                    "fingerprint": "sha256:changed-source",
+                },
+            },
+        }
+
+        with self.assertRaisesRegex(PipelineExecutionError, "fingerprint"):
+            execute_preview(
+                graph,
+                row_limit=500,
+                source_resolver=_source_resolver(directory, drifted_manifest),
+            )
+        cross_tenant_manifest = {
+            **manifest,
+            "sources": {
+                **manifest["sources"],
+                "snapshot:directory-2026-01": {
+                    **manifest["sources"]["snapshot:directory-2026-01"],
+                    "tenant_id": "tenant-other",
+                },
+            },
+        }
+        with self.assertRaisesRegex(PipelineExecutionError, "tenant"):
+            execute_preview(
+                graph,
+                row_limit=500,
+                source_resolver=_source_resolver(directory, cross_tenant_manifest),
+            )
+
+        unknown_source_graph = graph.model_copy(deep=True)
+        unknown_source_graph.nodes[0].config["source_ref"] = "connector:untrusted"
+        with self.assertRaisesRegex(PipelineExecutionError, "pinned provider set"):
+            execute_preview(
+                unknown_source_graph,
+                row_limit=500,
+                source_resolver=_source_resolver(directory, manifest),
+            )
+        with self.assertRaises(ExpressionError):
+            parse_expression("read_csv('/tmp/private.csv')")
 
 
 def _json(path: Path):
     return json.loads(path.read_text(encoding="utf-8"))
 
 
+def _source_resolver(directory: Path, manifest: dict):
+    sources = dict(manifest.get("sources") or {})
+    tenant_id = str(manifest.get("tenant_id") or "")
+
+    def resolve(node, limit: int) -> ResolvedSource:
+        source_ref = str(node.config.get("source_ref") or "")
+        if source_ref not in sources:
+            raise ValueError(f"Fixture source {source_ref!r} is outside the pinned provider set.")
+        source = sources[source_ref]
+        if str(source.get("tenant_id") or "") != tenant_id:
+            raise ValueError(
+                f"Fixture source {source_ref!r} belongs to a different tenant."
+            )
+        fingerprint = str(source["fingerprint"])
+        expected = str(node.config.get("expected_fingerprint") or "")
+        if expected and expected != fingerprint:
+            raise ValueError(
+                f"Fixture source fingerprint changed for {source_ref!r}."
+            )
+        rows = _json(directory / str(source["fixture"]))
+        bounded = rows[:limit]
+        return ResolvedSource(
+            rows=tuple(dict(item) for item in bounded),
+            source_ref=source_ref,
+            provider=str(source["provider"]),
+            fingerprint=fingerprint,
+            total_rows=len(rows),
+            truncated=len(bounded) < len(rows),
+        )
+
+    return resolve
+
+
+def _rows_hash(rows: object) -> str:
+    return hashlib.sha256(
+        json.dumps(
+            rows,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+        ).encode("utf-8")
+    ).hexdigest()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_manifest.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import unittest
 
 from govoplan_core.core.dataflows import (
+    CAPABILITY_DATAFLOW_DATASET_OUTPUT,
     CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
     CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
 )
@@ -31,6 +32,10 @@ class DataflowManifestTests(unittest.TestCase):
             CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
             manifest.capability_factories,
         )
+        self.assertIn(
+            CAPABILITY_DATAFLOW_DATASET_OUTPUT,
+            manifest.capability_factories,
+        )
         self.assertEqual(
             {
                 "dataflow.pipeline_catalog",