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