feat(compress): expose frozen_message_count in library-mode compress() (#2178)

## Description

`read_lifecycle.apply()` already supports a frozen message prefix
(`frozen_message_count`) — stale-Read replacements inside the prefix are
skipped so compression never rewrites messages the provider's prompt
cache has anchored. But only the proxy handlers can pass it:
`ContentRouter` reads it from transform kwargs, `CompressConfig` has no
such field, and the public `compress()` never forwards it.

Library-mode callers that manage their own conversation loop (SDK
integrations, offline evaluation, sidecar scoring) therefore can't stop
transforms from rewriting already-sent history. On cached Anthropic
traffic that's expensive: every byte after the first rewritten one stops
billing as a 0.1× cache read and re-bills as a cache write (1.25× at the
5-minute TTL, 2× at the 1-hour TTL) — measured on live coding-agent
traffic, retroactive stale-Read rewrites were the dominant cache-bust
source once tool injection went session-sticky (PR-B7).

Relates to #809 (cache-bust economics discussion); does not close it.

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] 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

- `CompressConfig.frozen_message_count: int = 0` — documented field;
default `0` preserves existing behavior exactly.
- `compress()` forwards it through `pipeline.apply()` to the transforms,
matching what the proxy handlers already do.
- `compress()` docstring: added to the kwargs shorthand list.
- CHANGELOG entry under Unreleased → Features.
- Four tests in `tests/test_compress_api.py` (`TestFrozenMessageCount`).

## Testing

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

### Test Output

```text
$ uv run pytest tests/test_compress_api.py tests/test_transforms/test_read_lifecycle.py \
    tests/test_compression_safety_rails.py tests/test_compress_failure.py -q
59 passed, 1 warning in 3.05s

$ uv run ruff check headroom/compress.py tests/test_compress_api.py
All checks passed!

$ uv run mypy headroom
Success: no issues found in 471 source files
```

## Real Behavior Proof

- Environment: Linux, Python 3.12.3, this branch installed via `uv sync
--extra dev`
- Exact command / steps: build an Anthropic-format conversation with a
stale Read (file read at message 2, edited at message 3), then:

  ```python
  r0 = compress(msgs, model="claude-sonnet-4-5-20250929")
r5 = compress(msgs, model="claude-sonnet-4-5-20250929",
frozen_message_count=5)
  ```

- Observed result: without frozen prefix the stale Read is rewritten;
with frozen_message_count=5 the Read remains byte-identical.

  ```text
  without frozen prefix: stale Read rewritten: True
    transforms: ['read_lifecycle:stale:/app/config.py']
  with frozen_message_count=5: Read byte-identical: True
    transforms: []
  ```

- Not tested: proxy-mode code paths (untouched — they already pass
`frozen_message_count` their own way); Rust crates (untouched).

## 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
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] 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 — library API change, no UI.

## Additional Notes

Default `0` makes this a strict superset of current behavior — no caller
sees any change without opting in. The motivation data comes from a
proxy-side measurement tool that prices compression's cache effects on
live Anthropic agent traffic (per-request cache-adjusted dollars); happy
to share methodology in #809 if useful.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
Co-authored-by: Tejas Chopra <chopratejas@gmail.com>

CHANGELOG.md

@@ -100,6 +100,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Features
 
+* **compress:** expose `frozen_message_count` in library-mode `compress()` via a new `CompressConfig` field (default `0`, unchanged behavior). `read_lifecycle.apply()` already skips stale-Read replacements inside a frozen message prefix, but only the proxy handlers could pass it — `ContentRouter` reads it from transform kwargs and the public API never forwarded it. Library-mode callers that manage their own conversation loop can now stop transforms from rewriting messages already anchored in the provider's prompt cache, which would otherwise convert 0.1x cached prefix reads into full-price cache writes ([#2178](https://github.com/headroomlabs-ai/headroom/pull/2178)).
+
 * **proxy:** report a new-content-relative input savings rate in `/stats`: `tokens.new_input_tokens` (provider-billed non-cache-read input: uncached + cache-write tokens, from response usage) and `tokens.new_input_savings_percent` (savings as a fraction of new input plus the tokens compression removed before they could be billed). The existing whole-request ratios recount the full transcript on every turn, so a 200-turn session counts its history 200x into the denominator and long-running cached sessions (especially 1M-context models, which never compact) dilute toward ~0% regardless of how well compression performs on content newly entering context. Purely additive; existing fields unchanged. Reports 0 when no cache usage data exists (e.g. providers without cache metrics) rather than dividing savings by themselves.
 * **transforms:** first-class C# support in `CodeAwareCompressor` via the tree-sitter `csharp` grammar already shipped in the pinned `tree-sitter-language-pack` — no new dependencies ([#1664](https://github.com/headroomlabs-ai/headroom/issues/1664)). Parity with Java/C++/Rust: signatures preserved verbatim, method/constructor/destructor/operator/local-function bodies compressed; block-scoped and file-scoped namespaces, records, structs, interfaces, and enums handled; C#-distinctive auto-detection. Preprocessor conditionals (`#if`…`#endif`) are preserved verbatim as opaque regions (blocks wrapping only `using` directives stay with the imports), `#region` markers no longer swallow the following line during class-member extraction, and top-of-file license banners / `#region License` headers stay on top instead of being relocated below the code. Real-repo runs: 16.1% tokens saved on Newtonsoft.Json (945 files), 37.8% on Polly (797 files), output syntax-valid for 1742/1742 files.
 * **proxy:** add provider-only HTTP proxy routing via `--http-proxy` and `HEADROOM_HTTP_PROXY`. Upstream LLM provider calls can now use an HTTP proxy without setting process-wide `HTTP_PROXY`/`HTTPS_PROXY` variables that are inherited by tool executions; proxied provider clients use HTTP/1.1 so HTTPS provider APIs can tunnel through CONNECT.

headroom/compress.py

@@ -115,6 +115,15 @@ class CompressConfig:
     protect_analysis_context: bool = True
     """Detect 'analyze'/'review' intent and protect code from compression."""
 
+    frozen_message_count: int = 0
+    """Number of leading messages already anchored in the provider's prompt
+    cache. Transforms will not rewrite messages inside this frozen prefix
+    (read_lifecycle skips stale-Read replacements there), so compression
+    never converts 0.1x cached prefix reads into full-price rewrites.
+    Default 0 = no frozen prefix. The proxy handlers compute and pass this
+    automatically; library-mode callers that manage their own conversation
+    loop should pass the message count of the previous request."""
+
     # How aggressive
     target_ratio: float | None = None
     """Keep ratio for Kompress. None = model decides (~15% kept, aggressive).
@@ -182,7 +191,7 @@ def compress(
         config: Compression options (CompressConfig). Overrides defaults.
         **kwargs: Shorthand for CompressConfig fields. These override config:
             compress_user_messages, target_ratio, protect_recent,
-            protect_analysis_context, kompress_model.
+            protect_analysis_context, kompress_model, frozen_message_count.
 
     Returns:
         CompressResult with compressed messages and metrics.
@@ -256,6 +265,7 @@ def compress(
             protect_analysis_context=cfg.protect_analysis_context,
             min_tokens_to_compress=cfg.min_tokens_to_compress,
             kompress_model=cfg.kompress_model,
+            frozen_message_count=cfg.frozen_message_count,
         )
 
         tokens_before = result.tokens_before

tests/test_compress_api.py

@@ -306,3 +306,111 @@ class TestLiteLLMCallback:
 
         result = asyncio.run(callback.async_pre_call_hook("key", data, "embedding"))
         assert result is data
+
+
+# =============================================================================
+# Tests: frozen_message_count through library-mode compress()
+# =============================================================================
+
+
+class TestFrozenMessageCount:
+    """The frozen prefix must be reachable from library mode.
+
+    Proxy handlers pass frozen_message_count so transforms never rewrite
+    messages already anchored in the provider's prompt cache. Library-mode
+    callers manage their own conversation loop and need the same control —
+    without it, read_lifecycle rewrites sent history and converts cached
+    prefix reads into full-price rewrites.
+    """
+
+    @staticmethod
+    def _stale_read_conversation() -> list[dict]:
+        """Anthropic-format conversation with a stale Read: file read early,
+        edited later. read_lifecycle should classify the Read as STALE."""
+        big_content = "\n".join(f"line {i}: some file content here" for i in range(80))
+        return [
+            {"role": "user", "content": "read then edit the config"},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "t_read",
+                        "name": "Read",
+                        "input": {"file_path": "/app/config.py"},
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "t_read", "content": big_content}
+                ],
+            },
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "t_edit",
+                        "name": "Edit",
+                        "input": {
+                            "file_path": "/app/config.py",
+                            "old_string": "old",
+                            "new_string": "new",
+                        },
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [{"type": "tool_result", "tool_use_id": "t_edit", "content": "ok"}],
+            },
+            {"role": "assistant", "content": "edited."},
+            {"role": "user", "content": "now summarize the change"},
+        ]
+
+    @staticmethod
+    def _read_result_content(messages: list[dict]) -> str:
+        for msg in messages:
+            content = msg.get("content")
+            if not isinstance(content, list):
+                continue
+            for block in content:
+                if isinstance(block, dict) and block.get("tool_use_id") == "t_read":
+                    return str(block.get("content"))
+        raise AssertionError("t_read tool_result not found")
+
+    def test_stale_read_rewritten_without_frozen_prefix(self):
+        """Baseline: with no frozen prefix, the stale Read is rewritten."""
+        messages = self._stale_read_conversation()
+        original = self._read_result_content(messages)
+        result = compress(messages, model="claude-sonnet-4-5-20250929")
+        assert self._read_result_content(result.messages) != original
+
+    def test_frozen_prefix_blocks_stale_read_rewrite(self):
+        """frozen_message_count as kwarg: messages inside the frozen prefix
+        must come back byte-identical, even though the Read is stale."""
+        messages = self._stale_read_conversation()
+        original = self._read_result_content(messages)
+        result = compress(
+            messages,
+            model="claude-sonnet-4-5-20250929",
+            frozen_message_count=5,
+        )
+        assert self._read_result_content(result.messages) == original
+
+    def test_frozen_prefix_via_config_object(self):
+        """frozen_message_count set on CompressConfig behaves identically."""
+        messages = self._stale_read_conversation()
+        original = self._read_result_content(messages)
+        cfg = CompressConfig(frozen_message_count=5)
+        result = compress(messages, model="claude-sonnet-4-5-20250929", config=cfg)
+        assert self._read_result_content(result.messages) == original
+
+    def test_frozen_zero_is_legacy_behavior(self):
+        """Explicit 0 matches the default: stale Read gets rewritten."""
+        messages = self._stale_read_conversation()
+        original = self._read_result_content(messages)
+        result = compress(messages, model="claude-sonnet-4-5-20250929", frozen_message_count=0)
+        assert self._read_result_content(result.messages) != original