fix(rewrite): close record-drop gaps flagged in review

Two small-model robustness gaps surfaced by Greptile on the disposition path:

- _pessimistic_fallback_disposition could still drop a row: with empty entity
  context it built an empty SensitivityDispositionSchema, which raises on the
  min_length=1 tripwire (and again in the downstream parser), and both
  call-sites in _reconstruct_full_disposition_column were unguarded. It now
  emits a single no-op (leave_as_is/low) disposition in that pipeline-invariant
  case, logging loudly — the row survives and the no-op never reaches the
  rewrite (excluded from protected_entities). Also removes the dead duplicate
  return branch.
- SimpleDispositionItem._coerce_scalar_to_str now coerces unexpected container
  types (list/dict) to "" instead of returning them unchanged, so one drifted
  field no longer fails the whole item (which would discard every disposition
  for the row and force a pessimistic fallback).

Adds regression tests for the empty-context no-op guarantee (function + column
level) and container-value coercion.

Author: lipikaramaswamy

src/anonymizer/engine/rewrite/sensitivity_disposition.py

@@ -335,11 +335,32 @@ def _pessimistic_fallback_disposition(
             )
         )
     if not items:
-        # Genuinely no entities at all in context — the orchestrator should
-        # have short-circuited before this step. Still better to raise here
-        # than to silently emit garbage; SensitivityDispositionSchema's
-        # min_length=1 invariant will surface the bug.
-        return SensitivityDispositionSchema(sensitivity_disposition=items)
+        # Genuinely no entities at all in context. The orchestrator should have
+        # short-circuited rows with no detected entities before this step, so
+        # this is a pipeline-invariant violation — but this is the last-resort
+        # path whose contract is "never drop the row." Emitting an empty list
+        # would raise on SensitivityDispositionSchema's (and the downstream
+        # parser's) min_length=1 invariant and drop the record, so we log loudly
+        # and emit a single no-op (leave_as_is/low) disposition instead. It is
+        # excluded from protected_entities, so it never reaches the rewrite.
+        logger.error(
+            "pessimistic fallback: empty entity context at the disposition step "
+            "(orchestrator should have short-circuited entity-free rows); emitting "
+            "a single no-op disposition so the row is not dropped"
+        )
+        items.append(
+            EntityDispositionSchema(
+                id=1,
+                source="tagged",
+                category="quasi_identifier",
+                sensitivity="low",
+                entity_label="",
+                entity_value="",
+                protection_method_suggestion="leave_as_is",
+                combined_risk_level=derive_combined_risk_level("quasi_identifier", "leave_as_is", "low"),
+                protection_reason=template_protection_reason("quasi_identifier", "leave_as_is", "low"),
+            )
+        )
     return SensitivityDispositionSchema(sensitivity_disposition=items)
 
 

src/anonymizer/engine/schemas/rewrite.py

@@ -465,9 +465,14 @@ class SimpleDispositionItem(BaseModel):
     def _coerce_scalar_to_str(cls, v: object) -> str:
         if v is None:
             return ""
+        if isinstance(v, str):
+            return v
         if isinstance(v, (int, float, bool)):
             return str(v)
-        return v
+        # Unexpected container (list/dict) from a drifted response: coerce to ""
+        # rather than letting pydantic raise on the whole SimpleDispositionItem.
+        # The reconstructor recovers the true value from trusted entity context.
+        return ""
 
 
 class SimpleDispositionResult(BaseModel):

tests/engine/test_disposition_reconstructor.py

@@ -345,6 +345,24 @@ def test_unmapped_label_falls_back_to_quasi(self) -> None:
         item = result.sensitivity_disposition[0]
         assert item.category == "quasi_identifier"
 
+    def test_empty_context_returns_valid_noop_instead_of_raising(self) -> None:
+        """Empty context (a pipeline-invariant violation) must not raise on the
+        SensitivityDispositionSchema min_length=1 tripwire and drop the row;
+        the fallback emits a single no-op (leave_as_is/low) disposition."""
+        result = _pessimistic_fallback_disposition([], [])
+        assert len(result.sensitivity_disposition) == 1
+        item = result.sensitivity_disposition[0]
+        assert item.protection_method_suggestion == "leave_as_is"
+        assert item.combined_risk_level == "low"
+        assert result.protected_entities == []  # no-op never reaches the rewrite
+
+    def test_all_blank_slots_returns_valid_noop(self) -> None:
+        """Slots whose label/value strip to empty are skipped; if that empties
+        the disposition, the no-op guarantee still holds (no raise/no drop)."""
+        result = _pessimistic_fallback_disposition([{"value": "", "labels": [""]}], [])
+        assert len(result.sensitivity_disposition) == 1
+        assert result.sensitivity_disposition[0].protection_method_suggestion == "leave_as_is"
+
 
 # ---------------------------------------------------------------------------
 # _reconstruct_full_disposition_column (workflow glue)
@@ -394,6 +412,16 @@ def test_empty_simple_falls_back_to_pessimistic(self) -> None:
         assert items[0]["entity_label"] == "first_name"
         assert items[0]["protection_method_suggestion"] == "replace"
 
+    def test_empty_simple_and_empty_context_does_not_drop_row(self) -> None:
+        """Both unguarded fallback call-sites: empty simple output AND empty
+        entity context must still yield a valid row (the column generator must
+        never raise out and drop the record)."""
+        row = self._row({"sensitivity_disposition": []}, ebv=[], latent=[])
+        out = _reconstruct_full_disposition_column(row)
+        items = out[COL_SENSITIVITY_DISPOSITION]["sensitivity_disposition"]
+        assert len(items) == 1
+        assert items[0]["protection_method_suggestion"] == "leave_as_is"
+
     def test_invalid_simple_payload_falls_back(self) -> None:
         """If ``SimpleDispositionResult.model_validate`` raises, fall back
         to pessimistic disposition rather than dropping the row."""

tests/engine/test_small_model_drift.py

@@ -120,6 +120,18 @@ def test_none_in_str_fields_coerces_to_empty(self) -> None:
         assert item.entity_label == ""
         assert item.category == ""
 
+    def test_container_values_in_str_fields_coerce_to_empty(self) -> None:
+        """A model emitting a list/dict for a scalar str field must not fail the
+        whole item (which would discard every disposition for the row); coerce
+        to "" and let the reconstructor recover from trusted context."""
+        result = SimpleDispositionResult.model_validate(
+            [{"id": 1, "entity_label": ["first", "name"], "category": {"x": 1}, "entity_value": "Alice"}]
+        )
+        item = result.sensitivity_disposition[0]
+        assert item.entity_label == ""
+        assert item.category == ""
+        assert item.entity_value == "Alice"
+
 
 # ---------------------------------------------------------------------------
 # MeaningUnits — bare list, aspect normalize, importance default, id renumber