fix(proxy): keep OpenAI tool observations mutable in cache mode (#1884)

## Description

Diagnoses and fixes the low-savings OpenAI-compatible cache-mode path
reported in #1696.

OpenAI-compatible tool-calling clients can end a turn with `role:
"tool"` (or legacy `role: "function"`) rather than `role: "user"`. The
OpenAI chat handler's cache-mode freeze boundary treated those tails as
non-mutable, and because `HeadroomProxy` resolves
`_strict_previous_turn_frozen_count` from the Anthropic mixin first, the
OpenAI-specific helper was not used in production. That froze the entire
conversation before `ContentRouter` ran, leaving no live tool
observation to compress and producing near-pass-through savings on long
coding sessions.

This PR keeps final OpenAI tool/function observations mutable in cache
mode, explicitly calls the OpenAI helper to avoid the mixin-name
collision, and clamps negative token-savings artifacts at the
metrics/cost aggregation boundary so stats cannot under-report actual
forwarded savings.

Closes #1696

## 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 final OpenAI `user`, `tool`, and `function` messages as the
mutable cache-mode live zone.
- Route OpenAI cache-boundary calls through
`OpenAIHandlerMixin._strict_previous_turn_frozen_count` explicitly so
the Anthropic mixin method cannot shadow it in `HeadroomProxy`'s MRO.
- Preserve cache-mode live-tail boundaries even when compression-cache
state would otherwise freeze the whole request.
- Clamp negative `tokens_saved` artifacts in `CostTracker.record_tokens`
and `PrometheusMetrics.record_request`.
- Add regression coverage for OpenAI final `tool`/`function` tails,
over-frozen tracker state, and non-negative savings aggregation.

## Testing

- [ ] 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
$ maturin build --profile ci --out dist --interpreter python
Built wheel for abi3 Python >= 3.10 to dist\headroom_ai-0.29.0-cp310-abi3-win_amd64.whl

$ python -m pytest tests\test_proxy_handler_helpers.py tests\test_proxy_openai_cache_stability.py tests\test_observability_metrics.py tests\test_cost_tracker_counterfactual.py
49 passed in 10.27s

$ python -m ruff check .
All checks passed!

$ python -m mypy headroom
Success: no issues found in 407 source files

$ python -m pytest
53 failed, 7703 passed, 488 skipped, 5893 warnings, 131 errors in 595.18s (0:09:55)
```

Full-suite note: the full local `pytest` run was attempted on
Windows/Python 3.13 after building `headroom._core`. It did not complete
green due to broad pre-existing/local-environment failures outside this
change area, dominated by SQLite/memory persistence permission/path
errors plus unrelated adapter/cache/tool tests. The focused regression
suite for this PR passes, and repo-level lint/type gates pass.

## Real Behavior Proof

- Environment: Windows, Python 3.13.13, Rust/Cargo available, local
`headroom._core` wheel built with `maturin build --profile ci`.
- Exact command / steps: ran the OpenAI cache-stability tests with final
`role: "tool"` and `role: "function"` chat tails.
- Observed result:
`test_openai_cache_mode_keeps_final_tool_observation_mutable[tool]` and
`[function]` pass, proving the pipeline receives `frozen_message_count
== 2` for a 3-message request instead of freezing all 3 messages.
- Not tested: live Lemonade/KiloCode upstream session; no local Lemonade
Server was available.

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

## Screenshots (if applicable)

N/A

## Additional Notes

Docs and CHANGELOG are N/A for this narrow proxy bug fix. The broad
local `pytest` checkbox is intentionally left unchecked because the full
suite had unrelated local-environment failures; see the test output
above. Focused regression tests, `ruff check .`, and `mypy headroom` are
green.

headroom/proxy/cost.py

@@ -745,9 +745,7 @@ class CostTracker:
         # clamp it so `total_tokens_removed` reflects actually-forwarded bytes
         # instead of surfacing spurious negatives (verified clean on the wire).
         if tokens_saved < 0:
-            import logging as _lg
-
-            _lg.getLogger(__name__).debug(
+            logger.debug(
                 "record_tokens: clamping negative tokens_saved=%d to 0 for %s (artifact; wire not inflated)",
                 tokens_saved,
                 model,

headroom/proxy/handlers/openai.py

@@ -1070,7 +1070,7 @@ class OpenAIHandlerMixin:
             return base_frozen_count
         final_idx = len(messages) - 1
         if messages[final_idx].get("role") in ("user", "tool", "function"):
-            return max(base_frozen_count, final_idx)
+            return final_idx
         return len(messages)
 
     @staticmethod
@@ -2363,8 +2363,8 @@ class OpenAIHandlerMixin:
 
         openai_frozen_count = openai_prefix_tracker.get_frozen_message_count()
         if is_cache_mode(self.config.mode):
-            openai_frozen_count = self._strict_previous_turn_frozen_count(
-                original_client_messages,
+            openai_frozen_count = OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+                messages,
                 openai_frozen_count,
             )
 
@@ -2399,8 +2399,14 @@ class OpenAIHandlerMixin:
                     # Zone 1: Swap cached compressed versions
                     working_messages = comp_cache.apply_cached(messages)
 
-                    # Re-freeze boundary
-                    openai_frozen_count = comp_cache.compute_frozen_count(messages)
+                    # Re-freeze boundary. Token mode can use the compression
+                    # cache's positional frozen count. Cache mode must keep the
+                    # latest observation mutable even when the compression
+                    # cache has no compressible entry for it yet; otherwise
+                    # OpenAI-compatible tool-call clients freeze the entire
+                    # conversation and report near-zero savings.
+                    if not is_cache_mode(self.config.mode):
+                        openai_frozen_count = comp_cache.compute_frozen_count(messages)
 
                     result = await self._run_compression_in_executor(
                         lambda: self.openai_pipeline.apply(
@@ -2408,7 +2414,14 @@ class OpenAIHandlerMixin:
                             model=model,
                             model_limit=context_limit,
                             context=extract_user_query(working_messages),
-                            frozen_message_count=openai_frozen_count,
+                            frozen_message_count=(
+                                OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+                                    working_messages,
+                                    openai_frozen_count,
+                                )
+                                if is_cache_mode(self.config.mode)
+                                else openai_frozen_count
+                            ),
                             biases=_hook_biases,
                             compression_policy=compression_policy,
                             # Thread the savings-profile knobs (e.g.
@@ -2434,13 +2447,21 @@ class OpenAIHandlerMixin:
                     # so tokens_saved captures both Zone 1 + Zone 2 savings.
                     optimized_tokens = result.tokens_after
                 else:
+                    apply_frozen_count = (
+                        OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+                            messages,
+                            openai_frozen_count,
+                        )
+                        if is_cache_mode(self.config.mode)
+                        else openai_frozen_count
+                    )
                     result = await self._run_compression_in_executor(
                         lambda: self.openai_pipeline.apply(
                             messages=messages,
                             model=model,
                             model_limit=context_limit,
                             context=extract_user_query(messages),
-                            frozen_message_count=openai_frozen_count,
+                            frozen_message_count=apply_frozen_count,
                             biases=_hook_biases,
                             compression_policy=compression_policy,
                             # Same savings-profile threading as the token-mode

headroom/proxy/prometheus_metrics.py

@@ -597,9 +597,7 @@ class PrometheusMetrics:
         # model. Clamp so total_tokens_removed / avg_compression_pct reflect the
         # actually-forwarded bytes instead of surfacing spurious negatives.
         if tokens_saved < 0:
-            import logging as _lg
-
-            _lg.getLogger(__name__).debug(
+            logger.debug(
                 "metrics.record: clamping negative tokens_saved=%d to 0 for %s (artifact; wire not inflated)",
                 tokens_saved,
                 model,

tests/test_cost_tracker_counterfactual.py

@@ -80,6 +80,19 @@ def test_savings_zero_when_no_tokens_saved():
     assert stats["total_tokens_saved"] == 0
 
 
+def test_negative_token_savings_are_clamped_to_zero():
+    """Estimator artifacts must not reduce cumulative savings below reality."""
+    from headroom.proxy.server import CostTracker
+
+    ct = CostTracker()
+
+    ct.record_tokens("openai-compatible", tokens_saved=-500, tokens_sent=5_000)
+    stats = ct.stats()
+
+    assert stats["total_tokens_saved"] == 0
+    assert stats["per_model"]["openai-compatible"]["tokens_saved"] == 0
+
+
 def test_multi_model_savings():
     """Savings across multiple models use each model's own list price."""
     from headroom.proxy.server import CostTracker

tests/test_observability_metrics.py

@@ -198,3 +198,20 @@ async def test_prometheus_metrics_reads_late_configured_otel_metrics() -> None:
         assert spy.rate_limited_calls == [{"provider": "anthropic", "model": "claude-sonnet"}]
     finally:
         reset_otel_metrics()
+
+
+@pytest.mark.asyncio
+async def test_prometheus_metrics_clamps_negative_token_savings() -> None:
+    metrics = PrometheusMetrics()
+
+    await metrics.record_request(
+        provider="openai",
+        model="openai-compatible",
+        input_tokens=100,
+        output_tokens=5,
+        tokens_saved=-25,
+        latency_ms=1.0,
+    )
+
+    assert metrics.tokens_saved_total == 0
+    assert metrics.savings_history[-1][1] == 0

tests/test_proxy_handler_helpers.py

@@ -221,6 +221,27 @@ def test_openai_handler_prefix_helpers_cover_edge_cases() -> None:
         )
         == 1
     )
+    assert (
+        OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+            [{"role": "assistant"}, {"role": "tool", "content": "observation"}],
+            0,
+        )
+        == 1
+    )
+    assert (
+        OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+            [{"role": "user"}, {"role": "assistant"}, {"role": "tool", "content": "obs"}],
+            3,
+        )
+        == 2
+    )
+    assert (
+        OpenAIHandlerMixin._strict_previous_turn_frozen_count(
+            [{"role": "assistant"}, {"role": "function", "content": "legacy observation"}],
+            0,
+        )
+        == 1
+    )
     assert (
         OpenAIHandlerMixin._strict_previous_turn_frozen_count(
             [{"role": "user"}, {"role": "assistant"}],

tests/test_proxy_openai_cache_stability.py

@@ -112,6 +112,85 @@ def test_openai_cache_mode_freezes_previous_turns() -> None:
         assert captured["frozen_message_count"] == 2
 
 
+@pytest.mark.parametrize("tail_role", ["tool", "function"])
+def test_openai_cache_mode_keeps_final_tool_observation_mutable(tail_role: str) -> None:
+    captured = {}
+    with _make_proxy_client() as client:
+        proxy = client.app.state.proxy
+        proxy.config.optimize = True
+        proxy.config.mode = "cache"
+
+        fake_tracker = _FakePrefixTracker(frozen_count=0)
+        proxy.session_tracker_store.compute_session_id = lambda request, model, messages: (
+            "stable-session"
+        )
+        proxy.session_tracker_store.get_or_create = lambda session_id, provider: fake_tracker
+
+        def _fake_apply(**kwargs):
+            captured.setdefault("calls", []).append(
+                {
+                    "frozen_message_count": kwargs.get("frozen_message_count"),
+                    "roles": [msg.get("role") for msg in kwargs["messages"]],
+                    "mode": proxy.config.mode,
+                }
+            )
+            return SimpleNamespace(
+                messages=kwargs["messages"],
+                transforms_applied=["test:compress-tail"],
+                timing={},
+                tokens_before=120,
+                tokens_after=80,
+                waste_signals=None,
+            )
+
+        proxy.openai_pipeline.apply = _fake_apply
+
+        async def _fake_retry(method, url, headers, body, stream=False, **kwargs):  # noqa: ANN001
+            return httpx.Response(
+                200,
+                json={
+                    "id": "chatcmpl_tool_tail",
+                    "choices": [
+                        {
+                            "index": 0,
+                            "message": {"role": "assistant", "content": "ok"},
+                            "finish_reason": "stop",
+                        }
+                    ],
+                    "usage": {"prompt_tokens": 80, "completion_tokens": 3, "total_tokens": 83},
+                },
+            )
+
+        proxy._retry_request = _fake_retry
+
+        tail = {
+            "role": tail_role,
+            "content": "large command observation " * 200,
+        }
+        if tail_role == "tool":
+            tail["tool_call_id"] = "call_1"
+        else:
+            tail["name"] = "bash"
+
+        response = client.post(
+            "/v1/chat/completions",
+            headers={"authorization": "Bearer test-key"},
+            json={
+                "model": "gpt-4o-mini",
+                "messages": [
+                    {"role": "user", "content": "turn1"},
+                    {"role": "assistant", "content": "run command"},
+                    tail,
+                ],
+            },
+        )
+
+        assert response.status_code == 200
+        assert any(call["frozen_message_count"] == 2 for call in captured["calls"]), captured[
+            "calls"
+        ]
+
+
 def test_openai_cache_mode_restores_mutated_frozen_prefix() -> None:
     captured = {}
     with _make_proxy_client() as client: