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
headroom/tests/test_ccr_feedback.py
Abhay Singh eecb81e847
fix(cache/ccr): don't count a successful eviction as a retrieval (#2106) 
## Description

The compression feedback learner treats a *successful* compression as
evidence that it should compress less, which inverts the learning
signal.

When `CompressionStore` evicts an entry that was never retrieved, it
emits a synthetic event to tell the learner the compression was fine
(the model never needed the original):

```python
success_event = RetrievalEvent(..., retrieval_type="eviction_success")
self._pending_feedback_events.append(success_event)
```

`process_pending_feedback` forwards every pending event to
`CompressionFeedback.record_retrieval` unconditionally. But
`record_retrieval` has no branch for `"eviction_success"` — and since
that string isn't `"full"`, it lands in the `else`:

```python
self._total_retrievals += 1
pattern.total_retrievals += 1
if event.retrieval_type == "full":
    pattern.full_retrievals += 1
else:
    pattern.search_retrievals += 1   # <-- eviction_success counted here
```

So a compression that worked is booked as a *search retrieval*, which
raises the tool's `retrieval_rate` and `search_rate`.
`get_compression_hints` reads a high retrieval rate as "we're
compressing too aggressively" and recommends larger `max_items` / lower
aggressiveness (or `skip_compression`). Net effect: the more often
compression succeeds, the more the learner backs off from compressing. A
standalone repro books a single successful eviction as a 100% retrieval
rate.

Every sibling consumer of the event distinguishes the type — telemetry
and TOIN both receive `retrieval_type="eviction_success"` and handle it
as its own thing. Only the local feedback counter ignores the
distinction.

## Fix

Recognize `"eviction_success"` in `record_retrieval` and leave it out of
the retrieval counters. The compression itself is already counted by
`record_compression` at store time, so an entry that is compressed and
never retrieved already yields a low retrieval rate — which is the
correct "compression worked" signal. Genuine `full`/`search` retrievals
are unchanged.

Closes #

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `headroom/cache/compression_feedback.py`: early-return in
`record_retrieval` for `retrieval_type == "eviction_success"` so it is
not counted as a retrieval, with a comment explaining the signal.
- `tests/test_ccr_feedback.py`: add
`test_eviction_success_is_not_counted_as_retrieval` (asserts the
counters stay at zero after a successful eviction, and that a genuine
retrieval afterward still counts).
- `CHANGELOG.md`: Bug Fixes entry.

## Testing

- [ ] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
$ uvx ruff@0.15.17 check headroom/cache/compression_feedback.py tests/test_ccr_feedback.py
All checks passed!
$ python -m py_compile headroom/cache/compression_feedback.py tests/test_ccr_feedback.py
OK
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.12, `uvx ruff@0.15.17`. Importing
`headroom` pulls in the torch/transformers stack and a full `pytest`
gets OOM-killed on this box, so I verified the counting logic with a
dependency-free script that replicates
`record_compression`/`record_retrieval` and the
`retrieval_rate`/`search_rate` properties, and left the full pytest to
CI.
- Exact command / steps: recorded one compression, then a
`retrieval_type="eviction_success"` event, under the old counting (no
branch) and the new counting (early return), plus a genuine `search`
retrieval as a control.
- Observed result: old counting books the successful eviction as a
retrieval — `retrieval_rate=1.0`, `search_rate=1.0` — so the learner
would back off from compressing; new counting leaves
`retrieval_rate=0.0` and `total_retrievals=0`; a real retrieval
afterward still increments to 1. The new test asserts exactly this.
- Not tested: an end-to-end store-evict-then-hint cycle through
`CompressionStore.process_pending_feedback`; full local `pytest`
deferred to CI (OOM, per above).

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

The "unit tests pass locally" and "type checking" boxes are unchecked
because the full suite imports the ML stack, which I can't run in this
environment; the fix is a single early-return in a pure counting method,
verified by the standalone proof and the new regression test (which
reuses the existing `test_ccr_feedback.py` pattern) for CI. Scope is
deliberately limited to the local feedback learner — telemetry and TOIN
already receive the `eviction_success` type and handle it separately.

Co-authored-by: Tejas Chopra <chopratejas@gmail.com>
2026-07-13 10:53:55 -04:00

399 lines
13 KiB
Python
Raw Permalink Blame History

"""Tests for CCR feedback loop and pattern learning."""
import time
import pytest
from headroom.cache.compression_feedback import (
CompressionFeedback,
LocalToolPattern,
get_compression_feedback,
reset_compression_feedback,
)
from headroom.cache.compression_store import (
CompressionStore,
RetrievalEvent,
reset_compression_store,
)
@pytest.fixture(autouse=True)
def reset_globals():
"""Reset global state before each test."""
reset_compression_feedback()
reset_compression_store()
yield
reset_compression_feedback()
reset_compression_store()
class TestCompressionFeedback:
"""Test CompressionFeedback analyzer."""
def test_record_compression(self):
"""Recording compression events updates tool patterns."""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
feedback.record_compression("test_tool", 200, 20)
patterns = feedback.get_all_patterns()
assert "test_tool" in patterns
assert patterns["test_tool"].total_compressions == 2
def test_record_retrieval(self):
"""Recording retrieval events updates patterns."""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
event = RetrievalEvent(
hash="abc123",
query="find errors",
items_retrieved=50,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
feedback.record_retrieval(event)
patterns = feedback.get_all_patterns()
assert patterns["test_tool"].total_retrievals == 1
assert patterns["test_tool"].search_retrievals == 1
def test_retrieval_rate_calculation(self):
"""Retrieval rate is calculated correctly."""
feedback = CompressionFeedback()
# 10 compressions
for _ in range(10):
feedback.record_compression("test_tool", 100, 10)
# 5 retrievals (50% retrieval rate)
for _ in range(5):
event = RetrievalEvent(
hash="abc123",
query=None,
items_retrieved=100,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="full",
)
feedback.record_retrieval(event)
pattern = feedback.get_all_patterns()["test_tool"]
assert pattern.retrieval_rate == 0.5
assert pattern.full_retrieval_rate == 1.0 # All were full retrievals
def test_eviction_success_is_not_counted_as_retrieval(self):
"""An eviction-without-retrieval is a compression success, not a retrieval.
The event arrives with retrieval_type="eviction_success". Because that
isn't "full" it used to fall into the search_retrievals branch and
inflate retrieval_rate/search_rate, driving get_compression_hints toward
less aggressive compression — the inverse of the intended signal. It must
leave the retrieval counters untouched.
"""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
event = RetrievalEvent(
hash="abc123",
query=None,
items_retrieved=0,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="eviction_success",
)
feedback.record_retrieval(event, strategy="smart")
pattern = feedback.get_all_patterns()["test_tool"]
assert pattern.total_retrievals == 0
assert pattern.search_retrievals == 0
assert pattern.retrieval_rate == 0.0 # a successful compression, not a retrieval
# A genuine retrieval afterward is still counted.
feedback.record_retrieval(
RetrievalEvent(
hash="def456",
query="find errors",
items_retrieved=50,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
)
pattern = feedback.get_all_patterns()["test_tool"]
assert pattern.total_retrievals == 1
assert pattern.search_retrievals == 1
def test_hints_default_with_no_data(self):
"""Default hints returned when no data exists."""
feedback = CompressionFeedback()
hints = feedback.get_compression_hints("unknown_tool")
assert hints.max_items == 15 # Default
assert hints.skip_compression is False
assert "No pattern data" in hints.reason
def test_hints_insufficient_samples(self):
"""Default hints returned with insufficient samples."""
feedback = CompressionFeedback()
# Only 3 compressions (need 5 for hints)
for _ in range(3):
feedback.record_compression("test_tool", 100, 10)
hints = feedback.get_compression_hints("test_tool")
assert hints.max_items == 15 # Default
assert "Insufficient data" in hints.reason
def test_hints_high_retrieval_rate_less_aggressive(self):
"""High retrieval rate results in less aggressive compression."""
feedback = CompressionFeedback()
# 10 compressions
for _ in range(10):
feedback.record_compression("test_tool", 100, 10)
# 6 retrievals (60% retrieval rate - HIGH)
for _ in range(6):
event = RetrievalEvent(
hash="abc123",
query="search query",
items_retrieved=50,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
feedback.record_retrieval(event)
hints = feedback.get_compression_hints("test_tool")
assert hints.max_items > 15 # Should be more than default
assert hints.aggressiveness < 0.7 # Less aggressive
assert "High retrieval rate" in hints.reason or "less aggressive" in hints.reason.lower()
def test_hints_very_high_full_retrieval_skips_compression(self):
"""Very high full retrieval rate recommends skipping compression."""
feedback = CompressionFeedback()
# 10 compressions
for _ in range(10):
feedback.record_compression("test_tool", 100, 10)
# 9 FULL retrievals (90% retrieval rate, all full)
for _ in range(9):
event = RetrievalEvent(
hash="abc123",
query=None,
items_retrieved=100,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="full",
)
feedback.record_retrieval(event)
hints = feedback.get_compression_hints("test_tool")
assert hints.skip_compression is True
assert "skip compression" in hints.reason.lower()
def test_hints_low_retrieval_rate_aggressive(self):
"""Low retrieval rate means current compression is effective."""
feedback = CompressionFeedback()
# 10 compressions
for _ in range(10):
feedback.record_compression("test_tool", 100, 10)
# Only 1 retrieval (10% - LOW)
event = RetrievalEvent(
hash="abc123",
query=None,
items_retrieved=100,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="full",
)
feedback.record_retrieval(event)
hints = feedback.get_compression_hints("test_tool")
assert hints.max_items == 15 # Default/aggressive
assert "effective" in hints.reason.lower() or "Low retrieval" in hints.reason
def test_common_queries_tracked(self):
"""Common search queries are tracked per tool."""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
queries = ["find errors", "find errors", "status:failed", "error"]
for q in queries:
event = RetrievalEvent(
hash="abc123",
query=q,
items_retrieved=10,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
feedback.record_retrieval(event)
pattern = feedback.get_all_patterns()["test_tool"]
assert "find errors" in pattern.common_queries
assert pattern.common_queries["find errors"] == 2
def test_queried_fields_extracted(self):
"""Field names are extracted from queries."""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
# Query with field:value patterns
event = RetrievalEvent(
hash="abc123",
query="status:error id=12345",
items_retrieved=10,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
feedback.record_retrieval(event)
pattern = feedback.get_all_patterns()["test_tool"]
assert "status" in pattern.queried_fields
assert "id" in pattern.queried_fields
def test_preserve_fields_in_hints(self):
"""Frequently queried fields appear in hints."""
feedback = CompressionFeedback()
# Multiple compressions
for _ in range(10):
feedback.record_compression("test_tool", 100, 10)
# Multiple queries with same fields
for _ in range(5):
event = RetrievalEvent(
hash="abc123",
query="status:error code:500",
items_retrieved=10,
total_items=100,
tool_name="test_tool",
timestamp=time.time(),
retrieval_type="search",
)
feedback.record_retrieval(event)
hints = feedback.get_compression_hints("test_tool")
# Even if retrieval rate triggers hints, preserve_fields should be populated
assert len(hints.preserve_fields) > 0
def test_stats_returns_overview(self):
"""get_stats returns comprehensive overview."""
feedback = CompressionFeedback()
feedback.record_compression("tool_a", 100, 10)
feedback.record_compression("tool_b", 200, 20)
stats = feedback.get_stats()
assert stats["total_compressions"] == 2
assert stats["tools_tracked"] == 2
assert "tool_a" in stats["tool_patterns"]
assert "tool_b" in stats["tool_patterns"]
def test_clear_resets_state(self):
"""clear() removes all learned patterns."""
feedback = CompressionFeedback()
feedback.record_compression("test_tool", 100, 10)
feedback.clear()
assert len(feedback.get_all_patterns()) == 0
stats = feedback.get_stats()
assert stats["total_compressions"] == 0
class TestLocalToolPattern:
"""Test LocalToolPattern dataclass."""
def test_retrieval_rate_zero_compressions(self):
"""Retrieval rate is 0 when no compressions."""
pattern = LocalToolPattern(tool_name="test")
assert pattern.retrieval_rate == 0.0
def test_full_retrieval_rate_zero_retrievals(self):
"""Full retrieval rate is 0 when no retrievals."""
pattern = LocalToolPattern(tool_name="test")
assert pattern.full_retrieval_rate == 0.0
def test_search_rate_calculation(self):
"""Search rate is calculated correctly."""
pattern = LocalToolPattern(
tool_name="test",
total_retrievals=10,
full_retrievals=3,
search_retrievals=7,
)
assert pattern.search_rate == 0.7
class TestGlobalFeedback:
"""Test global feedback singleton."""
def test_singleton_returns_same_instance(self):
"""get_compression_feedback returns same instance."""
fb1 = get_compression_feedback()
fb2 = get_compression_feedback()
assert fb1 is fb2
def test_reset_clears_singleton(self):
"""reset_compression_feedback creates new instance."""
fb1 = get_compression_feedback()
fb1.record_compression("test", 100, 10)
reset_compression_feedback()
fb2 = get_compression_feedback()
assert len(fb2.get_all_patterns()) == 0
class TestFeedbackIntegrationWithStore:
"""Test feedback integration with CompressionStore."""
def test_store_notifies_feedback_on_retrieval(self):
"""CompressionStore adds events to pending for feedback processing."""
store = CompressionStore()
# Store content
hash_key = store.store(
original='[{"id": 1}, {"id": 2}]',
compressed='[{"id": 1}]',
original_item_count=2,
compressed_item_count=1,
tool_name="test_tool",
)
# Retrieve (should log event)
store.retrieve(hash_key)
# Process pending events (uses global feedback)
store.process_pending_feedback()
# Now global feedback should have the retrieval
feedback = get_compression_feedback()
patterns = feedback.get_all_patterns()
assert "test_tool" in patterns
assert patterns["test_tool"].total_retrievals == 1
Reference in a new issue View git blame Copy permalink