Fix ruff formatting

devto-article.md

@@ -0,0 +1,295 @@
+# I Was Wasting 85% of My LLM Tokens on JSON Boilerplate
+
+I recently built an agent to handle some SRE tasks—fetching logs, querying databases, searching code. It worked, but when I looked at the traces, I was annoyed.
+
+It wasn't just that it was expensive (though the bill was climbing). It was the sheer **inefficiency**.
+
+I looked at a single tool output—a search for Python files. It was 40,000 tokens.
+About 35,000 of those tokens were just `"type": "file"` and `"language": "python"` repeated 2,000 times.
+
+We are paying premium compute prices to force state-of-the-art models to read standard JSON boilerplate.
+
+I couldn't find a tool that solved this without breaking the agent, so I wrote one. It's called **Headroom**. It's a context optimization layer that sits between your app and your LLM. It compresses context by ~85% without losing semantic meaning.
+
+It's open source (Apache-2.0). If you just want the code:
+**[github.com/chopratejas/headroom](https://github.com/chopratejas/headroom)**
+
+---
+
+## Why Truncation and Summarization Don't Work
+
+When your context window fills up, the standard industry solution is **truncation** (chopping off the oldest messages or the middle of the document).
+
+But for an agent, truncation is dangerous.
+
+* If you chop the middle of a log file, you might lose the one error line that explains the crash.
+* If you chop a file list, you might lose the exact config file the user asked for.
+
+I tried **summarization** (using a cheaper model to summarize the data first), but that introduced hallucination. I had a summarizer tell me a deployment "looked fine" because it ignored specific error codes in the raw log.
+
+I needed a third option: **Lossless compression.** Or at least, "intent-lossless."
+
+---
+
+## The Core Idea: Statistical Analysis, Not Blind Truncation
+
+I realized that 90% of the data in a tool output is just schema scaffolding. The LLM doesn't need to see `status: active` repeated a thousand times. It needs the **anomalies**.
+
+Headroom's SmartCrusher runs statistical analysis before touching your data:
+
+**1. Constant Factoring**
+If every item in an array has `"type": "file"`, it doesn't repeat that 2,000 times. It extracts constants once.
+
+**2. Outlier Detection**
+It calculates standard deviation of numerical fields. It preserves the spikes—the values that are >2σ from the mean. Those are usually what matters.
+
+**3. Error Preservation**
+Hard rule: never discard strings that look like stack traces, error messages, or failures. Errors are sacred.
+
+**4. Relevance Scoring**
+If you searched for "auth", items containing "auth" get preserved. Uses BM25 + semantic embeddings (hybrid scoring) to match items against the user's query context.
+
+**5. First/Last Retention**
+Always keeps first few and last few items. The LLM expects to see some examples, and recency matters.
+
+The result: 40,000 tokens → 4,000 tokens. Same information density. No hallucination risk.
+
+---
+
+## CCR: Making Compression Reversible
+
+Here's the insight that changed everything: **compression should be reversible**.
+
+I call the architecture **CCR** (Compress-Cache-Retrieve):
+
+### 1. Compress
+SmartCrusher compresses the tool output from 2,000 items to 20.
+
+### 2. Cache
+The original 2,000 items are cached locally (5-minute TTL, LRU eviction).
+
+### 3. Retrieve
+Headroom injects a tool called `headroom_retrieve()` into the LLM's context. If the model looks at the compressed summary and decides it needs more data—maybe the user asked a follow-up question—it can call that tool. Headroom fetches from the cache and returns the relevant items.
+
+This changes the risk calculus. You can compress aggressively (90%+) because **nothing is ever truly lost**. The model can always "unzip" what it needs.
+
+I've had conversations like this:
+
+```
+Turn 1: "Search for all Python files"
+        → 1000 files returned, compressed to 15
+
+Turn 5: "Actually, what was that file handling JWT tokens?"
+        → LLM calls headroom_retrieve("jwt")
+        → Returns jwt_handler.py from cached data
+```
+
+No extra API calls. No "sorry, I don't have that information anymore."
+
+---
+
+## TOIN: The Network Effect
+
+Here's where it gets interesting. Headroom learns from compression patterns.
+
+**TOIN** (Tool Output Intelligence Network) tracks—anonymously—what happens after compression:
+- Which fields get retrieved most often?
+- Which tool types have high retrieval rates?
+- What query patterns trigger retrievals?
+
+This data feeds back into compression recommendations. If TOIN learns that users frequently retrieve `error_code` fields after compression, it tells SmartCrusher to preserve `error_code` more aggressively next time.
+
+Privacy is built in:
+- No actual data values stored
+- Tool names are structure hashes
+- Field names are SHA256[:8] hashes
+- No user identifiers
+
+The network effect: more users → more compression events → better recommendations for everyone.
+
+---
+
+## Memory: Cross-Conversation Learning
+
+Agents often need to remember things across conversations. "I prefer dark mode." "My timezone is PST." "I'm working on the auth refactor."
+
+Headroom has a memory system that extracts and stores these facts automatically.
+
+Two approaches:
+
+**Fast Memory (Recommended)**
+Zero extra latency. The LLM outputs a `<memory>` block inline with its response. Headroom parses it out and stores the memory.
+
+```python
+from headroom.memory import with_fast_memory
+client = with_fast_memory(OpenAI(), user_id="alice")
+
+# Memories extracted automatically from responses
+# Injected automatically into future requests
+```
+
+**Background Memory**
+Separate LLM call extracts memories asynchronously. More accurate but adds latency.
+
+```python
+from headroom import with_memory
+client = with_memory(OpenAI(), user_id="alice")
+```
+
+Memories are stored locally (SQLite) and injected into future conversations. The model remembers that Alice prefers dark mode without you managing state.
+
+---
+
+## The Transform Pipeline
+
+Headroom runs four transforms on each request:
+
+### 1. CacheAligner
+LLM providers offer cached token pricing (Anthropic: 90% off, OpenAI: 50% off). But caching only works if your prompt prefix is stable.
+
+Problem: your system prompt probably has a timestamp. `Current time: 2024-01-15 10:32:45`. That breaks caching.
+
+CacheAligner extracts dynamic content and moves it to the end, stabilizing the prefix. Same information, better cache hits.
+
+### 2. SmartCrusher
+The statistical compression engine. Analyzes arrays, detects patterns, preserves anomalies, factors constants.
+
+### 3. ContentRouter
+Different content needs different compression. Code isn't JSON isn't logs isn't prose.
+
+ContentRouter uses ML-based content detection to route data to specialized compressors:
+- **Code** → AST-aware compression (tree-sitter)
+- **JSON** → SmartCrusher
+- **Logs** → LogCompressor (clusters similar messages)
+- **Text** → Optional LLMLingua integration (20x compression, adds latency)
+
+### 4. RollingWindow
+When context exceeds the model limit, something has to go. RollingWindow drops oldest tool calls + responses together (never orphans data), preserves system prompt and recent turns.
+
+---
+
+## Three Ways to Use It
+
+### Option 1: Proxy Server (Zero Code Changes)
+
+```bash
+pip install headroom-ai
+headroom proxy --port 8787
+```
+
+Point your OpenAI client to `http://localhost:8787/v1`. Done.
+
+```python
+from openai import OpenAI
+client = OpenAI(base_url="http://localhost:8787/v1")
+# No other changes
+```
+
+Works with Claude Code, Cursor, any OpenAI-compatible client.
+
+### Option 2: SDK Wrapper
+
+```python
+from headroom import HeadroomClient
+from openai import OpenAI
+
+client = HeadroomClient(OpenAI())
+
+response = client.chat.completions.create(
+    model="gpt-4o",
+    messages=[...],
+    headroom_mode="optimize"  # or "audit" or "simulate"
+)
+```
+
+Three modes:
+- **audit**: Observe only. Logs what would be optimized, doesn't change anything.
+- **optimize**: Apply compression. This is what saves tokens.
+- **simulate**: Dry run. Returns the optimized messages without calling the API.
+
+Start with `audit` to see potential savings, then flip to `optimize` when you're confident.
+
+### Option 3: Framework Integrations
+
+**LangChain:**
+```python
+from langchain_openai import ChatOpenAI
+from headroom.integrations.langchain import HeadroomChatModel
+
+base_model = ChatOpenAI(model="gpt-4o")
+model = HeadroomChatModel(base_model, mode="optimize")
+
+# Use in any chain or agent
+chain = prompt | model | parser
+```
+
+**Agno:**
+```python
+from agno.agent import Agent
+from headroom.integrations.agno import HeadroomAgnoModel
+
+model = HeadroomAgnoModel(original_model, mode="optimize")
+agent = Agent(model=model, tools=[...])
+```
+
+**MCP (Model Context Protocol):**
+```python
+from headroom.integrations.mcp import compress_tool_result
+
+# Compress any tool result before returning to LLM
+compressed = compress_tool_result(tool_name, result_data)
+```
+
+---
+
+## Real Numbers
+
+I've been running this in production for months. Here's what the token reduction looks like:
+
+| Workload | Before | After | Savings |
+|----------|--------|-------|---------|
+| Log Analysis | 22,000 | 3,300 | 85% |
+| Code Search | 45,000 | 4,500 | 90% |
+| Database Queries | 18,000 | 2,700 | 85% |
+| Long Conversations | 80,000 | 32,000 | 60% |
+
+Latency overhead: 3-5ms per request. No extra LLM calls.
+
+---
+
+## What's Coming Next
+
+This is actively maintained. On the roadmap:
+
+**More Frameworks**
+- CrewAI integration
+- AutoGen integration
+- Semantic Kernel integration
+
+**Managed Storage**
+- Cloud-hosted TOIN backend (opt-in)
+- Cross-device memory sync
+- Team-shared compression patterns
+
+**Better Compression**
+- Domain-specific profiles (SRE, coding, data analysis)
+- Custom compressor plugins
+- Streaming compression for real-time tools
+
+---
+
+## Why I Built This
+
+I'm a believer that we're in the "optimization phase" of the AI hype cycle. Getting things to work is table stakes; getting them to work cheaply and reliably is the actual engineering work.
+
+Headroom is my attempt to fix the "context bloat" problem properly. Not with heuristics or truncation, but with statistical analysis and reversible compression.
+
+It runs entirely locally. No data leaves your machine (except to OpenAI/Anthropic as usual). Apache-2.0 licensed.
+
+**Repo:** [github.com/chopratejas/headroom](https://github.com/chopratejas/headroom)
+
+If you find bugs or have ideas, open an issue. I'm actively maintaining this.
+
+---
+
+*Tags: #llm #ai #python #openai #anthropic #agents #optimization*

headroom/cache/compression_store.py

@@ -741,16 +741,13 @@ class CompressionStore:
                         # Handle both direct arrays and wrapped arrays
                         if isinstance(parsed, list):
                             # Filter to dicts only (field learning needs dict items)
-                            retrieved_items = [
-                                item for item in parsed if isinstance(item, dict)
-                            ]
+                            retrieved_items = [item for item in parsed if isinstance(item, dict)]
                         elif isinstance(parsed, dict):
                             # Check for common wrapper patterns: {"items": [...], "results": [...]}
                             for key in ("items", "results", "data", "records"):
                                 if key in parsed and isinstance(parsed[key], list):
                                     retrieved_items = [
-                                        item for item in parsed[key]
-                                        if isinstance(item, dict)
+                                        item for item in parsed[key] if isinstance(item, dict)
                                     ]
                                     break
                     except (json.JSONDecodeError, TypeError):

headroom/telemetry/models.py

@@ -436,7 +436,7 @@ class FieldSemantics:
             "field_hash": self.field_hash,
             "inferred_type": self.inferred_type,
             "confidence": self.confidence,
-            "important_value_hashes": self.important_value_hashes[:self.MAX_IMPORTANT_VALUES],
+            "important_value_hashes": self.important_value_hashes[: self.MAX_IMPORTANT_VALUES],
             "default_value_hash": self.default_value_hash,
             "value_retrieval_frequency": dict(
                 sorted(
@@ -514,9 +514,7 @@ class FieldSemantics:
                 )[: self.MAX_IMPORTANT_VALUES]
 
         # Track query operators
-        self.query_operator_frequency[operator] = (
-            self.query_operator_frequency.get(operator, 0) + 1
-        )
+        self.query_operator_frequency[operator] = self.query_operator_frequency.get(operator, 0) + 1
 
     def record_compression_stats(
         self,
@@ -543,9 +541,7 @@ class FieldSemantics:
         self.total_unique_values_seen = int(
             (self.total_unique_values_seen * (n - 1) + unique_values) / n
         )
-        self.total_values_seen = int(
-            (self.total_values_seen * (n - 1) + total_values) / n
-        )
+        self.total_values_seen = int((self.total_values_seen * (n - 1) + total_values) / n)
         self.most_common_value_frequency = (
             self.most_common_value_frequency * (n - 1) + most_common_frequency
         ) / n
@@ -569,9 +565,7 @@ class FieldSemantics:
             return
 
         # Calculate metrics
-        uniqueness_ratio = (
-            self.total_unique_values_seen / max(1, self.total_values_seen)
-        )
+        uniqueness_ratio = self.total_unique_values_seen / max(1, self.total_values_seen)
         has_dominant_default = self.most_common_value_frequency > 0.7
         retrieval_diversity = len(self.value_retrieval_frequency) / max(1, self.retrieval_count)
 
@@ -598,15 +592,15 @@ class FieldSemantics:
         # ERROR_INDICATOR: Has dominant default + retrievals are for non-default values
         elif has_dominant_default and self.default_value_hash:
             # Check if retrieved values are different from default
-            default_retrieval_count = self.value_retrieval_frequency.get(
-                self.default_value_hash, 0
-            )
+            default_retrieval_count = self.value_retrieval_frequency.get(self.default_value_hash, 0)
             non_default_retrieval_ratio = 1 - (
                 default_retrieval_count / max(1, self.retrieval_count)
             )
             if non_default_retrieval_ratio > 0.7:
                 inferred = "error_indicator"
-                confidence = min(0.9, non_default_retrieval_ratio * self.most_common_value_frequency)
+                confidence = min(
+                    0.9, non_default_retrieval_ratio * self.most_common_value_frequency
+                )
 
         # STATUS: Low uniqueness + specific values retrieved
         elif uniqueness_ratio < 0.2 and retrieval_diversity < 0.5:

headroom/telemetry/toin.py

@@ -169,9 +169,7 @@ class ToolPattern:
             "skip_compression_recommended": self.skip_compression_recommended,
             "preserve_fields": self.preserve_fields,
             # Field-level semantics (TOIN Evolution)
-            "field_semantics": {
-                k: v.to_dict() for k, v in self.field_semantics.items()
-            },
+            "field_semantics": {k: v.to_dict() for k, v in self.field_semantics.items()},
             "sample_size": self.sample_size,
             "user_count": self.user_count,
             "confidence": self.confidence,

headroom/transforms/smart_crusher.py

@@ -645,7 +645,8 @@ def _detect_items_by_learned_semantics(
     # Build a quick lookup for field_hash -> FieldSemantics
     # Pre-filter to fields with sufficient confidence
     confident_semantics = {
-        fh: fs for fh, fs in field_semantics.items()
+        fh: fs
+        for fh, fs in field_semantics.items()
         if fs.confidence >= 0.3 and fs.inferred_type != "unknown"
     }
 
@@ -1599,7 +1600,9 @@ class SmartCrusher(Transform):
             return keep_indices
 
         # Use provided field_semantics or fall back to instance variable (set by crush())
-        effective_field_semantics = field_semantics or getattr(self, "_current_field_semantics", None)
+        effective_field_semantics = field_semantics or getattr(
+            self, "_current_field_semantics", None
+        )
 
         # Identify error items using KEYWORD detection (preservation guarantee)
         # This ensures ALL error items are kept, regardless of frequency
@@ -2024,7 +2027,9 @@ class SmartCrusher(Transform):
         # === TOIN Evolution: Extract field semantics for signal detection ===
         # Store temporarily on instance for use in _prioritize_indices
         # This enables learned signal detection without changing all method signatures
-        self._current_field_semantics = toin_hint.field_semantics if toin_hint.field_semantics else None
+        self._current_field_semantics = (
+            toin_hint.field_semantics if toin_hint.field_semantics else None
+        )
 
         # Local feedback hints (if TOIN didn't apply)
         if not toin_hint_applied and self.config.use_feedback_hints and tool_name:

tests/test_cache/test_client_integration.py

@@ -388,9 +388,9 @@ class TestCacheOptimizerInvocation:
         headroom_meta = result.headroom
 
         # Check that cache optimizer was reported
-        assert headroom_meta.cache_optimizer_used is not None or \
-            any("cache_optimizer" in t for t in (headroom_meta.transforms_applied or [])), \
-            "Cache optimizer usage should be reported in metadata"
+        assert headroom_meta.cache_optimizer_used is not None or any(
+            "cache_optimizer" in t for t in (headroom_meta.transforms_applied or [])
+        ), "Cache optimizer usage should be reported in metadata"
 
     @patch("headroom.storage.sqlite.SQLiteStorage.save")
     def test_optimizer_not_called_in_audit_mode(self, mock_save, temp_db):
@@ -430,9 +430,7 @@ class TestCacheOptimizerInvocation:
         )
 
         # Optimizer should NOT be called in AUDIT mode
-        assert not spy_optimize.called, (
-            "Cache optimizer should NOT be called in AUDIT mode"
-        )
+        assert not spy_optimize.called, "Cache optimizer should NOT be called in AUDIT mode"
 
 
 class TestSemanticCacheIntegration:
@@ -444,9 +442,7 @@ class TestSemanticCacheIntegration:
     """
 
     @patch("headroom.storage.sqlite.SQLiteStorage.save")
-    def test_semantic_cache_hit_returns_cached_response_without_api_call(
-        self, mock_save, temp_db
-    ):
+    def test_semantic_cache_hit_returns_cached_response_without_api_call(self, mock_save, temp_db):
         """CRITICAL: Verify semantic cache hit returns cached response without API call.
 
         This test catches the gap where semantic cache is enabled but cached
@@ -565,8 +561,7 @@ class TestSessionStatsTracking:
         after_requests = after_stats["session"]["requests_total"]
 
         assert after_requests == initial_requests + 1, (
-            f"requests_total should increment. "
-            f"Before: {initial_requests}, After: {after_requests}"
+            f"requests_total should increment. Before: {initial_requests}, After: {after_requests}"
         )
         assert after_stats["session"]["requests_audit"] >= 1, (
             "requests_audit should be at least 1 after AUDIT mode request"

tests/test_proxy_ccr.py

@@ -386,7 +386,9 @@ class TestEndToEndTOINIntegration:
             yield client
         reset_compression_store()
 
-    def test_pipeline_compresses_tool_output_and_records_toin(self, fresh_toin, client_with_optimization):
+    def test_pipeline_compresses_tool_output_and_records_toin(
+        self, fresh_toin, client_with_optimization
+    ):
         """CRITICAL: Verify SmartCrusher compression records events in TOIN.
 
         This tests the production code path:
@@ -467,8 +469,7 @@ class TestEndToEndTOINIntegration:
 
         # Verify SmartCrusher was invoked (transform name starts with smart_crush)
         smart_crush_applied = any(
-            t.startswith("smart_crush") or t.startswith("smart:")
-            for t in result.transforms_applied
+            t.startswith("smart_crush") or t.startswith("smart:") for t in result.transforms_applied
         )
         assert smart_crush_applied, (
             f"SmartCrusher should be in transforms: {result.transforms_applied}"

tests/test_toin_field_learning.py

@@ -245,7 +245,9 @@ class TestFieldSemanticsLearning:
 
         # 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"
+        assert not fs.is_value_important(unknown_hash), (
+            "Never-retrieved value should NOT be important"
+        )
 
 
 class TestTOINFieldLearningIntegration:
@@ -275,8 +277,7 @@ class TestTOINFieldLearningIntegration:
         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. "
-            f"Got: {pattern.field_semantics}"
+            f"field_semantics should be populated after retrieval. Got: {pattern.field_semantics}"
         )
 
         # Check that field hashes match expected fields
@@ -327,7 +328,9 @@ class TestTOINFieldLearningIntegration:
 
         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}"
+        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."""
@@ -513,9 +516,10 @@ class TestEndToEndFieldLearning:
             # 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"
+            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."""
@@ -549,8 +553,7 @@ class TestEndToEndFieldLearning:
         # VERIFY
         assert hint is not None
         assert len(hint.field_semantics) > 0, (
-            f"Recommendation should include field_semantics. "
-            f"Got: {hint.field_semantics}"
+            f"Recommendation should include field_semantics. Got: {hint.field_semantics}"
         )
         assert status_hash in hint.field_semantics