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"