fix(guard,explain): survive duplicate index labels and unhashable cells

src/freshdata/explain.py

@@ -39,6 +39,19 @@ def _require_distinct_label_names(df: pd.DataFrame, func: str) -> None:
         )
 
 
+def _nunique(series: pd.Series) -> int:
+    """Distinct non-null values, counting unhashable cells by their text form.
+
+    ``nunique`` hashes every value, so a list or dict cell raised
+    ``TypeError`` although ``fd.clean`` and ``fd.profile`` accept the same
+    frame (#450).
+    """
+    try:
+        return int(series.nunique(dropna=True))
+    except TypeError:
+        return int(series.dropna().map(repr).nunique())
+
+
 def _column_stats(df: pd.DataFrame) -> dict[str, dict[str, Any]]:
     stats: dict[str, dict[str, Any]] = {}
     for col in df.columns:
@@ -47,7 +60,7 @@ def _column_stats(df: pd.DataFrame) -> dict[str, dict[str, Any]]:
             "dtype": str(s.dtype),
             "null_count": int(s.isna().sum()),
             "null_pct": round(float(s.isna().mean()), 4),
-            "nunique": int(s.nunique(dropna=True)),
+            "nunique": _nunique(s),
         }
         if is_numeric_dtype(s):
             nonnull = s.dropna()

src/freshdata/guard.py

@@ -27,6 +27,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterable, Mapping
+from typing import Any
 
 import pandas as pd
 
@@ -143,6 +144,46 @@ def snapshot_protected(
     return snapshot
 
 
+def _cell_equal(left: Any, right: Any) -> bool:
+    """Value equality that treats two missing values as equal."""
+    left_na, right_na = _is_missing(left), _is_missing(right)
+    if left_na or right_na:
+        return left_na and right_na
+    try:
+        return bool(left == right)
+    except Exception:  # noqa: BLE001 - exotic cell types compare however they like
+        return left is right
+
+
+def _is_missing(value: Any) -> bool:
+    try:
+        return bool(pd.isna(value))
+    except (TypeError, ValueError):  # containers: never missing
+        return False
+
+
+def _kept_in_order(before: pd.Series, after: pd.Series) -> str | None:
+    """Check surviving values positionally when index labels cannot align them.
+
+    With repeated labels ``before.loc[after.index]`` multiplies rows instead of
+    selecting them, so the comparison could never succeed (#449). Row-level
+    steps only drop rows and never reorder or rewrite a protected column, so
+    every surviving value must still appear, in order, in the original column.
+    """
+    values = list(before)
+    position = 0
+    for n, value in enumerate(after):
+        while position < len(values) and not _cell_equal(values[position], value):
+            position += 1
+        if position == len(values):
+            return (
+                "surviving cell values do not match the original column "
+                f"(first at position {n})"
+            )
+        position += 1
+    return None
+
+
 def _series_identical(before: pd.Series, after: pd.Series) -> str | None:
     """Return a human explanation of the first difference, or ``None`` if none."""
     if str(before.dtype) != str(after.dtype):
@@ -153,6 +194,8 @@ def _series_identical(before: pd.Series, after: pd.Series) -> str | None:
         # Row-level steps (dedupe, empty-row drops) legitimately remove rows;
         # surviving rows must still hold their original values, aligned by
         # index label.
+        if not before.index.is_unique or not after.index.is_unique:
+            return _kept_in_order(before, after)
         try:
             before = before.loc[after.index]
         except KeyError:

tests/test_explain.py

@@ -110,3 +110,14 @@ def test_explain_clean_dedupe_reports_zero_cell_changes(messy=None):
     rep = fd.explain_clean(df, drop_duplicates=True)
     assert rep.rows_after < rep.rows_before
     assert all(v == 0 for v in rep.cell_changes.values()), rep.cell_changes
+
+
+def test_explain_clean_accepts_unhashable_cells():
+    # Regression (#450): nunique() hashes every value, so a list cell raised
+    # TypeError although fd.clean and fd.profile accept the same frame.
+    df = pd.DataFrame({"payload": [[1], [1], {"k": 2}, None], "n": [1, 2, 3, 4]})
+    report = fd.explain_clean(df)
+    assert report is not None
+    stats = report.to_dict()["before_stats"]["payload"]
+    assert stats["nunique"] == 2  # [1] twice, {"k": 2} once
+    assert stats["null_count"] == 1

tests/test_guard_protected.py

@@ -8,8 +8,10 @@
 import pytest
 
 import freshdata as fd
+from freshdata.context.types import ColumnConstraint, ContextPolicy
 from freshdata.guard import (
     ProtectedColumnError,
+    _series_identical,
     hard_protected_columns,
     protected_column_set,
     snapshot_protected,
@@ -181,3 +183,47 @@ def test_guard_report_metadata_names_protected_columns():
     )
     guard_actions = [a for a in report if a.step == "guard"]
     assert guard_actions and guard_actions[0].metadata["protected_columns"] == ["rev"]
+
+
+# ── #449: duplicate index labels ────────────────────────────────────────────────
+
+
+def _dup_frame() -> pd.DataFrame:
+    return pd.DataFrame(
+        {"a": pd.array([None, None, None, 0], dtype="Int64"),
+         "b": pd.array([None] * 4, dtype="Int64")},
+        index=[0, 0, 0, 0],
+    )
+
+
+def _never_modify_policy(column: str) -> ContextPolicy:
+    return ContextPolicy(constraints=(ColumnConstraint(
+        id="c1", column=column, resolved_from=column, resolution_confidence=1.0,
+        rule="protected", action="never_modify", enforcement="hard"),))
+
+
+def test_protected_column_survives_a_duplicate_index():
+    # Regression (#449): the guard aligned the snapshot with before.loc[after.index],
+    # which multiplies rows when labels repeat, so dropping any row made the
+    # comparison impossible and raised on an unmodified column.
+    df = _dup_frame()
+    expected = fd.clean(df.copy(), verbose=False)["a"].tolist()
+    out = fd.clean(df.copy(), policy=_never_modify_policy("a"), verbose=False)
+    assert out["a"].tolist() == expected
+
+
+def test_guard_still_detects_a_modified_cell_on_a_duplicate_index():
+    before = pd.Series([1.0, 2.0, 3.0, 4.0], index=[0, 0, 0, 0])
+    assert _series_identical(before, pd.Series([1.0, 4.0], index=[0, 0])) is None
+    for broken in (
+        pd.Series([1.0, 9.0], index=[0, 0]),          # value rewritten
+        pd.Series([4.0, 1.0], index=[0, 0]),          # rows reordered
+        pd.Series([1.0, 2.0, 3.0, 4.0, 5.0], index=[0] * 5),  # row gained
+    ):
+        assert _series_identical(before, broken) is not None
+
+
+def test_guard_treats_two_missing_values_as_equal_on_a_duplicate_index():
+    before = pd.Series([1.0, None, 3.0], index=[0, 0, 0])
+    assert _series_identical(before, pd.Series([None, 3.0], index=[0, 0])) is None
+    assert _series_identical(before, pd.Series([7.0, 3.0], index=[0, 0])) is not None