test(streaming,memory): batch-boundary invariance and memory-replay collision safety

[kevincostner17]

Production-readiness lane: Phase 15 (streaming batch-boundary invariance) and
Phase 16 (learning/memory replay safety). Tests only — no source changes.

Two new files, 55 tests:

• tests/test_streaming_boundaries.py (33) — feeds the same rows through
  StreamingCleaner as 1 / 10 / 100 / 1000 batches and pins down exactly what is
  batch-independent and what is not, plus duplicates across boundaries, schema
  changes mid-stream, pathological batch sizes, and the constant-memory claim.
• tests/test_memory_conflicts.py (22) — the categorical one-token replay trap
  (M -> male learned on dataset A, replayed onto a dataset B where M means
  medium) and the collision axes around it. tests/test_cleaning_memory.py:237
  already covers the DATE version of a memory/deterministic disagreement; the
  categorical case did not exist.

Tests whose docstring starts with FINDING pin current behaviour that this
lane reports as a defect or a specification gap, not as desired behaviour.

What IS batch-invariant

Outcome	Invariance
emitted row count, column set, column order	exact
reservoir median / Q1 / Q3, min / max	bit-identical, even with the reservoir far under capacity
Space-Saving mode + mode ratio (unsaturated), total	bit-identical
per-column missing / non_null, finalize() accounting	exact
Welford mean / std	equal to ~1 ULP only — Chan's parallel update sums in a different order per batch size

What is NOT batch-invariant (characterised, not defects)

Which value lands in a missing cell (imputation reads the running state as of
that batch); how many cells get imputed at all (warmup deferral + the cumulative
missing-band gate both move with batch size, and warmup-deferred cells are never
back-filled); the Space-Saving mode once max_categories saturates; recent-window
cross-batch dedup; the drift-event count.

Findings

#	Sev	Area	Finding
1	S2	memory + profile replay	A one-token categorical repair replays onto a same-named column whose meaning changed and is auto-applied: M -> male learned on a gender column rewrites S/M/L sizes at confidence 0.95, risk="medium", human_review=False. The deterministic expert correctly abstains there (gender_like is false), but an abstention is not a disagreement, so _merge_proposals passes the memory proposal straight through. Normalized matching also catches a dominant lowercase m. Reproduces through profile= value maps (semantic:allowed_value_map:profile), so the hole is in replay, not in CleaningMemory. The stored column_signature/value_signature are never compared at retrieval and would not have helped (both frames profile identically); what is missing is a re-check that the evidence behind the learned repair still holds.
2	S2	streaming memory bound	StreamingState.drift_log (_state.py:224) is an uncapped list. On a persistently drifting stream it grows strictly linearly: 50 batches → 74 entries / 23 KB; 4000 batches → 5999 entries / 1.85 MB (~310 B/entry). Proposed fix: back it with deque(maxlen=...) exactly like _trust_history (cap 10_000). Both state_["n_drift_events"] and finalize().streaming["drift_events"] are currently derived from len(drift_log), and finalize() replays one warning per retained entry, so a cap needs a separate monotonic total beside it and should disclose how many entries it dropped.
3	S2	streaming memory bound	StreamingCleaner._exception_batches (_cleaner.py:93) retains the full DataFrame of every quarantined row for the whole stream: 20 batches → 361 rows / 29 KB; 400 batches → 7581 rows / 614 KB. A 100M-row stream with a few percent late data holds millions of rows in memory — the largest hole in "feeding 100M rows costs the same memory as feeding 100k". Proposed fix: a max_retained_exceptions cap on StreamingCleanConfig and/or a spill-to-parquet sink, keeping late_quarantined_total exact; last_exceptions_ already covers the per-batch case.
4	S2	streaming context policy	The policy is compiled once against batch 1, so context="Never modify bonus." is silently inert when bonus first appears in a later batch — the streaming imputer then writes into it. The only signal is a batch-1 unresolved-reference warning; batch 2 says nothing. docs/streaming.md documents the converse ("a batch that happens to be missing a column can't make the policy drift") but not this direction.
5	S3	streaming trust gate	An empty micro-batch (idle Kafka poll, trailing chunk) scores batch_trust_score == 0.0, fails fail_under_trust, counts as a gate failure (non-zero CLI exit) and drags the unweighted rolling trust down. Only the row-weighted cumulative score ignores it.
6	S3	streaming memory bound	StreamingState.columns / StreamingCleaner._roles are keyed by column name with no eviction, and each entry owns a full reservoir (20k floats by default) plus a counter. A schema-churning stream (partition- or date-named columns) grows state linearly in distinct column names.
7	S3	streaming dedup (spec gap)	global_duplicates=True compares a row only against earlier batches. A single batch de-duplicates nothing (12 identical-pattern rows → 12 out), and the surviving row count is a function of how the caller chunked the stream (2 batches → 6, 4 → 3, 12 → 3). The name reads as a global guarantee.
8	S3	memory robustness	config_overrides() copies any value stored under an allowed threshold key straight into CleanConfig unvalidated. A memory holding {"missing_threshold_low": "not-a-number"} takes down fd.clean with a bare TypeError: '<' not supported between instances of 'float' and 'str', naming neither the memory nor the field.
9	S3	memory identity	The SQLite store is keyed on dataset_id alone with ON CONFLICT DO UPDATE. Saving a memory for an unrelated schema under an id already present destroys the previous memory — different signature hash, different learned repairs — with no warning and no recovery. The plain-JSON path does guard identity.
10	S3	memory hints	Declaring a semantic_type does not block a colliding replay. Only allowed_values, mutable=False, or target/id/preserve protection do.
11	S4	memory versioning (spec gap)	CleaningMemory JSON carries only freshdata_version, which is written on learn and never read again — an ancient or future memory replays exactly like a current one. from_dict is a bare payload.get(...) per field with no type checking, so a foreign JSON document loads as an empty-but-valid memory (accepted stays a str). It is caught downstream by match(), so nothing is corrupted, but the diagnostic blames data drift rather than a bad file. Contrast freshdata.learning, which ships ProfileManifest / ProfileVersionError for exactly this.
12	S4	streaming time series	freshdata.streaming._timeseries.to_timedelta("1s") emits a numpy ≥ 2.5 DeprecationWarning ("generic unit for NumPy timedelta"), which will become an error. CI does not see it: filterwarnings only escalates warnings whose module is freshdata, and this one is raised inside pandas.

Verified safe (suspected, disproved)

• Reservoir quantiles are not batch-sensitive. Vitter-R here draws exactly one variate per post-fill element and keys on the global 1-indexed position, so median/Q1/Q3 are bit-identical at 1 / 10 / 100 / 1000 batches even with a 100-slot reservoir over 5000 rows.
• Space-Saving is exact and invariant until it saturates; total stays exact even after.
• All five memory-provenance signals hold under the collision (memory_influenced, the policy status, model_id suffix :memory, SemanticEvidence(kind="memory_replay"), backend="memory"), and survive report.to_dict() → JSON. Note the status nuance: a semantic replay carries the policy-gate status ("automatic"), while status="approved" is what annotate_report stamps on replayed non-semantic decisions and its own step="memory" summary.
• Target / id / preserve protection is never overridden by a replayed repair, and a memory whose stale roles say the column is an id protects it from its own replay.
• A renamed column blocks replay twice (signature match() fails and retrieval is keyed on the column name).
• A genuinely conflicting value is handled correctly: deterministic M -> m vs memory M -> male produces one unsafe_ambiguous record, risk="high", status="suggested", human_review=True, value never mutated.
• A corrupted or foreign memory never mutates data — match() refuses it with an explanatory warning.

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 935c5315-fe8e-46bc-b008-f815f695732e

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}