fix(semantic): a replayed repair needs corroboration in the frame it lands on

src/freshdata/semantic/apply.py

@@ -8,6 +8,8 @@
 
 from __future__ import annotations
 
+from dataclasses import replace
+
 import pandas as pd
 
 from .._numeric import safe_to_numeric
@@ -69,6 +71,47 @@ def _describe(decision: SemanticPolicyDecision) -> str:
     return f"Skipped semantic repair {pair}"
 
 
+#: Issue types whose correct repair depends on what the *column* means, not on
+#: the token alone. ``"M"`` expands to ``"male"`` in a gender column and to
+#: ``"medium"`` in a size column; ``"1"`` is True in a flag column and the
+#: number one in a count. A repair of one of these kinds, learned on one
+#: dataset, is not evidence about a different dataset's column.
+_CONTEXT_DEPENDENT_ISSUES = frozenset(
+    {"category_synonym", "boolean_synonym", "reference_value"}
+)
+
+
+def _uncorroborated_memory_proposal(mem: SemanticProposal, memory: object) -> SemanticProposal:
+    """Demote a replayed context-dependent repair that nothing in this frame supports.
+
+    The deterministic expert did not merely disagree -- it produced no proposal
+    at all for this value, because the evidence that justified the repair on the
+    learning dataset is absent here. An abstention is not corroboration, so the
+    replayed repair keeps its provenance but is routed to a human instead of
+    being applied on authority carried over from another frame.
+    """
+    dataset_id = getattr(memory, "dataset_id", "?")
+    note = SemanticEvidence(
+        "memory_uncorroborated",
+        (
+            f"no deterministic expert proposed {mem.raw_value!r} -> "
+            f"{mem.proposed_value!r} for column {mem.column!r} in this frame"
+        ),
+        0.0,
+    )
+    return replace(
+        mem,
+        risk="high",
+        human_review=True,
+        evidence=(*mem.evidence, note),
+        rationale=(
+            f"cleaning memory {dataset_id!r} learned {mem.raw_value!r} -> "
+            f"{mem.proposed_value!r}, but nothing in this frame corroborates it "
+            f"for column {mem.column!r}; held for review"
+        ),
+    )
+
+
 def _conflict_proposal(
     det: SemanticProposal, mem: SemanticProposal, memory: object
 ) -> SemanticProposal:
@@ -106,7 +149,11 @@ def _merge_proposals(
 ) -> list[SemanticProposal]:
     """Merge deterministic + memory-retrieved proposals, deduping same-key repairs.
 
-    Non-colliding proposals from either source pass through unchanged. When both
+    A memory proposal with no deterministic counterpart is demoted to a
+    review-required suggestion when its issue type is context-dependent (see
+    :data:`_CONTEXT_DEPENDENT_ISSUES`): an abstention by every expert means the
+    evidence that justified the repair is absent from this frame. Other
+    non-colliding proposals pass through unchanged. When both
     a deterministic expert and memory propose the same ``(column, raw_value)``:
     if they agree on the proposed value, keep whichever has the higher
     confidence (memory wins ties); if they disagree, replace both with one
@@ -128,7 +175,17 @@ def _merge_proposals(
         key = (mem_p.column, mem_p.raw_value)
         det_list = det_by_key.get(key)
         if not det_list:
-            merged.append(mem_p)
+            # Nothing in this frame proposed the same repair. For a
+            # context-dependent issue type that silence is meaningful: the
+            # evidence behind the learned repair is gone, so replaying it would
+            # rewrite a column whose meaning may have changed.
+            if (
+                mem_p.proposed_value is not None
+                and mem_p.issue_type in _CONTEXT_DEPENDENT_ISSUES
+            ):
+                merged.append(_uncorroborated_memory_proposal(mem_p, memory))
+            else:
+                merged.append(mem_p)
             continue
         touched.add(key)
         # A flag (proposed_value=None) is an abstention — "this value looks

tests/test_memory_conflicts.py

@@ -71,11 +71,12 @@ def semantic_actions(report) -> list:
 # --------------------------------------------------------------------------- #
 
 
-def test_one_token_categorical_replay_rewrites_a_column_that_changed_meaning(
+def test_one_token_categorical_replay_does_not_rewrite_a_column_that_changed_meaning(
     gender_memory: CleaningMemory,
 ) -> None:
-    """FINDING (S2): memory replay auto-applies ``M -> male`` onto a column of
-    clothing sizes.
+    """A learned ``M -> male`` must not be replayed onto clothing sizes.
+
+    This test was written to pin the defect: replay auto-applied the repair.
 
     Nothing in the retrieval path re-validates *why* the repair was learned.
     ``CategorySynonymExpert`` only proposed ``M -> male`` on dataset A because
@@ -87,18 +88,26 @@ def test_one_token_categorical_replay_rewrites_a_column_that_changed_meaning(
 
     The stored ``column_signature``/``value_signature`` on the memory record
     would not have helped either: both datasets profile as
-    ``role="categorical", semantic_type=None, free_text=False``. What is missing
-    is a check that the *evidence* behind the learned repair still holds.
+    ``role="categorical", semantic_type=None, free_text=False``.
+
+    The fix supplies the missing check: a context-dependent repair replayed with
+    no deterministic corroboration in *this* frame is demoted to a
+    review-required suggestion, so memory stays evidence rather than authority.
     """
     out, report = fd.clean(sizes_frame(), semantic_mode="auto", memory=gender_memory, **CLEAN)
 
-    assert out["segment"].tolist() == ["S", "male", "L", "male"] * 3
-    applied = [a for a in semantic_actions(report) if a.status == "automatic"]
-    assert len(applied) == 1
-    action = applied[0]
-    assert action.metadata["raw_value"] == "M"
-    assert action.metadata["proposed_value"] == "male"
-    assert action.human_review is False
+    assert out["segment"].tolist() == sizes_frame()["segment"].tolist(), (
+        "clothing sizes must survive a gender memory untouched"
+    )
+    assert not [a for a in semantic_actions(report) if a.status == "automatic"]
+    held = [
+        a for a in semantic_actions(report)
+        if a.metadata.get("raw_value") == "M" and a.metadata.get("proposed_value") == "male"
+    ]
+    assert len(held) == 1
+    assert held[0].status == "suggested"
+    assert held[0].risk == "high"
+    assert held[0].human_review is True
 
     # Control: without the memory, the deterministic layer leaves B alone.
     plain, plain_report = fd.clean(sizes_frame(), semantic_mode="auto", **CLEAN)
@@ -109,14 +118,16 @@ def test_one_token_categorical_replay_rewrites_a_column_that_changed_meaning(
 def test_normalized_match_also_catches_the_lowercase_spelling(
     gender_memory: CleaningMemory,
 ) -> None:
-    """FINDING (S2, same root cause): retrieval matches on the *normalized*
-    value, so a learned ``"M"`` also rewrites a dominant lowercase ``"m"`` —
-    and it does so while the deterministic expert is proposing the opposite
-    direction (``"M" -> "m"``, aligning to the dominant spelling)."""
+    """Retrieval matches on the *normalized* value, so a learned ``"M"`` also
+    reaches a dominant lowercase ``"m"``. That replay is context-dependent and
+    uncorroborated here, so it is held for review rather than applied, while the
+    genuinely *conflicting* value still takes the ``unsafe_ambiguous`` path."""
     df = frame(["m"] * 8 + ["M"] * 2 + ["S"] * 3 + ["L"] * 3)
     out, report = fd.clean(df, semantic_mode="auto", memory=gender_memory, **CLEAN)
 
-    assert out["segment"].tolist()[:8] == ["male"] * 8  # the dominant spelling, rewritten
+    assert out["segment"].tolist()[:8] != ["male"] * 8, (
+        "an uncorroborated gender replay must not rewrite the dominant spelling"
+    )
     # The genuinely *conflicting* value is handled correctly, though:
     conflicts = [
         a for a in semantic_actions(report)
@@ -179,19 +190,23 @@ def test_allowed_values_hint_blocks_the_collision(gender_memory: CleaningMemory)
     assert not [a for a in semantic_actions(report) if a.status == "automatic"]
 
 
-def test_semantic_type_hint_alone_does_not_block_the_collision(
+def test_a_semantic_type_hint_does_not_revive_the_collision(
     gender_memory: CleaningMemory,
 ) -> None:
-    """FINDING (S3): declaring a *semantic type* is not enough — only
-    ``allowed_values``, ``mutable=False`` or column-level protection stops the
-    replay. The stored ``column_signature.semantic_type`` is never compared
-    against the live one at retrieval time."""
+    """Declaring a semantic type must not re-enable the uncorroborated replay.
+
+    The stored ``column_signature.semantic_type`` is still never compared
+    against the live one at retrieval time -- that remains a gap -- but it no
+    longer matters for safety here, because the repair is held for review on
+    the absence of corroboration rather than on any signature check.
+    """
     context = {"columns": {"segment": {"semantic_type": "category"}}}
-    out, _ = fd.clean(
+    out, report = fd.clean(
         sizes_frame(), semantic_mode="auto", memory=gender_memory,
         semantic_context=context, **CLEAN,
     )
-    assert out["segment"].tolist() == ["S", "male", "L", "male"] * 3
+    assert out["segment"].tolist() == sizes_frame()["segment"].tolist()
+    assert not [a for a in semantic_actions(report) if a.status == "automatic"]
 
 
 def test_mutable_false_blocks_the_collision_and_audits_the_skip(
@@ -254,20 +269,24 @@ def test_every_memory_derived_decision_is_distinguishable_in_the_audit_trail(
     gender_memory: CleaningMemory,
 ) -> None:
     """All five signals hold on the colliding replay, so a reviewer can tell a
-    memory-derived mutation from a deterministic one without guessing.
+    memory-derived decision from a deterministic one without guessing.
 
-    Note the status nuance: a *semantic* replay carries the policy-gate status
-    (``"automatic"``), while ``status="approved"`` is what
-    :func:`freshdata.memory.annotate_report` stamps on replayed **non**-semantic
-    decisions plus its own ``step="memory"`` summary action.
+    The provenance must survive the demotion: a replay that is held for review
+    rather than applied is exactly the case a reviewer has to understand, so
+    losing the memory markers there would be worse than losing them on an
+    auto-applied one.
     """
     _, report = fd.clean(sizes_frame(), semantic_mode="auto", memory=gender_memory, **CLEAN)
-    applied = [a for a in semantic_actions(report) if a.status == "automatic"]
-    assert len(applied) == 1
-    action = applied[0]
+    replayed = [
+        a for a in semantic_actions(report)
+        if a.metadata.get("backend") == "memory"
+    ]
+    assert len(replayed) == 1
+    action = replayed[0]
 
     assert action.memory_influenced is True                      # signal 1
-    assert action.status == "automatic"                          # signal 2 (see docstring)
+    assert action.status == "suggested"                          # signal 2: held, not applied
+    assert action.human_review is True
     assert action.model_id == "semantic:category_synonym:memory"  # signal 3
     kinds = [e["kind"] for e in action.metadata["evidence"]]
     assert "memory_replay" in kinds                              # signal 4
@@ -390,7 +409,10 @@ def test_version_mismatch_is_recorded_but_never_checked(
         {**gender_memory.to_dict(), "freshdata_version": "0.0.1"}
     )
     out, report = fd.clean(sizes_frame(), semantic_mode="auto", memory=ancient, **CLEAN)
-    assert out["segment"].tolist() == ["S", "male", "L", "male"] * 3
+    # The version is still never checked -- that is the finding. The values
+    # survive only because the replay is uncorroborated here, not because the
+    # stale version was noticed.
+    assert out["segment"].tolist() == sizes_frame()["segment"].tolist()
     assert not [w for w in report.warnings if "version" in w.lower()]
 
 

tests/test_memory_uncorroborated_replay.py

@@ -0,0 +1,109 @@
+"""Memory is evidence, not authority (#FD2-006).
+
+A repair learned on one dataset was replayed onto another whose column meant
+something different, and applied automatically:
+
+    A: segment = M/F        -> learns  M -> male
+    B: segment = S/M/L      -> becomes S/male/L/male
+
+The conflict machinery did exist and did work, but only fired on
+*disagreement*: ``_conflict_proposal`` turns a deterministic-vs-memory clash
+into a high-risk ``unsafe_ambiguous`` record. On B the ``CategorySynonymExpert``
+correctly **abstains** -- the column is not gender-like -- and an abstention is
+not a disagreement, so the replayed repair passed through carrying the full
+confidence it had earned on a different column.
+
+The stored signatures would not have caught it: both columns profile as
+``role="categorical", semantic_type=None, free_text=False``.
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.semantic.apply import _CONTEXT_DEPENDENT_ISSUES
+
+CLEAN = {"verbose": False, "return_report": True}
+
+
+def _frame(values) -> pd.DataFrame:
+    return pd.DataFrame(
+        {"id": [f"r{i}" for i in range(len(values))], "segment": list(values)}
+    )
+
+
+@pytest.fixture
+def gender_memory():
+    """Memory learned where ``segment`` genuinely holds genders."""
+    learn_df = _frame(["M", "F"] * 6)
+    _, report = fd.clean(learn_df, semantic_mode="auto", **CLEAN)
+    return fd.learn_cleaning_memory(learn_df, decisions=report, dataset_id="hr")
+
+
+def _semantic(report):
+    return [a for a in report.actions if a.step == "semantic"]
+
+
+def test_a_gender_memory_does_not_rewrite_clothing_sizes(gender_memory):
+    """The headline trap: S/M/L must survive a memory that learned M -> male."""
+    sizes = _frame(["S", "M", "L", "M"] * 3)
+    out, report = fd.clean(sizes, semantic_mode="auto", memory=gender_memory, **CLEAN)
+
+    assert out["segment"].tolist() == sizes["segment"].tolist()
+    assert not [a for a in _semantic(report) if a.status == "automatic"]
+
+
+def test_the_held_repair_is_routed_to_a_human_with_its_reason(gender_memory):
+    """Holding it silently would only trade one failure for another."""
+    sizes = _frame(["S", "M", "L", "M"] * 3)
+    _, report = fd.clean(sizes, semantic_mode="auto", memory=gender_memory, **CLEAN)
+
+    held = [a for a in _semantic(report) if a.metadata.get("backend") == "memory"]
+    assert len(held) == 1
+    action = held[0]
+    assert action.status == "suggested"
+    assert action.risk == "high"
+    assert action.human_review is True
+    assert "corroborate" in action.rationale
+    assert action.metadata["raw_value"] == "M"
+    assert action.metadata["proposed_value"] == "male"
+
+
+def test_replay_onto_the_same_kind_of_column_still_works(gender_memory):
+    """The guard must not disable the feature it is guarding.
+
+    A second genuinely gender-shaped frame is exactly what memory is for, and
+    there the deterministic expert corroborates the repair.
+    """
+    genders = _frame(["M", "F", "M", "F"] * 3)
+    out, report = fd.clean(genders, semantic_mode="auto", memory=gender_memory, **CLEAN)
+
+    assert out["segment"].tolist() == ["male", "female"] * 6
+    assert [a for a in _semantic(report) if a.status == "automatic"]
+
+
+def test_provenance_survives_the_demotion(gender_memory):
+    """A held replay is precisely the case a reviewer must be able to trace."""
+    sizes = _frame(["S", "M", "L", "M"] * 3)
+    _, report = fd.clean(sizes, semantic_mode="auto", memory=gender_memory, **CLEAN)
+
+    action = next(a for a in _semantic(report) if a.metadata.get("backend") == "memory")
+    assert action.memory_influenced is True
+    assert action.model_id.endswith(":memory")
+    kinds = [e["kind"] for e in action.metadata["evidence"]]
+    assert "memory_replay" in kinds
+    assert "memory_uncorroborated" in kinds, "the reason for holding must be recorded"
+
+
+def test_a_context_free_repair_is_not_demoted(gender_memory):
+    """Only context-dependent issue types need corroboration.
+
+    An encoding or format repair means the same thing in every column, so
+    demoting it would make memory useless without making anything safer.
+    """
+    assert "category_synonym" in _CONTEXT_DEPENDENT_ISSUES
+    assert "boolean_synonym" in _CONTEXT_DEPENDENT_ISSUES
+    for context_free in ("encoding_repair", "format_alignment", "numeric_format"):
+        assert context_free not in _CONTEXT_DEPENDENT_ISSUES