Add version-pinned reconciliation decisions

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

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 |

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
   }
 ]

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"
     }
   ]

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
+  }
+]

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]

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,
     }

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 "

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",

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,
     }

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,
     }

tests/test_node_library.py

@@ -30,6 +30,7 @@ class DataflowNodeLibraryTests(unittest.TestCase):
                 "limit",
                 "quality.rules",
                 "reconcile.compare",
+                "reconcile.decisions",
                 "subflow",
                 "output",
             },

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,

webui/src/features/dataflow/NodeInspector.tsx

@@ -712,6 +712,73 @@ export default function NodeInspector({
             </FormField>
           </>
         ) : null}
+        {node.type === "reconcile.decisions" ? (
+          <>
+            <FormField label="Decision key hash column">
+              <input
+                value={textValue(node.config.decision_key_column)}
+                onChange={(event) => updateConfig({ decision_key_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Decision input hash column">
+              <input
+                value={textValue(node.config.decision_input_column)}
+                onChange={(event) => updateConfig({ decision_input_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Decision reference column">
+              <input
+                value={textValue(node.config.decision_ref_column)}
+                onChange={(event) => updateConfig({ decision_ref_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Action column">
+              <input
+                value={textValue(node.config.action_column)}
+                onChange={(event) => updateConfig({ action_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Actor reference column">
+              <input
+                value={textValue(node.config.actor_column)}
+                onChange={(event) => updateConfig({ actor_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Decision time column">
+              <input
+                value={textValue(node.config.decided_at_column)}
+                onChange={(event) => updateConfig({ decided_at_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Reason column">
+              <input
+                value={textValue(node.config.reason_column)}
+                onChange={(event) => updateConfig({ reason_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Correction column">
+              <input
+                value={textValue(node.config.correction_column)}
+                onChange={(event) => updateConfig({ correction_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Allowed actions">
+              <input
+                value={stringList(node.config.allowed_actions).join(", ")}
+                onChange={(event) => updateConfig({ allowed_actions: commaList(event.target.value) })}
+                disabled={readOnly}
+              />
+            </FormField>
+          </>
+        ) : null}
         {node.type === "subflow" ? (
           <>
             <FormField label="Template reference">

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"