Add durable reconciliation decisions

2026-08-06 12:42:20 +02:00
parent 3dae80b6b8
commit b14c693bde
14 changed files with 2037 additions and 19 deletions
@@ -17,6 +17,7 @@ from sqlalchemy import (
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship

+from govoplan_core.core.concurrency import strong_resource_etag
 from govoplan_core.db.base import Base, TimestampMixin


@@ -121,6 +122,11 @@ class DataflowPipeline(Base, TimestampMixin):
        cascade="all, delete-orphan",
        order_by="DataflowTrigger.created_at",
    )
+    decision_sets: Mapped[list["DataflowReconciliationDecisionSet"]] = relationship(
+        back_populates="pipeline",
+        cascade="all, delete-orphan",
+        order_by="DataflowReconciliationDecisionSet.name",
+    )


 class DataflowPipelineRevision(Base, TimestampMixin):
@@ -153,6 +159,99 @@ class DataflowPipelineRevision(Base, TimestampMixin):
    pipeline: Mapped[DataflowPipeline] = relationship(back_populates="revisions")


+class DataflowReconciliationDecisionSet(Base, TimestampMixin):
+    __tablename__ = "dataflow_reconciliation_decision_sets"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "pipeline_id",
+            "name",
+            name="uq_dataflow_decision_sets_pipeline_name",
+        ),
+        Index(
+            "ix_dataflow_decision_sets_tenant_pipeline",
+            "tenant_id",
+            "pipeline_id",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    pipeline_id: Mapped[str] = mapped_column(
+        ForeignKey("dataflow_pipelines.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    name: Mapped[str] = mapped_column(String(300), nullable=False)
+    node_id: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    resource_revision: Mapped[int] = mapped_column(
+        Integer,
+        default=1,
+        nullable=False,
+    )
+    created_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
+
+    pipeline: Mapped[DataflowPipeline] = relationship(back_populates="decision_sets")
+    decisions: Mapped[list["DataflowReconciliationDecision"]] = relationship(
+        back_populates="decision_set",
+        cascade="all, delete-orphan",
+        order_by="DataflowReconciliationDecision.revision",
+    )
+
+    @property
+    def strong_etag(self) -> str:
+        return strong_resource_etag(
+            "dataflow_reconciliation_decision_set",
+            self.id,
+            self.resource_revision,
+        )
+
+
+class DataflowReconciliationDecision(Base, TimestampMixin):
+    __tablename__ = "dataflow_reconciliation_decisions"
+    __table_args__ = (
+        UniqueConstraint(
+            "decision_set_id",
+            "revision",
+            name="uq_dataflow_reconciliation_decision_revision",
+        ),
+        Index(
+            "ix_dataflow_reconciliation_decisions_current",
+            "tenant_id",
+            "decision_set_id",
+            "key_hash",
+            "revision",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    decision_set_id: Mapped[str] = mapped_column(
+        ForeignKey(
+            "dataflow_reconciliation_decision_sets.id",
+            ondelete="CASCADE",
+        ),
+        nullable=False,
+        index=True,
+    )
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    key_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    input_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    action: Mapped[str] = mapped_column(String(20), nullable=False)
+    reason: Mapped[str] = mapped_column(Text, nullable=False)
+    correction: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    actor_ref: Mapped[str] = mapped_column(String(255), nullable=False)
+    decided_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+    )
+
+    decision_set: Mapped[DataflowReconciliationDecisionSet] = relationship(
+        back_populates="decisions"
+    )
+
+
 class DataflowRun(Base, TimestampMixin):
    __tablename__ = "dataflow_runs"
    __table_args__ = (
@@ -200,7 +200,12 @@ DOCUMENTATION = (
            "Every graph node declares typed inputs, configuration, output schema, and validation rules. "
            "Source nodes pin inline content or governed Datasource references; combine, filter, transform, "
            "quality, reconciliation, reusable-subflow, and output nodes remain explicit in the canonical graph. "
-            "Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes. A separate decision-table input can annotate exact matches, invalidate changed inputs, and report orphaned decisions without silently rewriting business data. "
+            "Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes. The "
+            "review dialog records accept, reject, correct, or defer decisions in tenant-owned immutable decision "
+            "sets. Their current projection is a fingerprinted Dataflow source; every superseded revision retains "
+            "the actor, reason, exact input hash, time, and optional correction. A reconcile.decisions node can "
+            "annotate exact matches, invalidate changed inputs, and report orphaned decisions without silently "
+            "rewriting business data. "
            "Expressions use the typed Dataflow expression language and never execute arbitrary host or database "
            "code. Selecting a node may request a bounded intermediate preview; preview rows are transient, "
            "privacy-filtered for the actor, and are not retained as run output. SQL editing compiles into the same "
@@ -218,6 +223,7 @@ DOCUMENTATION = (
                "dataflow.field.expression",
                "dataflow.field.schema",
                "dataflow.action.preview-node",
+                "dataflow.action.review-decisions",
            ],
        },
    ),
@@ -232,6 +238,8 @@ DOCUMENTATION = (
            "the revision and authorization grant, then re-evaluate authority for every delivery. 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; "
+            "it never mutates the reviewed business row. "
            "Deletion prevents future use while retained run, deployment, lineage, audit, and recovery evidence "
            "continues under its retention policy."
        ),
@@ -247,12 +255,14 @@ DOCUMENTATION = (
                "dataflow.action.save",
                "dataflow.action.derive",
                "dataflow.action.trigger",
+                "dataflow.action.record-decision",
                "dataflow.action.delete",
            ],
            "consequence_classes": {
                "save_revision": "Appends an immutable pipeline definition revision.",
                "derive_copy": "Creates a separately governed copy pinned to the source revision and hash.",
                "configure_trigger": "Creates or changes an automation command with revision and authorization evidence.",
+                "record_decision": "Appends an actor-attributed immutable decision revision against an exact input hash.",
                "delete_pipeline": "Prevents future use while retained evidence remains governed.",
            },
        },
@@ -362,6 +372,22 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
            )
            .count()
        ),
+        "dataflow_reconciliation_decision_sets": (
+            session.query(dataflow_models.DataflowReconciliationDecisionSet)
+            .filter(
+                dataflow_models.DataflowReconciliationDecisionSet.tenant_id
+                == tenant_id
+            )
+            .count()
+        ),
+        "dataflow_reconciliation_decisions": (
+            session.query(dataflow_models.DataflowReconciliationDecision)
+            .filter(
+                dataflow_models.DataflowReconciliationDecision.tenant_id
+                == tenant_id
+            )
+            .count()
+        ),
    }


@@ -532,6 +558,14 @@ manifest = ModuleManifest(
                parent_id="dataflow.page",
                order=50,
            ),
+            ViewSurface(
+                id="dataflow.decisions",
+                module_id=MODULE_ID,
+                kind="action",
+                label="Reconciliation decisions",
+                parent_id="dataflow.results",
+                order=55,
+            ),
            ViewSurface(
                id="dataflow.triggers",
                module_id=MODULE_ID,
@@ -580,6 +614,8 @@ manifest = ModuleManifest(
        script_location=str(Path(__file__).with_name("migrations") / "versions"),
        retirement_supported=True,
        retirement_provider=drop_table_retirement_provider(
+            dataflow_models.DataflowReconciliationDecision,
+            dataflow_models.DataflowReconciliationDecisionSet,
            dataflow_models.DataflowTriggerDelivery,
            dataflow_models.DataflowTrigger,
            dataflow_models.DataflowRun,
@@ -597,6 +633,8 @@ manifest = ModuleManifest(
        persistent_table_uninstall_guard(
            dataflow_models.DataflowPipeline,
            dataflow_models.DataflowPipelineRevision,
+            dataflow_models.DataflowReconciliationDecisionSet,
+            dataflow_models.DataflowReconciliationDecision,
            dataflow_models.DataflowPipelineDeployment,
            dataflow_models.DataflowRun,
            dataflow_models.DataflowTrigger,
@@ -614,7 +652,13 @@ manifest = ModuleManifest(
        known_limits=(
            "Execution adapters do not yet cover every declared node family.",
        ),
-        owned_concepts=("dataflow definition", "dataflow revision", "dataflow run", "transformation graph"),
+        owned_concepts=(
+            "dataflow definition",
+            "dataflow revision",
+            "dataflow run",
+            "reconciliation decision set",
+            "transformation graph",
+        ),
        non_owned_concepts=("datasource binding", "connector transport", "report presentation", "workflow task"),
        recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"),
        security_docs=("README.md",),
@@ -0,0 +1,118 @@
+"""Add durable Dataflow reconciliation decision sets.
+
+Revision ID: a3d7f1c5e9b2
+Revises: f6c2a9d4e7b1
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "a3d7f1c5e9b2"
+down_revision = "f6c2a9d4e7b1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "dataflow_reconciliation_decision_sets",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("pipeline_id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=300), nullable=False),
+        sa.Column("node_id", sa.String(length=100), nullable=True),
+        sa.Column("resource_revision", sa.Integer(), nullable=False),
+        sa.Column("created_by", sa.String(length=255), nullable=True),
+        sa.Column("updated_by", sa.String(length=255), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["pipeline_id"],
+            ["dataflow_pipelines.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "pipeline_id",
+            "name",
+            name="uq_dataflow_decision_sets_pipeline_name",
+        ),
+    )
+    op.create_index(
+        "ix_dataflow_decision_sets_tenant_id",
+        "dataflow_reconciliation_decision_sets",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_dataflow_decision_sets_pipeline_id",
+        "dataflow_reconciliation_decision_sets",
+        ["pipeline_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_dataflow_decision_sets_tenant_pipeline",
+        "dataflow_reconciliation_decision_sets",
+        ["tenant_id", "pipeline_id"],
+        unique=False,
+    )
+    op.create_table(
+        "dataflow_reconciliation_decisions",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("decision_set_id", sa.String(length=36), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("key_hash", sa.String(length=64), nullable=False),
+        sa.Column("input_hash", sa.String(length=64), nullable=False),
+        sa.Column("action", sa.String(length=20), nullable=False),
+        sa.Column("reason", sa.Text(), nullable=False),
+        sa.Column("correction", sa.JSON(), nullable=True),
+        sa.Column("actor_ref", sa.String(length=255), nullable=False),
+        sa.Column("decided_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["decision_set_id"],
+            ["dataflow_reconciliation_decision_sets.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "decision_set_id",
+            "revision",
+            name="uq_dataflow_reconciliation_decision_revision",
+        ),
+    )
+    op.create_index(
+        "ix_dataflow_reconciliation_decisions_tenant_id",
+        "dataflow_reconciliation_decisions",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_dataflow_reconciliation_decisions_decision_set_id",
+        "dataflow_reconciliation_decisions",
+        ["decision_set_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_dataflow_reconciliation_decisions_key_hash",
+        "dataflow_reconciliation_decisions",
+        ["key_hash"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_dataflow_reconciliation_decisions_current",
+        "dataflow_reconciliation_decisions",
+        ["tenant_id", "decision_set_id", "key_hash", "revision"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("dataflow_reconciliation_decisions")
+    op.drop_table("dataflow_reconciliation_decision_sets")
@@ -0,0 +1,380 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections.abc import Sequence
+
+from sqlalchemy import and_, func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, joinedload, selectinload
+
+from govoplan_core.core.concurrency import claim_revision
+from govoplan_core.security.time import utc_now
+from govoplan_dataflow.backend.db.models import (
+    DataflowPipeline,
+    DataflowReconciliationDecision,
+    DataflowReconciliationDecisionSet,
+)
+
+
+DECISION_SET_REF_PREFIX = "dataflow-decision-set:"
+DECISION_REF_PREFIX = "dataflow-decision:"
+DECISION_ACTIONS = frozenset({"accept", "reject", "correct", "defer"})
+
+
+class ReconciliationDecisionError(ValueError):
+    pass
+
+
+class ReconciliationDecisionNotFoundError(ReconciliationDecisionError):
+    pass
+
+
+class ReconciliationDecisionConflictError(ReconciliationDecisionError):
+    pass
+
+
+def list_decision_sets(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline_id: str | None = None,
+    include_decisions: bool = True,
+) -> tuple[DataflowReconciliationDecisionSet, ...]:
+    options = [joinedload(DataflowReconciliationDecisionSet.pipeline)]
+    if include_decisions:
+        options.append(selectinload(DataflowReconciliationDecisionSet.decisions))
+    statement = (
+        select(DataflowReconciliationDecisionSet)
+        .options(*options)
+        .where(DataflowReconciliationDecisionSet.tenant_id == tenant_id)
+        .order_by(
+            DataflowReconciliationDecisionSet.name,
+            DataflowReconciliationDecisionSet.id,
+        )
+    )
+    if pipeline_id:
+        statement = statement.where(
+            DataflowReconciliationDecisionSet.pipeline_id == pipeline_id
+        )
+    return tuple(session.scalars(statement).all())
+
+
+def create_decision_set(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline_id: str,
+    name: str,
+    node_id: str | None,
+    actor_ref: str,
+) -> DataflowReconciliationDecisionSet:
+    pipeline = session.get(DataflowPipeline, pipeline_id)
+    if (
+        pipeline is None
+        or pipeline.tenant_id not in {None, tenant_id}
+        or pipeline.deleted_at is not None
+    ):
+        raise ReconciliationDecisionNotFoundError("Dataflow pipeline not found.")
+    item = DataflowReconciliationDecisionSet(
+        tenant_id=tenant_id,
+        pipeline_id=pipeline_id,
+        name=name.strip(),
+        node_id=node_id,
+        resource_revision=1,
+        created_by=actor_ref,
+        updated_by=actor_ref,
+    )
+    session.add(item)
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        raise ReconciliationDecisionConflictError(
+            "A decision set with this name already exists for the pipeline."
+        ) from exc
+    return item
+
+
+def get_decision_set(
+    session: Session,
+    *,
+    tenant_id: str,
+    decision_set_id: str,
+    include_decisions: bool = True,
+) -> DataflowReconciliationDecisionSet:
+    options = [joinedload(DataflowReconciliationDecisionSet.pipeline)]
+    if include_decisions:
+        options.append(selectinload(DataflowReconciliationDecisionSet.decisions))
+    item = session.scalar(
+        select(DataflowReconciliationDecisionSet)
+        .options(*options)
+        .where(
+            DataflowReconciliationDecisionSet.id == decision_set_id,
+            DataflowReconciliationDecisionSet.tenant_id == tenant_id,
+        )
+    )
+    if item is None:
+        raise ReconciliationDecisionNotFoundError(
+            "Reconciliation decision set not found."
+        )
+    return item
+
+
+def record_decision(
+    session: Session,
+    *,
+    tenant_id: str,
+    decision_set_id: str,
+    expected_revision: int,
+    key_hash: str,
+    input_hash: str,
+    action: str,
+    reason: str,
+    correction: dict[str, object] | None,
+    actor_ref: str,
+) -> DataflowReconciliationDecisionSet:
+    item = get_decision_set(
+        session,
+        tenant_id=tenant_id,
+        decision_set_id=decision_set_id,
+    )
+    if action not in DECISION_ACTIONS:
+        raise ReconciliationDecisionError("Unsupported reconciliation action.")
+    next_revision = claim_revision(
+        session,
+        model=DataflowReconciliationDecisionSet,
+        filters=(
+            DataflowReconciliationDecisionSet.id == decision_set_id,
+            DataflowReconciliationDecisionSet.tenant_id == tenant_id,
+        ),
+        revision_attribute="resource_revision",
+        expected_revision=expected_revision,
+        resource_type="dataflow_reconciliation_decision_set",
+        resource_id=decision_set_id,
+    )
+    item.resource_revision = next_revision
+    item.updated_by = actor_ref
+    item.decisions.append(
+        DataflowReconciliationDecision(
+            tenant_id=tenant_id,
+            revision=next_revision,
+            key_hash=key_hash,
+            input_hash=input_hash,
+            action=action,
+            reason=reason.strip(),
+            correction=dict(correction) if correction else None,
+            actor_ref=actor_ref,
+            decided_at=utc_now(),
+        )
+    )
+    session.flush()
+    return item
+
+
+def current_decisions(
+    item: DataflowReconciliationDecisionSet,
+) -> tuple[DataflowReconciliationDecision, ...]:
+    current: dict[str, DataflowReconciliationDecision] = {}
+    for decision in sorted(item.decisions, key=lambda value: value.revision):
+        current[decision.key_hash] = decision
+    return tuple(
+        sorted(
+            current.values(),
+            key=lambda value: (value.key_hash, value.revision),
+        )
+    )
+
+
+def decision_rows(
+    item: DataflowReconciliationDecisionSet,
+) -> tuple[dict[str, object], ...]:
+    return tuple(_decision_row(value) for value in current_decisions(item))
+
+
+def decision_set_fingerprint(
+    item: DataflowReconciliationDecisionSet,
+) -> str:
+    payload = {
+        "decision_set_id": item.id,
+        "resource_revision": item.resource_revision,
+    }
+    return "sha256:" + hashlib.sha256(
+        json.dumps(
+            payload,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+        ).encode("utf-8")
+    ).hexdigest()
+
+
+def decision_set_payload(
+    item: DataflowReconciliationDecisionSet,
+    *,
+    include_decisions: bool = True,
+) -> dict[str, object]:
+    current = current_decisions(item) if include_decisions else ()
+    history = (
+        sorted(item.decisions, key=lambda value: value.revision)
+        if include_decisions
+        else ()
+    )
+    return {
+        "ref": f"{DECISION_SET_REF_PREFIX}{item.id}",
+        "id": item.id,
+        "pipeline_id": item.pipeline_id,
+        "name": item.name,
+        "node_id": item.node_id,
+        "resource_revision": item.resource_revision,
+        "etag": item.strong_etag,
+        "fingerprint": decision_set_fingerprint(item),
+        "decisions_included": include_decisions,
+        "current_decisions": [_decision_payload(value) for value in current],
+        "history": [
+            _decision_payload(value)
+            for value in history
+        ],
+        "created_by": item.created_by,
+        "updated_by": item.updated_by,
+        "created_at": item.created_at,
+        "updated_at": item.updated_at,
+    }
+
+
+def decision_set_source_payload(
+    item: DataflowReconciliationDecisionSet,
+) -> dict[str, object]:
+    return {
+        "ref": f"{DECISION_SET_REF_PREFIX}{item.id}",
+        "provider": "dataflow.reconciliation_decisions",
+        "source_name": _source_name(item.name, item.id),
+        "name": item.name,
+        "description": "Current immutable reconciliation decisions; prior revisions remain in decision history.",
+        "mode": "static",
+        "columns": [
+            {"name": "key_hash", "data_type": "string", "nullable": False},
+            {"name": "input_hash", "data_type": "string", "nullable": False},
+            {"name": "decision_ref", "data_type": "string", "nullable": False},
+            {"name": "action", "data_type": "string", "nullable": False},
+            {"name": "actor_ref", "data_type": "string", "nullable": False},
+            {"name": "decided_at", "data_type": "datetime", "nullable": False},
+            {"name": "reason", "data_type": "string", "nullable": False},
+            {"name": "correction", "data_type": "object", "nullable": True},
+        ],
+        "schema_version": "1",
+        "fingerprint": decision_set_fingerprint(item),
+        "row_count": None,
+        "byte_count": None,
+        "updated_at": item.updated_at,
+        "capabilities": ["preview", "read", "immutable_history"],
+    }
+
+
+def current_decision_rows(
+    session: Session,
+    *,
+    decision_set_id: str,
+    limit: int,
+) -> tuple[tuple[dict[str, object], ...], int]:
+    bounded_limit = max(1, limit)
+    latest = (
+        select(
+            DataflowReconciliationDecision.key_hash.label("key_hash"),
+            func.max(DataflowReconciliationDecision.revision).label("revision"),
+        )
+        .where(
+            DataflowReconciliationDecision.decision_set_id == decision_set_id
+        )
+        .group_by(DataflowReconciliationDecision.key_hash)
+        .subquery()
+    )
+    statement = (
+        select(DataflowReconciliationDecision)
+        .join(
+            latest,
+            and_(
+                DataflowReconciliationDecision.key_hash == latest.c.key_hash,
+                DataflowReconciliationDecision.revision == latest.c.revision,
+            ),
+        )
+        .where(
+            DataflowReconciliationDecision.decision_set_id == decision_set_id
+        )
+        .order_by(DataflowReconciliationDecision.key_hash)
+        .limit(bounded_limit)
+    )
+    records = tuple(session.scalars(statement))
+    total = int(
+        session.scalar(
+            select(func.count(func.distinct(DataflowReconciliationDecision.key_hash))).where(
+                DataflowReconciliationDecision.decision_set_id == decision_set_id
+            )
+        )
+        or 0
+    )
+    return tuple(_decision_row(value) for value in records), total
+
+
+def decision_set_id_from_ref(value: str) -> str | None:
+    cleaned = str(value or "").strip()
+    if not cleaned.startswith(DECISION_SET_REF_PREFIX):
+        return None
+    identifier = cleaned[len(DECISION_SET_REF_PREFIX) :]
+    return identifier or None
+
+
+def _decision_row(
+    value: DataflowReconciliationDecision,
+) -> dict[str, object]:
+    return {
+        "key_hash": value.key_hash,
+        "input_hash": value.input_hash,
+        "decision_ref": f"{DECISION_REF_PREFIX}{value.id}",
+        "action": value.action,
+        "actor_ref": value.actor_ref,
+        "decided_at": value.decided_at.isoformat(),
+        "reason": value.reason,
+        "correction": dict(value.correction) if value.correction else None,
+    }
+
+
+def _decision_payload(
+    value: DataflowReconciliationDecision,
+) -> dict[str, object]:
+    return {
+        "ref": f"{DECISION_REF_PREFIX}{value.id}",
+        "id": value.id,
+        "revision": value.revision,
+        "key_hash": value.key_hash,
+        "input_hash": value.input_hash,
+        "action": value.action,
+        "reason": value.reason,
+        "correction": dict(value.correction) if value.correction else None,
+        "actor_ref": value.actor_ref,
+        "decided_at": value.decided_at,
+    }
+
+
+def _source_name(name: str, identifier: str) -> str:
+    slug = re.sub(r"[^a-z0-9]+", "_", name.casefold()).strip("_")
+    return (f"decisions_{slug}" if slug else f"decisions_{identifier[:8]}")[:120]
+
+
+__all__: Sequence[str] = (
+    "DECISION_SET_REF_PREFIX",
+    "ReconciliationDecisionConflictError",
+    "ReconciliationDecisionError",
+    "ReconciliationDecisionNotFoundError",
+    "create_decision_set",
+    "current_decision_rows",
+    "current_decisions",
+    "decision_rows",
+    "decision_set_fingerprint",
+    "decision_set_id_from_ref",
+    "decision_set_payload",
+    "decision_set_source_payload",
+    "get_decision_set",
+    "list_decision_sets",
+    "record_decision",
+)
@@ -10,6 +10,7 @@ from govoplan_core.api.v1.schemas import (
 from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
 from govoplan_core.core.automation import AutomationInvocation
+from govoplan_core.core.concurrency import RevisionConflictError
 from govoplan_core.core.dataflows import (
    DataflowPublicationTarget,
    DataflowRunRequest,
@@ -71,6 +72,10 @@ from govoplan_dataflow.backend.schemas import (
    PipelineSqlResponse,
    PipelineUpdateRequest,
    PipelineValidationResponse,
+    ReconciliationDecisionSetCreateRequest,
+    ReconciliationDecisionSetListResponse,
+    ReconciliationDecisionSetResponse,
+    ReconciliationDecisionWriteRequest,
    TabularSnapshotCreateRequest,
    TabularSourceColumnResponse,
    TabularSourceListResponse,
@@ -110,6 +115,17 @@ from govoplan_dataflow.backend.service import (
    validate_draft,
 )
 from govoplan_dataflow.backend.run_worker import run_metrics
+from govoplan_dataflow.backend.reconciliation_decisions import (
+    ReconciliationDecisionConflictError,
+    ReconciliationDecisionError,
+    ReconciliationDecisionNotFoundError,
+    create_decision_set,
+    decision_set_payload,
+    decision_set_source_payload,
+    get_decision_set,
+    list_decision_sets,
+    record_decision,
+)
 from govoplan_dataflow.backend.recovery import dataflow_run_recovery_states
 from govoplan_dataflow.backend.triggers import (
    create_trigger,
@@ -145,6 +161,18 @@ def _actor_id(principal: ApiPrincipal) -> str | None:
    )


+def _decision_actor_ref(principal: ApiPrincipal) -> str:
+    actor_id = _actor_id(principal)
+    if not actor_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="A stable actor reference is required to record a decision.",
+        )
+    principal_ref = principal.to_platform_principal()
+    prefix = "service-account" if principal_ref.service_account_id else "account"
+    return f"{prefix}:{actor_id}"
+
+
 def _http_error(exc: DataflowError) -> HTTPException:
    if isinstance(exc, DataflowNotFoundError):
        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
@@ -161,6 +189,17 @@ def _http_error(exc: DataflowError) -> HTTPException:
    return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))


+def _decision_http_error(exc: ReconciliationDecisionError) -> HTTPException:
+    if isinstance(exc, ReconciliationDecisionNotFoundError):
+        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+    if isinstance(exc, ReconciliationDecisionConflictError):
+        return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
+    return HTTPException(
+        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+        detail=str(exc),
+    )
+
+
 def _governance_http_error(exc: PermissionError | ValueError) -> HTTPException:
    return HTTPException(
        status_code=(
@@ -329,21 +368,42 @@ def api_list_sources(
    registry = get_registry()
    provider = datasource_catalogue(registry)
    writer = datasource_lifecycle(registry)
-    if provider is None:
-        return TabularSourceListResponse(available=False, writable=False, sources=[])
-    try:
-        sources = provider.list_datasources(
-            session,
-            principal,
-            query=query,
-            limit=100,
-        )
-    except DatasourceError as exc:
-        raise _source_http_error(exc) from exc
+    sources = ()
+    if provider is not None:
+        try:
+            sources = provider.list_datasources(
+                session,
+                principal,
+                query=query,
+                limit=100,
+            )
+        except DatasourceError as exc:
+            raise _source_http_error(exc) from exc
+    decision_sources = []
+    for item in list_decision_sets(
+        session,
+        tenant_id=principal.tenant_id,
+        include_decisions=False,
+    ):
+        if query and query.casefold() not in item.name.casefold():
+            continue
+        try:
+            require_definition_action(
+                item.pipeline,
+                principal=principal,
+                registry=registry,
+                action="view",
+            )
+        except PermissionError:
+            continue
+        decision_sources.append(decision_set_source_payload(item))
    return TabularSourceListResponse(
-        available=True,
+        available=provider is not None or bool(decision_sources),
        writable=writer is not None,
-        sources=[_source_response(source) for source in sources],
+        sources=[
+            *(_source_response(source) for source in sources),
+            *(TabularSourceResponse.model_validate(item) for item in decision_sources),
+        ],
    )


@@ -504,6 +564,212 @@ def api_create_pipeline(
    return response


+@router.get(
+    "/pipelines/{pipeline_id}/decision-sets",
+    response_model=ReconciliationDecisionSetListResponse,
+)
+def api_list_reconciliation_decision_sets(
+    pipeline_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> ReconciliationDecisionSetListResponse:
+    _require_any_scope(principal, READ_SCOPE, WRITE_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        pipeline = get_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+        )
+        require_definition_action(
+            pipeline,
+            principal=principal,
+            registry=get_registry(),
+            action="view",
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    return ReconciliationDecisionSetListResponse(
+        decision_sets=[
+            ReconciliationDecisionSetResponse.model_validate(
+                decision_set_payload(item, include_decisions=False)
+            )
+            for item in list_decision_sets(
+                session,
+                tenant_id=principal.tenant_id,
+                pipeline_id=pipeline_id,
+                include_decisions=False,
+            )
+        ]
+    )
+
+
+@router.post(
+    "/pipelines/{pipeline_id}/decision-sets",
+    response_model=ReconciliationDecisionSetResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_reconciliation_decision_set(
+    pipeline_id: str,
+    payload: ReconciliationDecisionSetCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> ReconciliationDecisionSetResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        pipeline = get_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+        )
+        require_definition_action(
+            pipeline,
+            principal=principal,
+            registry=get_registry(),
+            action="edit",
+        )
+        item = create_decision_set(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+            name=payload.name,
+            node_id=payload.node_id,
+            actor_ref=_decision_actor_ref(principal),
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    except ReconciliationDecisionError as exc:
+        session.rollback()
+        raise _decision_http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="dataflow.reconciliation_decision_set.created",
+        object_type="dataflow_reconciliation_decision_set",
+        object_id=item.id,
+        details={"pipeline_id": pipeline_id, "node_id": payload.node_id},
+    )
+    response = ReconciliationDecisionSetResponse.model_validate(
+        decision_set_payload(item)
+    )
+    session.commit()
+    return response
+
+
+@router.get(
+    "/decision-sets/{decision_set_id}",
+    response_model=ReconciliationDecisionSetResponse,
+)
+def api_get_reconciliation_decision_set(
+    decision_set_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> ReconciliationDecisionSetResponse:
+    _require_any_scope(principal, READ_SCOPE, WRITE_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        item = get_decision_set(
+            session,
+            tenant_id=principal.tenant_id,
+            decision_set_id=decision_set_id,
+        )
+        require_definition_action(
+            item.pipeline,
+            principal=principal,
+            registry=get_registry(),
+            action="view",
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except ReconciliationDecisionError as exc:
+        raise _decision_http_error(exc) from exc
+    return ReconciliationDecisionSetResponse.model_validate(
+        decision_set_payload(item)
+    )
+
+
+@router.post(
+    "/decision-sets/{decision_set_id}/decisions",
+    response_model=ReconciliationDecisionSetResponse,
+)
+def api_record_reconciliation_decision(
+    decision_set_id: str,
+    payload: ReconciliationDecisionWriteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> ReconciliationDecisionSetResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        item = get_decision_set(
+            session,
+            tenant_id=principal.tenant_id,
+            decision_set_id=decision_set_id,
+        )
+        pipeline = get_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=item.pipeline_id,
+        )
+        require_definition_action(
+            pipeline,
+            principal=principal,
+            registry=get_registry(),
+            action="edit",
+        )
+        item = record_decision(
+            session,
+            tenant_id=principal.tenant_id,
+            decision_set_id=decision_set_id,
+            expected_revision=payload.base_revision,
+            key_hash=payload.key_hash,
+            input_hash=payload.input_hash,
+            action=payload.action,
+            reason=payload.reason,
+            correction=payload.correction,
+            actor_ref=_decision_actor_ref(principal),
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    except RevisionConflictError as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=exc.as_dict(),
+        ) from exc
+    except ReconciliationDecisionError as exc:
+        session.rollback()
+        raise _decision_http_error(exc) from exc
+    decision = item.decisions[-1]
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="dataflow.reconciliation_decision.recorded",
+        object_type="dataflow_reconciliation_decision",
+        object_id=decision.id,
+        details={
+            "decision_set_id": item.id,
+            "pipeline_id": item.pipeline_id,
+            "revision": decision.revision,
+            "key_hash": decision.key_hash,
+            "input_hash": decision.input_hash,
+            "decision_action": decision.action,
+        },
+    )
+    response = ReconciliationDecisionSetResponse.model_validate(
+        decision_set_payload(item)
+    )
+    session.commit()
+    return response
+
+
@router.get("/pipelines/{pipeline_id}", response_model=PipelineResponse)
 def api_get_pipeline(
    pipeline_id: str,
@@ -145,6 +145,68 @@ class PipelineListResponse(BaseModel):
    pipelines: list[PipelineResponse]


+class ReconciliationDecisionSetCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=300)
+    node_id: str | None = Field(
+        default=None,
+        min_length=1,
+        max_length=100,
+        pattern=r"^[A-Za-z0-9_.:-]+$",
+    )
+
+
+class ReconciliationDecisionWriteRequest(BaseModel):
+    base_revision: int = Field(ge=1)
+    key_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    action: Literal["accept", "reject", "correct", "defer"]
+    reason: str = Field(min_length=3, max_length=4_000)
+    correction: dict[str, Any] | None = None
+
+    @model_validator(mode="after")
+    def validate_correction(self) -> "ReconciliationDecisionWriteRequest":
+        if self.action == "correct" and not self.correction:
+            raise ValueError("Correct decisions require corrected field values.")
+        if self.action != "correct" and self.correction:
+            raise ValueError("Only correct decisions may include corrected field values.")
+        return self
+
+
+class ReconciliationDecisionResponse(BaseModel):
+    ref: str
+    id: str
+    revision: int
+    key_hash: str
+    input_hash: str
+    action: Literal["accept", "reject", "correct", "defer"]
+    reason: str
+    correction: dict[str, Any] | None
+    actor_ref: str
+    decided_at: datetime
+
+
+class ReconciliationDecisionSetResponse(BaseModel):
+    ref: str
+    id: str
+    pipeline_id: str
+    name: str
+    node_id: str | None
+    resource_revision: int = Field(ge=1)
+    etag: str
+    fingerprint: str
+    decisions_included: bool
+    current_decisions: list[ReconciliationDecisionResponse]
+    history: list[ReconciliationDecisionResponse]
+    created_by: str | None
+    updated_by: str | None
+    created_at: datetime
+    updated_at: datetime
+
+
+class ReconciliationDecisionSetListResponse(BaseModel):
+    decision_sets: list[ReconciliationDecisionSetResponse]
+
+
 class PipelineCreateRequest(BaseModel):
    name: str = Field(min_length=1, max_length=300)
    description: str | None = Field(default=None, max_length=4000)
@@ -86,6 +86,12 @@ from govoplan_dataflow.backend.recovery import (
    begin_dataflow_run_recovery,
    dataflow_run_recovery_state,
 )
+from govoplan_dataflow.backend.reconciliation_decisions import (
+    current_decision_rows,
+    decision_set_fingerprint,
+    decision_set_id_from_ref,
+    get_decision_set,
+)
 from govoplan_dataflow.backend.sql_compiler import (
    SqlCompilationError,
    compile_sql,
@@ -1352,8 +1358,12 @@ def _run_authorization_payload(
    graph = PipelineGraph.model_validate(revision.graph)
    scopes = {RUN_SCOPE}
    if any(
-        node.type == "source.reference"
-        or bool(node.config.get("source_ref"))
+        (
+            node.type == "source.reference"
+            or bool(node.config.get("source_ref"))
+        )
+        and decision_set_id_from_ref(str(node.config.get("source_ref") or ""))
+        is None
        for node in graph.nodes
    ):
        scopes.add(DATASOURCE_READ_SCOPE)
@@ -1880,6 +1890,49 @@ def _datasource_source_resolver(
    provider = datasource_catalogue(registry)

    def resolve_source(node: GraphNode, limit: int) -> ResolvedSource:
+        source_ref = str(node.config.get("source_ref") or "")
+        decision_set_id = decision_set_id_from_ref(source_ref)
+        if decision_set_id is not None:
+            try:
+                decision_set = get_decision_set(
+                    session,
+                    tenant_id=principal.tenant_id,
+                    decision_set_id=decision_set_id,
+                    include_decisions=False,
+                )
+                require_definition_action(
+                    decision_set.pipeline,
+                    principal=principal,
+                    registry=registry,
+                    action="view",
+                )
+            except (PermissionError, ValueError) as exc:
+                raise PipelineExecutionError(
+                    str(exc),
+                    node_id=node.id,
+                ) from exc
+            rows, total_rows = current_decision_rows(
+                session,
+                decision_set_id=decision_set.id,
+                limit=limit,
+            )
+            fingerprint = decision_set_fingerprint(decision_set)
+            expected_fingerprint = _clean_optional(
+                node.config.get("expected_fingerprint")
+            )
+            if expected_fingerprint and expected_fingerprint != fingerprint:
+                raise PipelineExecutionError(
+                    "Reconciliation decisions changed; refresh the decision source before running it.",
+                    node_id=node.id,
+                )
+            return ResolvedSource(
+                rows=rows,
+                source_ref=source_ref,
+                provider="dataflow.reconciliation_decisions",
+                fingerprint=fingerprint,
+                total_rows=total_rows,
+                truncated=total_rows > len(rows),
+            )
        if provider is None:
            raise PipelineExecutionError(
                "Datasource-backed execution requires the Datasources "