fix(code): validate Python compressed syntax (#1302)

## Description

Fix a Python code-compression validity gap from #1233 where tree-sitter
parsing could mark compressed output as syntactically valid even when
Python compile-time syntax rules reject it.

This keeps `from __future__ import ...` statements in the
import-preservation bucket so they stay before executable definitions,
and adds Python `compile(..., "exec")` verification after `ast.parse`.
It also keeps the earlier conservative class-method decorator
indentation hardening from this branch.

Refs #1233.

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

- Treat Python `future_import_statement` nodes as preserved imports.
- Verify Python compressed output with both `ast.parse` and
`compile(..., "exec")`.
- Preserve original source-line indentation for decorators attached to
class methods.
- Add a regression fixture covering `from __future__ import
annotations`, class decorators, property decorators, async methods, and
`match` statements.
- Add a direct regression assertion that future imports stay before
executable definitions.
- Document the user-visible fix in `CHANGELOG.md`.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ /tmp/headroom-issue-1233-venv/bin/python -m pytest tests/test_transforms/test_code_compressor.py::TestTreeSitterIntegration::test_python_future_import_stays_at_module_start -q
1 passed, 1 warning

$ /tmp/headroom-issue-1233-venv/bin/python -m pytest tests/test_transforms/test_code_compressor.py -q
61 passed, 1 warning

$ uv run --extra dev ruff check headroom/transforms/code_compressor.py tests/test_transforms/test_code_compressor.py
All checks passed!

$ uv run --extra dev ruff format --check headroom/transforms/code_compressor.py tests/test_transforms/test_code_compressor.py
2 files already formatted

$ git diff --check
# no output
```

## Real Behavior Proof

- Environment: macOS, Python 3.11.14, local checkout with `[code]`
dependencies installed in `/tmp/headroom-issue-1233-venv`.
- Exact command / steps: added
`test_python_future_import_stays_at_module_start`, ran it before the fix
to confirm the compressed output failure, then reran the focused test
and full `tests/test_transforms/test_code_compressor.py` after the
patch.
- Observed result: before this patch, the regression fixture produced
compressed Python with `from __future__ import annotations` after
class/function definitions. `result.syntax_valid` was `True`, but
`compile(result.compressed, "<test>", "exec")` failed with `SyntaxError:
from __future__ imports must occur at the beginning of the file`. After
this patch, the focused regression and full code-compressor test file
pass locally, and the regression now directly asserts that the future
import appears before executable definitions.
- Not tested: full repository pytest, `mypy headroom`, and a broad
corpus run over third-party source files.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project 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

## Screenshots (if applicable)

N/A

## Additional Notes

This PR is now scoped to the stable compile-time failure path in #1233.
The broader syntax-failure rate from the issue may still need
corpus-level follow-up.

Co-authored-by: JD Davis <mxjerrett@gmail.com>

CHANGELOG.md

@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug Fixes
 
+* **code:** keep Python `from __future__` imports before executable code during AST compression and validate compressed Python with `compile(..., "exec")` so compile-time syntax rules are enforced ([#1233](https://github.com/chopratejas/headroom/issues/1233)).
 * **proxy:** report real input tokens on the streaming `message_start` event for LiteLLM/Bedrock-backed requests. LiteLLM streaming never surfaces prompt tokens mid-stream, so `message_start.usage.input_tokens` was always `0`; Anthropic clients (e.g. Claude Code) read input-token metrics from that event, underreporting token usage by ~99% in OTel/CloudWatch dashboards. The Bedrock streamer now backfills `input_tokens` with the count Headroom actually sent upstream when the backend leaves it unset, preserving any non-zero value the backend genuinely reports ([#1132](https://github.com/chopratejas/headroom/issues/1132)).
 * **proxy:** force Responses API `store=true` when Headroom injects memory tools so `previous_response_id` continuations work after memory tool calls from clients that requested `store=false` ([#1103](https://github.com/chopratejas/headroom/pull/1103)).
 * **proxy:** build SSL contexts for custom CA bundles so enterprise/private PKI roots work with Python/OpenSSL strict verification.

headroom/transforms/code_compressor.py

@@ -261,7 +261,9 @@ class LangConfig:
 
 _LANG_CONFIGS: dict[CodeLanguage, LangConfig] = {
     CodeLanguage.PYTHON: LangConfig(
-        import_nodes=frozenset({"import_statement", "import_from_statement"}),
+        import_nodes=frozenset(
+            {"future_import_statement", "import_statement", "import_from_statement"}
+        ),
         function_nodes=frozenset({"function_definition"}),
         class_nodes=frozenset({"class_definition"}),
         type_nodes=frozenset({"type_alias_statement"}),
@@ -1660,7 +1662,9 @@ class CodeAwareCompressor(Transform):
                 method_compressed = None
                 for deco_child in child.children:
                     if deco_child.type == "decorator":
-                        decorator_lines.append(_get_node_text(deco_child, code))
+                        deco_start = deco_child.start_point[0]
+                        deco_end = deco_child.end_point[0]
+                        decorator_lines.append("\n".join(code_lines[deco_start : deco_end + 1]))
                     elif deco_child.type in lang_config.function_nodes:
                         method_compressed = self._compress_function_ast(
                             deco_child, code, language, lang_config, body_limits, analysis
@@ -1765,6 +1769,12 @@ class CodeAwareCompressor(Transform):
         (tokens the parser expected but didn't find).
         """
         try:
+            if language == CodeLanguage.PYTHON:
+                import ast
+
+                ast.parse(code)
+                compile(code, "<headroom-compressed>", "exec")
+
             parser = _get_parser(language.value)
             tree = parser.parse(bytes(code, "utf-8"))
             return not _has_syntax_issues(tree.root_node)

tests/test_transforms/test_code_compressor.py

@@ -9,6 +9,7 @@ Comprehensive tests covering:
 - Edge cases: Empty content, unavailable dependency, fallbacks
 """
 
+import textwrap
 from unittest.mock import patch
 
 import pytest
@@ -852,6 +853,85 @@ def main():
         except SyntaxError:
             pytest.fail("Compressed output has invalid Python syntax")
 
+    def test_python_future_import_stays_at_module_start(self):
+        """Compressed Python keeps future imports before executable statements."""
+        config = CodeCompressorConfig(
+            min_tokens_for_compression=10,
+            target_compression_rate=0.2,
+            max_body_lines=3,
+            enable_ccr=False,
+        )
+        compressor = CodeAwareCompressor(config)
+        code = textwrap.dedent(
+            """
+            from __future__ import annotations
+
+            from dataclasses import dataclass
+            from typing import Any, Callable, Iterable
+
+
+            def traced(label: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
+                def decorate(fn: Callable[..., Any]) -> Callable[..., Any]:
+                    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+                        return await fn(*args, **kwargs)
+
+                    return wrapper
+
+                return decorate
+
+
+            @dataclass(slots=True)
+            class Event:
+                kind: str
+                payload: dict[str, Any]
+                retries: int = 0
+
+                @property
+                def important(self) -> bool:
+                    return self.kind in {"error", "retry"} or self.retries > 2
+
+
+            class EventRouter:
+                def __init__(self, sinks: dict[str, Callable[[Event], Any]]) -> None:
+                    self.sinks = sinks
+                    self.history: list[tuple[str, bool]] = []
+
+                @traced("route")
+                async def route(self, events: Iterable[Event]) -> list[str]:
+                    accepted: list[str] = []
+                    for event in events:
+                        match event:
+                            case Event(kind="error", payload={"code": code, "message": msg}, retries=r) if r > 1:
+                                destination = "pager"
+                                accepted.append(f"{destination}:{code}:{msg}")
+                            case Event(kind=kind, payload=payload) if (route := payload.get("route")):
+                                destination = str(route)
+                                accepted.append(f"{destination}:{kind}")
+                            case _:
+                                destination = "dead_letter"
+                                accepted.append(destination)
+
+                        self.history.append((destination, event.important))
+
+                    return [item for item in accepted if item]
+            """
+        )
+
+        result = compressor.compress(code, language="python")
+
+        assert result.syntax_valid is True
+        future_import_index = result.compressed.index("from __future__ import annotations")
+        first_executable_index = min(
+            result.compressed.index("@dataclass"),
+            result.compressed.index("def traced"),
+            result.compressed.index("class EventRouter"),
+        )
+        assert future_import_index < first_executable_index
+        try:
+            compile(result.compressed, "<test>", "exec")
+        except SyntaxError as exc:
+            pytest.fail(f"Compressed output has invalid Python syntax: {exc}\n{result.compressed}")
+
     def test_tree_sitter_loaded_after_compression(self):
         """Parser is loaded after compression."""
         config = CodeCompressorConfig(