From c5493ea93bae798d489a82167c1f7bcff79eaecb Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Fri, 3 Jul 2026 15:04:07 -0700 Subject: [PATCH] fix(content-router): token-measure lossless folds at the acceptance gate (#1772) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Unit-mismatch bug in the compression acceptance gate. `router.apply()` computes `compression_ratio` from `len(text.split())` (word count), but a **lossless** search/log fold (`compact_lossless`) saves **bytes** by collapsing a repeated path prefix into a single heading — word count stays flat or even *rises* (the heading adds a word). So the gate saw `ratio ≥ 1.0` and discarded every free, byte-recoverable win as `ratio_too_high`. (Raising the floor to 1.0 in #1771 did **not** fix this — the word-ratio was already ≥ 1.0.) Measure lossless results (those whose `strategy_chain` carries a `lossless_*` entry) by **byte ratio** at the gate and in the result cache — the real saving. Lossy strategies are unchanged (word count tracks their token savings), and the reversibility gate is untouched (`LOG`/`SEARCH`/`DIFF` aren't in `LOSSY_UNMARKED_STRATEGIES`). The excluded-tool and bash-search paths already bypass this gate via `continue`; this fixes the **main strategy dispatch** (the lossless-mode `LOG`/`SEARCH`/`DIFF` path). Follow-up to #1771. Closes # ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) ## Changes Made - At the `apply()` acceptance gate: compute `accept_ratio` = byte ratio for lossless results (`strategy_chain` has `lossless_*`), else the existing word ratio. Gate + result-cache entry now use `accept_ratio`. - Added an end-to-end regression test that drives the full `router.apply()` path. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text tests/test_lossless_mode.py::test_router_apply_accepts_lossless_search_byte_measured PASSED tests/test_content_router_tool_role_reversibility.py .......... (10 passed) # broader (pre-move) sweep on the same change: tests/test_lossless_mode.py / test_transforms/test_content_router.py / test_lossless_excluded_compaction.py / test_bash_search_lossless_fold.py — 121 passed ruff check headroom/transforms/content_router.py -> All checks passed! mypy headroom/transforms/content_router.py -> Success: no issues found ``` ## Real Behavior Proof - Environment: local worktree, Python 3.12, `PYTHONPATH` pinned to the branch. - Exact command / steps: new regression test constructs a single-file grep result, runs it through `ContentRouter(lossless=True).apply(...)`, and asserts the tool output is byte-smaller and recovers exactly (`search_unheading(out) == original`). - Observed result: before this fix the fold was rejected (`out == original`, counted `ratio_too_high`); after, it's applied (`len(out) < len(original)`, marker-free, byte-exact recovery). The test also asserts the fold's word count is ≥ the original's, so the test is meaningless if "fixed" by word count. - Not tested: no live end-to-end proxy run; validated via the full `apply()` path in unit tests. ## 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 - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable (handled at release time) ## Additional Notes Why prior tests missed it: `compress()` and `_apply_strategy_to_content` return the folded result directly and never touch the `apply()` acceptance gate, so the existing lossless-mode unit tests (which call those) passed while the real proxy path silently discarded the fold. The new test exercises `apply()` end-to-end. --- headroom/transforms/content_router.py | 29 ++++++++++++++++----- tests/test_lossless_mode.py | 36 +++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index df5aabdff..4a4bbc8d0 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -3410,7 +3410,26 @@ class ContentRouter(Transform): compressor_timing.get(strategy_key, 0.0) + compress_ms ) - if result.compression_ratio < min_ratio: + # Lossless folds (search/log/diff via compact_lossless) shrink by + # collapsing repeated path prefixes, but the gate's default ratio + # is word count — which barely moves (a heading line can push it + # >1.0), discarding a free, recoverable win. Measure lossless + # results by REAL TOKEN count (what actually costs money/context), + # not words and not bytes: accept iff tokens genuinely drop. The + # excluded/bash paths already bypass this gate; this fixes the + # main strategy dispatch. + is_lossless = any( + s.startswith("lossless_") + for s in (getattr(result, "strategy_chain", None) or []) + ) + if is_lossless and getattr(result, "original", None): + orig_tok = tokenizer.count_text(result.original) + accept_ratio = ( + tokenizer.count_text(result.compressed) / orig_tok if orig_tok else 1.0 + ) + else: + accept_ratio = result.compression_ratio + if accept_ratio < min_ratio: # tool ground truth must stay reversible — a lossy summarizer # (kompress/text/code) that emitted no CCR retrieve marker is # unrecoverable, so the agent would act on a fabricated summary @@ -3433,7 +3452,7 @@ class ContentRouter(Transform): self._cache.put( content_key, result.compressed, - result.compression_ratio, + accept_ratio, result.strategy_used.value, ) if netcost_enabled and not self._net_cost_allows( @@ -3450,11 +3469,9 @@ class ContentRouter(Transform): continue result_slots[slot_idx] = {**message, "content": result.compressed} transforms_applied.append( - f"router:{result.strategy_used.value}:{result.compression_ratio:.2f}" - ) - compressed_details.append( - f"{result.strategy_used.value}:{result.compression_ratio:.2f}" + f"router:{result.strategy_used.value}:{accept_ratio:.2f}" ) + compressed_details.append(f"{result.strategy_used.value}:{accept_ratio:.2f}") if slot_idx in frozen_unlock_slots: transforms_applied.append("router:netcost_frozen_unlock") route_counts.setdefault("netcost_frozen_unlocked", 0) diff --git a/tests/test_lossless_mode.py b/tests/test_lossless_mode.py index 65bddd325..9e7a71294 100644 --- a/tests/test_lossless_mode.py +++ b/tests/test_lossless_mode.py @@ -279,6 +279,42 @@ def test_router_lossless_never_emits_marker_various_inputs() -> None: _assert_no_marker(router.compress(s, context="").compressed) +def test_router_apply_accepts_lossless_search_token_measured() -> None: + """Regression: the acceptance gate in router.apply() measured WORD count, so + a lossless search fold — which cuts TOKENS by collapsing a repeated path + prefix while word count stays flat or *rises* (the heading adds a word) — was + wrongly discarded as ratio_too_high. The gate now measures lossless results + by real token count, so the free, recoverable win is applied. (compress()/ + _apply_strategy_to_content bypass this gate, which is why unit tests above + never caught it — the bug only appears through the full apply() path.) + """ + from headroom.providers import OpenAIProvider + from headroom.tokenizer import Tokenizer + from headroom.transforms.lossless_compaction import search_heading + + tok = Tokenizer(OpenAIProvider().get_token_counter("gpt-4o"), "gpt-4o") + router = ContentRouter(ContentRouterConfig(lossless=True)) + grep = "".join( + f"headroom/transforms/content_router.py:{i}: identifier_{i} = compute(value)\n" + for i in range(1, 60) + ) + # The fold does NOT reduce word count (the heading even adds one) — this is + # exactly what made the old word-count gate reject it. + assert len(search_heading(grep).split()) >= len(grep.split()) + + messages = [ + { + "role": "assistant", + "tool_calls": [{"id": "c1", "function": {"name": "find_refs", "arguments": "{}"}}], + }, + {"role": "tool", "tool_call_id": "c1", "content": grep}, + ] + out = router.apply(messages, tok).messages[1]["content"] + assert tok.count_text(out) < tok.count_text(grep) # accepted: fewer TOKENS + assert search_unheading(out) == grep # byte-exact recovery + _assert_no_marker(out) + + # -------------------------------------------------------------------------- # Token-delta measurement (informational) # --------------------------------------------------------------------------