From 855b1617504eff12c323e493152e72a5b7f935bb Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:52:18 +0530
Subject: [PATCH 1/5] fix(privacy): keep pd.NA and NaT cells missing when
 masking

The rule path and detection path of anonymize(), detect_pii() and
apply_privacy_policy() treated a cell as missing only when it was None
or a float NaN. pd.NA (nullable string, Int64 and boolean columns) and
NaT fell through that check, were stringified to "<NA>"/"NaT" and then
redacted, tokenized or pseudonymised like real values. Missing cells
came back as placeholders or tokens and cells_changed counted them.

Add _is_missing_scalar(), which accepts None and any scalar that
pd.isna() reports as missing, and use it for the four null checks.
Missing cells are now passed through unchanged before masking; the
masking functions themselves are unchanged, so non-null output is
identical to before.

Tests cover string, Int64, boolean, datetime64[ns] and object columns
holding pd.NA across every rule strategy, the tokenize, pseudonymize,
redact and quarantine policy actions, the detection path and the three
reproductions from the issue.

Closes #243
---
 src/freshdata/enterprise/privacy.py        |  16 ++-
 src/freshdata/enterprise/privacy_policy.py |   3 +-
 tests/test_privacy_missing_and_labels.py   | 152 +++++++++++++++++++++
 3 files changed, 167 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_privacy_missing_and_labels.py

diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py
index b790fdde..969194a1 100644
--- a/src/freshdata/enterprise/privacy.py
+++ b/src/freshdata/enterprise/privacy.py
@@ -55,6 +55,16 @@
 _PREVIEW_LEN = 24
 
 
+def _is_missing_scalar(value: Any) -> bool:
+    """True for ``None`` and any scalar missing marker (``NaN``, ``pd.NA``, ``NaT``).
+
+    Nullable ``string``/``Int64``/``boolean`` columns hold ``pd.NA`` and datetime
+    columns hold ``NaT``; both must be passed through like ``None`` rather than
+    stringified to ``"<NA>"``/``"NaT"`` and masked as if they were values.
+    """
+    return value is None or (pd.api.types.is_scalar(value) and bool(pd.isna(value)))
+
+
 # =====================================================================
 # Entity patterns, context keywords, and HIPAA/GDPR maps
 # =====================================================================
@@ -479,7 +489,7 @@ def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanR
             continue
         scanned.append(str(col))
         for row, value in series.items():
-            if value is None or (isinstance(value, float) and pd.isna(value)):
+            if _is_missing_scalar(value):
                 continue
             text = str(value)
             cell_entities = detect_in_text(text, column=str(col), config=cfg)
@@ -1083,7 +1093,7 @@ def _apply_rule_column(
     n_changed = 0
     new_values: list[Any] = []
     for row, value in series.items():
-        if value is None or (isinstance(value, float) and pd.isna(value)):
+        if _is_missing_scalar(value):
             new_values.append(value)
             continue
         original = str(value)
@@ -1152,7 +1162,7 @@ def _anonymize_detected(
         touched = False
         new_values: list[Any] = []
         for row, value in series.items():
-            if value is None or (isinstance(value, float) and pd.isna(value)):
+            if _is_missing_scalar(value):
                 new_values.append(value)
                 continue
             text = str(value)
diff --git a/src/freshdata/enterprise/privacy_policy.py b/src/freshdata/enterprise/privacy_policy.py
index 304ea51f..20b38e10 100644
--- a/src/freshdata/enterprise/privacy_policy.py
+++ b/src/freshdata/enterprise/privacy_policy.py
@@ -48,6 +48,7 @@
     MaskingEvent,
     PrivacyReport,
     TokenVault,
+    _is_missing_scalar,
     _luhn_ok,
     detect_in_text,
     detokenize_value,
@@ -688,7 +689,7 @@ def apply_privacy_policy(
                 used_vault = tok_vault
                 reversible = rule is None or rule.reversible
             for value in series:
-                if value is None or (isinstance(value, float) and pd.isna(value)):
+                if _is_missing_scalar(value):
                     new_values.append(value)
                     continue
                 original = str(value)
diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py
new file mode 100644
index 00000000..f8583970
--- /dev/null
+++ b/tests/test_privacy_missing_and_labels.py
@@ -0,0 +1,152 @@
+"""Privacy engine correctness: missing values, categorical k-anonymity, labels and metadata.
+
+* #243: ``pd.NA``/``NaT`` cells stay missing under every strategy and action.
+"""
+
+from __future__ import annotations
+
+import sys
+
+import pandas as pd
+import pytest
+
+from freshdata.enterprise import (
+    MaskingRule,
+    PIIDetectionConfig,
+    PrivacyPolicy,
+    PrivacyRule,
+    anonymize,
+    apply_privacy_policy,
+    detect_pii,
+    load_compliance_pack,
+)
+
+KEY = "unit-test-key"
+
+
+@pytest.fixture
+def no_pyffx(monkeypatch):
+    """Force the surrogate fallback of ``fpe`` whether or not pyffx is installed."""
+    monkeypatch.setitem(sys.modules, "pyffx", None)
+
+
+# --------------------------------------------------------------------------
+# #243: missing cells are passed through, not masked
+# --------------------------------------------------------------------------
+
+#: dtype label -> column with a missing cell at position 1 and two real values.
+_NULLABLE_COLUMNS = {
+    "string": lambda: pd.array(["alice@example.com", None, "bob@example.com"], dtype="string"),
+    "Int64": lambda: pd.array([1234567890, None, 9876543210], dtype="Int64"),
+    "boolean": lambda: pd.array([True, None, False], dtype="boolean"),
+    "datetime64[ns]": lambda: pd.to_datetime(["2024-01-02", None, "2024-03-04"]),
+    "object_pd_NA": lambda: pd.Series(
+        ["alice@example.com", pd.NA, "bob@example.com"], dtype=object
+    ),
+}
+
+_RULES = {
+    "hash": MaskingRule(name="r", columns=("c",), strategy="hash", salt="fixed-salt"),
+    "redact": MaskingRule(name="r", columns=("c",), strategy="redact"),
+    "partial": MaskingRule(name="r", columns=("c",), strategy="partial"),
+    "regex_scrub": MaskingRule(
+        name="r", columns=("c",), strategy="regex_scrub", scrub_patterns=(), regexes=(r".+",)
+    ),
+    "tokenize": MaskingRule(name="r", columns=("c",), strategy="tokenize", key=KEY),
+    "surrogate": MaskingRule(name="r", columns=("c",), strategy="surrogate"),
+    "fpe": MaskingRule(name="r", columns=("c",), strategy="fpe", key=KEY),
+}
+
+
+@pytest.mark.parametrize("dtype", list(_NULLABLE_COLUMNS))
+@pytest.mark.parametrize("strategy", list(_RULES))
+def test_rule_strategies_keep_missing_cells_missing(dtype, strategy, no_pyffx):
+    df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype]()})
+    out, report = anonymize(df, rules=(_RULES[strategy],))
+    assert pd.isna(out["c"].iloc[1])
+    assert not pd.isna(out["c"].iloc[0]) and not pd.isna(out["c"].iloc[2])
+    assert report.cells_changed == 2
+    assert sorted(e.row for e in report.events) == [0, 2]
+
+
+@pytest.mark.parametrize("dtype", list(_NULLABLE_COLUMNS))
+def test_drop_counts_only_present_cells(dtype):
+    df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype](), "keep": [1, 2, 3]})
+    out, report = anonymize(df, rules=(MaskingRule(name="r", columns=("c",), strategy="drop"),))
+    assert "c" not in out.columns
+    assert report.cells_changed == 2
+
+
+_POLICY_CASES = [
+    (dtype, action)
+    for action in ("tokenize", "pseudonymize", "redact", "quarantine")
+    for dtype in _NULLABLE_COLUMNS
+    # quarantine writes a string placeholder with Series.where, which masked
+    # Int64/boolean arrays reject; that is separate from missing-value handling.
+    if not (action == "quarantine" and dtype in ("Int64", "boolean"))
+]
+
+
+@pytest.mark.parametrize(("dtype", "action"), _POLICY_CASES)
+def test_policy_actions_keep_missing_cells_missing(dtype, action, no_pyffx):
+    df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype]()})
+    rule = PrivacyRule(id="r", action=action, columns=("c",), classification="s")
+    out, report = apply_privacy_policy(df, PrivacyPolicy(name="p", rules=(rule,), key=KEY))
+    assert pd.isna(out["c"].iloc[1])
+    assert not pd.isna(out["c"].iloc[0]) and not pd.isna(out["c"].iloc[2])
+    assert report.cells_changed == 2
+
+
+@pytest.mark.parametrize("dtype", ["string", "object_pd_NA"])
+def test_detection_path_keeps_missing_cells_missing(dtype):
+    values = ["call 123-45-6789 today", None, "ssn 987-65-4321"]
+    series = (
+        pd.array(values, dtype="string")
+        if dtype == "string"
+        else pd.Series([values[0], pd.NA, values[2]], dtype=object)
+    )
+    df = pd.DataFrame({"notes": series})
+    out, report = anonymize(df, detection_config=PIIDetectionConfig())
+    assert pd.isna(out["notes"].iloc[1])
+    assert "<SSN>" in out["notes"].iloc[0] and "<SSN>" in out["notes"].iloc[2]
+    assert report.entities_found == 2
+    scan = detect_pii(df)
+    assert sorted(e.metadata["row"] for e in scan.entities) == [0, 2]
+
+
+def test_missing_masked_output_matches_object_none_control():
+    rule = MaskingRule(name="h", columns=("c",), strategy="hash", salt="fixed-salt")
+    nullable, _ = anonymize(
+        pd.DataFrame({"c": pd.array(["a@b.com", None], dtype="string")}), rules=(rule,)
+    )
+    control, _ = anonymize(pd.DataFrame({"c": ["a@b.com", None]}), rules=(rule,))
+    assert nullable["c"].iloc[0] == control["c"].iloc[0]
+
+
+def test_issue_243_redact_repro():
+    df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")})
+    out, report = anonymize(
+        df, rules=(MaskingRule(name="r", columns=("email",), strategy="redact"),)
+    )
+    assert out["email"].iloc[0] == "***"
+    assert pd.isna(out["email"].iloc[1])
+    assert report.cells_changed == 1
+
+
+def test_issue_243_tokenize_repro():
+    df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")})
+    out, report = anonymize(
+        df, rules=(MaskingRule(name="t", columns=("email",), strategy="tokenize", key="k"),)
+    )
+    assert out["email"].iloc[0].startswith("tok_")
+    assert pd.isna(out["email"].iloc[1])
+    assert report.cells_changed == 1
+
+
+def test_issue_243_policy_repro():
+    df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")})
+    policy = PrivacyPolicy(packs=(load_compliance_pack("hipaa"),), jurisdiction="US")
+    out, report = apply_privacy_policy(df, policy)
+    assert out["email"].iloc[0] == "<EMAIL>"
+    assert pd.isna(out["email"].iloc[1])
+    assert report.cells_changed == 1

From 5c8fe0f5f56934bf7fba3815db53c0431147f6f6 Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:53:52 +0530
Subject: [PATCH 2/5] fix(privacy): ignore empty category combinations in
 check_k_anonymity

check_k_anonymity grouped rows with groupby(..., dropna=False).size()
and the default observed=False. For categorical quasi-identifiers pandas
then emits every combination of categories, including combinations with
no rows. Those size-0 groups set smallest_class_size to 0, inflated
n_equivalence_classes, turned ok to False and were listed in
high_risk_groups, while the same data as object dtype passed. The same
report feeds clean_enterprise with KAnonymityConfig.

Group on the quasi-identifier columns with categorical columns cast to
object, so only observed value combinations form groups, and drop any
size-0 group. observed=True is not used because it mishandles
dropna=False on pandas 1.5. Missing quasi-identifier values still form
their own class.

Tests cover the reproduction, a categorical quasi-identifier holding
NaN, unused categories, mixed categorical and object quasi-identifiers,
and clean_enterprise with KAnonymityConfig.

Closes #244
---
 src/freshdata/enterprise/privacy.py      | 12 +++-
 tests/test_privacy_missing_and_labels.py | 75 ++++++++++++++++++++++++
 2 files changed, 86 insertions(+), 1 deletion(-)

diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py
index 969194a1..79026aff 100644
--- a/src/freshdata/enterprise/privacy.py
+++ b/src/freshdata/enterprise/privacy.py
@@ -1258,7 +1258,17 @@ def check_k_anonymity(
         raise ValueError("check_k_anonymity requires at least one quasi-identifier")
 
     n_rows = len(frame)
-    sizes = frame.groupby(list(quasi_identifiers), dropna=False).size()
+    # Categorical keys would make groupby emit every category combination,
+    # including empty ones. Group on the observed values instead; observed=True
+    # is avoided because it mishandles dropna=False on pandas 1.5.
+    keys = frame[list(quasi_identifiers)]
+    categorical = {
+        col: object for col, dtype in keys.dtypes.items() if isinstance(dtype, pd.CategoricalDtype)
+    }
+    if categorical:
+        keys = keys.astype(categorical)
+    sizes = keys.groupby(list(quasi_identifiers), dropna=False).size()
+    sizes = sizes[sizes > 0]
     n_classes = int(len(sizes))
     smallest = int(sizes.min()) if n_classes else 0
     violating_mask = sizes < k
diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py
index f8583970..6368b49a 100644
--- a/tests/test_privacy_missing_and_labels.py
+++ b/tests/test_privacy_missing_and_labels.py
@@ -1,22 +1,28 @@
 """Privacy engine correctness: missing values, categorical k-anonymity, labels and metadata.
 
 * #243: ``pd.NA``/``NaT`` cells stay missing under every strategy and action.
+* #244: categorical quasi-identifiers do not produce empty equivalence classes.
 """
 
 from __future__ import annotations
 
 import sys
 
+import numpy as np
 import pandas as pd
 import pytest
 
 from freshdata.enterprise import (
+    EnterpriseConfig,
+    KAnonymityConfig,
     MaskingRule,
     PIIDetectionConfig,
     PrivacyPolicy,
     PrivacyRule,
     anonymize,
     apply_privacy_policy,
+    check_k_anonymity,
+    clean_enterprise,
     detect_pii,
     load_compliance_pack,
 )
@@ -150,3 +156,72 @@ def test_issue_243_policy_repro():
     assert out["email"].iloc[0] == "<EMAIL>"
     assert pd.isna(out["email"].iloc[1])
     assert report.cells_changed == 1
+
+
+# --------------------------------------------------------------------------
+# #244: k-anonymity with categorical quasi-identifiers
+# --------------------------------------------------------------------------
+
+
+def _zip_sex_frame() -> pd.DataFrame:
+    return pd.DataFrame({"zip": ["10001"] * 5 + ["20002"] * 5, "sex": ["F"] * 5 + ["M"] * 5})
+
+
+def test_issue_244_categorical_matches_object():
+    df = _zip_sex_frame()
+    obj = check_k_anonymity(df, ["zip", "sex"], k=5)
+    cat = check_k_anonymity(df.astype("category"), ["zip", "sex"], k=5)
+    for report in (obj, cat):
+        assert report.ok
+        assert report.smallest_class_size == 5
+        assert report.n_equivalence_classes == 2
+        assert report.rows_violating_k == 0
+        assert report.high_risk_groups == []
+    assert cat.to_dict() == obj.to_dict()
+
+
+def test_categorical_quasi_identifier_with_missing_value():
+    df = pd.DataFrame({"zip": ["10001", "10001", np.nan, "20002"], "sex": ["F", "F", "M", "M"]})
+    obj = check_k_anonymity(df, ["zip", "sex"], k=2)
+    cat = check_k_anonymity(df.astype("category"), ["zip", "sex"], k=2)
+    assert cat.n_equivalence_classes == obj.n_equivalence_classes == 3
+    assert cat.smallest_class_size == obj.smallest_class_size == 1
+    assert cat.rows_violating_k == obj.rows_violating_k == 2
+    assert {"zip": None, "sex": "M"} in [g["group"] for g in cat.high_risk_groups]
+    assert all(g["size"] > 0 for g in cat.high_risk_groups)
+
+
+def test_unused_categories_are_not_classes():
+    df = pd.DataFrame(
+        {
+            "zip": pd.Categorical(["10001"] * 3, categories=["10001", "20002", "30003"]),
+            "sex": pd.Categorical(["F"] * 3, categories=["F", "M", "X"]),
+        }
+    )
+    report = check_k_anonymity(df, ["zip", "sex"], k=3)
+    assert report.ok
+    assert report.n_equivalence_classes == 1
+    assert report.smallest_class_size == 3
+
+
+def test_mixed_categorical_and_object_quasi_identifiers():
+    df = _zip_sex_frame()
+    df["zip"] = df["zip"].astype("category")
+    report = check_k_anonymity(df, ["zip", "sex"], k=5)
+    assert report.ok
+    assert report.n_equivalence_classes == 2
+    assert report.smallest_class_size == 5
+
+
+def test_clean_enterprise_k_anonymity_with_categorical_quasi_identifiers():
+    df = _zip_sex_frame().astype("category")
+    df["value"] = range(len(df))
+    config = EnterpriseConfig(
+        k_anonymity=KAnonymityConfig(enabled=True, quasi_identifiers=("zip", "sex"), k=5)
+    )
+    result = clean_enterprise(df, enterprise=config)
+    report = result.k_anonymity_report
+    assert report is not None
+    assert report.ok
+    assert report.n_equivalence_classes == 2
+    assert report.smallest_class_size == 5

From 5eeb04763a1a9b71becf0ce2acfff7c6bc1283c6 Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:55:55 +0530
Subject: [PATCH 3/5] fix(privacy): reject duplicate column labels in
 detect_pii and anonymize

detect_pii and the detection pass of anonymize loop over frame.columns
and read frame[col].dtype. For a duplicated label frame[col] returns a
DataFrame rather than a Series, so both functions failed with an
unhelpful AttributeError. A masking rule that selected a duplicated
label had the same problem, and _resolve_columns listed the label once
per occurrence.

Raise a ValueError naming the duplicated labels instead:

- detect_pii raises up front when any label is duplicated.
- anonymize raises when detection is enabled and any label is
  duplicated, or when a rule resolves to a duplicated label. The check
  runs before any rule is applied.

A rule that targets a unique column still works on a frame that has
duplicate labels elsewhere, as long as detection is not enabled.

Tests cover both reproductions, a rule aimed at a duplicated label, and
rules on a unique column next to duplicated ones with detection absent
or disabled.

Refs #265
---
 src/freshdata/enterprise/privacy.py      | 29 +++++++++++++++
 tests/test_privacy_missing_and_labels.py | 45 ++++++++++++++++++++++++
 2 files changed, 74 insertions(+)

diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py
index 79026aff..8a074c99 100644
--- a/src/freshdata/enterprise/privacy.py
+++ b/src/freshdata/enterprise/privacy.py
@@ -473,14 +473,27 @@ def _get_presidio_analyzer() -> Any:  # pragma: no cover - requires optional Pre
     return _PRESIDIO_ANALYZER
 
 
+def _duplicated_labels(frame: pd.DataFrame) -> list[Any]:
+    """Column labels that occur more than once, in first-seen order."""
+    return list(dict.fromkeys(frame.columns[frame.columns.duplicated()]))
+
+
 def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanReport:
     """Scan the text columns of *df* for PII; return a :class:`PIIScanReport`.
 
     Read-only. Only object/string columns are scanned. Raw matched substrings
     are redacted in the report unless ``config.redact_samples=False``.
+
+    Raises :class:`ValueError` when *df* has duplicate column labels, because a
+    duplicated label does not identify a single column to scan.
     """
     cfg = config or PIIDetectionConfig()
     frame = to_pandas(df)
+    duplicated = _duplicated_labels(frame)
+    if duplicated:
+        raise ValueError(
+            f"detect_pii requires unique column labels; duplicated: {duplicated}"
+        )
     entities: list[PIIEntity] = []
     scanned: list[str] = []
     for col in frame.columns:
@@ -965,6 +978,22 @@ def anonymize(
             "detection_config=PIIDetectionConfig() to say what to mask."
         )
     frame = to_pandas(df).copy()
+    duplicated = _duplicated_labels(frame)
+    if duplicated:
+        # A duplicated label selects several columns at once, so neither the
+        # detection pass nor a rule aimed at it can address a single column.
+        if detection_config is not None and detection_config.enabled:
+            raise ValueError(
+                "anonymize requires unique column labels for PII detection; "
+                f"duplicated: {duplicated}"
+            )
+        for rule in rules:
+            targeted = _resolve_columns(rule, duplicated)
+            if targeted:
+                raise ValueError(
+                    f"anonymize requires unique column labels; rule {rule.name!r} "
+                    f"targets duplicated: {targeted}"
+                )
     events: list[MaskingEvent] = []
     changed_cols: list[str] = []
     cells_changed = 0
diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py
index 6368b49a..bbd18180 100644
--- a/tests/test_privacy_missing_and_labels.py
+++ b/tests/test_privacy_missing_and_labels.py
@@ -2,6 +2,7 @@
 
 * #243: ``pd.NA``/``NaT`` cells stay missing under every strategy and action.
 * #244: categorical quasi-identifiers do not produce empty equivalence classes.
+* #265: duplicate column labels raise a clear ``ValueError``.
 """
 
 from __future__ import annotations
@@ -225,3 +226,47 @@ def test_clean_enterprise_k_anonymity_with_categorical_quasi_identifiers():
     assert report.ok
     assert report.n_equivalence_classes == 2
     assert report.smallest_class_size == 5
+
+
+# --------------------------------------------------------------------------
+# #265 part 1: duplicate column labels in detect_pii / anonymize
+# --------------------------------------------------------------------------
+
+
+def _duplicate_email_frame() -> pd.DataFrame:
+    return pd.DataFrame([["a@b.com", "c@d.com"]], columns=["email", "email"])
+
+
+def test_issue_265_detect_pii_rejects_duplicate_labels():
+    with pytest.raises(ValueError, match=r"detect_pii requires unique column labels.*'email'"):
+        detect_pii(_duplicate_email_frame())
+
+
+def test_issue_265_anonymize_detection_rejects_duplicate_labels():
+    df = _duplicate_email_frame()
+    with pytest.raises(ValueError, match=r"requires unique column labels.*'email'"):
+        anonymize(df, detection_config=PIIDetectionConfig())
+    assert df.iloc[0].tolist() == ["a@b.com", "c@d.com"]
+
+
+def test_anonymize_rule_targeting_duplicated_label_raises():
+    df = _duplicate_email_frame()
+    rule = MaskingRule(name="r", columns=("email",), strategy="redact")
+    with pytest.raises(ValueError, match=r"rule 'r'.*'email'"):
+        anonymize(df, rules=(rule,))
+
+
+def test_anonymize_rule_on_unique_column_with_duplicates_elsewhere():
+    df = pd.DataFrame([["x", "y", "a@b.com"]], columns=["dup", "dup", "email"])
+    rule = MaskingRule(name="r", columns=("email",), strategy="redact")
+    out, report = anonymize(df, rules=(rule,))
+    assert out.columns.tolist() == ["dup", "dup", "email"]
+    assert out.iloc[0].tolist() == ["x", "y", "***"]
+    assert report.cells_changed == 1
+
+
+def test_anonymize_disabled_detection_ignores_duplicates_elsewhere():
+    df = pd.DataFrame([["x", "y", "a@b.com"]], columns=["dup", "dup", "email"])
+    rule = MaskingRule(name="r", columns=("email",), strategy="redact")
+    out, _ = anonymize(df, rules=(rule,), detection_config=PIIDetectionConfig(enabled=False))
+    assert out.iloc[0].tolist() == ["x", "y", "***"]

From 082f1542215ae4b6bef96aa1266f5dca71a99014 Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:57:57 +0530
Subject: [PATCH 4/5] fix(privacy): report fpe reversibility and mode from what
 each cell used

anonymize() computed MaskingEvent.reversible from the rule alone, so an
fpe rule with reversible=True marked every event reversible even when
pyffx was unavailable and the cell was masked with the one-way surrogate
fallback. metadata["fpe_mode"] was also overwritten per cell, so a
column mixing real FPE and the surrogate fallback reported only the
mode of the last cell processed.

_apply_rule_column now sets each event's reversible flag from the mode
_mask_one returned for that cell: true only for tokenize, or for fpe
when the cell used crypto_fpe, and only when the rule asked for it. It
returns per-mode cell counts instead of the last mode. anonymize()
aggregates those counts: a single mode is still reported as the plain
fpe_mode string, and more than one mode is reported as
fpe_mode="mixed" with fpe_modes={column: {mode: count}}. Masked values
are unchanged.

Tests use pyffx=None and a stub pyffx that cannot encrypt some values to
cover the fallback, mixed modes within a column and across columns,
reversible tokenize, and a single-mode report that matches the previous
output.

Refs #281
---
 src/freshdata/enterprise/privacy.py      |  36 ++++++--
 tests/test_privacy_missing_and_labels.py | 103 +++++++++++++++++++++++
 2 files changed, 130 insertions(+), 9 deletions(-)

diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py
index 8a074c99..daaa4ab1 100644
--- a/src/freshdata/enterprise/privacy.py
+++ b/src/freshdata/enterprise/privacy.py
@@ -999,20 +999,32 @@ def anonymize(
     cells_changed = 0
     metadata: dict[str, Any] = {}
 
+    fpe_modes: dict[str, dict[str, int]] = {}
     for rule in rules:
         key = _resolve_key(rule)
         vault = _vault_for(rule)
         for column in _resolve_columns(rule, list(frame.columns)):
             if column not in frame.columns:
                 continue
-            n, fpe_mode = _apply_rule_column(
+            n, mode_counts = _apply_rule_column(
                 frame, column, rule, key, vault, events, audit_include_pii
             )
             if n:
                 cells_changed += n
                 changed_cols.append(str(column))
-            if fpe_mode:
-                metadata["fpe_mode"] = fpe_mode
+            if mode_counts:
+                per_column = fpe_modes.setdefault(str(column), {})
+                for mode, count in mode_counts.items():
+                    per_column[mode] = per_column.get(mode, 0) + count
+
+    # One mode overall keeps the plain mode string; a mix of modes (per cell,
+    # column or rule) is reported as "mixed" with per-column cell counts.
+    modes_used = {mode for per_column in fpe_modes.values() for mode in per_column}
+    if len(modes_used) == 1:
+        metadata["fpe_mode"] = next(iter(modes_used))
+    elif modes_used:
+        metadata["fpe_mode"] = "mixed"
+        metadata["fpe_modes"] = fpe_modes
 
     entities_found = 0
     if detection_config is not None and detection_config.enabled:
@@ -1098,7 +1110,8 @@ def _apply_rule_column(
     vault: TokenVault,
     events: list[MaskingEvent],
     include_pii: bool,
-) -> tuple[int, str | None]:
+) -> tuple[int, dict[str, int]]:
+    """Mask one column in place; return ``(cells_changed, {fpe_mode: cell_count})``."""
     if rule.strategy == "drop":
         n = int(frame[column].notna().sum())
         _record_event(
@@ -1107,16 +1120,15 @@ def _apply_rule_column(
             source="column", original="", masked="<dropped>", include_pii=include_pii,
         )
         frame.drop(columns=[column], inplace=True)
-        return n, None
+        return n, {}
 
-    reversible = rule.strategy in ("tokenize", "fpe") and rule.reversible
     format_preserving = rule.strategy in ("fpe", "surrogate") or rule.preserve_format
     if rule.strategy in ("tokenize", "fpe") and rule.reversible and not key:
         raise ValueError(
             f"masking rule {rule.name!r}: reversible {rule.strategy} requires key= or key_env="
         )
 
-    fpe_mode: str | None = None
+    mode_counts: dict[str, int] = {}
     series = frame[column]
     entity_type = _entity_for_rule(rule, str(column))
     n_changed = 0
@@ -1128,10 +1140,16 @@ def _apply_rule_column(
         original = str(value)
         masked, mode = _mask_one(original, rule, key, vault)
         if mode:
-            fpe_mode = mode
+            mode_counts[mode] = mode_counts.get(mode, 0) + 1
         new_values.append(masked)
         if masked != original:
             n_changed += 1
+            # Only a vault token or real FPE can be reversed; the surrogate
+            # fallback of ``fpe`` is one-way, whatever the rule asked for.
+            reversible = rule.reversible and (
+                rule.strategy == "tokenize"
+                or (rule.strategy == "fpe" and mode == "crypto_fpe")
+            )
             _record_event(
                 events, column=str(column), row=row, entity_type=entity_type, rule=rule,
                 strategy=rule.strategy, reversible=reversible,
@@ -1139,7 +1157,7 @@ def _apply_rule_column(
                 original=original, masked=masked, include_pii=include_pii,
             )
     frame[column] = pd.Series(new_values, index=series.index)
-    return n_changed, fpe_mode
+    return n_changed, mode_counts
 
 
 def _mask_one(
diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py
index bbd18180..6873d3c7 100644
--- a/tests/test_privacy_missing_and_labels.py
+++ b/tests/test_privacy_missing_and_labels.py
@@ -3,11 +3,13 @@
 * #243: ``pd.NA``/``NaT`` cells stay missing under every strategy and action.
 * #244: categorical quasi-identifiers do not produce empty equivalence classes.
 * #265: duplicate column labels raise a clear ``ValueError``.
+* #281: fpe audit metadata follows the mode each cell actually used.
 """
 
 from __future__ import annotations
 
 import sys
+import types
 
 import numpy as np
 import pandas as pd
@@ -270,3 +272,104 @@ def test_anonymize_disabled_detection_ignores_duplicates_elsewhere():
     rule = MaskingRule(name="r", columns=("email",), strategy="redact")
     out, _ = anonymize(df, rules=(rule,), detection_config=PIIDetectionConfig(enabled=False))
     assert out.iloc[0].tolist() == ["x", "y", "***"]
+
+
+# --------------------------------------------------------------------------
+# #281 parts 1-2: fpe audit metadata follows the mode actually used
+# --------------------------------------------------------------------------
+
+_SURROGATE_MODE = "surrogate_format_preserving_not_crypto_fpe"
+
+
+class _StubInteger:
+    """Stand-in for ``pyffx.Integer`` that cannot encrypt the number 999."""
+
+    def __init__(self, key, length):
+        self.length = length
+
+    def encrypt(self, n):
+        if n == 999:
+            raise ValueError("stub cannot encrypt 999")
+        return (n * 7 + 3) % (10**self.length)
+
+
+@pytest.fixture
+def stub_pyffx(monkeypatch):
+    module = types.ModuleType("pyffx")
+    module.Integer = _StubInteger
+    monkeypatch.setitem(sys.modules, "pyffx", module)
+
+
+def test_issue_281_surrogate_fallback_is_not_reported_reversible(no_pyffx):
+    df = pd.DataFrame({"ssn": ["123-45-6789"]})
+    rule = MaskingRule(name="r", columns=("ssn",), strategy="fpe", reversible=True, key="k")
+    out, report = anonymize(df, rules=(rule,))
+    assert out["ssn"].iloc[0] == "119-85-2634"
+    assert [e.reversible for e in report.events] == [False]
+    assert report.metadata == {"fpe_mode": _SURROGATE_MODE}
+
+
+def test_issue_281_mixed_modes_in_one_column_are_counted(stub_pyffx):
+    df = pd.DataFrame({"x": ["abc-def", "123-45-6789", "999"]})
+    rule = MaskingRule(name="r", columns=("x",), strategy="fpe", reversible=True, key="k")
+    out, report = anonymize(df, rules=(rule,))
+    assert out["x"].iloc[0] == "xhs-rps"
+    assert out["x"].iloc[1] == "864-19-7526"
+    assert report.metadata["fpe_mode"] == "mixed"
+    assert report.metadata["fpe_modes"] == {"x": {_SURROGATE_MODE: 2, "crypto_fpe": 1}}
+    assert {e.row: e.reversible for e in report.events} == {0: False, 1: True, 2: False}
+    assert "fpe_mode: mixed" in report.summary()
+
+
+def test_mixed_modes_across_columns_are_counted_per_column(stub_pyffx):
+    df = pd.DataFrame({"a": ["123-45-6789"], "b": ["123-45-6789"]})
+    rules = (
+        MaskingRule(name="fa", columns=("a",), strategy="fpe", key="k"),
+        MaskingRule(name="sb", columns=("b",), strategy="surrogate"),
+    )
+    _, report = anonymize(df, rules=rules)
+    assert report.metadata == {
+        "fpe_mode": "mixed",
+        "fpe_modes": {"a": {"crypto_fpe": 1}, "b": {_SURROGATE_MODE: 1}},
+    }
+
+
+def test_crypto_fpe_single_mode_reports_mode_string(stub_pyffx):
+    df = pd.DataFrame({"x": ["123-45-6789", "555-12-3456"]})
+    rule = MaskingRule(name="r", columns=("x",), strategy="fpe", key="k")
+    _, report = anonymize(df, rules=(rule,))
+    assert report.metadata == {"fpe_mode": "crypto_fpe"}
+    assert [e.reversible for e in report.events] == [False, False]
+
+
+def test_reversible_tokenize_events_stay_reversible():
+    df = pd.DataFrame({"email": ["a@b.com", "c@d.com"]})
+    rule = MaskingRule(name="t", columns=("email",), strategy="tokenize", reversible=True, key=KEY)
+    _, report = anonymize(df, rules=(rule,))
+    assert [e.reversible for e in report.events] == [True, True]
+    assert "fpe_mode" not in report.metadata
+
+
+def test_single_mode_report_is_unchanged(no_pyffx):
+    df = pd.DataFrame({"ssn": ["123-45-6789", "987-65-4321"], "acct": ["1234567890", None]})
+    rules = (
+        MaskingRule(
+            name="s", columns=("ssn",), strategy="surrogate", preserve_format=True, visible=4
+        ),
+        MaskingRule(name="f", columns=("acct",), strategy="fpe", key="K", preserve_format=True),
+        MaskingRule(name="d", columns=("ssn",), strategy="drop"),
+    )
+    _, report = anonymize(df, rules=rules)
+    assert report.metadata == {"fpe_mode": _SURROGATE_MODE}
+    assert report.summary().endswith(f"fpe_mode: {_SURROGATE_MODE}")
+    assert all(e.reversible is False for e in report.events)
+
+
+def test_rules_without_fpe_add_no_fpe_metadata():
+    df = pd.DataFrame({"a": ["x"], "b": ["y"]})
+    rules = (
+        MaskingRule(name="r", columns=("a",), strategy="redact"),
+        MaskingRule(name="d", columns=("b",), strategy="drop"),
+    )
+    _, report = anonymize(df, rules=rules)
+    assert report.metadata == {}

From 86cdf2ab7b67c9ab95ea3ded68616e47aab48568 Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 16:03:50 +0530
Subject: [PATCH 5/5] fix(privacy): report whether detect_pii's NER pass
 actually ran

detect_pii(config=PIIDetectionConfig(use_ner=True)) wrote
metadata={"ner": True} from the config flag alone. When
presidio_analyzer was importable but AnalyzerEngine() raised (for
example a missing language model), _get_presidio_analyzer swallowed the
exception without caching it. The NER pass then contributed nothing,
the report still claimed NER ran, no warning was emitted, and the
engine constructor was retried for every cell.

_get_presidio_analyzer now records the first failure as
"ExceptionType: message" in _PRESIDIO_ERROR and does not retry it.
detect_pii resolves the analyzer once per call when use_ner is set,
skips the per-cell NER pass when it is unavailable and emits one
UserWarning naming the error. The metadata reports the real state:
ner (now the active flag), ner_requested, ner_active, and ner_error
when NER was requested but could not run. The PIIDetectionConfig
docstring no longer says the pass is skipped silently.

Tests stub presidio_analyzer through sys.modules: an engine that fails
to start (one warning, correct metadata, constructor called once across
cells and calls), a working engine returning no results (ner_active
True), a missing package (import error recorded), and NER not requested.

Closes #282
---
 src/freshdata/enterprise/config.py       |  5 +-
 src/freshdata/enterprise/privacy.py      | 41 +++++++++--
 tests/test_privacy_missing_and_labels.py | 92 ++++++++++++++++++++++++
 3 files changed, 130 insertions(+), 8 deletions(-)

diff --git a/src/freshdata/enterprise/config.py b/src/freshdata/enterprise/config.py
index 433fcd7e..4d9762c8 100644
--- a/src/freshdata/enterprise/config.py
+++ b/src/freshdata/enterprise/config.py
@@ -294,7 +294,10 @@ class PIIDetectionConfig:
     The fallback detector (regex + context keywords) needs no extra
     dependencies. When ``use_ner`` is set and the optional
     ``freshdata-cleaner[privacy]`` extra (Presidio) is installed, an NER pass is
-    layered on top; otherwise it is skipped silently.
+    layered on top. If Presidio is not installed or its analyzer fails to start,
+    :func:`~freshdata.enterprise.detect_pii` emits one ``UserWarning``, runs the
+    regex/context detector alone and records ``ner_active=False`` plus the
+    ``ner_error`` in the report metadata.
     """
 
     enabled: bool = True
diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py
index daaa4ab1..c3f60eee 100644
--- a/src/freshdata/enterprise/privacy.py
+++ b/src/freshdata/enterprise/privacy.py
@@ -34,6 +34,7 @@
 import json
 import os
 import re
+import warnings
 from abc import ABC, abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass, field
@@ -459,17 +460,24 @@ def _ner_entities(
 
 
 _PRESIDIO_ANALYZER: Any = None
+#: ``"ExceptionType: message"`` from the first failed Presidio start-up. Cached
+#: so a missing package or language model is not retried for every cell.
+_PRESIDIO_ERROR: str | None = None
 
 
-def _get_presidio_analyzer() -> Any:  # pragma: no cover - requires optional Presidio
-    global _PRESIDIO_ANALYZER
-    if _PRESIDIO_ANALYZER is None:
+def _get_presidio_analyzer() -> Any:
+    """Return the shared Presidio analyzer, or ``None`` when it cannot start.
+
+    The first failure is recorded in :data:`_PRESIDIO_ERROR` and not retried.
+    """
+    global _PRESIDIO_ANALYZER, _PRESIDIO_ERROR
+    if _PRESIDIO_ANALYZER is None and _PRESIDIO_ERROR is None:
         try:
             from presidio_analyzer import AnalyzerEngine
 
             _PRESIDIO_ANALYZER = AnalyzerEngine()
-        except Exception:
-            _PRESIDIO_ANALYZER = None
+        except Exception as exc:
+            _PRESIDIO_ERROR = f"{type(exc).__name__}: {exc}"
     return _PRESIDIO_ANALYZER
 
 
@@ -494,6 +502,18 @@ def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanR
         raise ValueError(
             f"detect_pii requires unique column labels; duplicated: {duplicated}"
         )
+    ner_active = False
+    ner_error: str | None = None
+    if cfg.use_ner:
+        ner_active = _get_presidio_analyzer() is not None
+        if not ner_active:
+            ner_error = _PRESIDIO_ERROR or "presidio analyzer unavailable"
+            warnings.warn(
+                f"detect_pii: use_ner=True but the Presidio NER pass is unavailable "
+                f"({ner_error}); only the regex/context detector ran",
+                UserWarning,
+                stacklevel=2,
+            )
     entities: list[PIIEntity] = []
     scanned: list[str] = []
     for col in frame.columns:
@@ -506,15 +526,22 @@ def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanR
                 continue
             text = str(value)
             cell_entities = detect_in_text(text, column=str(col), config=cfg)
-            if cfg.use_ner:
+            if ner_active:
                 cell_entities = _merge_ner(cell_entities, _ner_entities(text, str(col), cfg))
             for e in cell_entities:
                 e.metadata["row"] = int(row) if isinstance(row, (int, float)) else row
             entities.extend(cell_entities)
+    metadata: dict[str, Any] = {
+        "ner": ner_active,
+        "ner_requested": bool(cfg.use_ner),
+        "ner_active": ner_active,
+    }
+    if ner_error is not None:
+        metadata["ner_error"] = ner_error
     return PIIScanReport(
         entities=entities,
         columns_scanned=tuple(scanned),
-        metadata={"ner": bool(cfg.use_ner)},
+        metadata=metadata,
     )
 
 
diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py
index 6873d3c7..c1dce6cf 100644
--- a/tests/test_privacy_missing_and_labels.py
+++ b/tests/test_privacy_missing_and_labels.py
@@ -4,6 +4,7 @@
 * #244: categorical quasi-identifiers do not produce empty equivalence classes.
 * #265: duplicate column labels raise a clear ``ValueError``.
 * #281: fpe audit metadata follows the mode each cell actually used.
+* #282: detect_pii reports whether the NER pass actually ran.
 """
 
 from __future__ import annotations
@@ -373,3 +374,94 @@ def test_rules_without_fpe_add_no_fpe_metadata():
     )
     _, report = anonymize(df, rules=rules)
     assert report.metadata == {}
+
+
+# --------------------------------------------------------------------------
+# #282: NER metadata when the Presidio analyzer cannot start
+# --------------------------------------------------------------------------
+
+
+@pytest.fixture
+def presidio_globals(monkeypatch):
+    """Start each test with no cached analyzer or start-up error."""
+    monkeypatch.setattr("freshdata.enterprise.privacy._PRESIDIO_ANALYZER", None)
+    monkeypatch.setattr("freshdata.enterprise.privacy._PRESIDIO_ERROR", None)
+
+
+def _install_presidio_stub(monkeypatch, engine_cls) -> None:
+    module = types.ModuleType("presidio_analyzer")
+    module.AnalyzerEngine = engine_cls
+    monkeypatch.setitem(sys.modules, "presidio_analyzer", module)
+
+
+def _freshdata_user_warnings(record) -> list:
+    return [w for w in record if issubclass(w.category, UserWarning)]
+
+
+def test_issue_282_failed_analyzer_start_is_reported(monkeypatch, presidio_globals):
+    constructed: list[int] = []
+
+    class FailingEngine:
+        def __init__(self):
+            constructed.append(1)
+            raise RuntimeError("spaCy model en_core_web_lg not installed")
+
+    _install_presidio_stub(monkeypatch, FailingEngine)
+    df = pd.DataFrame({"t": ["Alice Johnson visited", "Bob wrote to a@b.com", "Carol"]})
+    config = PIIDetectionConfig(use_ner=True)
+
+    with pytest.warns(UserWarning, match="NER pass is unavailable") as record:
+        report = detect_pii(df, config=config)
+    assert len(_freshdata_user_warnings(record)) == 1
+    assert report.metadata == {
+        "ner": False,
+        "ner_requested": True,
+        "ner_active": False,
+        "ner_error": "RuntimeError: spaCy model en_core_web_lg not installed",
+    }
+    assert report.entity_types == {"EMAIL"}
+    assert constructed == [1]
+
+    with pytest.warns(UserWarning, match="spaCy model"):
+        again = detect_pii(df, config=config)
+    assert again.metadata["ner_error"] == report.metadata["ner_error"]
+    assert constructed == [1]
+
+
+def test_working_analyzer_marks_ner_active(monkeypatch, presidio_globals):
+    analyzed: list[str] = []
+
+    class WorkingEngine:
+        def analyze(self, text, language):
+            analyzed.append(text)
+            return []
+
+    _install_presidio_stub(monkeypatch, WorkingEngine)
+    df = pd.DataFrame({"t": ["Alice Johnson visited", None]})
+    report = detect_pii(df, config=PIIDetectionConfig(use_ner=True))
+    assert report.metadata == {"ner": True, "ner_requested": True, "ner_active": True}
+    assert analyzed == ["Alice Johnson visited"]
+
+
+def test_missing_presidio_package_records_import_error(monkeypatch, presidio_globals):
+    monkeypatch.setitem(sys.modules, "presidio_analyzer", None)
+    with pytest.warns(UserWarning, match="presidio_analyzer") as record:
+        report = detect_pii(pd.DataFrame({"t": ["x"]}), config=PIIDetectionConfig(use_ner=True))
+    assert len(_freshdata_user_warnings(record)) == 1
+    assert report.metadata["ner"] is False
+    assert report.metadata["ner_active"] is False
+    assert report.metadata["ner_error"].split(":")[0] in ("ImportError", "ModuleNotFoundError")
+
+
+def test_ner_not_requested_never_starts_the_analyzer(monkeypatch, presidio_globals):
+    constructed: list[int] = []
+
+    class FailingEngine:
+        def __init__(self):
+            constructed.append(1)
+            raise RuntimeError("should not start")
+
+    _install_presidio_stub(monkeypatch, FailingEngine)
+    report = detect_pii(pd.DataFrame({"t": ["a@b.com"]}))
+    assert report.metadata == {"ner": False, "ner_requested": False, "ner_active": False}
+    assert constructed == []