fix(contracts): profile awkward cells, compare storage-independently, drop phantom drift (#446)

Four defects found by fuzzing, all on entry points documented as read-only
profiling of any pandas frame:

* list and dict cells crashed profiling, because nunique/unique hash every
  value. They are ordinary JSON-derived data, so such cells are counted by
  their text form instead.
* non-UTF-8 bytes crashed the categorical profile on pandas 2, whose string
  cast decodes strictly, and lone surrogates crashed the keyed-label HMAC.
  Bytes now decode with backslash escapes, and the HMAC encodes with
  "surrogatepass". Valid UTF-8 strings keep their previous label, so
  existing baselines still match.
* allowed_values raised a raw pyarrow error on pyarrow-backed columns, while
  the same rule reported a violation on numpy or masked storage. The
  membership test now falls back to object comparison, so the finding no
  longer depends on how the column is stored.
* a frame could fail PSI drift against a baseline built from that same frame.
  A quantile interpolated between two observed values sits in a gap where the
  baseline CDF is flat, so its nominal probability overstates the share of
  rows at or below it, and the bin between two such knots claimed mass no
  data could fill. Stored probabilities are now read as the baseline's own
  empirical CDF, (floor(p*(n-1)) + 1)/n.

Closes #434
Closes #435
Closes #439
Closes #441

Author: kevincostner17

src/freshdata/enterprise/contracts.py

@@ -157,9 +157,61 @@ def _resolve_label_key(label_key: str | bytes | None) -> bytes | None:
     return key
 
 
+def _cell_text(value: Any) -> str:
+    """Text for one cell; undecodable bytes become ``\\xNN`` escapes.
+
+    pandas 2 casts bytes to ``string`` by strict UTF-8 decoding, so a single
+    invalid byte raised ``UnicodeDecodeError`` while profiling (#435). Decoding
+    here is deterministic and works the same on every pandas version.
+    """
+    if isinstance(value, (bytes, bytearray)):
+        return bytes(value).decode("utf-8", errors="backslashreplace")
+    return str(value)
+
+
+def _text_series(series: pd.Series) -> pd.Series:
+    """``series.astype("string")`` that cannot fail on bytes cells (#435)."""
+    try:
+        return series.astype("string")
+    except (UnicodeDecodeError, TypeError, ValueError):
+        return pd.Series(
+            [_cell_text(v) for v in series], index=series.index, name=series.name, dtype="string"
+        )
+
+
+def _hashable_series(series: pd.Series) -> pd.Series:
+    """*series* with unhashable cells replaced by their text form (#434).
+
+    List and dict cells are ordinary JSON-derived data, but they cannot go
+    through ``nunique``/``unique``/``value_counts``, which hash every value.
+    """
+    return pd.Series(
+        [_cell_text(v) if isinstance(v, (list, dict, set, bytearray)) else v for v in series],
+        index=series.index,
+        name=series.name,
+        dtype=object,
+    )
+
+
+def _unhashable_safe(series: pd.Series) -> pd.Series:
+    """*series* if its cells can be hashed, else :func:`_hashable_series`."""
+    if series.dtype != object:
+        return series
+    try:
+        set(series)
+    except TypeError:
+        return _hashable_series(series)
+    return series
+
+
 def _keyed_label(key: bytes, value: str) -> str:
     """Pseudonymous category label: ``k:`` + HMAC-SHA256 (128 bits, hex)."""
-    digest = hmac.new(key, _LABEL_DOMAIN + value.encode("utf-8"), hashlib.sha256)
+    # "surrogatepass" keeps lone surrogates (from surrogateescape file reads)
+    # encodable; every valid UTF-8 string encodes exactly as before, so
+    # existing baselines keep matching (#435).
+    digest = hmac.new(
+        key, _LABEL_DOMAIN + value.encode("utf-8", "surrogatepass"), hashlib.sha256
+    )
     return "k:" + digest.hexdigest()[:32]
 
 
@@ -849,7 +901,7 @@ def _profile_column(
     dtype = str(series.dtype)
     n_missing = int(series.isna().sum())
     missing_ratio = (n_missing / n_rows) if n_rows else 0.0
-    non_null = series.dropna()
+    non_null = _unhashable_safe(series.dropna())
     n_unique = int(non_null.nunique())
 
     cb = ColumnBaseline(
@@ -864,7 +916,7 @@ def _profile_column(
 
     if include_samples and len(non_null):
         uniques = pd.unique(non_null)[:_MAX_SAMPLE_VALUES]
-        cb.sample_values = tuple(str(v) for v in uniques)
+        cb.sample_values = tuple(_cell_text(v) for v in uniques)
 
     if family in ("int", "float") and len(non_null):
         # float: Arrow decimals come back as Decimal objects, which can't take std().
@@ -883,7 +935,7 @@ def _profile_column(
             cb.min_timestamp = ts.min().isoformat()
             cb.max_timestamp = ts.max().isoformat()
     elif len(non_null):  # categorical / string / bool / other
-        counts = non_null.astype("string").value_counts()
+        counts = _text_series(non_null).value_counts()
         top = counts.head(_MAX_TOP_CATEGORIES)
         total = int(counts.sum())
         # Category labels can themselves be PII. Unless the caller opted into
@@ -1058,6 +1110,31 @@ def _cdf_between_knots(
     return out
 
 
+def _baseline_non_null(cb: ColumnBaseline) -> int:
+    """How many non-null rows the baseline was built from."""
+    n = cb.n_rows * (1.0 - (cb.missing_ratio or 0.0))
+    return int(round(n)) if n >= 1 else 0
+
+
+def _empirical_cut(probs: np.ndarray, cb: ColumnBaseline) -> np.ndarray:
+    """Stored quantile probabilities as the baseline's own CDF at those values.
+
+    A quantile interpolated between two observed values sits in a gap where the
+    baseline CDF is flat, so its nominal probability overstates the share of
+    rows at or below it, and bins between two such knots claim mass that no
+    data can fill — a frame then drifted against its own baseline (#441). With
+    *n* baseline rows, numpy's linear quantile for probability *p* lies between
+    order statistics *k* and *k+1* with ``k = floor(p * (n - 1))``, so the share
+    at or below it is ``(k + 1) / n``. Ties on a knot only raise that share, and
+    the caller already clips the current share into the observed range.
+    """
+    n = _baseline_non_null(cb)
+    if n < 2:
+        return probs
+    snapped = (np.floor(probs * (n - 1) + 1e-9) + 1.0) / n
+    return np.maximum.accumulate(np.clip(snapped, 0.0, 1.0))
+
+
 def _psi_numeric(cb: ColumnBaseline, current: pd.Series) -> float | None:
     """PSI over numeric bins defined by the baseline quantile edges."""
     pts = cb.cdf_points()
@@ -1076,7 +1153,7 @@ def _psi_numeric(cb: ColumnBaseline, current: pd.Series) -> float | None:
     # across it the same way, so a point mass on a quantile isn't counted
     # wholly in the upper bin.
     inner = knots[1:-1]
-    cut = p_lo[1:-1]
+    cut = _empirical_cut(p_lo[1:-1], cb)
     cur = np.sort(vals)
     below = np.searchsorted(cur, inner, side="left") / n
     at = np.searchsorted(cur, inner, side="right") / n
@@ -1919,12 +1996,26 @@ def _contract_lengths(
     return bool(ok)
 
 
+def _isin(series: pd.Series, values: list[Any]) -> pd.Series:
+    """``series.isin(values)`` that works on pyarrow-backed columns (#439).
+
+    A pyarrow-backed column raises ``ArrowTypeError``/``ArrowInvalid`` when the
+    values have another type, so the same rule reported a violation on an
+    ``Int64`` column and a raw pyarrow error on ``int64[pyarrow]``. Comparing as
+    Python objects gives the storage-independent result.
+    """
+    try:
+        return series.isin(values)
+    except Exception:  # noqa: BLE001 - pyarrow raises its own error types
+        return series.astype(object).isin(values)
+
+
 def _contract_values(findings: list[DriftFinding], cc: ColumnContract, series: pd.Series) -> bool:
     ok = True
     non_null = series.dropna()
     n_total = len(non_null)
     if cc.allowed_values:
-        bad_mask = ~non_null.isin(list(cc.allowed_values))
+        bad_mask = ~_isin(non_null, list(cc.allowed_values))
         n_bad = int(bad_mask.sum())
         if n_bad:
             offenders = sorted({str(v) for v in non_null[bad_mask].unique()})

tests/test_contracts_input_robustness.py

@@ -8,14 +8,16 @@
 
 from __future__ import annotations
 
+import hashlib
+import hmac
 import warnings
 
 import numpy as np
 import pandas as pd
 import pytest
 
 import freshdata as fd
-from freshdata.enterprise.contracts import _resolve_label
+from freshdata.enterprise.contracts import _LABEL_DOMAIN, _keyed_label, _resolve_label
 
 
 def _contract(*columns: fd.ColumnContract, **kwargs: object) -> fd.DataContract:
@@ -295,3 +297,101 @@ def test_validate_suite_rejects_duplicate_labels(duplicated):
     suite = fd.ValidationSuite(name="s", rules=[fd.ColumnRule("id", min_value=0)])
     with pytest.raises(ValueError, match="duplicated label"):
         fd.validate(duplicated, suite=suite)
+
+
+# ── #434: list / dict cells are data, not hash keys ─────────────────────────────
+
+
+@pytest.mark.parametrize("cell", [[1], {"k": 1}, {1, 2}, bytearray(b"ab")])
+def test_profiling_accepts_unhashable_cells(cell):
+    df = pd.DataFrame({"payload": [cell, "x", None]})
+    baseline = fd.build_baseline(df, name="b")
+    assert baseline.columns["payload"].cardinality == 2
+
+    suite = fd.ValidationSuite(name="s", rules=[fd.ColumnRule("payload", nullable=True)])
+    assert fd.validate(df, suite=suite) is not None
+    assert fd.enforce_contract(df, _contract(fd.ColumnContract("payload"))) is not None
+
+
+def test_unhashable_cells_keep_their_samples():
+    df = pd.DataFrame({"payload": [[1], [1], {"k": 2}]})
+    baseline = fd.build_baseline(df, name="b", include_samples=True)
+    assert set(baseline.columns["payload"].sample_values) == {"[1]", "{'k': 2}"}
+
+
+# ── #435: bytes and surrogate strings profile without raising ───────────────────
+
+
+def test_baseline_accepts_non_utf8_bytes():
+    df = pd.DataFrame({"blob": [b"\xff", "x", None, b"ok"]})
+    baseline = fd.build_baseline(df, name="b", include_samples=True)
+    values = baseline.columns["blob"].sample_values
+    assert any("\\xff" in v for v in values)
+    assert "ok" in values
+
+
+def test_baseline_accepts_surrogate_strings_with_a_label_key():
+    df = pd.DataFrame({"name": ["x\udcff", "y", None]})
+    baseline = fd.build_baseline(df, name="b", label_key="k")
+    assert baseline.columns["name"].top_values
+    assert all(v.startswith("k:") for v in baseline.columns["name"].top_values)
+
+
+def test_keyed_labels_are_unchanged_for_valid_utf8():
+    # Existing baselines must keep matching: the surrogate fix may not move
+    # the HMAC of any string that already encoded (#435).
+    for value in ["alice", "Ünicode ✓", "", '"quoted"', "日本語"]:
+        expected = hmac.new(
+            b"secret", _LABEL_DOMAIN + value.encode("utf-8"), hashlib.sha256
+        ).hexdigest()[:32]
+        assert _keyed_label(b"secret", value) == "k:" + expected
+
+
+# ── #439: allowed_values must not depend on column storage ──────────────────────
+
+
+@pytest.mark.parametrize("dtype", ["int64", "Int64", "int64[pyarrow]", "double[pyarrow]"])
+def test_allowed_values_reports_violations_on_any_storage(dtype):
+    pytest.importorskip("pyarrow")
+    df = pd.DataFrame({"code": pd.array([1, 2, 3], dtype=dtype)})
+    contract = _contract(fd.ColumnContract("code", allowed_values=("a", "b")))
+    report = fd.enforce_contract(df, contract)
+    assert "contract.allowed_values" in _check_ids(report)
+
+
+def test_allowed_values_passes_on_pyarrow_backed_matches():
+    pytest.importorskip("pyarrow")
+    df = pd.DataFrame({"code": pd.array([1, 2], dtype="int64[pyarrow]")})
+    contract = _contract(fd.ColumnContract("code", allowed_values=(1, 2)))
+    assert "contract.allowed_values" not in _check_ids(fd.enforce_contract(df, contract))
+
+
+# ── #441: a frame never drifts against its own baseline ─────────────────────────
+
+
+@pytest.mark.parametrize("kind", ["wide_int_with_na", "float_with_nan", "zero_inflated",
+                                  "constant", "few_rows", "all_null", "heavy_ties"])
+def test_frame_does_not_drift_against_its_own_baseline(kind):
+    rng = np.random.default_rng(20260916)
+    columns = {
+        "wide_int_with_na": pd.array(
+            [-(2**63) + 200, -(2**63) + 1, None, 2, -2, 0, 3, None, 1, -1] * 3, dtype="Int64"
+        ),
+        "float_with_nan": pd.array([1e18, -1e18, np.nan, 0.5, 2.5] * 6, dtype="float64"),
+        "zero_inflated": pd.array([0.0] * 25 + [7.5, 9.0, 0.0, 12.0, 0.0], dtype="float64"),
+        "constant": pd.array([4.0] * 30, dtype="float64"),
+        "few_rows": pd.array([1, 5, None], dtype="Int64"),
+        "all_null": pd.array([None] * 12, dtype="Int64"),
+        "heavy_ties": pd.array(list(rng.integers(0, 3, 60)), dtype="Int64"),
+    }
+    df = pd.DataFrame({"f": columns[kind]})
+    report = fd.compare_to_baseline(df, fd.build_baseline(df, name="b"))
+    assert report.passed, [f.message for f in report.errors]
+
+
+def test_real_drift_is_still_reported():
+    rng = np.random.default_rng(7)
+    baseline = fd.build_baseline(pd.DataFrame({"f": rng.normal(0, 1, 400)}), name="b")
+    report = fd.compare_to_baseline(pd.DataFrame({"f": rng.normal(4, 1, 400)}), baseline)
+    assert not report.passed
+    assert "drift.psi" in _check_ids(report)