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

feat(rust): retire python smart_crusher, ship rust-only via pyo3

Browse source 
Stage 3c.1b step 2 + cleanup. The python `SmartCrusher` (3669 lines)
is replaced by a thin pyo3-backed shim (~290 lines) that delegates
every byte to `headroom._core.SmartCrusher` (built from
`crates/headroom-py`, landed in the previous commit). There is no
python implementation and no env-var fallback — the wheel is a hard
import.

Why now: parity was already proven across 17 fixtures + the python-
side bridge test (1+17 in `test_smart_crusher_rust_parity.py`).
Keeping a shadow python impl behind a flag is a permanent maintenance
cost with no operational benefit. Stage 3c.1b deletes ~3380 lines of
python parser/scorer/analyzer/orchestrator code; the rust crate has
its own coverage (388 unit tests + property tests in headroom-core).

Surface preserved (drop-in for every production caller):
- `headroom.transforms.smart_crusher.SmartCrusher` — same class name,
  same `__init__(config, relevance_config, scorer, ccr_config)`
  signature (the latter three are accepted for source-compat and
  silently dropped — rust port keeps those subsystems disabled in
  Stage 3c.1, they re-attach in Stage 3c.2).
- `SmartCrusherConfig` and `CrushResult` dataclasses kept as python
  dataclasses (callers use `asdict()` / dataclass matching on them).
- `crush(content, query, bias)`, `_smart_crush_content(content, ...)`,
  `apply(messages, tokenizer, **kwargs)`, and
  `_extract_context_from_messages(messages)` all preserved.
- `smart_crush_tool_output(content, config, ccr_config)` thin wrapper.

The transform-protocol `apply()` orchestration stays python (message
walking, digest-marker insertion, token counting); only the per-
message compression call delegates to rust.

Removed:
- Python parser / planner / scorer / analyzer / classifier (~3380 lines).
- Internal helpers `_classify_array`, `_detect_sequential_pattern`,
  `_detect_rare_status_values`, `_detect_items_by_learned_semantics`,
  `_percentile_linear`, `_compute_k_split`, `_crush_number_array`,
  `_process_value`, etc. — rust crate has parallel coverage.
- `SmartAnalyzer`, `ArrayType`, `CompressionStrategy`,
  `extract_query_anchors` — internals; not used by any production
  caller (only tests probed them).

Tests deleted (probed deleted internals — same precedent as Stage 3b):
- `tests/test_transforms/test_smart_crusher.py` (40 tests)
- `tests/test_transforms/test_universal_json_crush.py` (45)
- `tests/test_transforms/test_anchor_selector.py` (49)
- `tests/test_toin_field_learning.py` (21)
- `tests/test_crushability.py` (20)

Tests trimmed (removed methods/classes that probe deferred subsystems
— scorer injection, CCR marker injection, TOIN feedback recording —
all of which re-attach in Stage 3c.2):
- `tests/test_transforms/test_smart_crusher_bugs.py`:
  TestNumberArraySchemaPreservation, TestStage3c1BugFixes.
- `tests/test_relevance.py`: 2 scorer-injection tests.
- `tests/test_ccr.py`: TestSmartCrusherCCRIntegration class +
  test_custom_marker_template.
- `tests/test_toin_integration.py`: TestTOINIntegration +
  TestStoreToTOINHash classes.
- `tests/test_critical_fixes.py`: TestSmartCrusherTOINIntegration +
  test_full_feedback_loop.
- `tests/test_acceptance.py::TestQueryAnchorExtraction`: dropped the
  `extract_query_anchors` probe; kept the end-to-end "Alice
  preserved" assertion.

Bug fixes from Stage 3c.1 (#1 percentile linear interp, #2 zero-
padded sequential, #3 rare-status pareto, #4 k-split overshoot) are
pinned by the rust crate and the parity fixtures
(`tests/parity/fixtures/smart_crusher/`).

Tests:
- 517 passed in the smart_crusher-adjacent file set
  (test_transforms/, test_relevance*, test_ccr, test_toin_integration,
  test_quality_retention, test_acceptance, test_critical_fixes).
- 18 in `test_smart_crusher_rust_parity.py` (1 sanity + 17 fixtures).
- 388 rust unit tests still green.

One stale-error-message regex in `test_relevance_extra.py` updated
from "requires sentence-transformers" → "requires fastembed".
This commit is contained in:
chopratejas 2026-04-27 00:52:21 -07:00
parent 5328d87b1e
commit c765c53bf8
13 changed files with 222 additions and 8674 deletions
Download patch file Download diff file Expand all files Collapse all files

3802
headroom/transforms/smart_crusher.py
View file

File diff suppressed because it is too large Load diff

30
tests/test_acceptance.py
View file

@ -447,11 +447,7 @@ class TestQueryAnchorExtraction:
"""If user asks for 'Alice', item with Alice should be preserved."""
import json
from headroom.transforms.smart_crusher import (
SmartCrusher,
SmartCrusherConfig,
extract_query_anchors,
)
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
# User is searching for 'Alice'
messages = [
@ -478,38 +474,31 @@ class TestQueryAnchorExtraction:
},
]
# Verify anchor extraction works
anchors = extract_query_anchors("Find the user named 'Alice' in the system.")
assert "alice" in anchors
# Verify crushing preserves Alice
# End-to-end behavior: the relevance scorer (HybridScorer in
# the Rust port — BM25 + embedding) should pick up "Alice"
# from the user message and preserve the matching tool item
# even though it sits at index 50.
config = SmartCrusherConfig(
enabled=True,
min_items_to_analyze=5,
min_tokens_to_crush=100,
max_items_after_crush=10, # Should normally drop Alice at index 50
max_items_after_crush=10,
)
crusher = SmartCrusher(config)
tokenizer = get_tokenizer()
result = crusher.apply(messages, tokenizer)
# Find the crushed tool output
tool_msg = next(m for m in result.messages if m.get("role") == "tool")
crushed_content = tool_msg["content"]
# Alice should be preserved even though she's at index 50
assert "Alice" in crushed_content
def test_preserves_needle_by_uuid(self):
"""If user asks for a UUID, item with that UUID should be preserved."""
import json
from headroom.transforms.smart_crusher import (
SmartCrusher,
SmartCrusherConfig,
extract_query_anchors,
)
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
target_uuid = "550e8400-e29b-41d4-a716-446655440000"
@ -537,10 +526,6 @@ class TestQueryAnchorExtraction:
},
]
# Verify anchor extraction
anchors = extract_query_anchors(f"Get details for request {target_uuid}")
assert target_uuid.lower() in anchors
config = SmartCrusherConfig(
enabled=True,
min_items_to_analyze=5,
@ -555,7 +540,6 @@ class TestQueryAnchorExtraction:
tool_msg = next(m for m in result.messages if m.get("role") == "tool")
crushed_content = tool_msg["content"]
# UUID should be preserved
assert target_uuid in crushed_content

172
tests/test_ccr.py
View file

@ -235,160 +235,6 @@ class TestCompressionStore:
assert store.exists(h)
class TestSmartCrusherCCRIntegration:
"""Test SmartCrusher integration with CCR."""
@pytest.fixture(autouse=True)
def reset_store(self):
"""Reset global store before each test."""
reset_compression_store()
yield
reset_compression_store()
def test_compression_caches_original(self):
"""SmartCrusher caches original content when compressing."""
items = [{"id": i, "score": 100 - i, "data": f"item_{i}"} for i in range(100)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
ccr_config = CCRConfig(
enabled=True,
inject_retrieval_marker=False, # Don't add marker for this test
min_items_to_cache=10,
)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
assert was_modified
# Check that content was cached
store = get_compression_store()
stats = store.get_stats()
assert stats["entry_count"] >= 1
def test_retrieval_marker_injected(self):
"""CCR marker is injected when configured."""
items = [{"id": i, "score": 100 - i, "data": f"item_{i}"} for i in range(100)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
ccr_config = CCRConfig(
enabled=True,
inject_retrieval_marker=True,
min_items_to_cache=10,
)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
assert was_modified
# Marker should be present
assert "items compressed" in compressed_str or "hash=" in compressed_str
def test_small_arrays_not_cached(self):
"""Arrays smaller than min_items_to_cache are not cached."""
items = [{"id": i} for i in range(15)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=10)
ccr_config = CCRConfig(
enabled=True,
min_items_to_cache=50, # Require 50+ items
)
smart_crush_tool_output(content, config, ccr_config)
store = get_compression_store()
stats = store.get_stats()
# Should not cache because original has < 50 items
assert stats["entry_count"] == 0
def test_uncrushed_data_not_cached(self):
"""Data that doesn't get crushed is not cached."""
# DB results with unique IDs - shouldn't be crushed
items = [{"id": i, "name": f"User {i}", "email": f"user{i}@test.com"} for i in range(30)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=10)
ccr_config = CCRConfig(enabled=True, min_items_to_cache=10)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
# If not modified, shouldn't be cached
if not was_modified:
store = get_compression_store()
stats = store.get_stats()
assert stats["entry_count"] == 0
def test_can_retrieve_after_compression(self):
"""Can retrieve original content after compression."""
items = [
{"id": i, "score": 100 - i, "content": f"Document about topic {i}"} for i in range(100)
]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
ccr_config = CCRConfig(
enabled=True,
inject_retrieval_marker=True,
min_items_to_cache=10,
)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
assert was_modified
# Extract hash from marker
# Marker format: [100 items compressed to 15. Retrieve more: hash=abc123...]
import re
match = re.search(r"hash=([a-f0-9]+)", compressed_str)
assert match is not None, f"No hash found in: {compressed_str}"
hash_key = match.group(1)
# Retrieve original
store = get_compression_store()
entry = store.retrieve(hash_key)
assert entry is not None
original_items = json.loads(entry.original_content)
assert len(original_items) == 100
def test_search_after_compression(self):
"""Can search within original content after compression."""
items = [
{"id": 1, "content": "Authentication error: invalid token"},
{"id": 2, "content": "Database connection successful"},
{"id": 3, "content": "User login completed"},
] + [{"id": i, "content": f"Generic log entry {i}"} for i in range(4, 104)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
ccr_config = CCRConfig(
enabled=True,
inject_retrieval_marker=True,
min_items_to_cache=10,
)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
assert was_modified
# Extract hash
import re
match = re.search(r"hash=([a-f0-9]+)", compressed_str)
hash_key = match.group(1)
# Search for authentication items
store = get_compression_store()
results = store.search(hash_key, "authentication error token")
# Should find the authentication error item
assert len(results) >= 1
assert any("Authentication" in r.get("content", "") for r in results)
class TestCCRConfig:
"""Test CCR configuration options."""
@ -402,24 +248,6 @@ class TestCCRConfig:
assert config.feedback_enabled is True
assert config.min_items_to_cache == 20
def test_custom_marker_template(self):
"""Custom marker template is used."""
items = [{"id": i, "score": 100 - i} for i in range(100)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
ccr_config = CCRConfig(
enabled=True,
inject_retrieval_marker=True,
min_items_to_cache=10,
marker_template="\n[CUSTOM: {original_count} -> {compressed_count}, key={hash}]",
)
compressed_str, was_modified, _ = smart_crush_tool_output(content, config, ccr_config)
if was_modified:
assert "CUSTOM:" in compressed_str or "key=" in compressed_str
class TestCCRFeedbackLoop:
"""Test CCR feedback tracking for learning."""

143
tests/test_critical_fixes.py
View file

@ -11,8 +11,6 @@ These tests verify the before/after behavior of critical bug fixes:
import time
from unittest.mock import patch
import pytest
class TestTOINConfidenceMathFix:
"""Test for CRITICAL: Confidence calculation math error in toin.py:721.
@ -303,146 +301,5 @@ class TestUnboundedStrategyDicts:
)
class TestSmartCrusherTOINIntegration:
"""Test for CRITICAL: SmartCrusher not calling toin.record_compression().
BUG: SmartCrusher calls feedback.record_compression() but never calls
toin.record_compression(). This means TOIN only learns from retrieval events,
not from compression events - breaking the feedback loop.
FIX: Add toin.record_compression() call after compression in SmartCrusher.
"""
def test_smart_crusher_records_to_toin(self):
"""SmartCrusher should record compression events to TOIN."""
from headroom.telemetry.models import ToolSignature
from headroom.telemetry.toin import get_toin, reset_toin
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
reset_toin()
config = SmartCrusherConfig(
min_items_to_analyze=5,
max_items_after_crush=10,
use_feedback_hints=True,
)
crusher = SmartCrusher(config)
# Create test items that look like search results with a clear score field
# This pattern is crushable because:
# 1. Has a clear numeric score field in BOUNDED range [0,1]
# 2. Has repeated structure with some constant fields (type, language)
# 3. Score values vary within the bounded range
items = [
{
"name": f"repo_{i}",
"relevance_score": (50 - i) / 50.0, # Bounded [0,1] - descending order
"type": "repository", # Constant field
"language": "python" if i % 3 == 0 else "javascript", # Low cardinality
"description": f"Description {i % 5}", # Low cardinality
}
for i in range(50)
]
# Get TOIN instance and check initial state
toin = get_toin()
len(toin._patterns)
# Crush the array
result, info, markers, _summary = crusher._crush_array(
items, query_context="test query", tool_name="test_tool"
)
# Verify compression happened (not skipped)
assert "skip" not in info.lower(), (
f"Compression was skipped: {info}. Test needs crushable data."
)
# Get the signature that would have been created
sig = ToolSignature.from_items(items)
# Check TOIN was notified
with toin._lock:
pattern = toin._patterns.get(sig.structure_hash)
# After fix, TOIN should have a pattern for this tool's signature
assert pattern is not None, (
f"TOIN should have recorded the compression event. "
f"Info: {info}, pattern count: {len(toin._patterns)}"
)
if pattern:
assert pattern.total_compressions >= 1, (
f"Pattern should have at least 1 compression recorded, got {pattern.total_compressions}"
)
class TestAllFixesIntegrated:
"""Integration tests ensuring all fixes work together."""
def test_full_feedback_loop(self):
"""Test complete feedback loop: compress -> store -> retrieve -> learn."""
from headroom.cache.compression_feedback import (
reset_compression_feedback,
)
from headroom.cache.compression_store import reset_compression_store
from headroom.telemetry.models import ToolSignature
from headroom.telemetry.toin import get_toin, reset_toin
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
# Reset all singletons
reset_toin()
reset_compression_store()
reset_compression_feedback()
# Setup
config = SmartCrusherConfig(
min_items_to_analyze=5,
max_items_after_crush=10,
use_feedback_hints=True,
)
crusher = SmartCrusher(config)
# Create test items that look like API responses with scoring
# This pattern is crushable because:
# 1. Has a clear numeric score field in BOUNDED range [0,1]
# 2. Has constant fields (status, type)
# 3. Has enough items for compression (100)
items = [
{
"priority": (100 - i) / 100.0, # Bounded [0,1] - descending order
"status": "ok", # Constant field
"type": "response", # Constant field
"data": f"content_{i % 10}", # Low cardinality (only 10 unique values)
}
for i in range(100)
]
# Step 1: Compress
result, info, markers, _summary = crusher._crush_array(
items, query_context="find status", tool_name="api_response"
)
# Verify compression happened (not skipped)
assert "skip" not in info.lower(), (
f"Compression was skipped: {info}. Test needs crushable data."
)
# Step 2: Check TOIN was notified (after fix)
toin = get_toin()
sig = ToolSignature.from_items(items)
with toin._lock:
toin_pattern = toin._patterns.get(sig.structure_hash)
# After fix, TOIN should have the pattern
assert toin_pattern is not None, (
f"TOIN should have learned from the compression event. Info: {info}"
)
assert toin_pattern.total_compressions >= 1, (
f"TOIN pattern should have recorded compression, got {toin_pattern.total_compressions}"
)
# Run specific test to verify fix
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])

423
tests/test_crushability.py
View file

@ -1,423 +0,0 @@
"""Tests for SmartCrusher crushability analysis.
These tests verify that SmartCrusher correctly identifies when it's SAFE
to crush data vs when it should SKIP crushing.
The key insight: High variability + No importance signal = DON'T CRUSH.
Test scenarios:
1. DB results (unique entities, no signal) → SKIP
2. Search results (has score field) → CRUSH using score
3. Log entries (has errors) → CRUSH keeping errors
4. Time series (has anomalies) → CRUSH keeping anomalies
5. Repetitive data (low uniqueness) → CRUSH with sampling
"""
import json
import pytest
from headroom.transforms.smart_crusher import (
CompressionStrategy,
SmartAnalyzer,
SmartCrusherConfig,
smart_crush_tool_output,
)
class TestCrushabilityDetection:
"""Test the crushability analysis logic."""
@pytest.fixture
def analyzer(self):
"""Create a SmartAnalyzer instance."""
return SmartAnalyzer(SmartCrusherConfig())
def test_db_results_not_crushable(self, analyzer):
"""DB query results with unique IDs and no signal should NOT be crushed."""
# Simulate: SELECT * FROM users LIMIT 50
items = [
{
"id": i,
"name": f"User {i}",
"email": f"user{i}@example.com",
"department": "Engineering",
}
for i in range(50)
]
analysis = analyzer.analyze_array(items)
# Should detect unique entities with no importance signal
assert analysis.crushability is not None
assert not analysis.crushability.crushable, (
f"DB results should NOT be crushable. "
f"Reason: {analysis.crushability.reason}, "
f"Signals: {analysis.crushability.signals_present}"
)
assert analysis.recommended_strategy == CompressionStrategy.SKIP
assert "unique" in analysis.crushability.reason.lower()
def test_db_results_with_unique_uuid(self, analyzer):
"""DB results with UUID field should NOT be crushed."""
items = [
{
"uuid": f"550e8400-e29b-41d4-a716-44665544{i:04d}",
"name": f"Record {i}",
"value": i * 10,
}
for i in range(50)
]
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert not analysis.crushability.crushable
assert analysis.crushability.has_id_field
def test_search_results_crushable(self, analyzer):
"""Search results with score field SHOULD be crushed."""
items = [
{
"id": i,
"title": f"Document {i}",
"snippet": f"This is document {i} content...",
"score": 1.0 - (i * 0.01), # Decreasing relevance
}
for i in range(100)
]
analysis = analyzer.analyze_array(items)
# Should detect score field as importance signal
assert analysis.crushability is not None
assert analysis.crushability.crushable, (
f"Search results should be crushable. Reason: {analysis.crushability.reason}"
)
assert analysis.crushability.has_score_field
assert any("score" in s for s in analysis.crushability.signals_present)
def test_log_entries_with_errors_crushable(self, analyzer):
"""Log entries containing structural outliers SHOULD be crushed (outliers preserved)."""
items = []
for i in range(100):
item = {
"id": i,
"timestamp": f"2024-01-15T10:{i:02d}:00Z",
"message": f"Request processed successfully - {i}",
"level": "INFO",
}
# Add some errors - these are STRUCTURAL OUTLIERS (have extra "error" field)
if i % 20 == 0:
item["level"] = "ERROR"
item["message"] = f"Connection failed: timeout at {i}"
item["error"] = "TimeoutError" # Extra field that most items don't have
items.append(item)
analysis = analyzer.analyze_array(items)
# Should detect structural outliers (items with rare fields like "error")
assert analysis.crushability is not None
assert analysis.crushability.crushable
# Now uses structural_outliers instead of keyword-based error count
assert any(
"structural_outliers" in s or "outlier" in s.lower()
for s in analysis.crushability.signals_present
)
def test_time_series_with_anomalies_crushable(self, analyzer):
"""Time series with numeric anomalies SHOULD be crushed."""
items = []
for i in range(100):
value = 100.0 # Normal value
if i in [25, 50, 75]: # Anomaly points
value = 999.0
items.append(
{
"id": i,
"timestamp": i,
"cpu_usage": value,
}
)
analysis = analyzer.analyze_array(items)
# Should detect anomalies as importance signal
assert analysis.crushability is not None
assert analysis.crushability.crushable
assert analysis.crushability.anomaly_count > 0
def test_repetitive_data_crushable(self, analyzer):
"""Repetitive data (low uniqueness) SHOULD be crushable."""
# Same status repeated many times
items = [
{
"id": i,
"status": "success", # Same for all
"code": 200, # Same for all
"message": "OK", # Same for all
}
for i in range(100)
]
analysis = analyzer.analyze_array(items)
# Should detect low uniqueness - safe to sample
assert analysis.crushability is not None
assert analysis.crushability.crushable
# Can be "low_uniqueness" or "repetitive_content_with_ids"
assert (
"low_uniqueness" in analysis.crushability.reason
or "repetitive" in analysis.crushability.reason
)
def test_file_listing_not_crushable(self, analyzer):
"""File listing with unique paths should NOT be crushed."""
items = [
{
"id": i,
"path": f"/home/user/project/src/module{i}/file{i}.py",
"size": 1000 + i,
"modified": f"2024-01-{(i % 28) + 1:02d}",
}
for i in range(50)
]
analysis = analyzer.analyze_array(items)
# Paths are highly unique, no importance signal
assert analysis.crushability is not None
# Should NOT crush file listings
assert not analysis.crushability.crushable or analysis.crushability.confidence < 0.7
def test_order_list_not_crushable(self, analyzer):
"""Order list with unique order IDs should NOT be crushed."""
items = [
{
"order_id": f"ORD-2024-{i:05d}",
"customer": f"Customer {i}",
"total": 50.0 + i,
"status": "completed",
}
for i in range(50)
]
analysis = analyzer.analyze_array(items)
# Each order is a unique entity
assert analysis.crushability is not None
# order_id contains 'id' pattern
assert not analysis.crushability.crushable
class TestCrushabilityEndToEnd:
"""End-to-end tests for crushability-aware crushing."""
def test_db_results_preserved_completely(self):
"""DB results should be returned unchanged when not crushable."""
items = [{"id": i, "name": f"User {i}", "email": f"user{i}@test.com"} for i in range(30)]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=10)
crushed, was_modified, info = smart_crush_tool_output(content, config)
# Should NOT be modified (skip crushing)
if was_modified:
result = json.loads(crushed)
# If it was modified, all items should still be there
assert len(result) == 30, (
f"DB results should not lose items! Had 30, got {len(result)}. Info: {info}"
)
def test_search_results_crushed_by_score(self):
"""Search results should be crushed using score field."""
items = [
{
"id": i,
"title": f"Result {i}",
"score": 100 - i, # Higher score = more relevant
}
for i in range(100)
]
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=15)
crushed, was_modified, info = smart_crush_tool_output(content, config)
assert was_modified
result = json.loads(crushed)
assert len(result) < 100
# Top scores should be preserved
scores = [item.get("score", 0) for item in result]
assert max(scores) >= 90 # Top items preserved
def test_mixed_data_with_errors_preserves_errors(self):
"""Data with errors should crush but preserve ALL errors."""
items = []
error_ids = [5, 25, 45, 65, 85]
for i in range(100):
item = {"id": i, "data": f"value_{i}"}
if i in error_ids:
item["status"] = "failed"
item["error"] = f"Error at {i}"
items.append(item)
content = json.dumps(items)
config = SmartCrusherConfig(max_items_after_crush=20)
crushed, was_modified, info = smart_crush_tool_output(content, config)
result = json.loads(crushed)
# All errors must be preserved
error_count = sum(1 for item in result if item.get("error"))
assert error_count == len(error_ids), (
f"All {len(error_ids)} errors should be preserved, got {error_count}"
)
class TestCrushabilitySignals:
"""Test individual signal detection."""
@pytest.fixture
def analyzer(self):
return SmartAnalyzer(SmartCrusherConfig())
def test_detects_id_field_variations(self, analyzer):
"""Should detect various ID field naming patterns."""
test_cases = [
("id", [{"id": i} for i in range(20)]),
("uuid", [{"uuid": f"uuid-{i}"} for i in range(20)]),
("_id", [{"_id": f"mongo-{i}"} for i in range(20)]),
("pk", [{"pk": i} for i in range(20)]),
("key", [{"key": f"key-{i}"} for i in range(20)]),
("user_id", [{"user_id": i} for i in range(20)]),
]
for field_name, items in test_cases:
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert analysis.crushability.has_id_field, f"Should detect '{field_name}' as ID field"
def test_detects_score_field_variations(self, analyzer):
"""Should detect various score field naming patterns."""
test_cases = [
"score",
"rank",
"relevance",
"confidence",
"_score",
"rating",
]
for field_name in test_cases:
items = [{field_name: i * 0.1, "data": f"item_{i}"} for i in range(20)]
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert analysis.crushability.has_score_field, (
f"Should detect '{field_name}' as score field"
)
def test_detects_error_keywords(self, analyzer):
"""Should detect various error keyword patterns."""
error_keywords = ["error", "exception", "failed", "failure", "critical", "fatal"]
for keyword in error_keywords:
items = [{"id": i, "msg": "OK"} for i in range(20)]
items[10]["msg"] = f"Something {keyword} happened"
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert analysis.crushability.error_item_count >= 1, (
f"Should detect '{keyword}' as error indicator"
)
class TestCrushabilityEdgeCases:
"""Test edge cases in crushability analysis."""
@pytest.fixture
def analyzer(self):
return SmartAnalyzer(SmartCrusherConfig())
def test_empty_array(self, analyzer):
"""Empty array should not crash."""
analysis = analyzer.analyze_array([])
assert analysis.recommended_strategy == CompressionStrategy.NONE
def test_small_array_skipped(self, analyzer):
"""Arrays below min_items_to_analyze should be skipped."""
items = [{"id": i} for i in range(3)]
analysis = analyzer.analyze_array(items)
assert analysis.recommended_strategy == CompressionStrategy.NONE
def test_mixed_signals(self, analyzer):
"""Data with multiple signals should still be crushable."""
items = []
for i in range(100):
item = {
"id": i,
"score": 100 - i, # Score signal
"value": 50.0,
}
if i == 50:
item["error"] = "Test error" # Error signal
item["value"] = 999.0 # Anomaly signal
items.append(item)
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert analysis.crushability.crushable
assert len(analysis.crushability.signals_present) >= 2
def test_all_items_are_errors(self, analyzer):
"""When all items are errors, keyword detection finds them as a signal.
With keyword-based error detection (for the preservation guarantee),
when ALL items have error keywords, we detect error_keywords:50 as a
signal. This makes the data technically crushable.
However, since ALL items are errors, they will ALL be preserved due to
the preservation guarantee. The end result is the same - no data loss.
"""
items = [{"id": i, "error": f"Error {i}", "status": "failed"} for i in range(50)]
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
# With keyword-based error detection, all 50 items contain error keywords
# This IS a signal (error_keywords:50), making the data crushable.
# However, all 50 items will be preserved due to the preservation guarantee.
assert analysis.crushability.crushable
assert "error_keywords:50" in analysis.crushability.signals_present
class TestCrushabilityConfidence:
"""Test confidence scoring in crushability analysis."""
@pytest.fixture
def analyzer(self):
return SmartAnalyzer(SmartCrusherConfig())
def test_high_confidence_for_clear_cases(self, analyzer):
"""Clear-cut cases should have high confidence."""
# Low uniqueness - clearly safe
items = [{"status": "ok", "code": 200} for _ in range(100)]
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
assert analysis.crushability.confidence >= 0.8
def test_lower_confidence_for_ambiguous_cases(self, analyzer):
"""Ambiguous cases should have lower confidence."""
# Medium uniqueness with weak signal
items = [
{"id": i, "value": i % 10, "status": "active" if i % 2 == 0 else "inactive"}
for i in range(100)
]
# Add one error to provide weak signal
items[50]["error"] = "minor issue"
analysis = analyzer.analyze_array(items)
assert analysis.crushability is not None
# Should be lower confidence due to ambiguity
assert analysis.crushability.confidence <= 0.7

27
tests/test_relevance.py
View file

@ -273,33 +273,6 @@ class TestEmbeddingAvailable:
class TestSmartCrusherIntegration:
"""Integration tests for SmartCrusher with RelevanceScorer."""
def test_crusher_uses_scorer(self):
"""SmartCrusher uses relevance scorer for context matching."""
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
config = SmartCrusherConfig(
min_items_to_analyze=3,
min_tokens_to_crush=0,
max_items_after_crush=5,
)
crusher = SmartCrusher(config=config)
# Verify scorer is initialized
assert hasattr(crusher, "_scorer")
assert hasattr(crusher, "_relevance_threshold")
def test_custom_scorer_injection(self):
"""Custom scorer can be injected."""
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
custom_scorer = BM25Scorer(k1=2.0)
crusher = SmartCrusher(
config=SmartCrusherConfig(),
scorer=custom_scorer,
)
assert crusher._scorer is custom_scorer
def test_context_extraction(self):
"""Context is extracted from messages."""
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig

2
tests/test_relevance_extra.py
View file

@ -90,7 +90,7 @@ def test_embedding_numpy_and_model_error_paths(monkeypatch) -> None:
assert embedding._cosine_similarity([1, 0], [0, 1]) == 0.0
monkeypatch.setattr(EmbeddingScorer, "is_available", classmethod(lambda cls: False))
with pytest.raises(RuntimeError, match="requires sentence-transformers"):
with pytest.raises(RuntimeError, match="requires fastembed"):
EmbeddingScorer()._get_model()

737
tests/test_toin_field_learning.py
View file

@ -1,737 +0,0 @@
"""Tests for TOIN Field-Level Learning.
These tests PROVE that field-level learning actually works:
1. FieldSemantics correctly infers types from retrieval patterns
2. TOIN populates field_semantics from retrievals
3. SmartCrusher uses learned semantics to detect important items
4. End-to-end: important items are preserved based on learned behavior
No hardcoded patterns - all learning is behavior-based.
"""
import hashlib
import pytest
from headroom.telemetry import (
ToolIntelligenceNetwork,
ToolPattern,
ToolSignature,
reset_toin,
)
from headroom.telemetry.models import FieldSemantics
@pytest.fixture(autouse=True)
def reset_globals():
"""Reset global state before each test."""
reset_toin()
yield
reset_toin()
def _hash_value(value) -> str:
"""Hash a value the same way TOIN does."""
value_str = str(value)
return hashlib.sha256(value_str.encode()).hexdigest()[:8]
def _hash_field(field_name: str) -> str:
"""Hash a field name the same way TOIN does."""
return hashlib.sha256(field_name.encode()).hexdigest()[:8]
class TestFieldSemanticsLearning:
"""Test that FieldSemantics correctly learns from retrieval patterns."""
def test_identifier_type_inference(self):
"""PROVES: Field used with exact-match queries + high uniqueness = identifier."""
fs = FieldSemantics(field_hash="test123")
# Simulate: user retrieves items by unique IDs (exact match queries)
# Each ID is different - high uniqueness
for i in range(10):
fs.record_retrieval_value(_hash_value(f"id_{i}"), operator="=")
# Simulate compression stats: all values are unique
fs.record_compression_stats(
unique_values=100,
total_values=100, # uniqueness ratio = 1.0
most_common_value_hash=_hash_value("id_0"),
most_common_frequency=0.01, # No dominant value
)
fs.record_compression_stats(
unique_values=100,
total_values=100,
most_common_value_hash=_hash_value("id_1"),
most_common_frequency=0.01,
)
# Now infer the type
fs.infer_type()
# VERIFY: Should be classified as identifier
assert fs.inferred_type == "identifier", (
f"Expected 'identifier' but got '{fs.inferred_type}'. "
"High uniqueness + exact match queries should = identifier"
)
assert fs.confidence > 0.5, "Should have reasonable confidence"
def test_error_indicator_type_inference(self):
"""PROVES: Field with dominant default + retrievals for non-default = error_indicator."""
fs = FieldSemantics(field_hash="status_field")
# Simulate: most items have status="success" (the default)
# But user only retrieves items with status="error" or "failed"
error_hash = _hash_value("error")
failed_hash = _hash_value("failed")
success_hash = _hash_value("success")
# User retrieves "error" and "failed" values (non-default)
for _ in range(5):
fs.record_retrieval_value(error_hash, operator="=")
fs.record_retrieval_value(failed_hash, operator="=")
# Compression stats: 90% have "success" (the default)
fs.record_compression_stats(
unique_values=3, # "success", "error", "failed"
total_values=100,
most_common_value_hash=success_hash,
most_common_frequency=0.9, # 90% are "success"
)
fs.record_compression_stats(
unique_values=3,
total_values=100,
most_common_value_hash=success_hash,
most_common_frequency=0.9,
)
fs.infer_type()
# VERIFY: Should be error_indicator
assert fs.inferred_type == "error_indicator", (
f"Expected 'error_indicator' but got '{fs.inferred_type}'. "
"Dominant default + retrieval of non-default = error indicator"
)
assert fs.default_value_hash == success_hash, "Default should be 'success'"
assert fs.confidence > 0.5, "Should have reasonable confidence"
def test_status_type_inference(self):
"""PROVES: Low cardinality + specific values retrieved = status."""
fs = FieldSemantics(field_hash="state_field")
# Simulate: field has few unique values (low cardinality)
# User retrieves the same few values repeatedly
pending_hash = _hash_value("pending")
active_hash = _hash_value("active")
for _ in range(6):
fs.record_retrieval_value(pending_hash, operator="=")
for _ in range(4):
fs.record_retrieval_value(active_hash, operator="=")
# Compression stats: only 5 unique values across 100 items
fs.record_compression_stats(
unique_values=5,
total_values=100, # uniqueness ratio = 0.05 (very low)
most_common_value_hash=None,
most_common_frequency=0.3, # No overwhelming default
)
fs.record_compression_stats(
unique_values=5,
total_values=100,
most_common_value_hash=None,
most_common_frequency=0.3,
)
fs.infer_type()
# VERIFY: Should be status
assert fs.inferred_type == "status", (
f"Expected 'status' but got '{fs.inferred_type}'. "
"Low cardinality + specific values retrieved = status"
)
def test_score_type_inference(self):
"""PROVES: Range queries = score type."""
fs = FieldSemantics(field_hash="relevance_field")
# Simulate: user queries with range operators (top-N behavior)
for _ in range(8):
fs.record_retrieval_value(_hash_value("0.95"), operator=">")
for _ in range(4):
fs.record_retrieval_value(_hash_value("0.90"), operator=">=")
# Compression stats
fs.record_compression_stats(
unique_values=50,
total_values=100,
most_common_value_hash=None,
most_common_frequency=0.1,
)
fs.record_compression_stats(
unique_values=50,
total_values=100,
most_common_value_hash=None,
most_common_frequency=0.1,
)
fs.infer_type()
# VERIFY: Should be score
assert fs.inferred_type == "score", (
f"Expected 'score' but got '{fs.inferred_type}'. "
"Range queries (>, >=) should = score type"
)
def test_content_type_inference(self):
"""PROVES: Contains/text search queries = content type."""
fs = FieldSemantics(field_hash="description_field")
# Simulate: user does text search on this field
for i in range(10):
fs.record_retrieval_value(_hash_value(f"search_term_{i}"), operator="contains")
# Compression stats: high uniqueness (different descriptions)
fs.record_compression_stats(
unique_values=90,
total_values=100,
most_common_value_hash=None,
most_common_frequency=0.05,
)
fs.record_compression_stats(
unique_values=90,
total_values=100,
most_common_value_hash=None,
most_common_frequency=0.05,
)
fs.infer_type()
# VERIFY: Should be content
assert fs.inferred_type == "content", (
f"Expected 'content' but got '{fs.inferred_type}'. "
"Contains queries should = content type"
)
def test_is_value_important_for_error_indicator(self):
"""PROVES: For error_indicator, non-default values are important."""
fs = FieldSemantics(field_hash="status")
error_hash = _hash_value("error")
success_hash = _hash_value("success")
# Set up as error_indicator
fs.inferred_type = "error_indicator"
fs.confidence = 0.8
fs.default_value_hash = success_hash
fs.important_value_hashes = [error_hash]
# VERIFY
assert fs.is_value_important(error_hash), "Error value should be important"
assert not fs.is_value_important(success_hash), "Default value should NOT be important"
def test_is_value_important_for_status(self):
"""PROVES: For status fields, retrieved values are important."""
fs = FieldSemantics(field_hash="state")
pending_hash = _hash_value("pending")
unknown_hash = _hash_value("never_retrieved")
# Set up as status
fs.inferred_type = "status"
fs.confidence = 0.7
fs.value_retrieval_frequency = {pending_hash: 5}
# VERIFY
assert fs.is_value_important(pending_hash), "Retrieved value should be important"
assert not fs.is_value_important(unknown_hash), (
"Never-retrieved value should NOT be important"
)
class TestTOINFieldLearningIntegration:
"""Test that TOIN correctly integrates field-level learning."""
def test_record_retrieval_populates_field_semantics(self):
"""PROVES: record_retrieval with items actually populates field_semantics."""
toin = ToolIntelligenceNetwork()
# Create a tool signature
items = [
{"id": "123", "status": "ok", "value": 100},
{"id": "456", "status": "error", "value": 200},
]
sig = ToolSignature.from_items(items)
# Record retrieval with items - THIS IS WHERE LEARNING HAPPENS
toin.record_retrieval(
tool_signature_hash=sig.structure_hash,
retrieval_type="full",
query="status=error",
query_fields=["status"],
retrieved_items=items,
)
# VERIFY: pattern should have field_semantics populated
pattern = toin._patterns.get(sig.structure_hash)
assert pattern is not None, "Pattern should exist"
assert len(pattern.field_semantics) > 0, (
f"field_semantics should be populated after retrieval. Got: {pattern.field_semantics}"
)
# Check that field hashes match expected fields
expected_field_hashes = {_hash_field(f) for f in ["id", "status", "value"]}
actual_field_hashes = set(pattern.field_semantics.keys())
assert expected_field_hashes == actual_field_hashes, (
f"Expected field hashes {expected_field_hashes}, got {actual_field_hashes}"
)
def test_repeated_retrievals_trigger_type_inference(self):
"""PROVES: After multiple retrievals, TOIN infers field types."""
toin = ToolIntelligenceNetwork()
# Simulate items with status field
items = [
{"status": "success"},
{"status": "success"},
{"status": "success"},
{"status": "error"},
]
sig = ToolSignature.from_items(items)
# Simulate 6 retrievals (enough to trigger inference at retrieval 5)
for _ in range(6):
# User always retrieves items with error status
toin.record_retrieval(
tool_signature_hash=sig.structure_hash,
retrieval_type="full",
query="status=error",
query_fields=["status"],
retrieved_items=[{"status": "error"}],
)
# Also record compression to get enough data for inference
toin.record_compression(
tool_signature=sig,
original_count=100,
compressed_count=10,
original_tokens=1000,
compressed_tokens=100,
strategy="top_n",
items=items,
)
# VERIFY: After enough retrievals, type should be inferred
pattern = toin._patterns.get(sig.structure_hash)
status_hash = _hash_field("status")
assert status_hash in pattern.field_semantics, "Status field should be tracked"
status_sem = pattern.field_semantics[status_hash]
# The type should be inferred (not unknown) after enough data
assert status_sem.retrieval_count >= 6, (
f"Should have 6+ retrievals, got {status_sem.retrieval_count}"
)
def test_get_recommendation_includes_field_semantics(self):
"""PROVES: get_recommendation returns learned field_semantics."""
toin = ToolIntelligenceNetwork()
# Set up pattern with learned field_semantics
items = [{"id": "123", "status": "ok"}]
sig = ToolSignature.from_items(items)
# Record enough data
for i in range(10):
toin.record_retrieval(
tool_signature_hash=sig.structure_hash,
retrieval_type="full",
query=f"id={i}",
query_fields=["id"],
retrieved_items=[{"id": str(i), "status": "ok"}],
)
toin.record_compression(
tool_signature=sig,
original_count=100,
compressed_count=10,
original_tokens=1000,
compressed_tokens=100,
strategy="top_n",
items=items,
)
# Get recommendation
hint = toin.get_recommendation(sig)
# VERIFY: hint should include field_semantics
assert hint is not None, "Should get a recommendation"
# field_semantics might be empty if confidence is too low,
# but the attribute should exist
assert hasattr(hint, "field_semantics"), "Hint should have field_semantics attribute"
def test_field_semantics_persisted_correctly(self):
"""PROVES: field_semantics survives to_dict/from_dict round-trip."""
# ToolPattern is already imported at module level from headroom.telemetry
# Create pattern with field_semantics
pattern = ToolPattern(tool_signature_hash="test123")
# Add field semantics
fs = FieldSemantics(field_hash="field123")
fs.inferred_type = "error_indicator"
fs.confidence = 0.8
fs.important_value_hashes = ["value1", "value2"]
fs.default_value_hash = "default"
pattern.field_semantics["field123"] = fs
# Round-trip through dict
d = pattern.to_dict()
pattern2 = ToolPattern.from_dict(d)
# VERIFY: field_semantics preserved
assert "field123" in pattern2.field_semantics, "field_semantics should be preserved"
fs2 = pattern2.field_semantics["field123"]
assert fs2.inferred_type == "error_indicator"
assert fs2.confidence == 0.8
assert fs2.important_value_hashes == ["value1", "value2"]
class TestSmartCrusherUsesLearnedSemantics:
"""Test that SmartCrusher actually uses learned field semantics."""
def test_detect_items_by_learned_semantics_finds_important_items(self):
"""PROVES: _detect_items_by_learned_semantics correctly identifies items."""
from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
# Create field semantics that knows "error" is important
status_hash = _hash_field("status")
error_hash = _hash_value("error")
success_hash = _hash_value("success")
fs = FieldSemantics(field_hash=status_hash)
fs.inferred_type = "error_indicator"
fs.confidence = 0.8
fs.default_value_hash = success_hash
fs.important_value_hashes = [error_hash]
fs.value_retrieval_frequency = {error_hash: 10}
field_semantics = {status_hash: fs}
# Test items - index 1 has error status
items = [
{"status": "success", "message": "all good"},
{"status": "error", "message": "something failed"}, # <-- This should be detected
{"status": "success", "message": "also good"},
]
# VERIFY
important_indices = _detect_items_by_learned_semantics(items, field_semantics)
assert 1 in important_indices, (
f"Index 1 (error status) should be detected as important. "
f"Got indices: {important_indices}"
)
assert 0 not in important_indices, "Index 0 (success) should not be important"
assert 2 not in important_indices, "Index 2 (success) should not be important"
def test_detect_items_handles_empty_semantics(self):
"""PROVES: Function handles edge cases gracefully."""
from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
items = [{"status": "ok"}]
# Empty semantics
assert _detect_items_by_learned_semantics(items, {}) == []
assert _detect_items_by_learned_semantics(items, None) == []
assert _detect_items_by_learned_semantics([], {"x": FieldSemantics(field_hash="x")}) == []
def test_detect_items_requires_confidence(self):
"""PROVES: Low confidence semantics are ignored."""
from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
status_hash = _hash_field("status")
error_hash = _hash_value("error")
fs = FieldSemantics(field_hash=status_hash)
fs.inferred_type = "error_indicator"
fs.confidence = 0.1 # TOO LOW
fs.important_value_hashes = [error_hash]
items = [{"status": "error"}]
# VERIFY: Low confidence = ignored
result = _detect_items_by_learned_semantics(items, {status_hash: fs})
assert result == [], "Low confidence semantics should be ignored"
class TestEndToEndFieldLearning:
"""End-to-end tests proving the full learning pipeline works."""
def test_full_pipeline_learns_and_applies(self):
"""PROVES: End-to-end learning from retrieval to compression."""
toin = ToolIntelligenceNetwork()
# PHASE 1: Learning - User retrieves items with error status
# Simulating: "show me all failed items"
items_with_errors = [
{"id": "1", "status": "success", "data": "..."},
{"id": "2", "status": "error", "data": "..."}, # Retrieved
{"id": "3", "status": "success", "data": "..."},
{"id": "4", "status": "failed", "data": "..."}, # Retrieved
]
sig = ToolSignature.from_items(items_with_errors)
# User keeps retrieving error/failed items (learning behavior)
for _ in range(5):
toin.record_retrieval(
tool_signature_hash=sig.structure_hash,
retrieval_type="full",
query="status!=success",
query_fields=["status"],
retrieved_items=[
{"id": "2", "status": "error", "data": "..."},
{"id": "4", "status": "failed", "data": "..."},
],
)
toin.record_compression(
tool_signature=sig,
original_count=100,
compressed_count=10,
original_tokens=1000,
compressed_tokens=100,
strategy="top_n",
items=[
{"id": str(i), "status": "success" if i % 5 != 0 else "error", "data": "..."}
for i in range(100)
],
)
# PHASE 2: Verify learning occurred
pattern = toin._patterns.get(sig.structure_hash)
assert pattern is not None
assert len(pattern.field_semantics) > 0, "Should have learned field semantics"
# Check status field was learned
status_hash = _hash_field("status")
if status_hash in pattern.field_semantics:
status_sem = pattern.field_semantics[status_hash]
# Error value should be tracked
error_hash = _hash_value("error")
failed_hash = _hash_value("failed")
assert (
error_hash in status_sem.important_value_hashes
or failed_hash in status_sem.important_value_hashes
), "Error/failed values should be marked as important"
def test_recommendation_hint_includes_learned_semantics(self):
"""PROVES: TOIN recommendation includes learned field semantics for SmartCrusher."""
toin = ToolIntelligenceNetwork()
# Set up sufficient learning
items = [{"status": "ok", "id": "123"}]
sig = ToolSignature.from_items(items)
# Create pattern with confident field semantics by directly manipulating internal state
# (This is a test - in real code, patterns are created via record_* methods)
pattern = ToolPattern(tool_signature_hash=sig.structure_hash)
toin._patterns[sig.structure_hash] = pattern
status_hash = _hash_field("status")
fs = FieldSemantics(field_hash=status_hash)
fs.inferred_type = "error_indicator"
fs.confidence = 0.8 # High confidence
fs.retrieval_count = 10
fs.important_value_hashes = [_hash_value("error")]
pattern.field_semantics[status_hash] = fs
# Ensure pattern has enough data for recommendation
pattern.total_compressions = 10
pattern.sample_size = 10 # Required by min_samples_for_recommendation
pattern.confidence = 0.5
# Get recommendation
hint = toin.get_recommendation(sig)
# VERIFY
assert hint is not None
assert len(hint.field_semantics) > 0, (
f"Recommendation should include field_semantics. Got: {hint.field_semantics}"
)
assert status_hash in hint.field_semantics
class TestFieldSemanticsMemoryBounds:
"""Test that memory bounds are enforced."""
def test_important_values_bounded(self):
"""PROVES: important_value_hashes stays within bounds."""
fs = FieldSemantics(field_hash="test")
# Add more values than MAX_IMPORTANT_VALUES
for i in range(fs.MAX_IMPORTANT_VALUES + 20):
fs.record_retrieval_value(_hash_value(f"value_{i}"))
# VERIFY: bounded
assert len(fs.important_value_hashes) <= fs.MAX_IMPORTANT_VALUES
def test_value_frequency_bounded(self):
"""PROVES: value_retrieval_frequency stays within bounds."""
fs = FieldSemantics(field_hash="test")
# Add more values than MAX_VALUE_FREQUENCY_ENTRIES
for i in range(fs.MAX_VALUE_FREQUENCY_ENTRIES + 20):
fs.record_retrieval_value(_hash_value(f"value_{i}"))
# VERIFY: bounded
assert len(fs.value_retrieval_frequency) <= fs.MAX_VALUE_FREQUENCY_ENTRIES
class TestProductionCodePath:
"""Integration tests for the ACTUAL production code path.
These tests verify that CompressionStore -> TOIN integration works,
not just TOIN in isolation. This is critical because the unit tests
can pass while the production integration is broken.
"""
def test_compression_store_passes_items_to_toin(self):
"""PROVES: CompressionStore.process_pending_feedback passes retrieved_items to TOIN.
This is the integration test that would have caught the original bug
where retrieved_items was never passed to TOIN in production.
"""
import json
from headroom.cache.compression_store import CompressionStore
from headroom.telemetry.toin import get_toin, reset_toin
reset_toin()
toin = get_toin()
# Create a store with feedback enabled
store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
# Store some compressed content with items that have distinct field values
items = [
{"id": "123", "status": "success", "value": 100},
{"id": "456", "status": "error", "value": 200},
{"id": "789", "status": "success", "value": 300},
]
compressed_json = json.dumps(items)
original_json = json.dumps(items * 10) # Original was bigger
# Create a tool signature hash for this structure
sig = ToolSignature.from_items(items)
hash_key = store.store(
original=original_json,
compressed=compressed_json,
original_tokens=1000,
compressed_tokens=100,
original_item_count=30,
compressed_item_count=3,
tool_name="test_api",
tool_call_id="call_123",
tool_signature_hash=sig.structure_hash,
compression_strategy="top_n",
)
# Simulate a retrieval (this triggers the feedback loop)
store.retrieve(hash_key, query="status=error")
# Process pending feedback - THIS IS WHERE THE BUG WAS
store.process_pending_feedback()
# VERIFY: TOIN should have received the items and learned from them
pattern = toin._patterns.get(sig.structure_hash)
assert pattern is not None, "TOIN should have a pattern for this tool"
# The key assertion: field_semantics should be populated
# This would have FAILED before the fix because retrieved_items wasn't passed
assert len(pattern.field_semantics) > 0, (
"TOIN should have learned field semantics from the retrieved items. "
"If this fails, CompressionStore is not passing retrieved_items to TOIN."
)
# Verify specific fields were learned
id_hash = _hash_field("id")
status_hash = _hash_field("status")
value_hash = _hash_field("value")
learned_fields = set(pattern.field_semantics.keys())
expected_fields = {id_hash, status_hash, value_hash}
assert expected_fields == learned_fields, (
f"Expected fields {expected_fields}, got {learned_fields}"
)
def test_compression_store_handles_wrapped_arrays(self):
"""PROVES: CompressionStore correctly extracts items from wrapped arrays."""
import json
from headroom.cache.compression_store import CompressionStore
from headroom.telemetry.toin import get_toin, reset_toin
reset_toin()
toin = get_toin()
store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
# Content wrapped in {"results": [...]} pattern
items = [{"name": "test", "score": 0.95}]
wrapped_content = json.dumps({"results": items, "total": 1})
sig = ToolSignature.from_items(items)
hash_key = store.store(
original=wrapped_content,
compressed=wrapped_content,
original_tokens=100,
compressed_tokens=100,
original_item_count=1,
compressed_item_count=1,
tool_name="search_api",
tool_call_id="call_456",
tool_signature_hash=sig.structure_hash,
compression_strategy="top_n",
)
store.retrieve(hash_key)
store.process_pending_feedback()
# VERIFY: Items were extracted from wrapped structure
pattern = toin._patterns.get(sig.structure_hash)
assert pattern is not None
assert len(pattern.field_semantics) > 0, (
"TOIN should extract items from wrapped arrays like {'results': [...]}"
)
def test_compression_store_handles_invalid_json(self):
"""PROVES: CompressionStore gracefully handles invalid JSON."""
from headroom.cache.compression_store import CompressionStore
from headroom.telemetry.toin import get_toin, reset_toin
reset_toin()
get_toin() # Initialize TOIN for feedback loop
store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
# Store invalid JSON content
invalid_json = "not valid json {"
hash_key = store.store(
original=invalid_json,
compressed=invalid_json,
original_tokens=10,
compressed_tokens=10,
original_item_count=0,
compressed_item_count=0,
tool_name="broken_api",
tool_call_id="call_789",
tool_signature_hash="invalid123",
compression_strategy="none",
)
# This should not crash
store.retrieve(hash_key)
store.process_pending_feedback()
# VERIFY: No crash, pattern may or may not exist but no exception
# The main assertion is that we got here without exception

337
tests/test_toin_integration.py
View file

@ -8,7 +8,6 @@ Tests the complete flow:
5. Future compressions get improved recommendations
"""
import json
import tempfile
from pathlib import Path
@ -18,18 +17,11 @@ from headroom.cache.compression_store import (
get_compression_store,
reset_compression_store,
)
from headroom.config import CCRConfig
from headroom.telemetry import ToolSignature
from headroom.telemetry.toin import (
TOINConfig,
ToolIntelligenceNetwork,
get_toin,
reset_toin,
)
from headroom.transforms.smart_crusher import (
SmartCrusher,
SmartCrusherConfig,
)
@pytest.fixture
@ -55,332 +47,3 @@ def fresh_store():
store = get_compression_store(max_entries=100, default_ttl=300)
yield store
reset_compression_store()
class TestTOINIntegration:
"""Test the full TOIN feedback loop."""
def test_compression_records_correct_hash(self, fresh_toin, fresh_store):
"""Test that SmartCrusher records the correct tool_signature_hash in store."""
# Create test data
items = [{"id": i, "score": 100 - i, "name": f"Item {i}"} for i in range(50)]
content = json.dumps(items)
# Create a SmartCrusher with CCR enabled
ccr_config = CCRConfig(enabled=True, inject_retrieval_marker=False)
crusher = SmartCrusher(
SmartCrusherConfig(max_items_after_crush=10),
ccr_config=ccr_config,
)
# Compress the content
crushed, was_modified, info = crusher._smart_crush_content(content, tool_name="test_tool")
assert was_modified, "Content should be modified by compression"
# Verify the store has an entry with the correct tool_signature_hash
stats = fresh_store.get_stats()
assert stats["entry_count"] >= 1, "Store should have at least one entry"
# Get the entry and verify it has tool_signature_hash
# We need to find the hash key from the store
entries = [entry for _, entry in fresh_store._backend.items()]
assert len(entries) >= 1, "Should have at least one entry"
entry = entries[0]
assert entry.tool_signature_hash is not None, "Entry should have tool_signature_hash"
assert entry.compression_strategy is not None, "Entry should have compression_strategy"
# Verify the hash matches what ToolSignature would generate
expected_signature = ToolSignature.from_items(items)
assert entry.tool_signature_hash == expected_signature.structure_hash, (
"Stored hash should match ToolSignature.structure_hash"
)
def test_retrieval_updates_toin_strategy_success(self, fresh_toin, fresh_store):
"""Test that retrieval events update TOIN strategy success rates."""
# Create test data
items = [{"id": i, "score": 100 - i, "name": f"Item {i}"} for i in range(50)]
signature = ToolSignature.from_items(items)
# Record some compressions with fresh_toin
for _ in range(5):
fresh_toin.record_compression(
tool_signature=signature,
original_count=50,
compressed_count=10,
original_tokens=5000,
compressed_tokens=1000,
strategy="smart_sample",
)
# Get initial strategy success rate
pattern = fresh_toin._patterns.get(signature.structure_hash)
assert pattern is not None, "Pattern should exist after compressions"
initial_rate = pattern.strategy_success_rates.get("smart_sample", 1.0)
assert initial_rate > 0, "Initial success rate should be positive"
# Simulate retrieval events (which indicate compression was too aggressive)
for _ in range(3):
fresh_toin.record_retrieval(
tool_signature_hash=signature.structure_hash,
retrieval_type="full",
query="test query",
query_fields=["id"],
strategy="smart_sample",
)
# Verify success rate decreased
final_rate = pattern.strategy_success_rates.get("smart_sample", 1.0)
assert final_rate < initial_rate, (
f"Success rate should decrease after retrievals: {initial_rate} -> {final_rate}"
)
def test_full_feedback_loop(self, fresh_toin, fresh_store):
"""Test the complete feedback loop: compress → retrieve → learn → recommend."""
# Create test data
items = [{"id": i, "score": 100 - i, "name": f"Item {i}"} for i in range(50)]
content = json.dumps(items)
signature = ToolSignature.from_items(items)
# Step 1: Compress with SmartCrusher (records to TOIN)
ccr_config = CCRConfig(enabled=True, inject_retrieval_marker=False)
crusher = SmartCrusher(
SmartCrusherConfig(max_items_after_crush=10, use_feedback_hints=True),
ccr_config=ccr_config,
)
# Multiple compressions to build pattern
for _ in range(5):
crusher._smart_crush_content(content, tool_name="test_tool")
# Step 2: Verify TOIN has a pattern
pattern = fresh_toin._patterns.get(signature.structure_hash)
assert pattern is not None, "TOIN should have a pattern after compressions"
assert pattern.total_compressions >= 5, "Should have recorded 5 compressions"
# Step 3: Simulate retrievals (indicating compression was too aggressive)
# Find the stored entry hash
entries = [entry for _, entry in fresh_store._backend.items()]
assert len(entries) > 0, "Should have cached entries"
# Retrieve multiple times to trigger learning
for entry in entries[:3]:
fresh_store.retrieve(entry.hash, query="find all items")
# Step 4: Verify TOIN learned from retrievals
pattern = fresh_toin._patterns.get(signature.structure_hash)
assert pattern is not None, "Pattern should exist after retrievals"
# CRITICAL: Verify field-level learning actually happened
# This assertion would have caught the bug where compression_store
# wasn't passing retrieved_items to TOIN
assert len(pattern.field_semantics) > 0, (
"TOIN should learn field semantics from retrieved items. "
"If this fails, the production code path (CompressionStore -> TOIN) is broken."
)
# Verify retrieval stats were updated
assert pattern.total_retrievals >= 1, "Should have recorded retrievals"
recommendation = fresh_toin.get_recommendation(signature, "find all items")
# After many retrievals, TOIN should recommend more items
# Default is 15-20, but with high retrieval rate it should go higher
assert recommendation.confidence > 0, "Recommendation should have confidence"
def test_preserve_fields_used_in_compression(self, fresh_toin, fresh_store):
"""Test that TOIN preserve_fields are used during compression planning."""
# Create test data with specific fields
items = [{"id": i, "score": 100 - i, "category": f"cat_{i % 3}"} for i in range(50)]
signature = ToolSignature.from_items(items)
# Record compressions and retrievals that query the "category" field
for _ in range(10):
fresh_toin.record_compression(
tool_signature=signature,
original_count=50,
compressed_count=10,
original_tokens=5000,
compressed_tokens=1000,
strategy="smart_sample",
)
# Record retrievals that query "category"
for _ in range(5):
fresh_toin.record_retrieval(
tool_signature_hash=signature.structure_hash,
retrieval_type="search",
query="category:cat_1",
query_fields=["category"],
strategy="smart_sample",
)
# Get recommendation - should now preserve "category" field
recommendation = fresh_toin.get_recommendation(signature, "find all items in cat_1")
# Verify the recommendation reflects learning
assert recommendation.source in ("local", "network", "default"), (
f"Should have a valid source: {recommendation.source}"
)
def test_instance_id_stable_across_restarts(self):
"""Test that instance_id is stable across restarts."""
reset_toin()
with tempfile.TemporaryDirectory() as tmpdir:
storage_path = str(Path(tmpdir) / "toin_stable.json")
# Create first TOIN instance
toin1 = ToolIntelligenceNetwork(TOINConfig(storage_path=storage_path))
instance_id_1 = toin1._instance_id
# Create second instance with same path (simulating restart)
toin2 = ToolIntelligenceNetwork(TOINConfig(storage_path=storage_path))
instance_id_2 = toin2._instance_id
# Instance IDs should be the same since derived from path
assert instance_id_1 == instance_id_2, (
f"Instance ID should be stable: {instance_id_1} vs {instance_id_2}"
)
def test_atomic_save(self):
"""Test that save() uses atomic writes."""
reset_toin()
with tempfile.TemporaryDirectory() as tmpdir:
storage_path = str(Path(tmpdir) / "toin_atomic.json")
toin = ToolIntelligenceNetwork(TOINConfig(storage_path=storage_path))
# Create some data
items = [{"id": i, "name": f"test_{i}"} for i in range(10)]
signature = ToolSignature.from_items(items)
toin.record_compression(
tool_signature=signature,
original_count=10,
compressed_count=5,
original_tokens=1000,
compressed_tokens=500,
strategy="smart_sample",
)
# Save
toin.save()
# Verify file exists and is valid JSON
saved_path = Path(storage_path)
assert saved_path.exists(), "Save file should exist"
with open(saved_path) as f:
data = json.load(f)
assert "patterns" in data, "Saved data should have patterns"
assert len(data["patterns"]) > 0, "Should have at least one pattern"
def test_query_patterns_merged_on_import(self):
"""Test that query patterns are merged when importing patterns."""
reset_toin()
with tempfile.TemporaryDirectory() as tmpdir:
storage_path = str(Path(tmpdir) / "toin_merge.json")
toin = ToolIntelligenceNetwork(TOINConfig(storage_path=storage_path))
# Create local pattern
items = [{"id": i, "name": f"test_{i}"} for i in range(10)]
signature = ToolSignature.from_items(items)
toin.record_compression(
tool_signature=signature,
original_count=10,
compressed_count=5,
original_tokens=1000,
compressed_tokens=500,
strategy="smart_sample",
)
# Record local query pattern
toin.record_retrieval(
tool_signature_hash=signature.structure_hash,
retrieval_type="search",
query="local_pattern:value",
query_fields=["local_pattern"],
strategy="smart_sample",
)
# Create import data with different query pattern
import_data = {
"patterns": {
signature.structure_hash: {
"tool_signature_hash": signature.structure_hash,
"total_compressions": 100,
"total_retrievals": 20,
"avg_original_count": 50,
"avg_compressed_count": 10,
"avg_compression_ratio": 0.2,
"retrieval_rate": 0.2,
"common_query_patterns": ["imported_pattern:x"],
"strategy_success_rates": {"smart_sample": 0.8},
"preserve_fields": ["imported_field"],
"user_count": 50,
"last_updated": 1234567890.0,
}
}
}
# Import patterns
toin.import_patterns(import_data)
# Verify patterns were merged
pattern = toin._patterns.get(signature.structure_hash)
assert pattern is not None, "Pattern should exist after merge"
# Check that both query patterns exist
assert "imported_pattern:x" in pattern.common_query_patterns, (
"Imported query pattern should be present"
)
class TestStoreToTOINHash:
"""Test the hash correlation between compression_store and TOIN."""
def test_hash_matches_between_store_and_toin(self, fresh_toin, fresh_store):
"""Test that the hash stored in compression_store matches TOIN events."""
# Use 50 items with score field to ensure compression threshold is met
items = [{"id": i, "score": 100 - i, "name": f"Item {i}"} for i in range(50)]
content = json.dumps(items)
signature = ToolSignature.from_items(items)
# Compress with aggressive settings to ensure items are actually reduced
ccr_config = CCRConfig(enabled=True, inject_retrieval_marker=False)
crusher = SmartCrusher(
SmartCrusherConfig(max_items_after_crush=10),
ccr_config=ccr_config,
)
crushed, was_modified, info = crusher._smart_crush_content(content, tool_name="hash_test")
# Verify compression actually happened
assert was_modified, f"Content should be modified by compression: {info}"
# Get the stored hash
entries = [entry for _, entry in fresh_store._backend.items()]
assert len(entries) >= 1, (
f"Should have stored entry. Modified: {was_modified}, Info: {info}"
)
stored_hash = entries[0].tool_signature_hash
# Verify it matches ToolSignature
assert stored_hash == signature.structure_hash, (
f"Store hash {stored_hash} should match signature {signature.structure_hash}"
)
# Verify TOIN can receive events for this hash
fresh_toin.record_compression(
tool_signature=signature,
original_count=50,
compressed_count=10,
original_tokens=5000,
compressed_tokens=1000,
strategy="smart_sample",
)
pattern = fresh_toin._patterns.get(signature.structure_hash)
assert pattern is not None, "TOIN should have pattern for same hash"

1186
tests/test_transforms/test_anchor_selector.py
View file

File diff suppressed because it is too large Load diff

1359
tests/test_transforms/test_smart_crusher.py
View file

File diff suppressed because it is too large Load diff

189
tests/test_transforms/test_smart_crusher_bugs.py
View file

@ -20,6 +20,7 @@ from headroom.transforms.smart_crusher import SmartCrusher
def _make_crusher(max_items: int = 10, min_items: int = 3) -> SmartCrusher:
"""Build a SmartCrusher with deterministic small-K config for tests."""
config = SmartCrusherConfig(
enabled=True,
min_items_to_analyze=min_items,
@ -30,102 +31,13 @@ def _make_crusher(max_items: int = 10, min_items: int = 3) -> SmartCrusher:
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 #1 (number array schema preservation) — invariant pinned by the
# Rust port (`crates/headroom-core/src/transforms/smart_crusher/crushers.rs::
# crush_number_array` + its unit tests) and the parity fixtures
# (`tests/parity/fixtures/smart_crusher/number_array_40_changepoint*`).
# The Python `_crush_number_array` helper that the previous tests
# probed was removed when the Python implementation was retired in
# Stage 3c.1b.
# ---------------------------------------------------------------------------
# Bug 2: Race condition on _current_field_semantics
@ -212,79 +124,12 @@ class TestRecursionDepthLimit:
assert isinstance(parsed, list)
# ---------------------------------------------------------------------------
# Stage 3c.1 lockstep bug fixes (#1 percentile, #2 sequential, #3 rare-status,
# #4 k-split). Each test pins the Python behavior post-fix; Rust has matching
# tests so parity fixtures byte-equal both languages.
# ---------------------------------------------------------------------------
class TestStage3c1BugFixes:
"""Bugs fixed in lockstep with the Rust port at Stage 3c.1."""
# Bug #1 — percentile off-by-one (cosmetic, strategy string only).
def test_bug1_percentile_uses_linear_interpolation(self) -> None:
from headroom.transforms.smart_crusher import _percentile_linear
# n=10 [10..100]: p25 index = 0.25 * 9 = 2.25 → 30*0.75 + 40*0.25 = 32.5
sorted_vals = [10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0, 100.0]
assert _percentile_linear(sorted_vals, 0.25) == 32.5
assert _percentile_linear(sorted_vals, 0.75) == 77.5
# n=1 → return the single value.
assert _percentile_linear([42.0], 0.25) == 42.0
# n=0 → 0.0 (defensive).
assert _percentile_linear([], 0.25) == 0.0
# Bug #2 — zero-padded string IDs misclassified as sequential.
def test_bug2_zero_padded_strings_not_sequential(self) -> None:
from headroom.transforms.smart_crusher import _detect_sequential_pattern
# Pre-fix: int("001") → 1, so ["001",...,"005"] looked like 1..5
# and was classified as sequential. Post-fix: had_non_string_numeric
# stays False because every value came from a string → return False.
zero_padded = ["001", "002", "003", "004", "005"]
assert _detect_sequential_pattern(zero_padded, check_order=False) is False
# Real ints (not strings) still classify as sequential.
real_ints = [1, 2, 3, 4, 5]
assert _detect_sequential_pattern(real_ints, check_order=False) is True
# Bug #3 — rare-status detection cardinality cap.
def test_bug3_high_cardinality_pareto(self) -> None:
from headroom.transforms.smart_crusher import _detect_rare_status_values
# 60×INFO + 25×WARN + 15 distinct error codes (cardinality 17).
# Pre-fix: 17 > 10 → field skipped → 0 outliers.
# Post-fix: top-2 covers 85% (60+25=85), K=2 ≤ 5, the 15 rare codes flagged.
items = []
for _ in range(60):
items.append({"code": "INFO"})
for _ in range(25):
items.append({"code": "WARN"})
for i in range(15):
items.append({"code": f"ERR_{i}"})
outliers = _detect_rare_status_values(items, common_fields={"code"})
assert len(outliers) == 15
def test_bug3_uniform_distribution_no_outliers(self) -> None:
# 50 distinct values, 1 each → top-K never reaches 80% with K<=5.
# Field correctly identified as non-categorical.
from headroom.transforms.smart_crusher import _detect_rare_status_values
items = [{"code": f"CAT_{i}"} for i in range(50)]
assert _detect_rare_status_values(items, common_fields={"code"}) == []
# Bug #4 — k-split overshoot when k_total=1.
def test_bug4_k_split_no_overshoot_when_k_total_one(self) -> None:
from headroom import SmartCrusherConfig
from headroom.transforms.smart_crusher import SmartCrusher
config = SmartCrusherConfig(min_items_to_analyze=1)
crusher = SmartCrusher(config=config)
# Force k_total=1 by passing a single-item list — the n<=8 fast
# path returns n=1, so k_total=1.
k_total, k_first, k_last, k_importance = crusher._compute_k_split(["only"], 1.0)
assert k_total == 1
assert k_first + k_last <= k_total, (
f"BUG #4: k_first={k_first} + k_last={k_last} must not exceed k_total={k_total}"
)
# Stage 3c.1 lockstep bug-fix tests previously lived here; they probed
# Python helpers (`_percentile_linear`, `_detect_sequential_pattern`,
# `_detect_rare_status_values`, `_compute_k_split`) that were removed
# along with the Python implementation in Stage 3c.1b. The Rust port
# pins the same invariants — see the `bug1_*` / `bug2_*` / `bug3_*` /
# `bug4_*` tests in `crates/headroom-core/src/transforms/smart_crusher/`
# (notably `crushers.rs` and `analyzer.rs`). Parity fixtures
# (`tests/parity/fixtures/smart_crusher/`) byte-compare the post-fix
# behavior across the language boundary.

489
tests/test_transforms/test_universal_json_crush.py
View file

@ -1,489 +0,0 @@
"""Tests for universal JSON compression (all array types).
Verifies that SmartCrusher handles arrays of dicts, strings, numbers,
mixed types, and nested arrays — with consistent safety guarantees
across all types.
"""
from __future__ import annotations
import json
import pytest
from headroom.transforms.smart_crusher import (
ArrayType,
SmartCrusher,
SmartCrusherConfig,
_classify_array,
)
# =====================================================================
# Fixtures
# =====================================================================
@pytest.fixture
def crusher():
"""SmartCrusher configured for testing."""
return SmartCrusher(
config=SmartCrusherConfig(
min_items_to_analyze=5,
min_tokens_to_crush=0, # Always crush
max_items_after_crush=15,
)
)
@pytest.fixture
def crusher_large_k():
"""SmartCrusher with higher max items for larger test arrays."""
return SmartCrusher(
config=SmartCrusherConfig(
min_items_to_analyze=5,
min_tokens_to_crush=0,
max_items_after_crush=50,
)
)
# =====================================================================
# Type Classification
# =====================================================================
class TestClassifyArray:
def test_dict_array(self):
assert _classify_array([{"a": 1}, {"b": 2}]) == ArrayType.DICT_ARRAY
def test_string_array(self):
assert _classify_array(["hello", "world", "foo"]) == ArrayType.STRING_ARRAY
def test_number_array_int(self):
assert _classify_array([1, 2, 3]) == ArrayType.NUMBER_ARRAY
def test_number_array_float(self):
assert _classify_array([1.0, 2.5, 3.7]) == ArrayType.NUMBER_ARRAY
def test_number_array_mixed_int_float(self):
assert _classify_array([1, 2.5, 3]) == ArrayType.NUMBER_ARRAY
def test_bool_array(self):
assert _classify_array([True, False, True]) == ArrayType.BOOL_ARRAY
def test_nested_array(self):
assert _classify_array([[1, 2], [3, 4]]) == ArrayType.NESTED_ARRAY
def test_mixed_array(self):
assert _classify_array([{"a": 1}, "string", 42]) == ArrayType.MIXED_ARRAY
def test_empty(self):
assert _classify_array([]) == ArrayType.EMPTY
def test_single_dict(self):
assert _classify_array([{"key": "val"}]) == ArrayType.DICT_ARRAY
def test_single_string(self):
assert _classify_array(["only"]) == ArrayType.STRING_ARRAY
def test_none_values(self):
# Arrays with None mixed in are MIXED
assert _classify_array([1, None, 3]) == ArrayType.MIXED_ARRAY
def test_bool_not_confused_with_int(self):
# Python's True/False are int subclasses — we handle this
assert _classify_array([True, False]) == ArrayType.BOOL_ARRAY
# But mixed bools and ints should be MIXED or NUMBER depending on impl
result = _classify_array([True, 42])
assert result in (ArrayType.MIXED_ARRAY, ArrayType.NUMBER_ARRAY)
# =====================================================================
# String Array Compression
# =====================================================================
class TestCrushStringArray:
def test_basic_compression(self, crusher):
strings = [f"item_{i}" for i in range(100)]
crushed, strategy = crusher._crush_string_array(strings)
assert len(crushed) < len(strings)
assert "string:adaptive" in strategy
def test_errors_always_preserved(self, crusher):
strings = ["ok"] * 50 + ["error: connection timeout", "failed: auth denied"] + ["ok"] * 48
crushed, strategy = crusher._crush_string_array(strings)
assert any("error" in s for s in crushed)
assert any("failed" in s for s in crushed)
def test_first_last_kept(self, crusher):
strings = [f"item_{i}" for i in range(50)]
crushed, strategy = crusher._crush_string_array(strings)
# First item always present
assert strings[0] in crushed
# Last item always present
assert strings[-1] in crushed
def test_dedup_reduces_output(self, crusher):
# 95 identical + 5 unique
strings = ["repeated_value"] * 95 + [f"unique_{i}" for i in range(5)]
crushed, strategy = crusher._crush_string_array(strings)
# Should massively reduce — not keep 95 copies
assert len(crushed) < 20
# All 5 unique values should survive (they have high info value)
for i in range(5):
assert f"unique_{i}" in crushed
def test_below_threshold_passthrough(self, crusher):
strings = ["a", "b", "c"] # Below min_items_to_analyze=5
# Direct method call — should passthrough since <= 8
crushed, strategy = crusher._crush_string_array(strings)
assert crushed == strings
assert "passthrough" in strategy
def test_empty_strings_handled(self, crusher):
strings = [""] * 20
crushed, strategy = crusher._crush_string_array(strings)
# Should not crash
assert isinstance(crushed, list)
def test_unicode_strings(self, crusher):
strings = [f"日本語テスト_{i}" for i in range(50)]
crushed, strategy = crusher._crush_string_array(strings)
assert len(crushed) < len(strings)
assert all(isinstance(s, str) for s in crushed)
def test_length_anomalies_preserved(self, crusher_large_k):
# Most strings are short, one is very long
strings = ["short"] * 95 + ["x" * 10000] + ["short"] * 4
crushed, strategy = crusher_large_k._crush_string_array(strings)
assert any(len(s) > 1000 for s in crushed)
# =====================================================================
# Number Array Compression
# =====================================================================
class TestCrushNumberArray:
def test_basic_compression(self, crusher):
numbers = [42.0 + i * 0.1 for i in range(100)]
crushed, strategy = crusher._crush_number_array(numbers)
assert len(crushed) < len(numbers)
assert "number:adaptive" in strategy
def test_stats_in_strategy_not_array(self, crusher):
numbers = list(range(100))
crushed, strategy = crusher._crush_number_array(numbers)
# Stats should be in the strategy string, not in the array
assert "min=" in strategy
assert "max=" in strategy
# Array should contain only numbers (schema-preserving)
for item in crushed:
assert isinstance(item, int | float)
def test_outliers_preserved(self, crusher):
# Normal values around 50 with one extreme outlier
numbers = [50.0 + i * 0.01 for i in range(100)] + [999.9]
crushed, strategy = crusher._crush_number_array(numbers)
assert 999.9 in crushed
assert "outliers" in strategy
def test_all_identical(self, crusher):
numbers = [42.0] * 100
crushed, strategy = crusher._crush_number_array(numbers)
# With all identical, should compress heavily
# Summary + a few representatives
numeric_values = [v for v in crushed if isinstance(v, int | float)]
assert all(v == 42.0 for v in numeric_values)
def test_first_last_kept(self, crusher):
numbers = list(range(50))
crushed, strategy = crusher._crush_number_array(numbers)
numeric_values = [v for v in crushed if isinstance(v, int | float)]
assert 0 in numeric_values # First
assert 49 in numeric_values # Last
def test_change_point_preserved(self, crusher_large_k):
# Stable at 10, then jumps to 100
numbers = [10.0] * 50 + [100.0] * 50
crushed, strategy = crusher_large_k._crush_number_array(numbers)
numeric_values = [v for v in crushed if isinstance(v, int | float)]
# Both 10.0 and 100.0 should be present
assert 10.0 in numeric_values
assert 100.0 in numeric_values
def test_nan_inf_filtered(self, crusher):
numbers = [1.0, 2.0, float("nan"), float("inf"), 3.0] * 10
crushed, strategy = crusher._crush_number_array(numbers)
# Should not crash; stats in strategy based on finite values
assert "min=" in strategy
assert "max=" in strategy
def test_integers_preserved_as_int(self, crusher):
numbers = list(range(50))
crushed, strategy = crusher._crush_number_array(numbers)
numeric_values = [v for v in crushed if isinstance(v, int | float)]
# Integers should remain integers (not converted to float)
assert any(isinstance(v, int) for v in numeric_values)
def test_statistics_accuracy(self, crusher):
numbers = list(range(1, 101)) # 1 to 100
crushed, strategy = crusher._crush_number_array(numbers)
# Stats are in the strategy string
assert "min=1" in strategy
assert "max=100" in strategy
assert "mean=50.5" in strategy
# =====================================================================
# Mixed Array Compression
# =====================================================================
class TestCrushMixedArray:
def test_basic_compression(self, crusher_large_k):
mixed = [{"id": i} for i in range(30)] + [f"msg_{i}" for i in range(30)]
crushed, strategy = crusher_large_k._crush_mixed_array(mixed)
# With diversity-aware K, unique items may all be kept.
# Verify compression happened OR items are preserved due to high diversity.
assert len(crushed) <= len(mixed)
assert "mixed" in strategy
def test_small_groups_kept(self, crusher):
# 50 dicts + 3 strings (below threshold)
mixed = [{"id": i} for i in range(50)] + ["rare_1", "rare_2", "rare_3"]
crushed, strategy = crusher._crush_mixed_array(mixed)
# All 3 rare strings should be kept (below min_items threshold)
assert "rare_1" in crushed
assert "rare_2" in crushed
assert "rare_3" in crushed
def test_errors_across_types(self, crusher_large_k):
mixed = (
[{"status": "ok"}] * 30
+ [{"status": "error: timeout"}]
+ ["error: auth failed"]
+ [f"ok_{i}" for i in range(30)]
)
crushed, strategy = crusher_large_k._crush_mixed_array(mixed)
crushed_str = json.dumps(crushed)
assert "error: timeout" in crushed_str
assert "error: auth failed" in crushed_str
def test_original_order_preserved(self, crusher_large_k):
mixed = [{"id": i} for i in range(20)] + [f"str_{i}" for i in range(20)]
crushed, strategy = crusher_large_k._crush_mixed_array(mixed)
# Dicts should come before strings (original order)
first_str_idx = next(
(i for i, item in enumerate(crushed) if isinstance(item, str)), len(crushed)
)
last_dict_idx = max(
(i for i, item in enumerate(crushed) if isinstance(item, dict)), default=-1
)
assert last_dict_idx < first_str_idx
def test_passthrough_small(self, crusher):
mixed = [1, "two", {"three": 3}]
crushed, strategy = crusher._crush_mixed_array(mixed)
assert crushed == mixed
assert "passthrough" in strategy
# =====================================================================
# Adaptive K
# =====================================================================
class TestAdaptiveK:
def test_scales_with_n(self, crusher):
"""K grows sublinearly with collection size (or saturates at max_items)."""
small = [f"item_{i}" for i in range(20)]
large = [f"item_{i}" for i in range(500)]
k_small = crusher._compute_k_split(small)[0]
k_large = crusher._compute_k_split(large)[0]
# With diversity-aware K, both may saturate at max_items_after_crush
# for highly unique items. The key property: k never exceeds max_items.
assert k_large >= k_small
assert k_large <= crusher.config.max_items_after_crush
def test_respects_max_items(self, crusher):
items = [f"item_{i}" for i in range(1000)]
k_total, _, _, _ = crusher._compute_k_split(items)
assert k_total <= crusher.config.max_items_after_crush
def test_first_last_fractions(self, crusher):
items = [f"item_{i}" for i in range(100)]
k_total, k_first, k_last, k_importance = crusher._compute_k_split(items)
# First and last should be roughly the configured fractions
assert k_first >= 1
assert k_last >= 1
assert k_first + k_last + k_importance == k_total
def test_homogeneous_vs_diverse(self, crusher_large_k):
"""Homogeneous data should produce smaller K than diverse data."""
homogeneous = ["same value"] * 100
diverse = [f"unique_value_{i}_{'x' * (i * 10)}" for i in range(100)]
k_homo = crusher_large_k._compute_k_split(homogeneous)[0]
k_diverse = crusher_large_k._compute_k_split(diverse)[0]
# Diverse should need more items (or at least equal)
assert k_diverse >= k_homo
# =====================================================================
# Safety Guarantees (parametrized across types)
# =====================================================================
class TestSafetyGuarantees:
@pytest.mark.parametrize(
"items,item_type",
[
([{"status": "ok"}] * 50 + [{"status": "error: timeout"}], "dict"),
(["ok"] * 50 + ["error: connection failed"], "string"),
(
[{"id": i} for i in range(30)]
+ ["error: auth denied"]
+ [f"ok_{i}" for i in range(30)],
"mixed",
),
],
ids=["dict_array", "string_array", "mixed_array"],
)
def test_errors_never_dropped(self, crusher_large_k, items, item_type):
"""Error items must be preserved regardless of array type."""
if item_type == "dict":
crushed, _ = crusher_large_k._crush_string_array([json.dumps(i) for i in items])
crushed_text = " ".join(crushed)
elif item_type == "string":
crushed, _ = crusher_large_k._crush_string_array(items)
crushed_text = " ".join(crushed)
elif item_type == "mixed":
crushed, _ = crusher_large_k._crush_mixed_array(items)
crushed_text = json.dumps(crushed)
assert "error" in crushed_text.lower()
@pytest.mark.parametrize(
"items",
[
[f"item_{i}" for i in range(50)],
list(range(50)),
],
ids=["string_array", "number_array"],
)
def test_first_last_always_present(self, crusher, items):
"""First and last items must be present in output."""
if isinstance(items[0], str):
crushed, _ = crusher._crush_string_array(items)
assert items[0] in crushed
assert items[-1] in crushed
else:
crushed, _ = crusher._crush_number_array(items)
numeric = [v for v in crushed if isinstance(v, int | float)]
assert items[0] in numeric
assert items[-1] in numeric
@pytest.mark.parametrize(
"items",
[
["a", "b", "c"],
[1, 2, 3],
[{"x": 1}, "y", 2],
],
ids=["string_below", "number_below", "mixed_below"],
)
def test_passthrough_below_min_items(self, crusher, items):
"""Arrays below min_items_to_analyze pass through unchanged."""
if all(isinstance(i, str) for i in items):
crushed, strategy = crusher._crush_string_array(items)
elif all(isinstance(i, int | float) for i in items):
crushed, strategy = crusher._crush_number_array(items)
else:
crushed, strategy = crusher._crush_mixed_array(items)
assert "passthrough" in strategy
# =====================================================================
# Integration: Full pipeline
# =====================================================================
class TestFullPipelineIntegration:
"""Test that new types work through the compress() API."""
def test_string_array_via_compress(self):
from headroom import compress
strings = [f"log line {i}: GET /api 200" for i in range(100)]
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Show logs"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_logs", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": json.dumps(strings)},
]
result = compress(messages)
assert result.tokens_saved > 0
assert result.compression_ratio > 0
def test_number_array_via_compress(self):
from headroom import compress
numbers = [42.0 + i * 0.1 for i in range(200)]
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Show metrics"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_metrics", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": json.dumps(numbers)},
]
result = compress(messages)
assert result.tokens_saved > 0
def test_dict_array_unchanged(self):
"""Verify dict arrays still work (regression test)."""
from headroom import compress
data = [{"id": i, "name": f"user_{i}", "status": "active"} for i in range(100)]
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "List users"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "list_users", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": json.dumps(data)},
]
result = compress(messages)
assert result.tokens_saved > 0
assert result.compression_ratio > 0