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

Fix lint errors in text compression utilities

Browse source
This commit is contained in:
chopratejas 2026-01-12 17:32:51 -08:00
parent eac2890bce
commit 946ba4fab6
6 changed files with 57 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

6
headroom/transforms/__init__.py
View file

@ -3,16 +3,16 @@
from .base import Transform
from .cache_aligner import CacheAligner
from .content_detector import ContentType, DetectionResult, detect_content_type
from .log_compressor import LogCompressor, LogCompressorConfig, LogCompressionResult
from .log_compressor import LogCompressionResult, LogCompressor, LogCompressorConfig
from .pipeline import TransformPipeline
from .rolling_window import RollingWindow
from .search_compressor import (
SearchCompressionResult,
SearchCompressor,
SearchCompressorConfig,
SearchCompressionResult,
)
from .smart_crusher import SmartCrusher, SmartCrusherConfig
from .text_compressor import TextCompressor, TextCompressorConfig, TextCompressionResult
from .text_compressor import TextCompressionResult, TextCompressor, TextCompressorConfig
from .tool_crusher import ToolCrusher
__all__ = [

9
headroom/transforms/content_detector.py
View file

@ -46,9 +46,7 @@ _SEARCH_RESULT_PATTERN = re.compile(
r"^[^\s:]+:\d+:" # file:line: format (grep -n style)
)
_DIFF_HEADER_PATTERN = re.compile(
r"^(diff --git|--- a/|@@\s+-\d+,\d+\s+\+\d+,\d+\s+@@)"
)
_DIFF_HEADER_PATTERN = re.compile(r"^(diff --git|--- a/|@@\s+-\d+,\d+\s+\+\d+,\d+\s+@@)")
_DIFF_CHANGE_PATTERN = re.compile(r"^[+-][^+-]")
@ -332,7 +330,6 @@ def is_json_array_of_dicts(content: str) -> bool:
True if content is a JSON array where all items are dicts.
"""
result = detect_content_type(content)
return (
result.content_type == ContentType.JSON_ARRAY
and result.metadata.get("is_dict_array", False)
return result.content_type == ContentType.JSON_ARRAY and result.metadata.get(
"is_dict_array", False
)

40
headroom/transforms/log_compressor.py
View file

@ -140,9 +140,7 @@ class LogCompressor:
# Level detection patterns
_LEVEL_PATTERNS = {
LogLevel.ERROR: re.compile(
r"\b(ERROR|error|Error|FATAL|fatal|Fatal|CRITICAL|critical)\b"
),
LogLevel.ERROR: re.compile(r"\b(ERROR|error|Error|FATAL|fatal|Fatal|CRITICAL|critical)\b"),
LogLevel.FAIL: re.compile(r"\b(FAIL|FAILED|fail|failed|Fail|Failed)\b"),
LogLevel.WARN: re.compile(r"\b(WARN|WARNING|warn|warning|Warn|Warning)\b"),
LogLevel.INFO: re.compile(r"\b(INFO|info|Info)\b"),
@ -178,7 +176,7 @@ class LogCompressor:
"""
self.config = config or LogCompressorConfig()
def compress(self, content: str, context: str = "") -> "LogCompressionResult":
def compress(self, content: str, context: str = "") -> LogCompressionResult:
"""Compress log output.
Args:
@ -354,9 +352,7 @@ class LogCompressor:
# Select errors (first, last, highest scoring)
if errors:
selected_errors = self._select_with_first_last(
errors, self.config.max_errors
)
selected_errors = self._select_with_first_last(errors, self.config.max_errors)
selected.extend(selected_errors)
# Select fails
@ -371,7 +367,7 @@ class LogCompressor:
selected.extend(warnings[: self.config.max_warnings])
# Select stack traces
for i, stack in enumerate(stack_traces[: self.config.max_stack_traces]):
for stack in stack_traces[: self.config.max_stack_traces]:
selected.extend(stack[: self.config.stack_trace_max_lines])
# Always include summary lines
@ -393,9 +389,7 @@ class LogCompressor:
return selected
def _select_with_first_last(
self, lines: list[LogLine], max_count: int
) -> list[LogLine]:
def _select_with_first_last(self, lines: list[LogLine], max_count: int) -> list[LogLine]:
"""Select lines keeping first and last."""
if len(lines) <= max_count:
return lines
@ -411,7 +405,7 @@ class LogCompressor:
# Fill remaining with highest scoring
remaining = max_count - len(selected)
if remaining > 0:
candidates = [l for l in lines if l not in selected]
candidates = [line for line in lines if line not in selected]
candidates = sorted(candidates, key=lambda x: x.score, reverse=True)
selected.extend(candidates[:remaining])
@ -434,18 +428,14 @@ class LogCompressor:
return deduped
def _add_context(
self, all_lines: list[LogLine], selected: list[LogLine]
) -> list[LogLine]:
def _add_context(self, all_lines: list[LogLine], selected: list[LogLine]) -> list[LogLine]:
"""Add context lines around selected lines."""
selected_indices = {l.line_number for l in selected}
selected_indices = {line.line_number for line in selected}
context_indices: set[int] = set()
for idx in selected_indices:
# Add lines before
for i in range(
max(0, idx - self.config.error_context_lines), idx
):
for i in range(max(0, idx - self.config.error_context_lines), idx):
context_indices.add(i)
# Add lines after
for i in range(
@ -467,10 +457,10 @@ class LogCompressor:
"""Format selected lines with summary stats."""
# Count categories
stats: dict[str, int] = {
"errors": sum(1 for l in all_lines if l.level == LogLevel.ERROR),
"fails": sum(1 for l in all_lines if l.level == LogLevel.FAIL),
"warnings": sum(1 for l in all_lines if l.level == LogLevel.WARN),
"info": sum(1 for l in all_lines if l.level == LogLevel.INFO),
"errors": sum(1 for line in all_lines if line.level == LogLevel.ERROR),
"fails": sum(1 for line in all_lines if line.level == LogLevel.FAIL),
"warnings": sum(1 for line in all_lines if line.level == LogLevel.WARN),
"info": sum(1 for line in all_lines if line.level == LogLevel.INFO),
"total": len(all_lines),
"selected": len(selected),
}
@ -497,9 +487,7 @@ class LogCompressor:
return "\n".join(output_lines), stats
def _store_in_ccr(
self, original: str, compressed: str, original_count: int
) -> str | None:
def _store_in_ccr(self, original: str, compressed: str, original_count: int) -> str | None:
"""Store original in CCR for later retrieval."""
try:
from ..cache.compression_store import get_compression_store

3
headroom/transforms/search_compressor.py
View file

@ -20,7 +20,6 @@ Integrates with CCR for reversible compression.
from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass, field
@ -107,7 +106,7 @@ class SearchCompressor:
self,
content: str,
context: str = "",
) -> "SearchCompressionResult":
) -> SearchCompressionResult:
"""Compress search results.
Args:

24
headroom/transforms/text_compressor.py
View file

@ -64,7 +64,7 @@ class TextCompressor:
"""
self.config = config or TextCompressorConfig()
def compress(self, content: str, context: str = "") -> "TextCompressionResult":
def compress(self, content: str, context: str = "") -> TextCompressionResult:
"""Compress text content.
Args:
@ -98,11 +98,7 @@ class TextCompressor:
# Store in CCR if significant compression
cache_key = None
if (
self.config.enable_ccr
and len(lines) >= self.config.min_lines_for_ccr
and ratio < 0.7
):
if self.config.enable_ccr and len(lines) >= self.config.min_lines_for_ccr and ratio < 0.7:
cache_key = self._store_in_ccr(content, compressed, len(lines))
if cache_key:
compressed += f"\n[{len(lines)} lines compressed. hash={cache_key}]"
@ -116,13 +112,11 @@ class TextCompressor:
cache_key=cache_key,
)
def _score_lines(
self, lines: list[str], context: str
) -> list[tuple[int, str, float]]:
def _score_lines(self, lines: list[str], context: str) -> list[tuple[int, str, float]]:
"""Score lines by importance."""
context_lower = context.lower()
context_words = set(context_lower.split()) if context else set()
anchor_keywords = set(k.lower() for k in self.config.anchor_keywords)
anchor_keywords = {k.lower() for k in self.config.anchor_keywords}
scored: list[tuple[int, str, float]] = []
@ -179,7 +173,7 @@ class TextCompressor:
high_score_lines.sort(key=lambda x: x[2], reverse=True)
remaining_slots = self.config.max_total_lines - len(selected_indices)
for idx, line, score in high_score_lines[:remaining_slots]:
for idx, _line, _score in high_score_lines[:remaining_slots]:
selected_indices.add(idx)
remaining_slots -= 1
if remaining_slots <= 0:
@ -201,9 +195,7 @@ class TextCompressor:
selected = sorted(selected_indices)
return [(i, original_lines[i]) for i in selected]
def _format_output(
self, selected: list[tuple[int, str]], total_lines: int
) -> str:
def _format_output(self, selected: list[tuple[int, str]], total_lines: int) -> str:
"""Format selected lines with ellipsis markers."""
if not selected:
return f"[{total_lines} lines omitted]"
@ -227,9 +219,7 @@ class TextCompressor:
return "\n".join(output_lines)
def _store_in_ccr(
self, original: str, compressed: str, original_count: int
) -> str | None:
def _store_in_ccr(self, original: str, compressed: str, original_count: int) -> str | None:
"""Store original in CCR for later retrieval."""
try:
from ..cache.compression_store import get_compression_store

67
tests/test_text_compressors.py
View file

@ -3,11 +3,8 @@
Tests content detection, search compressor, log compressor, and text compressor.
"""
import pytest
from headroom.transforms import (
ContentType,
DetectionResult,
LogCompressor,
LogCompressorConfig,
SearchCompressor,
@ -31,7 +28,7 @@ class TestContentDetector:
def test_detect_json_array_non_dict(self):
"""JSON arrays of non-dicts are detected."""
content = '[1, 2, 3, 4, 5]'
content = "[1, 2, 3, 4, 5]"
result = detect_content_type(content)
assert result.content_type == ContentType.JSON_ARRAY
assert result.metadata.get("is_dict_array") is False
@ -133,9 +130,7 @@ class TestSearchCompressor:
def test_compress_search_results(self):
"""Search results are compressed."""
content = "\n".join(
[f"src/file{i}.py:{i * 10}:def function_{i}():" for i in range(100)]
)
content = "\n".join([f"src/file{i}.py:{i * 10}:def function_{i}():" for i in range(100)])
compressor = SearchCompressor()
result = compressor.compress(content, context="find function_50")
@ -146,9 +141,7 @@ class TestSearchCompressor:
def test_keeps_first_and_last(self):
"""First and last matches are preserved."""
content = "\n".join(
[f"src/file.py:{i}:line {i}" for i in range(1, 101)]
)
content = "\n".join([f"src/file.py:{i}:line {i}" for i in range(1, 101)])
compressor = SearchCompressor(
config=SearchCompressorConfig(
@ -192,17 +185,19 @@ class TestLogCompressor:
lines = ["=" * 40 + " test session starts " + "=" * 40]
lines.append("collected 100 items")
lines.extend([f"tests/test_{i}.py::test_case_{i} PASSED" for i in range(95)])
lines.extend([
"tests/test_fail.py::test_case_fail FAILED",
"",
"=" * 40 + " FAILURES " + "=" * 40,
"tests/test_fail.py::test_case_fail",
"AssertionError: expected True, got False",
"",
"=" * 40 + " short test summary " + "=" * 40,
"FAILED tests/test_fail.py::test_case_fail",
"1 failed, 95 passed",
])
lines.extend(
[
"tests/test_fail.py::test_case_fail FAILED",
"",
"=" * 40 + " FAILURES " + "=" * 40,
"tests/test_fail.py::test_case_fail",
"AssertionError: expected True, got False",
"",
"=" * 40 + " short test summary " + "=" * 40,
"FAILED tests/test_fail.py::test_case_fail",
"1 failed, 95 passed",
]
)
content = "\n".join(lines)
compressor = LogCompressor()
@ -241,9 +236,7 @@ INFO: Done
"""Small logs pass through unchanged."""
content = "INFO: Starting\nINFO: Done"
compressor = LogCompressor(
config=LogCompressorConfig(min_lines_for_ccr=100)
)
compressor = LogCompressor(config=LogCompressorConfig(min_lines_for_ccr=100))
result = compressor.compress(content)
assert result.compression_ratio == 1.0
@ -319,18 +312,14 @@ class TestSmartCrusherTextIntegration:
from headroom.transforms import SmartCrusher, SmartCrusherConfig
# Create search results content
search_results = "\n".join(
[f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)]
)
search_results = "\n".join([f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)])
messages = [
{"role": "user", "content": "Find all function definitions"},
{"role": "tool", "content": search_results},
]
crusher = SmartCrusher(
config=SmartCrusherConfig(min_tokens_to_crush=10)
)
crusher = SmartCrusher(config=SmartCrusherConfig(min_tokens_to_crush=10))
tokenizer = self._get_tokenizer()
result = crusher.apply(messages, tokenizer)
@ -359,9 +348,7 @@ class TestSmartCrusherTextIntegration:
{"role": "tool", "content": log_content},
]
crusher = SmartCrusher(
config=SmartCrusherConfig(min_tokens_to_crush=10)
)
crusher = SmartCrusher(config=SmartCrusherConfig(min_tokens_to_crush=10))
tokenizer = self._get_tokenizer()
result = crusher.apply(messages, tokenizer)
@ -373,9 +360,7 @@ class TestSmartCrusherTextIntegration:
def test_search_compressor_available_as_standalone(self):
"""SearchCompressor is available for explicit use by applications."""
# Create search results content
search_results = "\n".join(
[f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)]
)
search_results = "\n".join([f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)])
# Application explicitly chooses to compress
compressor = SearchCompressor()
@ -407,13 +392,19 @@ class TestSmartCrusherTextIntegration:
def test_smart_crusher_json_still_works(self):
"""SmartCrusher still handles JSON correctly."""
from headroom.transforms import SmartCrusher, SmartCrusherConfig
import json
import re
from headroom.transforms import SmartCrusher, SmartCrusherConfig
# Create JSON array content with larger items to trigger compression
items = [
{"id": i, "name": f"Item {i}", "value": i * 10, "description": f"This is item number {i}"}
{
"id": i,
"name": f"Item {i}",
"value": i * 10,
"description": f"This is item number {i}",
}
for i in range(500)
]
json_content = json.dumps(items)