From b14c693bdedb609a1fcb7612b485994d29d140b7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 6 Aug 2026 12:42:20 +0200 Subject: [PATCH] Add durable reconciliation decisions --- README.md | 10 + src/govoplan_dataflow/backend/db/models.py | 99 ++++ src/govoplan_dataflow/backend/manifest.py | 48 +- ...f1c5e9b2_v0119_reconciliation_decisions.py | 118 +++++ .../backend/reconciliation_decisions.py | 380 +++++++++++++++ src/govoplan_dataflow/backend/router.py | 292 +++++++++++- src/govoplan_dataflow/backend/schemas.py | 62 +++ src/govoplan_dataflow/backend/service.py | 57 ++- .../test_interface_documentation_contract.py | 4 + tests/test_migrations.py | 4 +- tests/test_reconciliation_decisions.py | 247 ++++++++++ webui/src/api/dataflow.ts | 85 ++++ webui/src/features/dataflow/DataflowPage.tsx | 446 +++++++++++++++++- webui/src/styles/dataflow.css | 204 ++++++++ 14 files changed, 2037 insertions(+), 19 deletions(-) create mode 100644 src/govoplan_dataflow/backend/migrations/versions/a3d7f1c5e9b2_v0119_reconciliation_decisions.py create mode 100644 src/govoplan_dataflow/backend/reconciliation_decisions.py create mode 100644 tests/test_reconciliation_decisions.py diff --git a/README.md b/README.md index a42a626..2bef201 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,16 @@ logical row disappeared. It never silently applies a correction to business data; a downstream governed transform or Workflow handoff must interpret the recorded action. +For saved reconciliation pipelines, the preview results provide a review +dialog for those rows. Reviewers create a tenant-owned decision set and append +accept, reject, correct, or defer decisions with a mandatory reason. Writes use +optimistic concurrency; updating a decision creates another immutable revision +rather than replacing history. The current projection appears in the ordinary +Dataflow source catalogue as `dataflow-decision-set:` and carries a content +fingerprint. A changed input hash is therefore shown as stale and cannot be +silently reused. Corrections remain annotations until an explicit downstream +transform applies them. + Reporting consumers may either evaluate a pinned pipeline revision or pin one successful published run. An exact run pin is immutable: it cannot be supplied new parameters, and Dataflow reads only the recorded Datasource materialization diff --git a/src/govoplan_dataflow/backend/db/models.py b/src/govoplan_dataflow/backend/db/models.py index 75a41e9..4716bf7 100644 --- a/src/govoplan_dataflow/backend/db/models.py +++ b/src/govoplan_dataflow/backend/db/models.py @@ -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__ = ( diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index ec66ee0..36621f1 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -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",), diff --git a/src/govoplan_dataflow/backend/migrations/versions/a3d7f1c5e9b2_v0119_reconciliation_decisions.py b/src/govoplan_dataflow/backend/migrations/versions/a3d7f1c5e9b2_v0119_reconciliation_decisions.py new file mode 100644 index 0000000..61ecc4b --- /dev/null +++ b/src/govoplan_dataflow/backend/migrations/versions/a3d7f1c5e9b2_v0119_reconciliation_decisions.py @@ -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") diff --git a/src/govoplan_dataflow/backend/reconciliation_decisions.py b/src/govoplan_dataflow/backend/reconciliation_decisions.py new file mode 100644 index 0000000..f6ef1dd --- /dev/null +++ b/src/govoplan_dataflow/backend/reconciliation_decisions.py @@ -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", +) diff --git a/src/govoplan_dataflow/backend/router.py b/src/govoplan_dataflow/backend/router.py index 275389d..2cf5a96 100644 --- a/src/govoplan_dataflow/backend/router.py +++ b/src/govoplan_dataflow/backend/router.py @@ -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, diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index 0a240fe..e24b331 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -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) diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index 5b18db9..b54b716 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -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 " diff --git a/tests/test_interface_documentation_contract.py b/tests/test_interface_documentation_contract.py index b429e3b..cc5bbc1 100644 --- a/tests/test_interface_documentation_contract.py +++ b/tests/test_interface_documentation_contract.py @@ -22,6 +22,7 @@ class DataflowInterfaceDocumentationContractTests(unittest.TestCase): "dataflow.sql", "dataflow.inspector", "dataflow.results", + "dataflow.decisions", "dataflow.triggers", "dataflow.runs", "dataflow.widget.pipelines", @@ -38,6 +39,7 @@ class DataflowInterfaceDocumentationContractTests(unittest.TestCase): "dataflow.runs", ): self.assertEqual("dataflow.page", surfaces[surface_id].parent_id) + self.assertEqual("dataflow.results", surfaces["dataflow.decisions"].parent_id) def test_help_and_consequence_metadata_remain_published(self) -> None: topics = {topic.id: topic for topic in get_manifest().documentation} @@ -48,8 +50,10 @@ class DataflowInterfaceDocumentationContractTests(unittest.TestCase): self.assertIn("dataflow.state.read-only", boundary.metadata["help_contexts"]) self.assertIn("dataflow.field.expression", nodes.metadata["help_contexts"]) + self.assertIn("dataflow.action.review-decisions", nodes.metadata["help_contexts"]) self.assertIn("save_revision", fields.metadata["consequence_classes"]) self.assertIn("delete_pipeline", fields.metadata["consequence_classes"]) + self.assertIn("record_decision", fields.metadata["consequence_classes"]) self.assertIn("publish_output", execution.metadata["consequence_classes"]) self.assertIn("promote_revision", execution.metadata["consequence_classes"]) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index f4508ec..cbb4ca0 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -24,7 +24,7 @@ class DataflowMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "f6c2a9d4e7b1", + "a3d7f1c5e9b2", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( @@ -32,6 +32,8 @@ class DataflowMigrationTests(unittest.TestCase): "dataflow_pipelines", "dataflow_pipeline_revisions", "dataflow_pipeline_deployments", + "dataflow_reconciliation_decision_sets", + "dataflow_reconciliation_decisions", "dataflow_runs", "dataflow_triggers", "dataflow_trigger_deliveries", diff --git a/tests/test_reconciliation_decisions.py b/tests/test_reconciliation_decisions.py new file mode 100644 index 0000000..3be78b0 --- /dev/null +++ b/tests/test_reconciliation_decisions.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +from types import SimpleNamespace +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.concurrency import RevisionConflictError +from govoplan_core.db.base import Base +from govoplan_dataflow.backend.db.models import ( + DataflowPipeline, + DataflowReconciliationDecision, + DataflowReconciliationDecisionSet, +) +from govoplan_dataflow.backend.executor import PipelineExecutionError +from govoplan_dataflow.backend.reconciliation_decisions import ( + create_decision_set, + current_decision_rows, + current_decisions, + decision_set_fingerprint, + decision_set_payload, + decision_set_source_payload, + record_decision, +) +from govoplan_dataflow.backend.schemas import GraphNode, GraphPosition +from govoplan_dataflow.backend.service import _datasource_source_resolver + + +TABLES = ( + DataflowPipeline.__table__, + DataflowReconciliationDecisionSet.__table__, + DataflowReconciliationDecision.__table__, +) + + +class ReconciliationDecisionTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine, tables=TABLES) + self.session = Session(self.engine) + self.pipeline = DataflowPipeline( + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + name="Monthly reconciliation", + status="active", + created_by="account-1", + updated_by="account-1", + ) + self.session.add(self.pipeline) + self.session.flush() + self.principal = ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + scopes=frozenset( + { + "dataflow:pipeline:read", + "dataflow:pipeline:write", + "dataflow:pipeline:run", + } + ), + ), + account=SimpleNamespace(id="account-1"), + user=SimpleNamespace(id="membership-1"), + ) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_decisions_are_immutable_current_rows_and_occ_protected(self) -> None: + decision_set = create_decision_set( + self.session, + tenant_id="tenant-1", + pipeline_id=self.pipeline.id, + name="July review", + node_id="reconcile", + actor_ref="account:account-1", + ) + first = record_decision( + self.session, + tenant_id="tenant-1", + decision_set_id=decision_set.id, + expected_revision=1, + key_hash="a" * 64, + input_hash="b" * 64, + action="accept", + reason="Checked against the source record.", + correction=None, + actor_ref="account:account-1", + ) + first_fingerprint = decision_set_fingerprint(first) + second = record_decision( + self.session, + tenant_id="tenant-1", + decision_set_id=decision_set.id, + expected_revision=2, + key_hash="a" * 64, + input_hash="c" * 64, + action="correct", + reason="The monthly source changed after review.", + correction={"amount": 25}, + actor_ref="account:account-1", + ) + + self.assertEqual(2, len(second.decisions)) + self.assertEqual(1, len(current_decisions(second))) + self.assertEqual("c" * 64, current_decisions(second)[0].input_hash) + self.assertNotEqual(first_fingerprint, decision_set_fingerprint(second)) + payload = decision_set_payload(second) + self.assertEqual(2, len(payload["history"])) + self.assertEqual(1, len(payload["current_decisions"])) + with self.assertRaises(RevisionConflictError): + record_decision( + self.session, + tenant_id="tenant-1", + decision_set_id=decision_set.id, + expected_revision=2, + key_hash="d" * 64, + input_hash="e" * 64, + action="defer", + reason="Needs another source.", + correction=None, + actor_ref="account:account-1", + ) + + def test_decision_set_is_a_fingerprinted_reference_source(self) -> None: + decision_set = create_decision_set( + self.session, + tenant_id="tenant-1", + pipeline_id=self.pipeline.id, + name="July review", + node_id="reconcile", + actor_ref="account:account-1", + ) + record_decision( + self.session, + tenant_id="tenant-1", + decision_set_id=decision_set.id, + expected_revision=1, + key_hash="a" * 64, + input_hash="b" * 64, + action="reject", + reason="The observed record belongs to another case.", + correction=None, + actor_ref="account:account-1", + ) + source = decision_set_source_payload(decision_set) + resolver = _datasource_source_resolver( + session=self.session, + principal=self.principal, + registry=None, + ) + node = GraphNode( + id="decisions", + type="source.reference", + label="Review decisions", + position=GraphPosition(x=0, y=0), + config={ + "source_ref": source["ref"], + "expected_fingerprint": source["fingerprint"], + }, + ) + + resolved = resolver(node, 100) + + self.assertEqual("dataflow.reconciliation_decisions", resolved.provider) + self.assertEqual(1, resolved.total_rows) + self.assertEqual("reject", resolved.rows[0]["action"]) + stale = node.model_copy(deep=True) + stale.config["expected_fingerprint"] = "sha256:" + "0" * 64 + with self.assertRaisesRegex(PipelineExecutionError, "changed"): + resolver(stale, 100) + + def test_tenant_can_keep_decisions_for_visible_system_pipeline(self) -> None: + system_pipeline = DataflowPipeline( + tenant_id=None, + scope_type="system", + scope_id=None, + name="Governed monthly reconciliation", + status="active", + created_by="system-admin", + updated_by="system-admin", + ) + self.session.add(system_pipeline) + self.session.flush() + + decision_set = create_decision_set( + self.session, + tenant_id="tenant-1", + pipeline_id=system_pipeline.id, + name="Tenant July review", + node_id=None, + actor_ref="account:account-1", + ) + + self.assertEqual("tenant-1", decision_set.tenant_id) + self.assertEqual(system_pipeline.id, decision_set.pipeline_id) + + def test_summary_and_current_projection_do_not_require_full_history(self) -> None: + decision_set = create_decision_set( + self.session, + tenant_id="tenant-1", + pipeline_id=self.pipeline.id, + name="Bounded review", + node_id="reconcile", + actor_ref="account:account-1", + ) + for expected_revision, key_hash, input_hash in ( + (1, "a" * 64, "b" * 64), + (2, "c" * 64, "d" * 64), + (3, "a" * 64, "e" * 64), + ): + record_decision( + self.session, + tenant_id="tenant-1", + decision_set_id=decision_set.id, + expected_revision=expected_revision, + key_hash=key_hash, + input_hash=input_hash, + action="accept", + reason="Reviewed against the current input.", + correction=None, + actor_ref="account:account-1", + ) + + summary = decision_set_payload(decision_set, include_decisions=False) + rows, total = current_decision_rows( + self.session, + decision_set_id=decision_set.id, + limit=1, + ) + + self.assertFalse(summary["decisions_included"]) + self.assertEqual([], summary["current_decisions"]) + self.assertEqual([], summary["history"]) + self.assertEqual(2, total) + self.assertEqual(1, len(rows)) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index b1a83ee..11adba1 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -205,6 +205,7 @@ export type TabularSource = { columns: TabularSourceColumn[]; schema_version: string; fingerprint: string; + decisions_included: boolean; row_count?: number | null; byte_count?: number | null; updated_at?: string | null; @@ -217,6 +218,38 @@ export type TabularSourceCatalogue = { sources: TabularSource[]; }; +export type ReconciliationDecisionAction = "accept" | "reject" | "correct" | "defer"; + +export type ReconciliationDecision = { + ref: string; + id: string; + revision: number; + key_hash: string; + input_hash: string; + action: ReconciliationDecisionAction; + reason: string; + correction?: Record | null; + actor_ref: string; + decided_at: string; +}; + +export type ReconciliationDecisionSet = { + ref: string; + id: string; + pipeline_id: string; + name: string; + node_id?: string | null; + resource_revision: number; + etag: string; + fingerprint: string; + current_decisions: ReconciliationDecision[]; + history: ReconciliationDecision[]; + created_by?: string | null; + updated_by?: string | null; + created_at: string; + updated_at: string; +}; + export type PipelineRun = { ref: string; pipeline_id: string; @@ -363,6 +396,58 @@ export function createDataflowSourceSnapshot( }); } +export async function listDataflowDecisionSets( + settings: ApiSettings, + pipelineId: string +): Promise { + const response = await apiFetch<{ decision_sets: ReconciliationDecisionSet[] }>( + settings, + `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/decision-sets` + ); + return response.decision_sets; +} + +export function createDataflowDecisionSet( + settings: ApiSettings, + pipelineId: string, + payload: { name: string; node_id?: string | null } +): Promise { + return apiFetch( + settings, + `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/decision-sets`, + { method: "POST", body: JSON.stringify(payload) } + ); +} + +export function getDataflowDecisionSet( + settings: ApiSettings, + decisionSetId: string +): Promise { + return apiFetch( + settings, + `/api/v1/dataflow/decision-sets/${encodeURIComponent(decisionSetId)}` + ); +} + +export function recordDataflowDecision( + settings: ApiSettings, + decisionSetId: string, + payload: { + base_revision: number; + key_hash: string; + input_hash: string; + action: ReconciliationDecisionAction; + reason: string; + correction?: Record | null; + } +): Promise { + return apiFetch( + settings, + `/api/v1/dataflow/decision-sets/${encodeURIComponent(decisionSetId)}/decisions`, + { method: "POST", body: JSON.stringify(payload) } + ); +} + export async function listDataflowPipelines(settings: ApiSettings): Promise { const response = await apiFetch<{ pipelines: Pipeline[] }>(settings, "/api/v1/dataflow/pipelines"); return response.pipelines; diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index aac1f1d..611079a 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -13,6 +13,7 @@ import { Code2, CopyPlus, DatabaseZap, + ListChecks, Network, Play, Plus, @@ -51,6 +52,7 @@ import { useLocation } from "react-router"; import { compileDataflowSql, cancelDataflowPipelineRun, + createDataflowDecisionSet, createDataflowTrigger, createDataflowPipeline, createDataflowSourceSnapshot, @@ -58,7 +60,9 @@ import { deleteDataflowTrigger, dataflowScopeReferenceProvider, deriveDataflowPipeline, + getDataflowDecisionSet, listDataflowNodeTypes, + listDataflowDecisionSets, listDataflowPipelineRuns, listDataflowPipelineDeployments, listDataflowPipelines, @@ -66,6 +70,7 @@ import { listDataflowTriggers, previewDataflowPipeline, promoteDataflowPipeline, + recordDataflowDecision, runDataflowPipeline, renderDataflowSql, updateDataflowPipeline, @@ -84,6 +89,8 @@ import { type PipelinePreview, type PipelineDeployment, type PipelineRun, + type ReconciliationDecisionAction, + type ReconciliationDecisionSet, type TabularSource } from "../../api/dataflow"; import DataflowCanvas, { updateGraphNode } from "./DataflowCanvas"; @@ -141,6 +148,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings const [definitionSettingsOpen, setDefinitionSettingsOpen] = useState(false); const [deriveOpen, setDeriveOpen] = useState(false); const [triggersOpen, setTriggersOpen] = useState(false); + const [decisionReviewOpen, setDecisionReviewOpen] = useState(false); const [nodeLibrary, setNodeLibrary] = useState(FALLBACK_NODE_LIBRARY); const [sources, setSources] = useState([]); const [sourceCatalogueAvailable, setSourceCatalogueAvailable] = useState(false); @@ -932,6 +940,16 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings diagnostics={diagnostics} nodeDiagnostics={nodeDiagnostics} selectedNodeId={selectedNodeId} + decisionReviewDisabledReason={ + !draft.id + ? "Save the pipeline before recording review decisions." + : dirty + ? "Save the current pipeline revision before recording review decisions." + : !canEdit + ? editBlockedReason + : undefined + } + onReviewDecisions={() => setDecisionReviewOpen(true)} onClose={() => setResultOpen(false)} /> ) : null} @@ -1068,6 +1086,22 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings editable={canManageTriggers} onClose={() => setTriggersOpen(false)} /> + setDecisionReviewOpen(false)} + onChanged={() => { + void listDataflowSources(settings).then((catalogue) => { + setSources(catalogue.sources); + setSourceCatalogueAvailable(catalogue.available); + setSourceCatalogueWritable(catalogue.writable); + }).catch((sourceError) => setError(apiErrorMessage(sourceError))); + }} + /> ); } @@ -2448,6 +2482,380 @@ function SourceSnapshotDialog({ ); } +type ReviewableReconciliationRow = { + keyHash: string; + inputHash: string; + row: Record; +}; + +function ReconciliationDecisionDialog({ + open, + settings, + pipeline, + node, + preview, + editable, + onClose, + onChanged +}: { + open: boolean; + settings: ApiSettings; + pipeline: { id: string; name: string } | null; + node: PipelineGraphNode | null; + preview: PipelinePreview | null; + editable: boolean; + onClose: () => void; + onChanged: () => void; +}) { + const [decisionSets, setDecisionSets] = useState([]); + const [selectedSetId, setSelectedSetId] = useState(""); + const [selectedKeyHash, setSelectedKeyHash] = useState(""); + const [newSetName, setNewSetName] = useState(""); + const [action, setAction] = useState("accept"); + const [reason, setReason] = useState(""); + const [correctionText, setCorrectionText] = useState("{}"); + const [loading, setLoading] = useState(false); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + const [success, setSuccess] = useState(""); + + const rows = useMemo(() => { + const result = preview?.node_preview ?? preview; + if (!result || node?.type !== "reconcile.compare") return []; + return result.rows.flatMap((row) => { + const keyHash = row._reconciliation_key_hash; + const inputHash = row._reconciliation_input_hash; + if (!isSha256Hex(keyHash) || !isSha256Hex(inputHash)) return []; + return [{ keyHash, inputHash, row }]; + }); + }, [node?.type, preview]); + + const selectedSet = decisionSets.find((item) => item.id === selectedSetId) ?? null; + const selectedRow = rows.find((item) => item.keyHash === selectedKeyHash) ?? null; + const currentDecision = selectedSet?.current_decisions.find( + (item) => item.key_hash === selectedRow?.keyHash + ) ?? null; + const exactDecision = currentDecision?.input_hash === selectedRow?.inputHash + ? currentDecision + : null; + + useEffect(() => { + if (!open) return; + setSelectedKeyHash((current) => rows.some((item) => item.keyHash === current) + ? current + : rows[0]?.keyHash ?? ""); + }, [open, rows]); + + useEffect(() => { + if (!open || !pipeline) { + setDecisionSets([]); + setSelectedSetId(""); + return; + } + let cancelled = false; + setLoading(true); + setError(""); + void listDataflowDecisionSets(settings, pipeline.id) + .then((items) => { + if (cancelled) return; + const relevant = items.filter((item) => !item.node_id || item.node_id === node?.id); + setDecisionSets(relevant); + setSelectedSetId((current) => relevant.some((item) => item.id === current) + ? current + : relevant[0]?.id ?? ""); + }) + .catch((loadError) => { + if (!cancelled) setError(apiErrorMessage(loadError)); + }) + .finally(() => { + if (!cancelled) setLoading(false); + }); + return () => { + cancelled = true; + }; + }, [node?.id, open, pipeline?.id, settings]); + + useEffect(() => { + if (!open || !selectedSetId) return; + let cancelled = false; + setLoading(true); + void getDataflowDecisionSet(settings, selectedSetId) + .then((item) => { + if (cancelled) return; + setDecisionSets((current) => current.map( + (candidate) => candidate.id === item.id ? item : candidate + )); + }) + .catch((loadError) => { + if (!cancelled) setError(apiErrorMessage(loadError)); + }) + .finally(() => { + if (!cancelled) setLoading(false); + }); + return () => { + cancelled = true; + }; + }, [open, selectedSetId, settings]); + + useEffect(() => { + setAction(exactDecision?.action ?? "accept"); + setReason(exactDecision?.reason ?? ""); + setCorrectionText(JSON.stringify(exactDecision?.correction ?? {}, null, 2)); + setError(""); + setSuccess(""); + }, [exactDecision?.id, selectedKeyHash, selectedSetId]); + + const createSet = async () => { + if (!pipeline || !node || !newSetName.trim() || !editable) return; + setBusy(true); + setError(""); + setSuccess(""); + try { + const created = await createDataflowDecisionSet(settings, pipeline.id, { + name: newSetName.trim(), + node_id: node.id + }); + setDecisionSets((current) => [created, ...current.filter((item) => item.id !== created.id)]); + setSelectedSetId(created.id); + setNewSetName(""); + setSuccess("Decision set created. It is now available as a governed Dataflow source."); + onChanged(); + } catch (createError) { + setError(apiErrorMessage(createError)); + } finally { + setBusy(false); + } + }; + + const saveDecision = async () => { + if (!selectedSet || !selectedRow || !editable) return; + if (reason.trim().length < 3) { + setError("Record a reason of at least three characters."); + return; + } + let correction: Record | null = null; + if (action === "correct") { + try { + const parsed = JSON.parse(correctionText) as unknown; + if (!isRecord(parsed) || !Object.keys(parsed).length) { + setError("Correct decisions require a JSON object with at least one corrected field."); + return; + } + correction = parsed; + } catch { + setError("Correction must be a valid JSON object."); + return; + } + } + setBusy(true); + setError(""); + setSuccess(""); + try { + const updated = await recordDataflowDecision(settings, selectedSet.id, { + base_revision: selectedSet.resource_revision, + key_hash: selectedRow.keyHash, + input_hash: selectedRow.inputHash, + action, + reason: reason.trim(), + correction + }); + setDecisionSets((current) => current.map((item) => item.id === updated.id ? updated : item)); + setSuccess(`Recorded decision revision ${updated.resource_revision}.`); + onChanged(); + } catch (saveError) { + setError(apiErrorMessage(saveError)); + } finally { + setBusy(false); + } + }; + + const reviewedCount = selectedSet + ? rows.filter((row) => selectedSet.current_decisions.some( + (decision) => decision.key_hash === row.keyHash && decision.input_hash === row.inputHash + )).length + : 0; + const staleCount = selectedSet + ? rows.filter((row) => selectedSet.current_decisions.some( + (decision) => decision.key_hash === row.keyHash && decision.input_hash !== row.inputHash + )).length + : 0; + + return ( + + + {selectedSet ? `${selectedSet.history.length} immutable decision revision(s)` : "No decision set selected"} + + + + )} + > +
+ {error ? {error} : null} + {success ? {success} : null} +
+ + + + +
+ setNewSetName(event.target.value)} + placeholder={`${pipeline?.name ?? "Pipeline"} review`} + disabled={!editable || busy} + /> + +
+ +
+ {reviewedCount} reviewed + {staleCount} stale + {Math.max(0, rows.length - reviewedCount - staleCount)} open +
+
+ {!rows.length ? ( +
+ Run a preview of a reconciliation comparison that emits stable key and input hashes. +
+ ) : ( +
+
+
+ RecordComparisonDecision +
+ {rows.map((item) => { + const decision = selectedSet?.current_decisions.find( + (candidate) => candidate.key_hash === item.keyHash + ); + const stale = Boolean(decision && decision.input_hash !== item.inputHash); + const statusLabel = stale ? "stale" : decision?.action ?? "open"; + return ( + + ); + })} +
+
+ {selectedRow && selectedSet ? ( + <> +
+
+ {reviewRowLabel(selectedRow.row)} + {selectedRow.keyHash} +
+ {currentDecision && !exactDecision ? ( + + ) : exactDecision ? ( + + ) : ( + + )} +
+ + + ariaLabel="Reconciliation decision" + options={[ + { id: "accept", label: "Accept" }, + { id: "reject", label: "Reject" }, + { id: "correct", label: "Correct" }, + { id: "defer", label: "Defer" } + ]} + value={action} + onChange={setAction} + disabled={!editable || busy} + /> + + +