fix(code): quarantine Perl parser from code-aware compression (#2204)

## Description

`headroom proxy` can wedge when code-aware compression enters the
tree-sitter Perl external scanner and the native scan keeps the GIL
indefinitely. Current main still has two routes into that scanner:
explicit `perl` or `pl` hints flow through `CodeAwareCompressor`, and
`detect_language()` can nominate Perl on non-Perl code because its
prefilter matches generic sigils such as decorators, JSDoc tags, and
shell variables before phase 2 parses every surviving candidate grammar.

This change quarantines Perl at the code-aware compression funnel
without widening scope. Perl remains recognized at the input boundary,
but live proxy compression no longer requests a Perl parser. Non-Perl
code keeps its existing code-aware behavior. Real Perl falls back
through the existing safe Kompress or passthrough contract instead of
entering tree-sitter. The diff stays inside
`headroom/transforms/code_compressor.py` plus focused parser-safety
regressions.

Refs #2185

## 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

- Added a Perl quarantine list in
`headroom/transforms/code_compressor.py` and used it to stop Perl
candidate parsing during language detection.
- Added a hard `_get_parser()` guard so no live code-aware path can
construct a Perl parser.
- Routed resolved explicit Perl hints through the existing safe fallback
or passthrough contract before AST compression.
- Added focused parser-safety regressions for non-Perl candidate bleed,
explicit `perl` and `pl`, inferred real Perl, mixed fenced Perl content,
fallback-disabled passthrough, and non-Perl negative space.

## Testing

- [x] Unit tests pass (`uv run pytest tests/test_perl_scanner_safety.py
-q`)
- [x] Linting passes (`uv run ruff check
headroom/transforms/code_compressor.py
tests/test_perl_scanner_safety.py`)
- [ ] Type checking passes (`uv run mypy headroom`)
- [x] New tests added for new functionality when applicable
- [ ] Manual testing performed

### Test Output

```text
uv run pytest tests/test_perl_scanner_safety.py -q
8 passed in 1.93s
uv run ruff check headroom/transforms/code_compressor.py tests/test_perl_scanner_safety.py
All checks passed!
uv run ruff format headroom/transforms/code_compressor.py tests/test_perl_scanner_safety.py --check
2 files already formatted
```

## Real Behavior Proof

- Environment: Windows, Python from the synced `uv` environment, `dev`
and `code` extras installed, no provider call
- Exact command / steps: Run `uv run pytest
tests/test_perl_scanner_safety.py -q`.
- Observed result: `8 passed in 1.93s`; the suite proves explicit `perl`
and `pl`, inferred real Perl, mixed fenced Perl, and non-Perl
negative-space routes all avoid Perl parser entry.
- Not tested: the reporter's macOS payload and long-running concurrent
workload

## 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
- [ ] 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
- [x] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

- Use `Refs #2185`, not `Closes #2185`. The wedge surface is this PR's
scope, but #2185 also carries a separate orphaned `headroom mcp serve`
report that this slice does not address.
- This is a reachability fix. It does not repair the upstream Perl
scanner and it does not harden other grammars against the same class of
native wedge.
- `CHANGELOG.md` remains unchanged because Headroom generates release
notes from conventional commits.
- Merged PR https://github.com/headroomlabs-ai/headroom/pull/2114
addresses a different cooperative compression stall and stays separate
from this native parser-entry slice.

---------

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
Rod Boev 2026-07-14 23:18:51 -04:00 committed by GitHub
parent 2a954b69b4
commit 8522fcbc40
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 209 additions and 4 deletions

View file

@ -1385,6 +1385,7 @@ class AnthropicHandlerMixin:
else "deferred:dropped"
]
timing = {}
waste_signals = None
result = _DeferredCompressionResult()
else:

View file

@ -102,6 +102,9 @@ def _tree_sitter_importable() -> bool:
return False
_UNSAFE_TREE_SITTER_LANGUAGES: frozenset[str] = frozenset({"perl"})
def _get_parser(language: str) -> Any:
"""Get a tree-sitter parser for the given language.
@ -129,6 +132,8 @@ def _get_parser(language: str) -> Any:
ImportError: If tree-sitter is not installed.
ValueError: If language is not supported.
"""
if language in _UNSAFE_TREE_SITTER_LANGUAGES:
raise ValueError(f"Language '{language}' is quarantined for code-aware compression.")
# NOTE: guard on importability (not _check_tree_sitter_available), because
# _check_tree_sitter_available now performs a real end-to-end parse via
# _get_parser; guarding on it here would recurse.
@ -161,7 +166,7 @@ def _get_parser(language: str) -> Any:
except Exception as e:
raise ValueError(
f"Language '{language}' is not supported by tree-sitter. "
f"Supported: python, javascript, typescript, go, rust, java, c, cpp, csharp, perl. "
f"Supported: python, javascript, typescript, go, rust, java, c, cpp, csharp. "
f"Error: {e}"
) from e
@ -714,6 +719,16 @@ def detect_language(code: str) -> tuple[CodeLanguage, float]:
if candidates[CodeLanguage.CPP] >= 2:
candidates[CodeLanguage.C] = 0
perl_score = candidates.get(CodeLanguage.PERL, 0)
if perl_score > 0:
best_non_perl = max(
(score for lang, score in candidates.items() if lang != CodeLanguage.PERL),
default=0,
)
if perl_score > best_non_perl:
return CodeLanguage.UNKNOWN, 0.0
candidates.pop(CodeLanguage.PERL, None)
# Phase 2: If tree-sitter available, parse with candidates and pick fewest errors
if _check_tree_sitter_available():
best_lang = CodeLanguage.UNKNOWN
@ -1142,6 +1157,19 @@ class CodeAwareCompressor(Transform):
language_confidence=0.0,
syntax_valid=True,
)
if detected_lang == CodeLanguage.PERL:
if self.config.fallback_to_kompress:
return self._fallback_compress(code, original_tokens)
return CodeCompressionResult(
compressed=code,
original=code,
original_tokens=original_tokens,
compressed_tokens=original_tokens,
compression_ratio=1.0,
language=detected_lang,
language_confidence=confidence,
syntax_valid=True,
)
# Check if tree-sitter is available
if not _check_tree_sitter_available():

View file

@ -9,13 +9,13 @@
"source": "github",
"id": "1129940957"
},
"version": "0.27.0",
"version": "0.31.0",
"packages": [
{
"registryType": "pypi",
"registryBaseUrl": "https://pypi.org",
"identifier": "headroom-ai",
"version": "0.27.0",
"version": "0.31.0",
"runtimeHint": "uvx",
"runtimeArguments": [
{

View file

@ -121,7 +121,7 @@ def test_wrap_prints_proxy_urls(
def fake_watcher(**kwargs): # noqa: ANN003
print_fn = kwargs.get("print_setup_lines")
if callable(print_fn):
print_fn()
print_fn(kwargs["port"])
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=fake_rtk):
with patch.object(wrap_mod, "_run_proxy_only_watcher", side_effect=fake_watcher):

View file

@ -38,6 +38,9 @@ class _DummyMetrics:
async def record_failed(self, **kwargs):
return None
def record_compression_failed(self, reason: str) -> None:
return None
async def record_rate_limited(self, **kwargs):
return None

View file

@ -0,0 +1,173 @@
from unittest.mock import patch
import pytest
import headroom.transforms.code_compressor as cc
from headroom.transforms.code_compressor import (
CodeAwareCompressor,
CodeCompressionResult,
CodeCompressorConfig,
CodeLanguage,
unload_tree_sitter,
)
from headroom.transforms.content_router import ContentRouter, ContentRouterConfig
@pytest.fixture(autouse=True)
def _reset_tree_sitter():
unload_tree_sitter()
yield
unload_tree_sitter()
def _compressor(**overrides) -> CodeAwareCompressor:
defaults = {
"min_tokens_for_compression": 1,
"fallback_to_kompress": False,
"enable_ccr": False,
}
defaults.update(overrides)
return CodeAwareCompressor(CodeCompressorConfig(**defaults))
def _record_parser_calls(monkeypatch):
calls: list[str] = []
real_get_parser = cc._get_parser
def spy(language: str):
calls.append(language)
return real_get_parser(language)
monkeypatch.setattr(cc, "_get_parser", spy)
return calls
def test_typescript_with_decorators_stays_typescript_without_perl(monkeypatch):
tree_sitter_installed = cc._tree_sitter_importable()
calls = _record_parser_calls(monkeypatch)
code = """
import { Component, Input } from "@angular/core";
@Component({
selector: "demo-card",
template: "<div>{{ title }}</div>",
})
export class DemoCardComponent {
@Input() title: string = "";
render(items: string[]): string {
return items.map((item) => item.trim()).join(",");
}
}
""".strip()
result = _compressor().compress(code)
assert result.language == CodeLanguage.TYPESCRIPT
assert "perl" not in calls
if tree_sitter_installed:
assert "typescript" in calls
def test_explicit_perl_uses_existing_fallback_without_requesting_parser(monkeypatch):
calls = _record_parser_calls(monkeypatch)
compressor = _compressor(fallback_to_kompress=True)
code = "use strict;\nsub demo {\n my $value = shift;\n return $value;\n}\n"
sentinel = CodeCompressionResult(
compressed="kompress sentinel",
original=code,
original_tokens=10,
compressed_tokens=3,
compression_ratio=0.3,
language=CodeLanguage.UNKNOWN,
language_confidence=0.0,
syntax_valid=False,
)
with patch.object(compressor, "_fallback_compress", return_value=sentinel) as fallback:
result = compressor.compress(code, language="perl")
assert result is sentinel
fallback.assert_called_once()
assert "perl" not in calls
@pytest.mark.parametrize("hint", ["perl", "pl"])
def test_explicit_perl_passthrough_keeps_original_without_parser(monkeypatch, hint):
calls = _record_parser_calls(monkeypatch)
code = "use strict;\nsub demo {\n my $value = shift;\n return $value;\n}\n"
result = _compressor().compress(code, language=hint)
assert result.compressed == code
assert result.language == CodeLanguage.PERL
assert "perl" not in calls
def test_real_perl_detects_unknown_without_requesting_parser(monkeypatch):
calls = _record_parser_calls(monkeypatch)
code = """
package Demo;
use strict;
use warnings;
sub greet {
my ($name) = @_;
return "hello $name";
}
1;
""".strip()
result = _compressor().compress(code)
assert result.language == CodeLanguage.UNKNOWN
assert result.compressed == code
assert "perl" not in calls
def test_fenced_perl_router_path_never_requests_parser(monkeypatch):
calls = _record_parser_calls(monkeypatch)
router = ContentRouter(ContentRouterConfig(enable_code_aware=True, min_section_tokens=1))
content = """
```perl
use strict;
use warnings;
sub greet {
my ($name) = @_;
return "hello $name";
}
```
""".strip()
result = router.compress(content)
assert "```perl" in result.compressed
assert "perl" not in calls
def test_go_still_uses_non_perl_parser(monkeypatch):
tree_sitter_installed = cc._tree_sitter_importable()
calls = _record_parser_calls(monkeypatch)
code = """
package main
import "fmt"
func main() {
fmt.Println("hello")
}
""".strip()
result = _compressor().compress(code)
assert result.language == CodeLanguage.GO
assert "perl" not in calls
if tree_sitter_installed:
assert "go" in calls
def test_get_parser_refuses_quarantined_perl():
with pytest.raises(ValueError, match="quarantined"):
cc._get_parser("perl")