# How to Debug Compression Issues in Headroom and View Saved Tokens

> Debug Headroom compression issues by enabling CCRConfig and inspecting token savings with get_compression_store().get_stats(). Verify cached entries using retrieve(hash).

- Repository: [Tejas Chopra/headroom](https://github.com/chopratejas/headroom)
- Tags: how-to-guide
- Published: 2026-06-09

---

**Enable `CCRConfig` with `ccr_enabled=True` and `ccr_inject_marker=True`, then use `get_compression_store().get_stats()` to inspect token savings and `retrieve(hash)` to verify cached entries in the Python `CompressionStore`.**

Headroom's compression pipeline optimizes token usage for LLM interactions through a multi-layered architecture involving content routing, strategy-specific compression, and reversible caching. When compression fails to deliver expected savings or CCR markers like `<<ccr:abc123…>>` cannot be retrieved, developers need systematic debugging approaches. This guide walks through diagnosing compression issues in the `chopratejas/headroom` repository using actual source file paths and runtime inspection techniques.

## Understanding Headroom's Compression Pipeline

### Transform Selection via ContentRouter

The process begins in [`headroom/transforms/content_router.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/content_router.py) where the `ContentRouter` class analyzes raw output and routes it to the appropriate compressor. The routing decision is recorded in a `RouterCompressionResult` containing the original text, compressed text, chosen `CompressionStrategy`, and a detailed `routing_log`.

### Compression Execution and Result Types

Each compressor returns specific dataclasses containing metrics. The `SmartCrusher` (for JSON arrays) and other compressors return objects like `CrushResult` or `CompressionResult` that include `original_tokens`, `compressed_tokens`, and `compression_ratio`. For JSON data, `SmartCrusher` in [`headroom/transforms/smart_crusher.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/smart_crusher.py) also emits CCR markers when configured.

### The CCR Store Architecture

The `CompressionStore` in [`headroom/cache/compression_store.py`](https://github.com/chopratejas/headroom/blob/main/headroom/cache/compression_store.py) provides in-process caching for dropped or substituted data. It tracks original and compressed token counts, retrieval statistics, and TTL (default 5 minutes) per entry. The store serves as the backbone for Headroom's "Compress-Cache-Retrieve" (CCR) pattern.

## Why Debugging Often Fails

Several architectural behaviors commonly mask compression issues:

- **Silent Fallbacks**: The `ContentRouter` may fall back to less aggressive compressors (e.g., `Kompress` after `SmartCrusher`) when compression ratios are too low, bypassing the CCR store.
- **Missing Markers**: CCR markers only appear when `CCRConfig` has `ccr_enabled=True` and `ccr_inject_marker=True` enabled.
- **TTL Eviction**: Entries disappear after 5 minutes or via LRU logic, causing "404" retrieval errors in long-running sessions.

## How to View Saved Tokens and Cache Statistics

Access the singleton store and inspect aggregate or entry-specific metrics:

1. **Obtain the store instance**:

   ```python
   from headroom.cache.compression_store import get_compression_store
   store = get_compression_store()
   ```

2. **Inspect aggregate statistics**:

   ```python
   stats = store.get_stats()
   print(f"Original tokens cached: {stats['total_original_tokens']:,}")
   print(f"Compressed tokens cached: {stats['total_compressed_tokens']:,}")
   print(f"Cache entries: {stats['entry_count']}")
   ```

3. **Inspect a specific entry** using the CCR hash from `<<ccr:HASH …>>` markers:

   ```python
   entry = store.retrieve('abc123')
   if entry:
       print(f"Original content size (tokens): {entry.original_tokens}")
       print(f"Compressed content size (tokens): {entry.compressed_tokens}")
       print(f"Retrieval count: {entry.retrieval_count}")
   ```

4. **List recent retrieval events** to confirm the LLM requested dropped data:

   ```python
   recent = store.get_retrieval_events(limit=5)
   for ev in recent:
       print(f"[{ev.retrieval_type}] hash={ev.hash[:8]} tokens={ev.items_retrieved}/{ev.total_items}")
   ```

5. **Clear the store (test environments)** to eliminate stale TTLs:

   ```python
   store.clear()
   ```

## Step-by-Step Debugging Workflow

Follow this sequence to isolate compression failures:

1. **Enable DEBUG logging** to capture routing decisions in `ContentRouter.compress` via `_log_router_debug`.

2. **Verify CCRConfig** settings ensure `ccr_enabled=True` and `ccr_inject_marker=True`.

3. **Locate CCR markers** like `<<ccr:HASH …>>` in the compressed output from `SmartCrusher._mirror_ccr_to_python_store`.

4. **Retrieve entries** using `CompressionStore.retrieve(hash)` to inspect token counts.

5. **Check aggregate stats** via `store.get_stats()`; zero values indicate passthrough fallback due to `min_compression_ratio_for_ccr` not being met.

6. **Inspect eviction logs** in `store._eviction_heap` or check `store._clean_expired` if entries are missing.

## Complete Debugging Code Example

```python
import logging
import re
from headroom.transforms.content_router import ContentRouter
from headroom.cache.compression_store import get_compression_store

# 1️⃣ Enable detailed debug output

logging.basicConfig(level=logging.DEBUG)

router = ContentRouter()

# Example payload that triggers SmartCrusher (JSON array)

payload = """
[
  {"id": 1, "name": "Alice", "role": "admin"},
  {"id": 2, "name": "Bob",   "role": "user"},
  {"id": 3, "name": "Carol", "role": "user"}
]
"""

# Run the router – it will log the chosen strategy and any CCR markers

result = router.compress(payload, context="list users")
print("Chosen strategy:", result.strategy_used.value)
print("Compressed output:", result.compressed)

# 2️⃣ Extract CCR hash from the output (if any)

match = re.search(r"<<ccr:([0-9a-f]+)", result.compressed)
if match:
    ccr_hash = match.group(1)
    store = get_compression_store()
    entry = store.retrieve(ccr_hash)
    if entry:
        print("\n--- CCR Entry ---")
        print("Original tokens:", entry.original_tokens)
        print("Compressed tokens:", entry.compressed_tokens)
        print("Retrieval count:", entry.retrieval_count)
    else:
        print("CCR entry not found (might have expired).")
else:
    print("No CCR marker emitted – compression probably fell back to passthrough.")

# 3️⃣ View aggregate stats

stats = store.get_stats()
print("\n--- Store Stats ---")
print(f"Total original tokens cached: {stats['total_original_tokens']:,}")
print(f"Total compressed tokens cached: {stats['total_compressed_tokens']:,}")
print(f"Cache entries: {stats['entry_count']}")

```

## Key Source Files for Reference

- [`headroom/transforms/content_router.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/content_router.py) – Orchestrates detection and strategy selection
- [`headroom/transforms/smart_crusher.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/smart_crusher.py) – Rust-backed JSON compression
- [`headroom/cache/compression_store.py`](https://github.com/chopratejas/headroom/blob/main/headroom/cache/compression_store.py) – In-process cache with TTL and retrieval logs
- [`headroom/utils.py`](https://github.com/chopratejas/headroom/blob/main/headroom/utils.py) – Helper functions for marker creation and hash computation
- [`wiki/text-compression.md`](https://github.com/chopratejas/headroom/blob/main/wiki/text-compression.md) – Architecture overview
- [`wiki/troubleshooting.md`](https://github.com/chopratejas/headroom/blob/main/wiki/troubleshooting.md) – Common failure diagnostics

## Summary

- Enable CCR markers and DEBUG logging to trace routing decisions in `ContentRouter`.
- Use `get_compression_store().get_stats()` to verify aggregate token savings across all cached entries.
- Retrieve specific entries by hash to confirm data persistence and view `compression_ratio`.
- Check for TTL expiration and fallback strategies when entries are missing from the store.
- Reference [`headroom/transforms/content_router.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/content_router.py), [`headroom/transforms/smart_crusher.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/smart_crusher.py), and [`headroom/cache/compression_store.py`](https://github.com/chopratejas/headroom/blob/main/headroom/cache/compression_store.py) for implementation details.

## Frequently Asked Questions

### Why are my CCR markers not appearing in the output?

Markers require explicit configuration in `CCRConfig` with both `ccr_enabled=True` and `ccr_inject_marker=True`. If the `ContentRouter` falls back to a passthrough strategy due to low compression ratios, no CCR entry is created and markers are omitted.

### How do I check if compression actually saved tokens?

Call `store.get_stats()` on the singleton `CompressionStore` instance returned by `get_compression_store()`. The dictionary contains `total_original_tokens` and `total_compressed_tokens` showing the aggregate savings across all cached entries.

### Why does retrieval return None for a valid CCR hash?

The `CompressionStore` uses TTL (5-minute default) and LRU eviction. Entries expire between creation and retrieval, especially in long-running sessions. Check `store._eviction_heap` or enable detailed retrieval logging to confirm expiration events.

### Can I clear the compression cache during testing?

Yes. Call `store.clear()` on the `CompressionStore` instance to reset all entries, eviction heaps, and statistics. This is useful when debugging to eliminate stale TTL concerns.