From 314e7c3edf747433fe352c1daadd4c6fa54f54e7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 11:45:35 +0200 Subject: [PATCH] Add version-pinned reconciliation decisions --- README.md | 8 + docs/HEICO_RELE_ASSESSMENT.md | 2 +- .../expected-output.json | 27 ++- .../golden/monthly-reconciliation/graph.json | 42 +++- .../inputs/review-decisions.json | 12 + src/govoplan_dataflow/backend/executor.py | 212 ++++++++++++++++++ src/govoplan_dataflow/backend/graph.py | 47 ++++ src/govoplan_dataflow/backend/manifest.py | 2 +- src/govoplan_dataflow/backend/node_library.py | 79 +++++++ .../backend/schema_validation.py | 62 +++++ src/govoplan_dataflow/backend/sql_compiler.py | 1 + tests/test_node_library.py | 1 + tests/test_operators.py | 143 ++++++++++++ webui/src/features/dataflow/NodeInspector.tsx | 67 ++++++ webui/src/features/dataflow/model.ts | 26 ++- 15 files changed, 724 insertions(+), 7 deletions(-) create mode 100644 fixtures/golden/monthly-reconciliation/inputs/review-decisions.json diff --git a/README.md b/README.md index 630068c..bce3fc9 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,14 @@ the set-based core of a RELE-style booking workflow with reviewable synthetic inputs and exact expected outputs. The detailed source-flow assessment is in [`docs/HEICO_RELE_ASSESSMENT.md`](docs/HEICO_RELE_ASSESSMENT.md). +The monthly fixture also consumes an explicit decision table. A decision pins +the reconciliation key hash and exact input hash plus its reference, actor, +time, reason, action, and optional correction. `reconcile.decisions` annotates +matching rows, marks changed-input decisions stale, and reports decisions whose +logical row disappeared. It never silently applies a correction to business +data; a downstream governed transform or Workflow handoff must interpret the +recorded action. + ## Development ```bash diff --git a/docs/HEICO_RELE_ASSESSMENT.md b/docs/HEICO_RELE_ASSESSMENT.md index 7d6d66c..835afc7 100644 --- a/docs/HEICO_RELE_ASSESSMENT.md +++ b/docs/HEICO_RELE_ASSESSMENT.md @@ -58,7 +58,7 @@ The correct GovOPlaN decomposition is: | Set-based normalization, joins, calculations, aggregation | Dataflow | Strong after this slice | | Ordered enrichments | Dataflow stages or reusable subflows | Modelable, manual decomposition | | Validation and reconciliation | Dataflow quality/reconciliation | Available for single-output checks | -| User verification, correction, rerun, resumability | Workflow | Foundation exists; flow-specific handoffs remain | +| User verification, correction, rerun, resumability | Workflow and Dataflow decision input | Version-pinned decision tables and invalidation are executable; flow-specific authoring/handoffs remain | | Database writes and cleanup | Governed Datasource output/effect capability | Missing explicit effect contract | | Warning/error side streams | Multi-output Dataflow nodes | Missing | | Fixed-width, CSV, and spreadsheet products | Reporting/Templates/Files | Missing integrated output profile | diff --git a/fixtures/golden/monthly-reconciliation/expected-output.json b/fixtures/golden/monthly-reconciliation/expected-output.json index 626d744..5b53b37 100644 --- a/fixtures/golden/monthly-reconciliation/expected-output.json +++ b/fixtures/golden/monthly-reconciliation/expected-output.json @@ -23,7 +23,14 @@ "case_id": "A-1", "status": "open", "amount": 10 - } + }, + "_decision_state": "unreviewed", + "_decision_action": null, + "_decision_ref": null, + "_decision_actor_ref": null, + "_decision_at": null, + "_decision_reason": null, + "_decision_correction": null }, { "case_id": "A-2", @@ -58,7 +65,14 @@ "case_id": "A-2", "status": "closed", "amount": 25 - } + }, + "_decision_state": "applied", + "_decision_action": "accept", + "_decision_ref": "decision:monthly-2026-07:A-2:1", + "_decision_actor_ref": "account:synthetic-reviewer", + "_decision_at": "2026-07-31T10:00:00Z", + "_decision_reason": "Synthetic fixture decision after source comparison.", + "_decision_correction": null }, { "monthly_case_id": "A-3", @@ -77,6 +91,13 @@ "case_id": "A-3", "status": "open", "amount": 30 - } + }, + "_decision_state": "unreviewed", + "_decision_action": null, + "_decision_ref": null, + "_decision_actor_ref": null, + "_decision_at": null, + "_decision_reason": null, + "_decision_correction": null } ] diff --git a/fixtures/golden/monthly-reconciliation/graph.json b/fixtures/golden/monthly-reconciliation/graph.json index 7f2d206..4193245 100644 --- a/fixtures/golden/monthly-reconciliation/graph.json +++ b/fixtures/golden/monthly-reconciliation/graph.json @@ -73,11 +73,39 @@ "right_prefix": "monthly_" } }, + { + "id": "decisions", + "type": "source.inline", + "label": "Review decisions", + "position": {"x": 1020, "y": 360}, + "config": { + "source_name": "review_decisions", + "fixture": "review-decisions.json", + "rows": [] + } + }, + { + "id": "apply-decisions", + "type": "reconcile.decisions", + "label": "Apply current decisions", + "position": {"x": 1260, "y": 160}, + "config": { + "decision_key_column": "key_hash", + "decision_input_column": "input_hash", + "decision_ref_column": "decision_ref", + "action_column": "action", + "actor_column": "actor_ref", + "decided_at_column": "decided_at", + "reason_column": "reason", + "correction_column": "correction", + "allowed_actions": ["accept", "reject", "correct", "defer"] + } + }, { "id": "output", "type": "output", "label": "Review differences", - "position": {"x": 1260, "y": 160}, + "position": {"x": 1500, "y": 160}, "config": {} } ], @@ -112,6 +140,18 @@ { "id": "e6", "source": "reconcile", + "target": "apply-decisions", + "target_port": "records" + }, + { + "id": "e7", + "source": "decisions", + "target": "apply-decisions", + "target_port": "decisions" + }, + { + "id": "e8", + "source": "apply-decisions", "target": "output" } ] diff --git a/fixtures/golden/monthly-reconciliation/inputs/review-decisions.json b/fixtures/golden/monthly-reconciliation/inputs/review-decisions.json new file mode 100644 index 0000000..e17a4b8 --- /dev/null +++ b/fixtures/golden/monthly-reconciliation/inputs/review-decisions.json @@ -0,0 +1,12 @@ +[ + { + "key_hash": "20f1552752978558506b159b92bc738d34f484d3d76068173b0c96e1fc4a8c2d", + "input_hash": "182de78bc3297d90cd4b02392bfb4a53fc9d9c29851d567f68e10be477a57dd8", + "decision_ref": "decision:monthly-2026-07:A-2:1", + "action": "accept", + "actor_ref": "account:synthetic-reviewer", + "decided_at": "2026-07-31T10:00:00Z", + "reason": "Synthetic fixture decision after source comparison.", + "correction": null + } +] diff --git a/src/govoplan_dataflow/backend/executor.py b/src/govoplan_dataflow/backend/executor.py index ab9167c..25a00ba 100644 --- a/src/govoplan_dataflow/backend/executor.py +++ b/src/govoplan_dataflow/backend/executor.py @@ -979,6 +979,205 @@ def _comparison_fields(value: object) -> tuple[tuple[str, str], ...]: return tuple(fields) +def _apply_reconciliation_decisions( + records: list[dict[str, Any]], + decisions: list[dict[str, Any]], + config: dict[str, Any], + *, + node_id: str, +) -> tuple[list[dict[str, Any]], tuple[str, ...]]: + columns = { + name: str(config[name]) + for name in ( + "decision_key_column", + "decision_input_column", + "decision_ref_column", + "action_column", + "actor_column", + "decided_at_column", + "reason_column", + "correction_column", + ) + } + allowed_actions = { + str(action) + for action in config.get("allowed_actions", ()) + } + decision_index: dict[str, dict[str, Any]] = {} + for row_number, decision in enumerate(decisions, start=1): + key_hash = _required_decision_hash( + decision.get(columns["decision_key_column"]), + row_number=row_number, + label="key hash", + node_id=node_id, + ) + _required_decision_hash( + decision.get(columns["decision_input_column"]), + row_number=row_number, + label="input hash", + node_id=node_id, + ) + for field_name, label in ( + ("decision_ref_column", "reference"), + ("action_column", "action"), + ("actor_column", "actor reference"), + ("decided_at_column", "decision time"), + ): + _required_decision_text( + decision.get(columns[field_name]), + row_number=row_number, + label=label, + node_id=node_id, + ) + action = str(decision[columns["action_column"]]).strip() + if action not in allowed_actions: + raise PipelineExecutionError( + f"Decision row {row_number} uses an action outside the governed action set.", + node_id=node_id, + ) + correction = decision.get(columns["correction_column"]) + if correction is not None and not isinstance(correction, dict): + raise PipelineExecutionError( + f"Decision row {row_number} correction must be an object or null.", + node_id=node_id, + ) + if action == "correct" and not correction: + raise PipelineExecutionError( + f"Decision row {row_number} requires a non-empty correction object.", + node_id=node_id, + ) + if key_hash in decision_index: + raise PipelineExecutionError( + "Decision rows contain more than one current decision for a reconciliation key.", + node_id=node_id, + ) + decision_index[key_hash] = decision + + matched_keys: set[str] = set() + stale_count = 0 + output: list[dict[str, Any]] = [] + for record in records: + key_hash = str(record.get("_reconciliation_key_hash") or "") + input_hash = str(record.get("_reconciliation_input_hash") or "") + if not _is_sha256(key_hash) or not _is_sha256(input_hash): + raise PipelineExecutionError( + "Decision application requires reconciliation key and input hashes.", + node_id=node_id, + ) + decision = decision_index.get(key_hash) + item = dict(record) + if decision is None: + _set_decision_fields(item, state="unreviewed") + output.append(item) + continue + matched_keys.add(key_hash) + decision_input_hash = str( + decision[columns["decision_input_column"]] + ).strip() + state = "applied" if decision_input_hash == input_hash else "stale" + if state == "stale": + stale_count += 1 + _set_decision_fields( + item, + state=state, + action=str(decision[columns["action_column"]]).strip(), + decision_ref=str( + decision[columns["decision_ref_column"]] + ).strip(), + actor_ref=str(decision[columns["actor_column"]]).strip(), + decided_at=str(decision[columns["decided_at_column"]]).strip(), + reason=_optional_decision_text( + decision.get(columns["reason_column"]) + ), + correction=decision.get(columns["correction_column"]), + ) + output.append(item) + + unmatched_count = len(decision_index.keys() - matched_keys) + messages = tuple( + message + for count, message in ( + ( + stale_count, + f"{stale_count} decision(s) are stale because reconciliation inputs changed.", + ), + ( + unmatched_count, + f"{unmatched_count} decision(s) no longer match a current reconciliation row.", + ), + ) + if count + ) + return output, messages + + +def _required_decision_hash( + value: object, + *, + row_number: int, + label: str, + node_id: str, +) -> str: + text = str(value or "").strip() + if not _is_sha256(text): + raise PipelineExecutionError( + f"Decision row {row_number} requires a SHA-256 {label}.", + node_id=node_id, + ) + return text + + +def _is_sha256(value: str) -> bool: + if len(value) != 64: + return False + try: + int(value, 16) + except ValueError: + return False + return True + + +def _required_decision_text( + value: object, + *, + row_number: int, + label: str, + node_id: str, +) -> str: + text = str(value or "").strip() + if not text: + raise PipelineExecutionError( + f"Decision row {row_number} requires a {label}.", + node_id=node_id, + ) + return text + + +def _optional_decision_text(value: object) -> str | None: + text = str(value or "").strip() + return text or None + + +def _set_decision_fields( + row: dict[str, Any], + *, + state: str, + action: str | None = None, + decision_ref: str | None = None, + actor_ref: str | None = None, + decided_at: str | None = None, + reason: str | None = None, + correction: object | None = None, +) -> None: + row["_decision_state"] = state + row["_decision_action"] = action + row["_decision_ref"] = decision_ref + row["_decision_actor_ref"] = actor_ref + row["_decision_at"] = decided_at + row["_decision_reason"] = reason + row["_decision_correction"] = correction + + def _unique_row_index( rows: list[dict[str, Any]], columns: list[str], @@ -1443,6 +1642,18 @@ def _execute_subflow_node( ) +def _execute_reconciliation_decisions( + context: OperatorExecutionContext, +) -> OperatorExecutionResult: + rows, messages = _apply_reconciliation_decisions( + context.outputs[context.inputs_by_port["records"][0]], + context.outputs[context.inputs_by_port["decisions"][0]], + context.node.config, + node_id=context.node.id, + ) + return OperatorExecutionResult(rows=rows, messages=messages) + + def _register_executors() -> None: executors = { "source.inline": _execute_inline_source, @@ -1546,6 +1757,7 @@ def _register_executors() -> None: node_id=context.node.id, ) ), + "reconcile.decisions": _execute_reconciliation_decisions, "subflow": _execute_subflow_node, "output": lambda context: OperatorExecutionResult( rows=[dict(row) for row in context.input_rows] diff --git a/src/govoplan_dataflow/backend/graph.py b/src/govoplan_dataflow/backend/graph.py index 6ae27fe..47d085e 100644 --- a/src/govoplan_dataflow/backend/graph.py +++ b/src/govoplan_dataflow/backend/graph.py @@ -1249,6 +1249,52 @@ def _validate_reconcile(node: GraphNode) -> list[DataflowDiagnostic]: return diagnostics +def _validate_reconciliation_decisions( + node: GraphNode, +) -> list[DataflowDiagnostic]: + diagnostics: list[DataflowDiagnostic] = [] + for field_name in ( + "decision_key_column", + "decision_input_column", + "decision_ref_column", + "action_column", + "actor_column", + "decided_at_column", + "reason_column", + "correction_column", + ): + if not _non_empty_text(node.config.get(field_name)): + diagnostics.append( + _node_field_error( + node, + "reconcile.decisions.column", + "Every decision mapping column must be named.", + field_name, + ) + ) + actions = node.config.get("allowed_actions") + if ( + not isinstance(actions, list) + or not actions + or len(actions) > 25 + or any( + not isinstance(action, str) or not action.strip() + for action in actions + ) + or len({action.strip() for action in actions if isinstance(action, str)}) + != len(actions) + ): + diagnostics.append( + _node_field_error( + node, + "reconcile.decisions.actions", + "Add between one and 25 unique allowed decision actions.", + "allowed_actions", + ) + ) + return diagnostics + + def _validate_subflow(node: GraphNode) -> list[DataflowDiagnostic]: diagnostics: list[DataflowDiagnostic] = [] for field_name, code, message in ( @@ -1415,6 +1461,7 @@ def _register_config_validators() -> None: "limit": _validate_limit, "quality.rules": _validate_quality, "reconcile.compare": _validate_reconcile, + "reconcile.decisions": _validate_reconciliation_decisions, "subflow": _validate_subflow, "output": _validate_no_config, } diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index ae6a346..34cbe97 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -200,7 +200,7 @@ 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 so a later governed human decision can be replayed only while its inputs still match. " + "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. " "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 " diff --git a/src/govoplan_dataflow/backend/node_library.py b/src/govoplan_dataflow/backend/node_library.py index 2421924..ad7e75a 100644 --- a/src/govoplan_dataflow/backend/node_library.py +++ b/src/govoplan_dataflow/backend/node_library.py @@ -530,6 +530,85 @@ _NODE_TYPES = ( }, sql_support="none", ), + NodeTypeDefinition( + type="reconcile.decisions", + category="quality", + label="Apply review decisions", + description="Attach versioned human decisions only while their reconciliation input hash still matches.", + icon="list-checks", + input_ports=( + NodePortDefinition(id="records", label="Reconciliation rows"), + NodePortDefinition(id="decisions", label="Decision rows"), + ), + config_fields=( + NodeConfigField( + id="decision_key_column", + label="Decision key hash column", + kind="text", + required=True, + ), + NodeConfigField( + id="decision_input_column", + label="Decision input hash column", + kind="text", + required=True, + ), + NodeConfigField( + id="decision_ref_column", + label="Decision reference column", + kind="text", + required=True, + ), + NodeConfigField( + id="action_column", + label="Action column", + kind="text", + required=True, + ), + NodeConfigField( + id="actor_column", + label="Actor reference column", + kind="text", + required=True, + ), + NodeConfigField( + id="decided_at_column", + label="Decision time column", + kind="text", + required=True, + ), + NodeConfigField( + id="reason_column", + label="Reason column", + kind="text", + required=True, + ), + NodeConfigField( + id="correction_column", + label="Correction column", + kind="text", + required=True, + ), + NodeConfigField( + id="allowed_actions", + label="Allowed actions", + kind="column_list", + required=True, + ), + ), + default_config={ + "decision_key_column": "key_hash", + "decision_input_column": "input_hash", + "decision_ref_column": "decision_ref", + "action_column": "action", + "actor_column": "actor_ref", + "decided_at_column": "decided_at", + "reason_column": "reason", + "correction_column": "correction", + "allowed_actions": ["accept", "reject", "correct", "defer"], + }, + sql_support="none", + ), NodeTypeDefinition( type="subflow", category="transform", diff --git a/src/govoplan_dataflow/backend/schema_validation.py b/src/govoplan_dataflow/backend/schema_validation.py index 968f7ca..dbe5783 100644 --- a/src/govoplan_dataflow/backend/schema_validation.py +++ b/src/govoplan_dataflow/backend/schema_validation.py @@ -640,6 +640,67 @@ def _reconcile(context: SchemaPropagationContext) -> SchemaPropagationResult: return SchemaPropagationResult(state, tuple(diagnostics)) +def _reconciliation_decisions( + context: SchemaPropagationContext, +) -> SchemaPropagationResult: + node = context.node + records = context.port_state("records") + decisions = context.port_state("decisions") + decision_columns = [ + str(node.config.get(field_name) or "") + for field_name in ( + "decision_key_column", + "decision_input_column", + "decision_ref_column", + "action_column", + "actor_column", + "decided_at_column", + "reason_column", + "correction_column", + ) + ] + diagnostics = [ + *_unknown_columns( + node, + records, + ["_reconciliation_key_hash", "_reconciliation_input_hash"], + field_name="records", + ), + *_unknown_columns( + node, + decisions, + decision_columns, + field_name="decisions", + ), + ] + state = SchemaState( + records.columns + | frozenset( + { + "_decision_state", + "_decision_action", + "_decision_ref", + "_decision_actor_ref", + "_decision_at", + "_decision_reason", + "_decision_correction", + } + ), + open=records.open, + types={ + **records.types, + "_decision_state": "string", + "_decision_action": "string", + "_decision_ref": "string", + "_decision_actor_ref": "string", + "_decision_at": "datetime", + "_decision_reason": "string", + "_decision_correction": "object", + }, + ) + return SchemaPropagationResult(state, tuple(diagnostics)) + + def _comparison_columns(value: object) -> tuple[list[str], list[str]]: left: list[str] = [] right: list[str] = [] @@ -982,6 +1043,7 @@ def register_schema_propagators() -> None: "limit": _identity, "quality.rules": _quality, "reconcile.compare": _reconcile, + "reconcile.decisions": _reconciliation_decisions, "subflow": _subflow, "output": _identity, } diff --git a/src/govoplan_dataflow/backend/sql_compiler.py b/src/govoplan_dataflow/backend/sql_compiler.py index a16d9b1..007b6fa 100644 --- a/src/govoplan_dataflow/backend/sql_compiler.py +++ b/src/govoplan_dataflow/backend/sql_compiler.py @@ -1953,6 +1953,7 @@ def _register_sql_renderers() -> None: "limit": _render_limit, "quality.rules": _render_unsupported, "reconcile.compare": _render_unsupported, + "reconcile.decisions": _render_unsupported, "subflow": _render_unsupported, "output": _render_noop, } diff --git a/tests/test_node_library.py b/tests/test_node_library.py index 724fcfd..6f45413 100644 --- a/tests/test_node_library.py +++ b/tests/test_node_library.py @@ -30,6 +30,7 @@ class DataflowNodeLibraryTests(unittest.TestCase): "limit", "quality.rules", "reconcile.compare", + "reconcile.decisions", "subflow", "output", }, diff --git a/tests/test_operators.py b/tests/test_operators.py index 3ba2fca..1bb0301 100644 --- a/tests/test_operators.py +++ b/tests/test_operators.py @@ -512,6 +512,149 @@ class DataflowOperatorTests(unittest.TestCase): self.assertNotIn("private-key", str(raised.exception)) + def test_reconciliation_decisions_apply_only_to_exact_input_evidence(self) -> None: + records = [ + { + "case_id": "A-1", + "_reconciliation_key_hash": "a" * 64, + "_reconciliation_input_hash": "b" * 64, + }, + { + "case_id": "A-2", + "_reconciliation_key_hash": "c" * 64, + "_reconciliation_input_hash": "d" * 64, + }, + ] + decisions = [ + { + "key_hash": "a" * 64, + "input_hash": "b" * 64, + "decision_ref": "decision:1", + "action": "accept", + "actor_ref": "account:reviewer", + "decided_at": "2026-08-04T09:00:00Z", + "reason": "Verified against source evidence.", + "correction": None, + }, + { + "key_hash": "c" * 64, + "input_hash": "e" * 64, + "decision_ref": "decision:2", + "action": "correct", + "actor_ref": "account:reviewer", + "decided_at": "2026-08-04T09:05:00Z", + "reason": "Prior amount was wrong.", + "correction": {"amount": 25}, + }, + { + "key_hash": "f" * 64, + "input_hash": "1" * 64, + "decision_ref": "decision:orphaned", + "action": "defer", + "actor_ref": "account:reviewer", + "decided_at": "2026-08-04T09:10:00Z", + "reason": "No current row.", + "correction": None, + }, + ] + graph = PipelineGraph( + nodes=[ + node( + "records", + "source.inline", + {"source_name": "records", "rows": records}, + x=0, + ), + node( + "decisions", + "source.inline", + {"source_name": "decisions", "rows": decisions}, + x=0, + ), + node( + "apply", + "reconcile.decisions", + { + "decision_key_column": "key_hash", + "decision_input_column": "input_hash", + "decision_ref_column": "decision_ref", + "action_column": "action", + "actor_column": "actor_ref", + "decided_at_column": "decided_at", + "reason_column": "reason", + "correction_column": "correction", + "allowed_actions": [ + "accept", + "reject", + "correct", + "defer", + ], + }, + x=300, + ), + node("output", "output", {}, x=600), + ], + edges=[ + GraphEdge( + id="e1", + source="records", + target="apply", + target_port="records", + ), + GraphEdge( + id="e2", + source="decisions", + target="apply", + target_port="decisions", + ), + GraphEdge(id="e3", source="apply", target="output"), + ], + ) + + result = execute_preview(graph, row_limit=100) + + self.assertEqual( + ["applied", "stale"], + [row["_decision_state"] for row in result.rows], + ) + self.assertEqual("decision:1", result.rows[0]["_decision_ref"]) + self.assertEqual( + {"amount": 25}, + result.rows[1]["_decision_correction"], + ) + apply_diagnostic = next( + item for item in result.node_diagnostics if item.node_id == "apply" + ) + self.assertEqual( + [ + "1 decision(s) are stale because reconciliation inputs changed.", + "1 decision(s) no longer match a current reconciliation row.", + ], + apply_diagnostic.messages, + ) + + decisions[1] = dict(decisions[0], decision_ref="decision:duplicate") + duplicate_nodes = [ + ( + graph_node.model_copy( + update={ + "config": {**graph_node.config, "rows": decisions} + } + ) + if graph_node.id == "decisions" + else graph_node + ) + for graph_node in graph.nodes + ] + with self.assertRaisesRegex( + PipelineExecutionError, + "more than one current decision", + ): + execute_preview( + graph.model_copy(update={"nodes": duplicate_nodes}), + row_limit=100, + ) + def test_parameterized_subflow_runs_a_pinned_graph(self) -> None: nested = { "schema_version": 1, diff --git a/webui/src/features/dataflow/NodeInspector.tsx b/webui/src/features/dataflow/NodeInspector.tsx index 1023e20..5d166ca 100644 --- a/webui/src/features/dataflow/NodeInspector.tsx +++ b/webui/src/features/dataflow/NodeInspector.tsx @@ -712,6 +712,73 @@ export default function NodeInspector({ ) : null} + {node.type === "reconcile.decisions" ? ( + <> + + updateConfig({ decision_key_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ decision_input_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ decision_ref_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ action_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ actor_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ decided_at_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ reason_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ correction_column: event.target.value })} + disabled={readOnly} + /> + + + updateConfig({ allowed_actions: commaList(event.target.value) })} + disabled={readOnly} + /> + + + ) : null} {node.type === "subflow" ? ( <> diff --git a/webui/src/features/dataflow/model.ts b/webui/src/features/dataflow/model.ts index 2e2848f..5a81090 100644 --- a/webui/src/features/dataflow/model.ts +++ b/webui/src/features/dataflow/model.ts @@ -243,6 +243,30 @@ export const FALLBACK_NODE_LIBRARY: NodeTypeDefinition[] = [ output, { left_keys: [""], right_keys: [""], compare_columns: [], right_prefix: "observed_" } ), + nodeType( + "reconcile.decisions", + "quality", + "Quality", + "Apply review decisions", + "Attach versioned decisions while reconciliation inputs still match.", + "list-checks", + [ + { id: "records", label: "Reconciliation rows", required: true, multiple: false, minimum_connections: 1 }, + { id: "decisions", label: "Decision rows", required: true, multiple: false, minimum_connections: 1 } + ], + output, + { + decision_key_column: "key_hash", + decision_input_column: "input_hash", + decision_ref_column: "decision_ref", + action_column: "action", + actor_column: "actor_ref", + decided_at_column: "decided_at", + reason_column: "reason", + correction_column: "correction", + allowed_actions: ["accept", "reject", "correct", "defer"] + } + ), nodeType( "subflow", "transform", @@ -446,7 +470,7 @@ function nodeType( config_fields: [], default_config: defaultConfig, sql_support: - ["replace", "quality.rules", "reconcile.compare", "subflow"].includes(type) + ["replace", "quality.rules", "reconcile.compare", "reconcile.decisions", "subflow"].includes(type) ? "none" : ["combine.union", "combine.join", "distinct", "derive", "expression", "convert", "filter.expression"].includes(type) ? "partial"