fix: harden edge cases, expand tool exclusions, and fix mypy errors

- Fix division-by-zero in OpenAI cache, BM25 scorer, BLEU metrics, and smart_crusher
- Add null safety for SQLite memory store JSON fields
- Add thread safety lock for CCR retrieval counter
- Add timeout/error handling for CCR stream collection
- Safer error logging and JSON serialization fallback in proxy server
- Return explicit zero stats on compression failure instead of None defaults
- Guard against missing "messages" key in ASGI and LiteLLM integrations
- Add aclose() for proper httpx client cleanup in ASGI middleware
- Expand DEFAULT_EXCLUDE_TOOLS to include Grep, Write, Edit
- Revert protect_recent_reads_fraction to 0.0 (protect all excluded-tool outputs)
- Fix whitespace waste detection to use tokenizer for both original and normalized
- Fix top-waste-requests sorting by tokens_saved instead of tokens_before
- Fix TOCTOU race in semantic cache _touch() method
- Fix mypy errors: type annotations for ASGI receive/send and proxy error logging

headroom/cache/openai.py

@@ -372,7 +372,7 @@ class OpenAICacheOptimizer(BaseCacheOptimizer):
 
         # Estimate cacheable portion (system + early messages)
         # OpenAI caches the longest matching prefix
-        cacheable_ratio = min(1.0, system_tokens / total_tokens)
+        cacheable_ratio = min(1.0, system_tokens / total_tokens) if total_tokens > 0 else 0.0
 
         # Check if prefix is stable
         current_hash = self._compute_prefix_hash(system_content)

headroom/cache/semantic.py

@@ -285,11 +285,13 @@ class SemanticCache:
 
     def _touch(self, key: str) -> None:
         """Update access time and move to end of LRU."""
-        if key in self._cache:
+        try:
             entry = self._cache.pop(key)
-            entry.last_accessed = time.time()
-            entry.access_count += 1
-            self._cache[key] = entry
+        except KeyError:
+            return
+        entry.last_accessed = time.time()
+        entry.access_count += 1
+        self._cache[key] = entry
 
     def _evict_oldest(self) -> None:
         """Evict the oldest (least recently used) entry."""

headroom/ccr/response_handler.py

@@ -96,6 +96,7 @@ class CCRResponseHandler:
     def __init__(self, config: ResponseHandlerConfig | None = None):
         self.config = config or ResponseHandlerConfig()
         self._retrieval_count = 0
+        self._retrieval_count_lock = __import__("threading").Lock()
 
     def has_ccr_tool_calls(
         self,
@@ -432,7 +433,8 @@ class CCRResponseHandler:
                 break
 
             rounds += 1
-            self._retrieval_count += len(ccr_calls)
+            with self._retrieval_count_lock:
+                self._retrieval_count += len(ccr_calls)
 
             logger.info(f"CCR: Handling {len(ccr_calls)} retrieval(s) in round {rounds}")
 
@@ -625,9 +627,16 @@ class StreamingCCRHandler:
         if self.buffer.detected_ccr:
             logger.info("CCR: Detected tool call in stream, switching to buffered mode")
 
-            # Collect rest of stream
-            async for chunk in stream_iterator:
-                self.buffer.add_chunk(chunk)
+            # Collect rest of stream with timeout to prevent indefinite blocking
+            import asyncio
+
+            try:
+                async for chunk in stream_iterator:
+                    self.buffer.add_chunk(chunk)
+            except asyncio.TimeoutError:
+                logger.warning("CCR: Timed out collecting rest of stream")
+            except Exception as e:
+                logger.error(f"CCR: Error collecting rest of stream: {e}")
 
             # Parse the complete response
             try:

headroom/compress.py

@@ -163,7 +163,13 @@ def compress(
 
     except Exception as e:
         logger.warning("Compression failed, returning original messages: %s", e)
-        return CompressResult(messages=messages)
+        return CompressResult(
+            messages=messages,
+            tokens_before=0,
+            tokens_after=0,
+            tokens_saved=0,
+            compression_ratio=0.0,
+        )
 
 
 def _get_pipeline() -> Any:

headroom/config.py

@@ -352,16 +352,23 @@ class AnchorConfig:
 #   Compressing would break the edit workflow.
 # Glob: Returns compact file path lists used for navigation. Low token count,
 #   not worth compressing.
-# Grep/Bash are NOT excluded - their outputs (search results, build logs,
-# test output) are ideal compression targets for SearchCompressor/LogCompressor,
-# and CCR provides safe retrieval if the LLM needs more detail.
+# Tool outputs that are reference data and must NOT be compressed.
+# Read/Glob/Grep contain exact file contents/search results the agent needs for edits.
+# Write/Edit record what changes were made — compressing them causes duplicate/conflicting edits.
+# Bash is NOT excluded — its outputs (build logs, test output) are ideal compression targets.
 DEFAULT_EXCLUDE_TOOLS: frozenset[str] = frozenset(
     {
         "Read",
         "Glob",
+        "Grep",
+        "Write",
+        "Edit",
         # Lowercase variants for case-insensitive matching
         "read",
         "glob",
+        "grep",
+        "write",
+        "edit",
     }
 )
 

headroom/evals/metrics.py

@@ -103,7 +103,8 @@ def compute_bleu(response_a: str, response_b: str, max_n: int = 4) -> float:
     # Geometric mean of precisions
     import math
 
-    log_sum = sum(math.log(p) for p in precisions if p > 0) / len(precisions)
+    nonzero_precisions = [p for p in precisions if p > 0]
+    log_sum = sum(math.log(p) for p in nonzero_precisions) / len(nonzero_precisions)
     return math.exp(log_sum)
 
 

headroom/integrations/asgi.py

@@ -29,6 +29,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+from collections.abc import MutableMapping
 from typing import Any
 
 from starlette.types import ASGIApp, Receive, Scope, Send
@@ -87,6 +88,12 @@ class CompressionMiddleware:
         """Whether cloud compression is enabled."""
         return self._api_key is not None
 
+    async def aclose(self) -> None:
+        """Close the underlying httpx.AsyncClient, if one was created."""
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] != "http":
             await self.app(scope, receive, send)
@@ -103,8 +110,8 @@ class CompressionMiddleware:
         # Buffer the request body
         body_chunks: list[bytes] = []
 
-        async def buffering_receive() -> dict[str, Any]:
-            message: dict[str, Any] = await receive()
+        async def buffering_receive() -> MutableMapping[str, Any]:
+            message = await receive()
             if message["type"] == "http.request":
                 chunk = message.get("body", b"")
                 if chunk:
@@ -135,7 +142,7 @@ class CompressionMiddleware:
                 else:
                     result = self._local_compress(messages, model)
 
-                if result and result.get("tokens_saved", 0) > 0:
+                if result and result.get("tokens_saved", 0) > 0 and "messages" in result:
                     body_json["messages"] = result["messages"]
                     full_body = json.dumps(body_json).encode("utf-8")
                     tokens_saved = result["tokens_saved"]
@@ -157,16 +164,16 @@ class CompressionMiddleware:
         # Create a new receive that returns the (possibly modified) body
         body_sent = False
 
-        async def modified_receive() -> dict[str, Any]:
+        async def modified_receive() -> MutableMapping[str, Any]:
             nonlocal body_sent
             if not body_sent:
                 body_sent = True
                 return {"type": "http.request", "body": full_body, "more_body": False}
-            result: dict[str, Any] = await receive()
+            result = await receive()
             return result
 
         # Wrap send to inject compression headers
-        async def metrics_send(message: dict[str, Any]) -> None:
+        async def metrics_send(message: MutableMapping[str, Any]) -> None:
             if message["type"] == "http.response.start" and tokens_saved > 0:
                 headers = list(message.get("headers", []))
                 headers.append((b"x-headroom-compressed", b"true"))

headroom/integrations/litellm_callback.py

@@ -103,7 +103,7 @@ class HeadroomCallback:
             else:
                 result = self._local_compress(messages, model)
 
-            if result and result.get("tokens_saved", 0) > 0:
+            if result and result.get("tokens_saved", 0) > 0 and "messages" in result:
                 data["messages"] = result["messages"]
                 self._total_saved += result["tokens_saved"]
                 logger.info(

headroom/memory/adapters/sqlite.py

@@ -217,14 +217,14 @@ class SQLiteMemoryStore:
             supersedes=row["supersedes"],
             superseded_by=row["superseded_by"],
             promoted_from=row["promoted_from"],
-            promotion_chain=json.loads(row["promotion_chain"]),
+            promotion_chain=json.loads(row["promotion_chain"]) if row["promotion_chain"] else [],
             access_count=row["access_count"],
             last_accessed=datetime.fromisoformat(row["last_accessed"])
             if row["last_accessed"]
             else None,
-            entity_refs=json.loads(row["entity_refs"]),
+            entity_refs=json.loads(row["entity_refs"]) if row["entity_refs"] else [],
             embedding=self._deserialize_embedding(row["embedding"]),
-            metadata=json.loads(row["metadata"]),
+            metadata=json.loads(row["metadata"]) if row["metadata"] else {},
         )
 
     async def save(self, memory: Memory) -> None:

headroom/parser.py

@@ -67,9 +67,10 @@ def detect_waste_signals(text: str, tokenizer: Tokenizer) -> WasteSignals:
     # Excessive whitespace
     ws_matches = WHITESPACE_PATTERN.findall(text)
     if ws_matches:
-        # Count tokens that could be saved by normalizing
+        # Count tokens that could be saved by normalizing whitespace to single spaces
         ws_text = "".join(ws_matches)
-        signals.whitespace_tokens = max(0, tokenizer.count_text(ws_text) - len(ws_matches))
+        normalized_text = " ".join(ws_matches)
+        signals.whitespace_tokens = max(0, tokenizer.count_text(ws_text) - tokenizer.count_text(normalized_text))
 
     # Large JSON blocks
     json_matches = JSON_BLOCK_PATTERN.findall(text)

headroom/proxy/server.py

@@ -2049,9 +2049,14 @@ class HeadroomProxy:
                             logger.info("CCR: Parsed JSON successfully")
                             return result
                         except Exception as e:
+                            resp_headers: str | dict[str, str] = "N/A"
+                            try:
+                                resp_headers = dict(cont_response.headers)
+                            except Exception:
+                                pass
                             logger.error(
                                 f"CCR: API call failed: {e}, "
-                                f"response headers: {dict(cont_response.headers) if 'cont_response' in dir() else 'N/A'}"
+                                f"response headers: {resp_headers}"
                             )
                             raise
 
@@ -2072,9 +2077,14 @@ class HeadroomProxy:
                             for k, v in response.headers.items()
                             if k.lower() not in ("content-encoding", "content-length")
                         }
+                        try:
+                            ccr_content = json.dumps(final_resp_json).encode()
+                        except (TypeError, ValueError) as json_err:
+                            logger.warning(f"[{request_id}] CCR: JSON serialization failed: {json_err}")
+                            ccr_content = json.dumps(resp_json).encode()
                         response = httpx.Response(
                             status_code=200,
-                            content=json.dumps(final_resp_json).encode(),
+                            content=ccr_content,
                             headers=ccr_response_headers,
                         )
                         logger.info(f"[{request_id}] CCR: Retrieval handled successfully")

headroom/relevance/bm25.py

@@ -134,7 +134,7 @@ class BM25Scorer(RelevanceScorer):
             return 0.0, []
 
         doc_len = len(doc_tokens)
-        avgdl = avg_doc_len or doc_len
+        avgdl = avg_doc_len or doc_len or 1
 
         doc_freq = Counter(doc_tokens)
         query_freq = Counter(query_tokens)

headroom/reporting/generator.py

@@ -471,7 +471,7 @@ def _get_top_waste_requests(
         )
 
     # Sort by tokens saved (waste potential)
-    requests.sort(key=lambda x: x["tokens_before"], reverse=True)
+    requests.sort(key=lambda x: x["tokens_saved"], reverse=True)
 
     return requests[:limit]
 

headroom/transforms/content_router.py

@@ -288,7 +288,7 @@ class ContentRouterConfig:
     # compression.  At 10 msgs, protects ~5 Reads.  At 100 msgs, protects ~10.
     # Old Reads beyond this window become compressible even though they are
     # in DEFAULT_EXCLUDE_TOOLS.  0.0 = always exclude all (old behavior).
-    protect_recent_reads_fraction: float = 0.5  # protect the most-recent 50% of messages
+    protect_recent_reads_fraction: float = 0.0  # 0.0 = protect ALL excluded-tool outputs (safest for coding agents)
 
     # Adaptive compression ratio: scales with context pressure.
     # At low pressure (<30% full), use the relaxed threshold (reject marginal).

headroom/transforms/smart_crusher.py

@@ -368,6 +368,10 @@ def _detect_sequential_pattern(values: list[Any], check_order: bool = True) -> b
     if len(nums) < 5:
         return False
 
+    # Need at least 2 elements for pairwise comparison
+    if len(nums) < 2:
+        return False
+
     # Check if sorted values form a near-sequence
     sorted_nums = sorted(nums)
     diffs = [sorted_nums[i + 1] - sorted_nums[i] for i in range(len(sorted_nums) - 1)]
@@ -500,12 +504,13 @@ def _detect_score_field_statistically(stats: FieldStats, items: list[dict]) -> t
                 values_in_order.append(float(val))
     if len(values_in_order) >= 5:
         # Check for descending sort
+        num_pairs = len(values_in_order) - 1
         descending_count = sum(
             1
-            for i in range(len(values_in_order) - 1)
+            for i in range(num_pairs)
             if values_in_order[i] >= values_in_order[i + 1]
         )
-        if descending_count / (len(values_in_order) - 1) > 0.7:
+        if num_pairs > 0 and descending_count / num_pairs > 0.7:
             confidence += 0.3
 
     # Score fields often have floating point values