Add durable reconciliation row evidence

src/govoplan_dataflow/backend/executor.py

@@ -878,12 +878,32 @@ def _reconcile_rows(
     right_keys = [str(item) for item in config["right_keys"]]
     right_prefix = str(config.get("right_prefix", "observed_"))
     comparisons = _comparison_fields(config.get("compare_columns"))
-    left_index = _unique_row_index(left_rows, left_keys, node_id=node_id)
-    right_index = _unique_row_index(right_rows, right_keys, node_id=node_id)
+    left_index = _unique_row_index(
+        left_rows,
+        left_keys,
+        node_id=node_id,
+        input_label="expected",
+    )
+    right_index = _unique_row_index(
+        right_rows,
+        right_keys,
+        node_id=node_id,
+        input_label="observed",
+    )
     output: list[dict[str, Any]] = []
     for key in dict.fromkeys((*left_index, *right_index)):
         left = left_index.get(key)
         right = right_index.get(key)
+        key_values = [
+            (left or {}).get(left_name)
+            if left is not None
+            else (right or {}).get(right_name)
+            for left_name, right_name in zip(
+                left_keys,
+                right_keys,
+                strict=True,
+            )
+        ]
         item = dict(left or {})
         if right is not None:
             item.update({f"{right_prefix}{name}": value for name, value in right.items()})
@@ -895,18 +915,55 @@ def _reconcile_rows(
             differences = []
         else:
             fields = comparisons or tuple((name, name) for name in left if name not in left_keys)
-            differences = [
-                left_name
+            changes = [
+                {
+                    "expected_field": left_name,
+                    "observed_field": right_name,
+                    "expected": left.get(left_name),
+                    "observed": right.get(right_name),
+                }
                 for left_name, right_name in fields
                 if left.get(left_name) != right.get(right_name)
             ]
+            differences = [str(change["expected_field"]) for change in changes]
             status = "changed" if differences else "match"
+        if left is None or right is None:
+            changes = []
         item["_reconciliation_status"] = status
         item["_reconciliation_differences"] = differences
+        item["_reconciliation_changes"] = changes
+        item["_reconciliation_key"] = key_values
+        item["_reconciliation_key_hash"] = _reconciliation_hash(
+            {
+                "left_keys": left_keys,
+                "right_keys": right_keys,
+                "values": key_values,
+            }
+        )
+        item["_reconciliation_input_hash"] = _reconciliation_hash(
+            {
+                "key": key_values,
+                "expected": left,
+                "observed": right,
+                "comparisons": comparisons,
+            }
+        )
+        item["_reconciliation_before"] = dict(left) if left is not None else None
+        item["_reconciliation_after"] = dict(right) if right is not None else None
         output.append(item)
     return output
 
 
+def _reconciliation_hash(value: object) -> str:
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
 def _comparison_fields(value: object) -> tuple[tuple[str, str], ...]:
     if not isinstance(value, list):
         return ()
@@ -927,13 +984,14 @@ def _unique_row_index(
     columns: list[str],
     *,
     node_id: str,
+    input_label: str,
 ) -> dict[tuple[Any, ...], dict[str, Any]]:
     index: dict[tuple[Any, ...], dict[str, Any]] = {}
     for row in rows:
         key = tuple(_hashable(row.get(column)) for column in columns)
         if key in index:
             raise PipelineExecutionError(
-                f"Reconciliation keys are not unique: {key!r}.",
+                f"Reconciliation keys are not unique in the {input_label} input.",
                 node_id=node_id,
             )
         index[key] = row