From 391ddf67eaa95ee60388209ec33dc1d04eb85729 Mon Sep 17 00:00:00 2001
From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com>
Date: Tue, 15 Sep 2026 14:31:59 +0530
Subject: [PATCH] fix(freshcore): fall back for missforest, per-column
 strategies, non-finite outliers and nullable bools; honour native duplicate
 counts

The FreshCore adapter ran several configs natively that its v1 kernels
silently get wrong. It now delegates them to the pandas reference through
the existing fallback path, so fallback_policy="error" raises FallbackError
and reports record the event.

- impute="missforest" and per-column impute_strategy: the native imputer
  ignores both, so nothing was filled. _unsupported_reason() now falls back
  with the same reasons as PlanGenerator.fallback_reason(), so fd.plan()
  and the recorded event agree. PlanGenerator now treats an empty
  impute_strategy mapping as a no-op, matching the pandas imputer.
- Outliers on float columns holding +/-inf: the native fences don't
  exclude non-finite values, so zscore and IQR flag or clip nothing. When
  outliers is set, a vectorised np.isinf scan of float columns triggers a
  fallback naming the columns.
- impute="mode"/"auto" with missing values in a nullable boolean column:
  the native kernel skips boolean data. Fall back when such a column has
  some (not all) values missing.
- Duplicate ratio: when the native result carries duplicates_detected, the
  adapter reports it through report_detected_duplicates(), in pandas action
  order, using the row count after empty-row removal. With
  drop_duplicates=True, the native drop count goes through
  check_duplicate_ratio(), so duplicate_ratio_action="error" raises and
  high ratios warn. A missing key skips detection, except under
  duplicate_ratio_action="error", where the run falls back so the
  escalation is not silently lost.

The fallback matrix, backends and FreshCore docs list the new triggers.

Closes #322
Closes #334
Closes #335
Refs #323 (Python adapter only; the native engine still needs to report duplicates)
---
 docs/backends.md                              |   5 +
 docs/fallback-matrix.md                       |   6 +
 docs/freshcore.md                             |   7 +-
 src/freshdata/execution/_plan.py              |   3 +-
 .../execution/backends/_freshcore.py          | 129 ++++++-
 .../test_freshcore_fallbacks.py               | 329 ++++++++++++++++++
 6 files changed, 473 insertions(+), 6 deletions(-)
 create mode 100644 tests/test_execution/test_freshcore_fallbacks.py

diff --git a/docs/backends.md b/docs/backends.md
index a64aec86..c62c0bee 100644
--- a/docs/backends.md
+++ b/docs/backends.md
@@ -161,6 +161,11 @@ Notes:
   match (Polars/DuckDB linear interpolation); Spark may flag a different count.
 - A non-default pandas index (e.g. a `DatetimeIndex`) forces a pandas fallback,
   since native frames carry no index.
+- FreshCore also falls back, based on the data, when `outliers` is set and a
+  float column holds `±inf`, or when `impute="mode"`/`"auto"` would fill a
+  nullable `boolean` column. With `drop_duplicates=False` and
+  `duplicate_ratio_action="error"`, it falls back unless the native module
+  reports a duplicate-row count (see the [fallback matrix](fallback-matrix.md)).
 - FreshCore v1 is a cleaning-first native engine, not an out-of-core engine. It
   supports pandas-compatible materialized outputs and records per-stage timings
   in `report.stage_timings`.
diff --git a/docs/fallback-matrix.md b/docs/fallback-matrix.md
index c39385f8..24566094 100644
--- a/docs/fallback-matrix.md
+++ b/docs/fallback-matrix.md
@@ -11,6 +11,9 @@ This table is transcribed from the single source of truth,
 `PlanGenerator.fallback_reason()` in `src/freshdata/execution/_plan.py`, plus
 the input checks in `pandas_ingest_fallback_reason()`
 (`src/freshdata/execution/_ingest.py`) — if you change either, change this page.
+FreshCore also runs its own config and data checks in
+`FreshCoreEngine._unsupported_reason()`
+(`src/freshdata/execution/backends/_freshcore.py`); keep those rows in sync too.
 
 | Operation / config | polars | duckdb | spark | freshcore | Why the fallback exists |
 |---|---|---|---|---|---|
@@ -20,10 +23,13 @@ the input checks in `pandas_ingest_fallback_reason()`
 | full-row dedup (`keep="first"/"last"`) | native | native | native | native | streaming polars dedup drops row order (disclosed); `streaming_dedup=False` restores it |
 | **subset dedup** (`duplicate_subset=`) | **native** | pandas | pandas | pandas | keep semantics are order-sensitive; Polars reproduces them via order-preserving `unique` (eager, not streaming — disclosed) |
 | dedup `keep="drop"/"aggregate"` | pandas | pandas | pandas | pandas | group-wise resolution isn't expressed natively yet |
+| detection-only dedup (`drop_duplicates=False`) with `duplicate_ratio_action="error"` | native | native | native | pandas unless the native module reports `duplicates_detected` | the escalation needs a duplicate-row count; FreshCore builds that don't report one would never raise |
 | global impute mean/median/mode | native | native | native | native | — |
+| impute `mode`/`auto` with missing values in a nullable `boolean` column | native | native | native | pandas | FreshCore v1 kernels do not impute boolean columns |
 | per-column `impute_strategy` | pandas | pandas | pandas | pandas | unimplemented natively (no fundamental blocker) |
 | `impute="missforest"` | pandas | pandas | pandas | pandas | scikit-learn model |
 | outliers `iqr` / `zscore` | native | native | native | native | — |
+| outliers when a float column holds `±inf` | native | native | native | pandas | FreshCore v1 fences don't exclude non-finite values, so they flag or clip nothing |
 | `outlier_action="auto"` / model methods | pandas | pandas | pandas | pandas | data-dependent / model-based selection |
 | `fix_dtypes=True` (default) | pandas | pandas | pandas | partial | sampled heuristics on the pandas reference; FreshCore casts bool/numeric natively, defers datetimes |
 | `drop_constant_columns` | pandas | pandas | pandas | pandas | needs a data scan before planning (two-phase plan not built) |
diff --git a/docs/freshcore.md b/docs/freshcore.md
index f4b1db1b..a1786b4f 100644
--- a/docs/freshcore.md
+++ b/docs/freshcore.md
@@ -52,7 +52,12 @@ FreshCore v1 runs native kernels for conservative, deterministic cleaning:
 FreshCore falls back for semantic cleaning, context/policy protection, cleaning
 memory, domain packs, contracts, non-default indexes, duplicate subsets,
 aggregate/drop duplicate modes, model-based outliers, constant-column dropping,
-memory downcasting, and the balanced/aggressive decision engine.
+memory downcasting, and the balanced/aggressive decision engine. It also falls
+back for `impute="missforest"`, per-column `impute_strategy`, outlier handling
+on float columns holding `±inf`, and mode/auto imputation of nullable boolean
+columns with missing values. Detection-only dedup under
+`duplicate_ratio_action="error"` falls back unless the native module reports
+`duplicates_detected`.
 
 ## Benchmarking
 
diff --git a/src/freshdata/execution/_plan.py b/src/freshdata/execution/_plan.py
index 2636ced1..dd5e207d 100644
--- a/src/freshdata/execution/_plan.py
+++ b/src/freshdata/execution/_plan.py
@@ -145,7 +145,8 @@ def fallback_reason(self) -> str | None:
             return "optimize_memory downcasting is evaluated by the pandas backend"
         if c.impute == "missforest":
             return "missforest imputation uses scikit-learn and is evaluated by the pandas backend"
-        if c.impute_strategy is not None:
+        if c.impute_strategy:
+            # An empty mapping is a no-op on the pandas reference as well.
             return "impute_strategy per-column overrides are evaluated by the pandas backend"
         if c.outliers is not None and c.outlier_method not in _NATIVE_OUTLIER_METHODS:
             return (
diff --git a/src/freshdata/execution/backends/_freshcore.py b/src/freshdata/execution/backends/_freshcore.py
index 77bcf9ff..26f07f1f 100644
--- a/src/freshdata/execution/backends/_freshcore.py
+++ b/src/freshdata/execution/backends/_freshcore.py
@@ -16,13 +16,15 @@
 import time
 from typing import TYPE_CHECKING, Any
 
+import numpy as np
 import pandas as pd
-from pandas.api.types import is_bool_dtype, is_numeric_dtype, is_object_dtype
+from pandas.api.types import is_bool_dtype, is_float_dtype, is_numeric_dtype, is_object_dtype
 
 from ..._util import memory_bytes
 from ...config import _DEFAULT_FACTOR
 from ...report import CleanReport
 from ...steps.columns import normalized_column_labels
+from ...steps.duplicates import check_duplicate_ratio, report_detected_duplicates
 from ...steps.strings import active_sentinels
 from .._base import ExecutionEngine
 from .._config import enforce_fallback_policy
@@ -71,8 +73,24 @@ def execute(
                 source, config, engine_config, "native_error", f"FreshCore failed: {exc}"
             )
 
+        if (
+            not config.drop_duplicates
+            and config.duplicate_ratio_action == "error"
+            and native.get("duplicates_detected") is None
+        ):
+            # Detection-only dedup needs the native duplicate count to honour
+            # the escalation; without it the error could never fire.
+            return self._fallback(
+                source,
+                config,
+                engine_config,
+                "drop_duplicates",
+                'duplicate_ratio_action="error" needs a duplicate-row count, '
+                "which this FreshCore native module does not report",
+            )
+
         cleaned = self._frame_from_native(native)
-        report = self._report_from_native(frame, cleaned, native, started)
+        report = self._report_from_native(frame, cleaned, native, started, config)
         return cleaned, report
 
     @staticmethod
@@ -113,6 +131,12 @@ def _unsupported_reason(self, frame: pd.DataFrame, config: CleanConfig) -> str |
             return "drop_constant_columns is not implemented in FreshCore v1"
         if config.optimize_memory:
             return "optimize_memory is not implemented in FreshCore v1"
+        # Same wording as PlanGenerator.fallback_reason() so fd.plan() and the
+        # recorded fallback event agree.
+        if config.impute == "missforest":
+            return "missforest imputation uses scikit-learn and is evaluated by the pandas backend"
+        if config.impute_strategy:
+            return "impute_strategy per-column overrides are evaluated by the pandas backend"
         if config.duplicate_subset is not None:
             return "duplicate_subset is not implemented in FreshCore v1"
         if config.duplicate_keep not in ("first", "last"):
@@ -130,8 +154,55 @@ def _unsupported_reason(self, frame: pd.DataFrame, config: CleanConfig) -> str |
             return "non-default pandas index semantics require the pandas reference path"
         if self._has_unsupported_object_values(frame):
             return "mixed object columns with non-string values require the pandas reference path"
+        if config.outliers is not None:
+            non_finite = self._non_finite_float_columns(frame)
+            if non_finite:
+                return (
+                    f"non-finite values in numeric column(s) {self._shown(non_finite)}: "
+                    "FreshCore v1 outlier fences do not exclude ±inf, so outlier "
+                    "handling requires the pandas reference path"
+                )
+        if config.impute in ("mode", "auto"):
+            bools = self._boolean_columns_to_impute(frame)
+            if bools:
+                return (
+                    f"missing values in boolean column(s) {self._shown(bools)}: "
+                    "FreshCore v1 does not impute boolean columns, so imputation "
+                    "requires the pandas reference path"
+                )
         return None
 
+    @staticmethod
+    def _shown(columns: list[str], limit: int = 5) -> str:
+        shown = ", ".join(repr(c) for c in columns[:limit])
+        extra = len(columns) - limit
+        return f"{shown} (+{extra} more)" if extra > 0 else shown
+
+    @staticmethod
+    def _non_finite_float_columns(frame: pd.DataFrame) -> list[str]:
+        """Columns holding ±inf. Only float dtypes can store infinities."""
+        found: list[str] = []
+        for i, col in enumerate(frame.columns):
+            s = frame.iloc[:, i]
+            if not is_float_dtype(s):
+                continue
+            if np.isinf(s.to_numpy(dtype="float64", na_value=np.nan)).any():
+                found.append(str(col))
+        return found
+
+    @staticmethod
+    def _boolean_columns_to_impute(frame: pd.DataFrame) -> list[str]:
+        """Boolean columns the pandas imputer would fill (some, not all, missing)."""
+        found: list[str] = []
+        for i, col in enumerate(frame.columns):
+            s = frame.iloc[:, i]
+            if not is_bool_dtype(s):
+                continue
+            n_missing = int(s.isna().sum())
+            if 0 < n_missing < len(s):
+                found.append(str(col))
+        return found
+
     @staticmethod
     def _has_unsupported_object_values(frame: pd.DataFrame) -> bool:
         for col in frame.columns:
@@ -218,6 +289,7 @@ def _report_from_native(
         cleaned: pd.DataFrame,
         native: dict[str, Any],
         started: float,
+        config: CleanConfig,
     ) -> CleanReport:
         report = CleanReport(
             rows_before=int(native.get("rows_before", len(original))),
@@ -235,13 +307,39 @@ def _report_from_native(
         report.outliers_handled = int(native.get("outliers_handled", 0))
         report.columns_dropped.extend(str(c) for c in native.get("columns_dropped", []))
         report.columns_imputed.extend(str(c) for c in native.get("columns_imputed", []))
+        detection_reported = False
         for action in native.get("actions", []):
+            step = str(action["step"])
+            count = int(action.get("count", 0))
+            if not detection_reported and step in ("impute", "outliers"):
+                # The pandas step records detection before imputation/outliers.
+                self._report_detected_duplicates(native, config, report)
+                detection_reported = True
+            high_ratio = False
+            if step == "drop_duplicates" and config.drop_duplicates:
+                # Raises under duplicate_ratio_action="error", like the pandas step.
+                high_ratio = check_duplicate_ratio(
+                    count, self._rows_entering_dedup(report), config
+                )
             report.add(
-                str(action["step"]),
+                step,
                 str(action["description"]),
                 column=action.get("column"),
-                count=int(action.get("count", 0)),
+                count=count,
+                risk="medium" if high_ratio else "low",
             )
+            if high_ratio:
+                n_before = self._rows_entering_dedup(report)
+                report.add_warning(
+                    f"duplicate ratio {100.0 * count / n_before:.1f}% exceeds "
+                    f"duplicate_threshold ({100 * config.duplicate_threshold:.0f}%); "
+                    "check for an upstream join or export problem"
+                )
+                report.add_recommendation(
+                    "review why so many rows were duplicated before trusting downstream stats"
+                )
+        if not detection_reported:
+            self._report_detected_duplicates(native, config, report)
         for stage, seconds in native.get("stage_timings", []):
             report.record_stage_timing("freshcore", str(stage), float(seconds))
         if any(a.step == "fix_dtypes" for a in report.actions):
@@ -253,3 +351,26 @@ def _report_from_native(
                 "datetime dtype parity is required.",
             )
         return report
+
+    @staticmethod
+    def _rows_entering_dedup(report: CleanReport) -> int:
+        """Rows reaching the dedup stage: the pandas step runs after empty-row removal."""
+        dropped = sum(a.count for a in report.actions if a.step == "drop_empty_rows")
+        return report.rows_before - dropped
+
+    def _report_detected_duplicates(
+        self, native: dict[str, Any], config: CleanConfig, report: CleanReport
+    ) -> None:
+        """Detection-only duplicate reporting from the native duplicate count.
+
+        Skipped when the native module does not report ``duplicates_detected``
+        (``execute`` falls back first if ``duplicate_ratio_action="error"``).
+        """
+        if config.drop_duplicates:
+            return
+        detected = native.get("duplicates_detected")
+        if detected is None:
+            return
+        report_detected_duplicates(
+            int(detected), self._rows_entering_dedup(report), config, report
+        )
diff --git a/tests/test_execution/test_freshcore_fallbacks.py b/tests/test_execution/test_freshcore_fallbacks.py
new file mode 100644
index 00000000..608f2f58
--- /dev/null
+++ b/tests/test_execution/test_freshcore_fallbacks.py
@@ -0,0 +1,329 @@
+"""FreshCore fallbacks and report parity for what the native v1 kernels miss.
+
+- #322: ``impute="missforest"`` / per-column ``impute_strategy`` imputed nothing.
+- #334: ``±inf`` made the native outlier fences unbounded, so nothing was flagged.
+- #335: missing values in nullable boolean columns were never imputed.
+- #323: duplicate detection was never reported, so ``duplicate_ratio_action``
+  could not escalate.
+
+The Rust extension is optional, so a fake native module stands in for it.
+"""
+
+from __future__ import annotations
+
+import re
+
+import numpy as np
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.execution.backends._freshcore import FreshCoreEngine
+from freshdata.steps.duplicates import DuplicateRatioError
+
+KW = {"strategy": "conservative", "fix_dtypes": False, "verbose": False}
+
+
+class _ForbiddenNative:
+    @staticmethod
+    def execute_plan(payload):  # pragma: no cover - must not be called
+        raise AssertionError("unexpected native call")
+
+
+class _EchoNative:
+    """Returns the input columns unchanged, plus whatever extra keys are set."""
+
+    calls = 0
+    extra: dict = {}
+
+    @classmethod
+    def execute_plan(cls, payload):
+        cls.calls += 1
+        return {"columns": payload["columns"], "actions": [], **cls.extra}
+
+
+def _use_native(monkeypatch, module) -> None:
+    monkeypatch.setattr(FreshCoreEngine, "_load_native", staticmethod(lambda: module))
+
+
+def _echo(monkeypatch, **extra) -> type[_EchoNative]:
+    native = type("EchoNative", (_EchoNative,), {"calls": 0, "extra": extra})
+    _use_native(monkeypatch, native)
+    return native
+
+
+def _inf_frame(values: list[float]) -> pd.DataFrame:
+    return pd.DataFrame({"x": values, "k": np.arange(len(values), dtype=float)})
+
+
+FALLBACK_CASES = [
+    pytest.param(
+        pd.DataFrame({"x": [1.0, None, 3.0] * 7, "y": np.arange(21.0)}),
+        {"impute": "missforest"},
+        "missforest imputation uses scikit-learn",
+        None,
+        id="322-missforest",
+    ),
+    pytest.param(
+        pd.DataFrame({"x": [1.0, None, 3.0] * 20, "y": np.arange(60.0)}),
+        {"impute_strategy": {"x": "mean"}},
+        "impute_strategy per-column overrides",
+        None,
+        id="322-impute-strategy",
+    ),
+    pytest.param(
+        _inf_frame([1.0, 2, 3, 4, 5, 6, 7, 8, 9, 100, np.inf]),
+        {"outliers": "flag", "outlier_method": "zscore", "outlier_factor": 2.0},
+        "non-finite values in numeric column(s) 'x'",
+        2,
+        id="334-zscore",
+    ),
+    pytest.param(
+        _inf_frame([1.0, 2, 3, 4, np.inf, np.inf, np.inf, np.inf]),
+        {"outliers": "flag", "outlier_method": "iqr", "outlier_factor": 1.5},
+        "non-finite values in numeric column(s) 'x'",
+        4,
+        id="334-iqr",
+    ),
+    pytest.param(
+        pd.DataFrame(
+            {"b": pd.array([True, True, None, False], dtype="boolean"), "k": [1.0, 2, 3, 4]}
+        ),
+        {"impute": "mode"},
+        "missing values in boolean column(s) 'b'",
+        None,
+        id="335-mode",
+    ),
+    pytest.param(
+        pd.DataFrame(
+            {"b": pd.array([True, True, None, False], dtype="boolean"), "k": [1.0, 2, 3, 4]}
+        ),
+        {"impute": "auto"},
+        "missing values in boolean column(s) 'b'",
+        None,
+        id="335-auto",
+    ),
+]
+
+
+@pytest.mark.parametrize(("df", "options", "reason", "outliers"), FALLBACK_CASES)
+def test_falls_back_to_pandas_with_identical_output(monkeypatch, df, options, reason, outliers):
+    _use_native(monkeypatch, _ForbiddenNative)
+    expected, expected_report = fd.clean(
+        df.copy(), engine="pandas", return_report=True, **KW, **options
+    )
+    out, report = fd.clean(df.copy(), engine="freshcore", return_report=True, **KW, **options)
+
+    assert report.backend == "pandas"
+    (event,) = report.fallback_events
+    assert event["backend"] == "freshcore"
+    assert event["fallback_step"] == "pipeline"
+    assert reason in event["fallback_reason"]
+    # engine="pandas" wraps its frame in CleanResult (a DataFrame subclass).
+    pd.testing.assert_frame_equal(out, pd.DataFrame(expected))
+    assert [(a.step, a.column, a.count) for a in report.actions] == [
+        (a.step, a.column, a.count) for a in expected_report.actions
+    ]
+    if outliers is not None:
+        assert report.outliers_handled == expected_report.outliers_handled == outliers
+
+
+@pytest.mark.parametrize(("df", "options", "reason", "outliers"), FALLBACK_CASES)
+def test_fallback_policy_error_refuses(monkeypatch, df, options, reason, outliers):
+    _use_native(monkeypatch, _ForbiddenNative)
+    with pytest.raises(fd.FallbackError, match=re.escape(reason)):
+        fd.clean(df, engine="freshcore", fallback_policy="error", **KW, **options)
+
+
+def test_imputation_fills_what_pandas_fills(monkeypatch):
+    _use_native(monkeypatch, _ForbiddenNative)
+    df = pd.DataFrame({"x": [1.0, None, 3.0] * 20, "y": np.arange(60.0)})
+    out = fd.clean(df, engine="freshcore", impute_strategy={"x": "mean"}, **KW)
+    assert int(out["x"].isna().sum()) == 0
+
+    bools = pd.DataFrame(
+        {"b": pd.array([True, True, None, False], dtype="boolean"), "k": [1.0, 2, 3, 4]}
+    )
+    out = fd.clean(bools, engine="freshcore", impute="mode", **KW)
+    assert out["b"].tolist() == [True, True, True, False]
+
+
+@pytest.mark.parametrize(
+    "options",
+    [{"impute": "missforest"}, {"impute_strategy": {"x": "median"}}],
+    ids=["missforest", "impute-strategy"],
+)
+def test_plan_preview_matches_recorded_fallback(monkeypatch, options):
+    _use_native(monkeypatch, _ForbiddenNative)
+    df = pd.DataFrame({"x": [1.0, None, 3.0, 4.0]})
+    preview = fd.plan(df, engine="freshcore", **KW, **options)
+    _, report = fd.clean(df, engine="freshcore", return_report=True, **KW, **options)
+    assert preview.fallback_reason == report.fallback_events[0]["fallback_reason"]
+
+
+@pytest.mark.parametrize(
+    ("df", "options"),
+    [
+        pytest.param(
+            _inf_frame([1.0, 2, 3, 4, 100.0]), {"outliers": "flag"}, id="finite-outliers"
+        ),
+        pytest.param(
+            _inf_frame([1.0, np.inf, 3.0]), {"impute": "median"}, id="inf-without-outliers"
+        ),
+        pytest.param(
+            pd.DataFrame({"b": pd.array([True, None, False], dtype="boolean")}),
+            {"impute": "median"},
+            id="boolean-median-is-a-skip-on-pandas-too",
+        ),
+        pytest.param(
+            pd.DataFrame({"b": pd.array([True, False, True], dtype="boolean")}),
+            {"impute": "mode"},
+            id="boolean-without-missing",
+        ),
+        pytest.param(
+            pd.DataFrame({"x": [1.0, None, 3.0]}),
+            {"impute": "mean", "impute_strategy": {}},
+            id="empty-impute-strategy",
+        ),
+    ],
+)
+def test_supported_shapes_stay_native(monkeypatch, df, options):
+    native = _echo(monkeypatch)
+    _, report = fd.clean(df, engine="freshcore", return_report=True, **KW, **options)
+    assert native.calls == 1
+    assert report.backend == "freshcore"
+    assert report.fallback_events == []
+
+
+# -- #323: duplicate detection and duplicate_ratio_action --------------------
+
+
+def _dup_frame() -> pd.DataFrame:
+    """12 rows: 2 all-missing, then 2 duplicates among the remaining 10 (20%)."""
+    return pd.DataFrame(
+        {
+            "a": [1.0, 1.0, 1.0, 2, 3, 4, 5, 6, 7, 8, None, None],
+            "b": ["x", "x", "x", "y", "y", "y", "y", "y", "y", "y", None, None],
+        }
+    )
+
+
+def _dup_native_result(**extra) -> dict:
+    return {
+        "rows_before": 12,
+        "rows_after": 10,
+        "actions": [
+            {
+                "step": "drop_empty_rows",
+                "column": None,
+                "description": "dropped 2 all-missing row(s)",
+                "count": 2,
+            },
+            {
+                "step": "impute",
+                "column": "a",
+                "description": "filled 0 missing value(s) with mean (1.0)",
+                "count": 0,
+            },
+        ],
+        **extra,
+    }
+
+
+def _dup_actions(report) -> list[str]:
+    return [a.description for a in report.actions if a.step == "drop_duplicates"]
+
+
+def test_native_duplicate_count_is_reported_like_pandas(monkeypatch):
+    _, expected = fd.clean(_dup_frame(), engine="pandas", return_report=True, **KW)
+    _echo(monkeypatch, **_dup_native_result(duplicates_detected=2))
+    _, report = fd.clean(_dup_frame(), engine="freshcore", return_report=True, **KW)
+
+    assert report.backend == "freshcore"
+    assert _dup_actions(report) == _dup_actions(expected)
+    assert _dup_actions(report) == ["detected 2 duplicate row(s) (20.0%), none removed"]
+    assert report.warnings == expected.warnings
+    assert report.recommendations == expected.recommendations
+    steps = [a.step for a in report.actions]
+    assert steps.index("drop_duplicates") < steps.index("impute")
+
+
+def test_native_duplicate_count_escalates_under_error(monkeypatch):
+    with pytest.raises(DuplicateRatioError):
+        fd.clean(_dup_frame(), engine="pandas", duplicate_ratio_action="error", **KW)
+    _echo(monkeypatch, **_dup_native_result(duplicates_detected=2))
+    with pytest.raises(DuplicateRatioError):
+        fd.clean(_dup_frame(), engine="freshcore", duplicate_ratio_action="error", **KW)
+
+
+def test_missing_native_duplicate_count_skips_detection(monkeypatch):
+    native = _echo(monkeypatch, **_dup_native_result())
+    _, report = fd.clean(_dup_frame(), engine="freshcore", return_report=True, **KW)
+
+    assert native.calls == 1
+    assert report.backend == "freshcore"
+    assert _dup_actions(report) == []
+    assert report.warnings == []
+
+
+def test_missing_native_duplicate_count_falls_back_under_error(monkeypatch):
+    native = _echo(monkeypatch, **_dup_native_result())
+    with pytest.raises(DuplicateRatioError):
+        fd.clean(_dup_frame(), engine="freshcore", duplicate_ratio_action="error", **KW)
+    assert native.calls == 1
+
+    below = fd.CleanConfig(**KW, duplicate_ratio_action="error", duplicate_threshold=0.5)
+    _, report = fd.clean(_dup_frame(), config=below, engine="freshcore", return_report=True)
+    assert report.backend == "pandas"
+    (event,) = report.fallback_events
+    assert event["fallback_step"] == "drop_duplicates"
+    assert 'duplicate_ratio_action="error"' in event["fallback_reason"]
+
+    with pytest.raises(fd.FallbackError, match="duplicate-row count"):
+        fd.clean(
+            _dup_frame(),
+            config=below,
+            engine="freshcore",
+            fallback_policy="error",
+        )
+
+
+def _dropped_native_result() -> dict:
+    result = _dup_native_result()
+    result["actions"].insert(
+        1,
+        {
+            "step": "drop_duplicates",
+            "column": None,
+            "description": 'dropped 2 duplicate row(s) (20.0% of rows, keep="first")',
+            "count": 2,
+        },
+    )
+    result["duplicates_removed"] = 2
+    return result
+
+
+def test_drop_duplicates_ratio_escalates_under_error(monkeypatch):
+    options = {"drop_duplicates": True, "duplicate_ratio_action": "error"}
+    with pytest.raises(DuplicateRatioError):
+        fd.clean(_dup_frame(), engine="pandas", **KW, **options)
+    _echo(monkeypatch, **_dropped_native_result())
+    with pytest.raises(DuplicateRatioError):
+        fd.clean(_dup_frame(), engine="freshcore", **KW, **options)
+
+
+def test_drop_duplicates_high_ratio_warns_like_pandas(monkeypatch):
+    _, expected = fd.clean(
+        _dup_frame(), engine="pandas", drop_duplicates=True, return_report=True, **KW
+    )
+    _echo(monkeypatch, **_dropped_native_result())
+    _, report = fd.clean(
+        _dup_frame(), engine="freshcore", drop_duplicates=True, return_report=True, **KW
+    )
+
+    assert report.backend == "freshcore"
+    (action,) = [a for a in report.actions if a.step == "drop_duplicates"]
+    (pandas_action,) = [a for a in expected.actions if a.step == "drop_duplicates"]
+    assert action.risk == pandas_action.risk == "medium"
+    assert report.warnings == expected.warnings
+    assert report.recommendations == expected.recommendations