akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix SmartCrusher bugs: schema violation, race condition, thread safety, recursion

Browse source 
- Number array compression no longer mixes types (string summary was
  prepended to numeric array, violating schema-preserving guarantee).
  Statistics now go in the strategy string instead.
- Replace instance-level _current_field_semantics with threading.local()
  to prevent cross-thread contamination in concurrent crushes.
- Add lock to module-level _within_compressor lazy init (was unprotected).
- Add _MAX_PROCESS_DEPTH=50 guard to _process_value to prevent
  RecursionError on deeply nested JSON.
- Remove dead expression (unused stats.max_val - stats.min_val).
- Fix all UP038 isinstance(x, (A, B)) -> isinstance(x, A | B) across file.
- Add 11 regression tests covering all fixes.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
chopratejas 2026-04-14 21:16:13 -07:00
parent 6cc8fd04b6
commit 14415dbbb5
2 changed files with 272 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

108
headroom/transforms/smart_crusher.py
View file

@ -180,27 +180,31 @@ def _hash_field_name(field_name: str) -> str:
# Minimum chars for a text field to be worth compressing within an item
_MIN_FIELD_CHARS_FOR_WITHIN = 200
# Lazy-loaded compressor for within-item text compression
# Lazy-loaded compressor for within-item text compression (thread-safe)
_within_compressor: Any = None
_within_compressor_checked = False
_within_compressor_lock = threading.Lock()
def _get_within_compressor() -> Any:
"""Get a text compressor for within-item field compression.
Returns Kompress if available (requires [ml] extra), else None.
Thread-safe via double-checked locking.
"""
global _within_compressor, _within_compressor_checked
if not _within_compressor_checked:
_within_compressor_checked = True
try:
from .kompress_compressor import KompressCompressor, is_kompress_available
with _within_compressor_lock:
if not _within_compressor_checked:
try:
from .kompress_compressor import KompressCompressor, is_kompress_available
if is_kompress_available():
_within_compressor = KompressCompressor()
logger.debug("Within-item compression: using Kompress")
except ImportError:
pass
if is_kompress_available():
_within_compressor = KompressCompressor()
logger.debug("Within-item compression: using Kompress")
except ImportError:
pass
_within_compressor_checked = True
return _within_compressor
@ -435,7 +439,7 @@ def _detect_sequential_pattern(values: list[Any], check_order: bool = True) -> b
# Get numeric values
nums = []
for v in values:
if isinstance(v, (int, float)) and not isinstance(v, bool):
if isinstance(v, int | float) and not isinstance(v, bool):
nums.append(v)
elif isinstance(v, str):
try:
@ -546,7 +550,6 @@ def _detect_score_field_statistically(stats: FieldStats, items: list[dict]) -> t
confidence = 0.0
# Check for bounded range typical of scores
stats.max_val - stats.min_val
min_val, max_val = stats.min_val, stats.max_val
# Common score ranges: [0,1], [0,10], [0,100], [-1,1], [0,5]
@ -578,7 +581,7 @@ def _detect_score_field_statistically(stats: FieldStats, items: list[dict]) -> t
for item in items:
if stats.name in item:
val = item.get(stats.name)
if isinstance(val, (int, float)) and math.isfinite(val):
if isinstance(val, int | float) and math.isfinite(val):
values_in_order.append(float(val))
if len(values_in_order) >= 5:
# Check for descending sort
@ -804,11 +807,11 @@ def _detect_items_by_learned_semantics(
value_canonical = "null"
elif isinstance(value, bool):
value_canonical = "true" if value else "false"
elif isinstance(value, (int, float)):
elif isinstance(value, int | float):
value_canonical = str(value)
elif isinstance(value, str):
value_canonical = value
elif isinstance(value, (list, dict)):
elif isinstance(value, list | dict):
try:
value_canonical = json.dumps(value, sort_keys=True, default=str)
except (TypeError, ValueError):
@ -1030,7 +1033,7 @@ class SmartAnalyzer:
first_val = non_null_values[0]
if isinstance(first_val, bool):
field_type = "boolean"
elif isinstance(first_val, (int, float)):
elif isinstance(first_val, int | float):
field_type = "numeric"
elif isinstance(first_val, str):
field_type = "string"
@ -1064,7 +1067,7 @@ class SmartAnalyzer:
# Numeric-specific analysis
if field_type == "numeric":
# Filter out NaN and Infinity which break statistics functions
nums = [v for v in non_null_values if isinstance(v, (int, float)) and math.isfinite(v)]
nums = [v for v in non_null_values if isinstance(v, int | float) and math.isfinite(v)]
if nums:
try:
stats.min_val = min(nums)
@ -1283,7 +1286,7 @@ class SmartAnalyzer:
threshold = self.config.variance_threshold * std
for i, item in enumerate(items):
val = item.get(stats.name)
if isinstance(val, (int, float)):
if isinstance(val, int | float):
if abs(val - stats.mean_val) > threshold:
anomaly_indices.add(i)
@ -1953,9 +1956,9 @@ class SmartCrusher(Transform):
if len(keep_indices) <= effective_max:
return keep_indices
# Use provided field_semantics or fall back to instance variable (set by crush())
# Use provided field_semantics or fall back to thread-local (set by _crush_array)
effective_field_semantics = field_semantics or getattr(
self, "_current_field_semantics", None
getattr(self, "_thread_local", None), "field_semantics", None
)
# Identify error items using KEYWORD detection (preservation guarantee)
@ -1976,7 +1979,7 @@ class SmartCrusher(Transform):
threshold = self.config.variance_threshold * std
for i, item in enumerate(items):
val = item.get(field_name)
if isinstance(val, (int, float)):
if isinstance(val, int | float):
if abs(val - stats.mean_val) > threshold:
anomaly_indices.add(i)
@ -2297,6 +2300,10 @@ class SmartCrusher(Transform):
return result, was_modified, info
# Maximum recursion depth for nested JSON processing.
# Prevents RecursionError on adversarial/deeply-nested input.
_MAX_PROCESS_DEPTH = 50
def _process_value(
self,
value: Any,
@ -2311,6 +2318,10 @@ class SmartCrusher(Transform):
Tuple of (processed_value, info_string, ccr_markers).
ccr_markers is a list of (hash, original_count, compressed_count, summary) tuples.
"""
# Guard against deeply nested JSON causing RecursionError
if depth >= self._MAX_PROCESS_DEPTH:
return value, "", []
info_parts = []
ccr_markers: list[tuple] = []
@ -2495,9 +2506,12 @@ class SmartCrusher(Transform):
)
# === TOIN Evolution: Extract field semantics for signal detection ===
# Store temporarily on instance for use in _prioritize_indices
# Store in thread-local storage for use in _prioritize_indices.
# This enables learned signal detection without changing all method signatures
self._current_field_semantics = (
# while remaining thread-safe (no cross-thread contamination).
if not hasattr(self, "_thread_local"):
self._thread_local = threading.local()
self._thread_local.field_semantics = (
toin_hint.field_semantics if toin_hint.field_semantics else None
)
@ -2661,12 +2675,14 @@ class SmartCrusher(Transform):
)
# Clean up temporary instance variable
self._current_field_semantics = None
if hasattr(self, "_thread_local"):
self._thread_local.field_semantics = None
return result, strategy_info, ccr_hash, dropped_summary
except Exception:
# Clean up temporary instance variable
self._current_field_semantics = None
if hasattr(self, "_thread_local"):
self._thread_local.field_semantics = None
# Re-raise any exceptions (removed finally block since we no longer mutate config)
raise
@ -2814,7 +2830,7 @@ class SmartCrusher(Transform):
return items, "number:passthrough"
# Filter out non-finite values for statistics
finite = [x for x in items if isinstance(x, (int, float)) and math.isfinite(x)]
finite = [x for x in items if isinstance(x, int | float) and math.isfinite(x)]
if not finite:
return items, "number:no_finite"
@ -2832,7 +2848,7 @@ class SmartCrusher(Transform):
outlier_indices: set[int] = set()
if std_val > 0:
for i, val in enumerate(items):
if isinstance(val, (int, float)) and math.isfinite(val):
if isinstance(val, int | float) and math.isfinite(val):
if abs(val - mean_val) > self.config.variance_threshold * std_val:
outlier_indices.add(i)
@ -2844,12 +2860,12 @@ class SmartCrusher(Transform):
left = [
items[j]
for j in range(i - window, i)
if isinstance(items[j], (int, float)) and math.isfinite(items[j])
if isinstance(items[j], int | float) and math.isfinite(items[j])
]
right = [
items[j]
for j in range(i, i + window)
if isinstance(items[j], (int, float)) and math.isfinite(items[j])
if isinstance(items[j], int | float) and math.isfinite(items[j])
]
if left and right:
left_mean = statistics.mean(left)
@ -2877,27 +2893,23 @@ class SmartCrusher(Transform):
if i not in keep_indices:
keep_indices.add(i)
# Build output: summary string + kept values in original order
stats_summary = (
f"[{n} numbers: min={min(finite)}, max={max(finite)}, "
f"mean={mean_val:.4g}, median={median_val:.4g}, "
f"stddev={std_val:.4g}, p25={p25:.4g}, p75={p75:.4g}"
# Build output: kept values only (schema-preserving — no generated text)
kept_values = [items[i] for i in sorted(keep_indices)]
# Encode statistics into the strategy string (not the array itself)
strategy = (
f"number:adaptive({n}->{len(kept_values)}"
f",min={min(finite)},max={max(finite)}"
f",mean={mean_val:.4g},median={median_val:.4g}"
f",stddev={std_val:.4g},p25={p25:.4g},p75={p75:.4g}"
)
if outlier_indices:
stats_summary += f", outliers={len(outlier_indices)}"
if change_indices:
stats_summary += f", change_points={len(change_indices)}"
stats_summary += "]"
kept_values = [items[i] for i in sorted(keep_indices)]
result: list = [stats_summary] + kept_values
strategy = f"number:adaptive({n}->{len(kept_values)}"
if outlier_indices:
strategy += f",outliers={len(outlier_indices)}"
if change_indices:
strategy += f",change_points={len(change_indices)}"
strategy += ")"
return result, strategy
return kept_values, strategy
def _crush_mixed_array(
self,
@ -2930,7 +2942,7 @@ class SmartCrusher(Transform):
key = "str"
elif isinstance(item, bool):
key = "bool"
elif isinstance(item, (int, float)):
elif isinstance(item, int | float):
key = "number"
elif isinstance(item, list):
key = "list"
@ -2979,13 +2991,13 @@ class SmartCrusher(Transform):
last_idx = set(indices[-k_last:])
keep_indices.update(first_idx | last_idx)
# Outliers
finite = [v for v in values if isinstance(v, (int, float)) and math.isfinite(v)]
finite = [v for v in values if isinstance(v, int | float) and math.isfinite(v)]
if len(finite) > 1:
mean_v = statistics.mean(finite)
std_v = statistics.stdev(finite)
if std_v > 0:
for idx, val in group_items:
if isinstance(val, (int, float)) and math.isfinite(val):
if isinstance(val, int | float) and math.isfinite(val):
if abs(val - mean_v) > self.config.variance_threshold * std_v:
keep_indices.add(idx)
strategy_parts.append(f"num:{len(values)}")
@ -3553,7 +3565,7 @@ class SmartCrusher(Transform):
threshold = self.config.variance_threshold * std
for i, item in enumerate(items):
val = item.get(name)
if isinstance(val, (int, float)):
if isinstance(val, int | float):
if abs(val - stats.mean_val) > threshold:
keep_indices.add(i)

212
tests/test_transforms/test_smart_crusher_bugs.py Normal file
View file

@ -0,0 +1,212 @@
"""Regression tests for SmartCrusher bugs.
Bug 1: _crush_number_array mixes types (string summary + numbers),
violating the schema-preserving guarantee.
Bug 2: _current_field_semantics is shared instance state, creating
a race condition when crushing concurrently.
"""
from __future__ import annotations
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from headroom import SmartCrusherConfig
from headroom.transforms.smart_crusher import SmartCrusher
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_crusher(max_items: int = 10, min_items: int = 3) -> SmartCrusher:
config = SmartCrusherConfig(
enabled=True,
min_items_to_analyze=min_items,
min_tokens_to_crush=0,
max_items_after_crush=max_items,
variance_threshold=2.0,
)
return SmartCrusher(config=config)
# ---------------------------------------------------------------------------
# Bug 1: Number array type mixing
# ---------------------------------------------------------------------------
class TestNumberArraySchemaPreservation:
"""_crush_number_array must return only original numeric values.
Previously it prepended a stats summary string, producing
[string, int, int, ...] which violates the schema-preserving
guarantee and breaks type-aware JSON consumers.
"""
def test_crushed_number_array_contains_only_numbers(self) -> None:
"""Every element of the crushed array must be int or float."""
crusher = _make_crusher(max_items=10)
numbers = list(range(50)) # 0..49, well above the n<=8 passthrough
crushed, strategy = crusher._crush_number_array(numbers)
for i, item in enumerate(crushed):
assert isinstance(item, int | float), (
f"Item {i} is {type(item).__name__} = {item!r}, expected int/float. "
f"Schema-preserving guarantee violated."
)
def test_crushed_number_array_subset_of_original(self) -> None:
"""Every value in the crushed array must exist in the original."""
crusher = _make_crusher(max_items=10)
numbers = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120]
crushed, _ = crusher._crush_number_array(numbers)
original_set = set(numbers)
for item in crushed:
assert item in original_set, (
f"Value {item!r} not in original array — generated content detected"
)
def test_stats_summary_in_strategy_not_in_array(self) -> None:
"""Statistics should be communicated via strategy string, not array content."""
crusher = _make_crusher(max_items=5)
numbers = list(range(100))
crushed, strategy = crusher._crush_number_array(numbers)
# Strategy should contain stats info
assert "number:" in strategy
# Array should not contain any strings
strings_in_result = [x for x in crushed if isinstance(x, str)]
assert strings_in_result == [], f"Found string(s) in numeric array: {strings_in_result}"
def test_number_array_passthrough_for_small(self) -> None:
"""Arrays with n <= 8 should pass through unchanged."""
crusher = _make_crusher()
small = [1, 2, 3, 4, 5]
crushed, strategy = crusher._crush_number_array(small)
assert crushed == small
assert strategy == "number:passthrough"
def test_number_array_preserves_outliers(self) -> None:
"""Outlier values should be preserved in the crushed output."""
crusher = _make_crusher(max_items=10)
# Normal range + extreme outlier
numbers = [10] * 20 + [10000]
crushed, strategy = crusher._crush_number_array(numbers)
assert 10000 in crushed, "Outlier value 10000 was dropped"
def test_number_array_preserves_boundaries(self) -> None:
"""First and last values should always be kept."""
crusher = _make_crusher(max_items=5)
numbers = list(range(100))
crushed, strategy = crusher._crush_number_array(numbers)
assert crushed[0] == 0, "First value not preserved"
assert numbers[-1] in crushed, "Last value not preserved"
def test_non_finite_passthrough(self) -> None:
"""All-NaN/Inf arrays should return unchanged."""
crusher = _make_crusher()
nans = [float("nan")] * 10
crushed, strategy = crusher._crush_number_array(nans)
assert strategy == "number:no_finite"
assert len(crushed) == 10
def test_full_crush_pipeline_number_array_types(self) -> None:
"""End-to-end: crushing a JSON number array via the public API."""
crusher = _make_crusher(max_items=10)
content = json.dumps(list(range(50)))
result, was_modified, info = crusher._smart_crush_content(content)
if was_modified:
parsed = json.loads(result)
assert isinstance(parsed, list)
for item in parsed:
assert isinstance(item, int | float), (
f"Public API returned non-numeric item {item!r} in number array"
)
# ---------------------------------------------------------------------------
# Bug 2: Race condition on _current_field_semantics
# ---------------------------------------------------------------------------
class TestFieldSemanticsThreadSafety:
"""_current_field_semantics must not leak between concurrent crushes.
Previously it was stored as instance state (self._current_field_semantics)
which created a race condition when the same SmartCrusher instance
was used from multiple threads.
"""
def test_concurrent_crushes_no_cross_contamination(self) -> None:
"""Two concurrent crushes must not share field_semantics state."""
crusher = _make_crusher(max_items=5)
# Two different array payloads
payload_a = json.dumps([{"name": f"item_{i}", "value": i} for i in range(20)])
payload_b = json.dumps([{"key": f"k_{i}", "score": i * 0.1} for i in range(20)])
results: dict[str, str] = {}
errors: list[Exception] = []
def crush_task(label: str, content: str) -> None:
try:
result, modified, info = crusher._smart_crush_content(content)
results[label] = result
except Exception as e:
errors.append(e)
with ThreadPoolExecutor(max_workers=4) as executor:
futures = []
# Run many concurrent crushes to increase race probability
for i in range(20):
futures.append(executor.submit(crush_task, f"a_{i}", payload_a))
futures.append(executor.submit(crush_task, f"b_{i}", payload_b))
for f in as_completed(futures):
f.result() # Re-raise exceptions
assert not errors, f"Concurrent crushes raised errors: {errors}"
# After all crushes, thread-local state must be clean
tl = getattr(crusher, "_thread_local", None)
if tl is not None:
semantics = getattr(tl, "field_semantics", None)
assert semantics is None, f"field_semantics leaked in thread-local: {semantics}"
# ---------------------------------------------------------------------------
# Issue 7: Recursion depth limit
# ---------------------------------------------------------------------------
class TestRecursionDepthLimit:
"""_process_value must not crash on deeply nested JSON."""
def test_deeply_nested_json_does_not_crash(self) -> None:
"""Nesting deeper than _MAX_PROCESS_DEPTH should return value unchanged."""
crusher = _make_crusher()
# Build a 100-level nested structure
nested: dict = {"leaf": "value"}
for _i in range(100):
nested = {"level": nested}
content = json.dumps(nested)
result, was_modified, info = crusher._smart_crush_content(content)
# Should not raise RecursionError
parsed = json.loads(result)
# The deep structure should be preserved (returned as-is past depth limit)
assert isinstance(parsed, dict)
def test_deeply_nested_list_does_not_crash(self) -> None:
"""Deeply nested lists should also be handled safely."""
crusher = _make_crusher()
nested: list = ["leaf"]
for _i in range(100):
nested = [nested]
content = json.dumps(nested)
result, was_modified, info = crusher._smart_crush_content(content)
parsed = json.loads(result)
assert isinstance(parsed, list)