diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py index eab35a381..64b07ef12 100644 --- a/headroom/cli/proxy.py +++ b/headroom/cli/proxy.py @@ -291,21 +291,16 @@ def dashboard(port: int, no_open: bool) -> None: help="Max tokens per minute. Env: HEADROOM_TPM. Default: 100000.", ) @click.option( - "--no-ccr-inject-tool", + "--no-ccr", is_flag=True, - envvar="HEADROOM_NO_CCR_INJECT_TOOL", + envvar="HEADROOM_NO_CCR", help=( - "Don't inject the CCR headroom_retrieve tool. Run compression-only — " - "for streaming / non-MCP clients that can't resolve the retrieve tool " - "and would otherwise error on it. Env: HEADROOM_NO_CCR_INJECT_TOOL." + "Disable CCR entirely: no retrieval markers in compressed content AND no " + "headroom_retrieve tool injected. Lossy compression with no recovery path " + "(maximum savings; also right for streaming / non-MCP clients that can't " + "resolve an injected tool). Env: HEADROOM_NO_CCR." ), ) -@click.option( - "--no-ccr-marker", - is_flag=True, - envvar="HEADROOM_NO_CCR_MARKER", - help=("Don't add CCR retrieval markers to compressed content. Env: HEADROOM_NO_CCR_MARKER."), -) @click.option( "--lossless", is_flag=True, @@ -862,8 +857,7 @@ def proxy( protect_tool_results: str | None, rpm: int | None, tpm: int | None, - no_ccr_inject_tool: bool, - no_ccr_marker: bool, + no_ccr: bool, lossless: bool, no_ccr_proactive_expansion: bool, proxy_extension: tuple[str, ...], @@ -1102,11 +1096,12 @@ def proxy( protect_recent=_get_env_int_optional("HEADROOM_PROTECT_RECENT"), protect_analysis_context=_get_env_bool_optional("HEADROOM_PROTECT_ANALYSIS_CONTEXT"), accuracy_guard=os.environ.get("HEADROOM_ACCURACY_GUARD") or None, - # CCR opt-outs for compression-only deployments (streaming / non-MCP - # clients that can't resolve the injected retrieve tool). Defaults keep - # CCR fully on; each flag flips one dataclass default to False. - ccr_inject_tool=not no_ccr_inject_tool, - ccr_inject_marker=not no_ccr_marker, + # CCR opt-out: --no-ccr disables both halves at once (markers in content + # AND the injected retrieve tool). Markers without a tool — or a tool + # without markers — are useless, so it is a single switch. Default keeps + # CCR fully on. + ccr_inject_tool=not no_ccr, + ccr_inject_marker=not no_ccr, lossless=lossless, ccr_proactive_expansion=not no_ccr_proactive_expansion, # Flatten repeat-flag tuple AND any comma-separated values inside it. diff --git a/headroom/proxy/models.py b/headroom/proxy/models.py index 53088756b..65474890f 100644 --- a/headroom/proxy/models.py +++ b/headroom/proxy/models.py @@ -133,8 +133,8 @@ class ProxyConfig: ccr_inject_tool: bool = True ccr_inject_system_instructions: bool = False # Proxy-level mirror of ContentRouterConfig.ccr_inject_marker, so retrieval - # markers can be toggled from the CLI (--no-ccr-marker). Threaded into the - # router in server.py; default preserves current behavior. + # markers can be toggled from the CLI (--no-ccr, which also drops the retrieve + # tool). Threaded into the router in server.py; default preserves current behavior. ccr_inject_marker: bool = True # CCR Response Handling diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index a6c0ab411..30b299b84 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -873,6 +873,22 @@ class ContentRouterConfig: # emits a `<>` / `Retrieve …` retrieval marker. SmartCrusher is # additionally forced marker-free via smart_crusher_lossless_only. lossless: bool = False + # Cross-turn (whole-conversation) verbatim de-dup. Replaces a contiguous span + # in a later tool output that already appeared verbatim in an earlier tool + # output with an in-context pointer. Prefix-monotonic (cache-safe) and + # information-preserving (the original stays in context). Env: HEADROOM_DEDUPE=1. + # Runs in both modes: lossless references verbatim/folded content; CCR mode + # references the earlier block's kompressed-but-CCR-recoverable form + # (deterministic content-hash → stable → still cache-safe, no added loss). + enable_cross_turn_dedup: bool = False + # Lossless-then-lossy. In lossy mode (not `lossless`), after a byte/data + # lossless fold (search/log/text) run the aggressive lossy compressor + # (Kompress) on the FOLDED remainder and keep it iff it removes a further + # meaningful chunk — recovering the semantic word-drop that plain lossless + # leaves on the table while never doing worse than the fold. DIFF folds are + # never lossy-chained (Kompressing hunks breaks `git apply`). No-op in + # lossless-only mode. Env: HEADROOM_LOSSLESS_THEN_LOSSY=1. + lossless_then_lossy: bool = False min_section_tokens: int = 20 # Min tokens to compress a section # Fallback: Kompress handles unknown/mixed content instead of passing through @@ -1250,6 +1266,16 @@ class ContentRouter(Transform): } ) + # Lossless-then-lossy gate: the lossy pass replaces the byte-exact fold only + # if it saves at least this fraction MORE tokens than the fold already did + # (default 0.05 => Kompress must cut >= 5% beyond the fold). Below that the + # marginal lossy win isn't worth the accuracy cost when a lossless fold is + # already in hand, so the pure fold is kept. Overridable at runtime via env + # HEADROOM_LOSSY_MIN_EXTRA_SAVINGS (read in __init__) so the gate can be tuned + # per deployment without a code edit + overlay rebuild. Higher = stricter + # (fewer lossy chains, safer); 0 = keep the lossy pass on any improvement. + _DEFAULT_LOSSY_MIN_EXTRA_SAVINGS = 0.05 + def __init__( self, config: ContentRouterConfig | None = None, @@ -1317,6 +1343,28 @@ class ContentRouter(Transform): "HEADROOM_TEXT_CRUSHER", "" ).strip().lower() in ("1", "true", "yes", "on") self._text_crusher: Any = None + # Cross-turn dedup: config field OR env HEADROOM_DEDUPE (robust to how the + # config was built). Effective only in lossless mode (guarded in apply()). + self._cross_turn_dedup_enabled: bool = ( + self.config.enable_cross_turn_dedup + or os.environ.get("HEADROOM_DEDUPE", "").strip().lower() in ("1", "true", "yes", "on") + ) + # Lossless-then-lossy. Config field OR env HEADROOM_LOSSLESS_THEN_LOSSY. + # Only takes effect in lossy mode (STAGE 0 guards on `not config.lossless`). + self._lossless_then_lossy: bool = self.config.lossless_then_lossy or os.environ.get( + "HEADROOM_LOSSLESS_THEN_LOSSY", "" + ).strip().lower() in ("1", "true", "yes", "on") + # Lossless-then-lossy gate: keep the lossy chain only if it saves at least + # this fraction MORE than the fold. Env override + # (HEADROOM_LOSSY_MIN_EXTRA_SAVINGS) falls back to the class default; a + # malformed value falls back rather than crashing. + try: + self._lossy_min_extra_savings: float = float( + os.environ.get("HEADROOM_LOSSY_MIN_EXTRA_SAVINGS") + or self._DEFAULT_LOSSY_MIN_EXTRA_SAVINGS + ) + except (TypeError, ValueError): + self._lossy_min_extra_savings = self._DEFAULT_LOSSY_MIN_EXTRA_SAVINGS # TOIN integration for cross-strategy learning self._toin: Any = None @@ -1748,6 +1796,78 @@ class ContentRouter(Transform): ], ) + def _lossless_first( + self, content: str, strategy: CompressionStrategy + ) -> tuple[str, str | None]: + """Byte/data-lossless first pass (intended design: always runs, pre-lossy). + + Maps the (content-detected) strategy to its format-native lossless fold — + SEARCH -> ripgrep --heading form, LOG -> run-collapse + ANSI strip, DIFF + -> drop ``index`` bookkeeping — and gives every other content type a + trivial blank-run collapse. ``compact_lossless`` is self-verifying (exact + inverse or unchanged) and returns the input when it cannot safely shrink, + so this never loses information and is a strict no-op when nothing folds. + + Returns ``(folded, "lossless_")`` when a real byte shrink happened, + else ``(content, None)``. + """ + from headroom.transforms.lossless_compaction import compact_lossless + + # Apply losslessness to the OUTPUT structure, not to the classification: + # try the fold implied by the detected strategy first, then the others. + # Each compact_lossless call is self-verifying (exact inverse or returns + # the input unchanged), so attempting a fold on non-matching content is a + # safe no-op — this recovers folds on content the detector misroutes + # (e.g. `grep -n` of .py files classified as SOURCE_CODE still gets the + # search fold). Keep the single fold that shrinks the most. + primary = { + CompressionStrategy.SEARCH: "search", + CompressionStrategy.LOG: "log", + CompressionStrategy.DIFF: "diff", + }.get(strategy) + order = ([primary] if primary else []) + [ + k for k in ("search", "log", "diff", "text") if k != primary + ] + best, best_label = content, None + for kind in order: + try: + cand = compact_lossless(content, kind) + except Exception: + continue + if len(cand) < len(best): + best, best_label = cand, f"lossless_{kind}" + return best, best_label + + @staticmethod + def _looks_like_diff(content: str) -> bool: + """Cheap structural sniff for unified/git-diff content. + + Used to keep the lossy-after-fold pass (Kompress) OFF diff content — + Kompressing hunks corrupts ``git apply``. This is defense-in-depth beyond the + DIFF-strategy and ``lossless_diff``-label checks: a diff can be folded + best under a non-diff label (e.g. blank-line collapse → ``lossless_text``) + or mis-detected, and must still never reach the lossy stage. + """ + return ( + "diff --git " in content + or "\n@@ " in content + or content.startswith("@@ ") + or content.startswith("--- ") + ) + + def _has_lossless_fold(self, content: str) -> bool: + """True if a byte/data-lossless fold shrinks ``content`` (any format). + + Lets small blocks bypass the lossy ``min_chars`` floor: a lossless fold + is byte-exact and cheap (stdlib regex), so there is no size threshold + below which it should be skipped. The floor exists only to keep the + expensive lossy compressors off marginal blocks — it must not gate the + free, recoverable fold. + """ + if not isinstance(content, str): + return False + return self._lossless_first(content, CompressionStrategy.PASSTHROUGH)[1] is not None + def _apply_strategy_to_content( self, content: str, @@ -1787,14 +1907,42 @@ class ContentRouter(Transform): strategy_chain: list[str] = [strategy.value] error: str | None = None - # Stage B/C: prompt-conditioned relevance split for LOG/SEARCH — keep - # relevant records verbatim, compress the low-value tail. Runs in BOTH - # modes; the tail's marker behavior follows the mode automatically via - # _try_ml_compressor: lossless → marker-free Kompress; CCR → Kompress - # with a retrieval marker so the dropped detail stays retrievable (a - # safety net if the scorer is wrong). DIFF is excluded — Kompressing - # hunks breaks `git apply`. Returns None (falls through to the normal - # path) when disabled, unavailable, or no better than plain compression. + # ── STAGE 0: LOSSLESS-FIRST (unconditional floor) ──────────────────── + # A byte/data-lossless fold has ZERO accuracy cost, so it ALWAYS runs + # first, in every mode — it banks a guaranteed, fully-recoverable win up + # front (search --heading, log run-collapse, diff index-strip; blank-run + # collapse otherwise). Detection is content-based (strategy is assigned by + # content_detector on the OUTPUT), so `cd DIR && rg …`, pipes and unknown + # tools route here by structure, not by command. `_lossless_first` is + # self-verifying (exact inverse or unchanged) → never loses information, + # and is a strict no-op returning (content, None) when nothing folds. + _ll_content, _ll_label = self._lossless_first(content, strategy) + + # ── LOSSLESS-ONLY mode: stop at the byte-exact fold ────────────────── + # HEADROOM_LOSSLESS=1 is an explicit no-unrecoverable-loss contract (the + # constructor forces markers off + SmartCrusher lossless-only). So we + # NEVER layer a lossy drop on top here — the fold IS the answer. When it + # folds, return it; otherwise leave the block verbatim (passthrough), + # never a marker-free lossy drop that could not be recovered. + if self.config.lossless: + if _ll_label is not None: + return _ll_content, len(_ll_content.split()), [_ll_label] + return content, original_tokens, [CompressionStrategy.PASSTHROUGH.value] + + # ── LOSSY / CCR mode: layer relevance-split + lossy ON TOP of the fold ─ + # The operator has opted into lossy compression, so we reclaim more than + # the fold's byte-exact floor. This is independent of the CCR-marker + # sub-setting: markers-on makes any drop recoverable; the no-CCR-lossy + # mode drops it unmarked by design. Either way STAGE 0 already banked the + # lossless win, so nothing below can do worse than the fold. + # + # Stage B/C — prompt-conditioned relevance split for LOG/SEARCH: keep the + # high-relevance records byte-verbatim (lossless-folded) and send only the + # low-value tail to the lossy compressor (Kompress; CCR-marked and thus + # recoverable when markers are on). It self-gates on beating the whole- + # block fold, so when it fires it is strictly smaller than the STAGE 0 + # floor; otherwise it returns None and we keep the fold below. DIFF is + # excluded — Kompressing hunks breaks `git apply`. if self.config.relevance_split and strategy in ( CompressionStrategy.LOG, CompressionStrategy.SEARCH, @@ -1802,33 +1950,45 @@ class ContentRouter(Transform): kind = "log" if strategy is CompressionStrategy.LOG else "search" split = self._relevance_split_compress(content, kind, context) if split is not None: - label = f"lossless_{kind}" if self.config.lossless else kind - return split, len(split.split()), [label, "relevance_split"] + return split, len(split.split()), [kind, "relevance_split"] - # No-CCR lossless mode: LOG/SEARCH/DIFF get format-native lossless - # compaction instead of the lossy Rust drop path, so the output stays - # marker-free (no `<>` / `Retrieve …`) and fully recoverable. - # SMART_CRUSHER relies on smart_crusher_lossless_only (wired elsewhere); - # KOMPRESS/TEXT/CODE_AWARE/PASSTHROUGH pass through unchanged here in - # Stage A. The reversibility + size gate lives in compact_lossless, - # which returns the original when it can't safely shrink it. - if self.config.lossless and strategy in ( - CompressionStrategy.LOG, - CompressionStrategy.SEARCH, - CompressionStrategy.DIFF, - ): - from headroom.transforms.lossless_compaction import compact_lossless + # No relevance split adopted → return the STAGE 0 lossless fold as the + # floor. Lossless-then-lossy: before returning, run the aggressive lossy + # compressor on the byte-folded remainder and keep it IFF it removes a + # further meaningful chunk (Kompress must save >= _lossy_min_extra_savings + # beyond the fold). Keeps the fold AND reclaims the semantic word-drop + # tail, never doing worse than the fold. DIFF folds are returned verbatim + # — Kompressing hunks corrupts `git apply`. + if _ll_label is not None: + _lossy_after_fold = ( + self._lossless_then_lossy + and strategy != CompressionStrategy.DIFF + and _ll_label != "lossless_diff" + and not self._looks_like_diff(content) + ) + if _lossy_after_fold: + _fold_tokens = len(_ll_content.split()) + try: + _komp, _komp_tokens = self._try_ml_compressor(_ll_content, context, question) + except Exception as exc: # noqa: BLE001 + logger.debug("lossy-after-fold failed: %s", exc) + _komp, _komp_tokens = None, None + if ( + _komp is not None + and _komp_tokens is not None + and _komp_tokens <= _fold_tokens * (1 - self._lossy_min_extra_savings) + and len(_komp) < len(_ll_content) + ): + return ( + _komp, + _komp_tokens, + [_ll_label, CompressionStrategy.KOMPRESS.value], + ) + return _ll_content, len(_ll_content.split()), [_ll_label] - kind = { - CompressionStrategy.LOG: "log", - CompressionStrategy.SEARCH: "search", - CompressionStrategy.DIFF: "diff", - }[strategy] - try: - compacted = compact_lossless(content, kind) - except Exception: - compacted = content - return compacted, len(compacted.split()), [f"lossless_{kind}"] + # CCR/lossy mode, nothing foldable (code/json/text/mixed) and no relevance + # split → fall through to the lossy compressors below (kompress / + # smart_crusher / code), which attach CCR retrieval markers when enabled. try: if strategy == CompressionStrategy.CODE_AWARE: @@ -3437,6 +3597,7 @@ class ContentRouter(Transform): # (#1307). Keep the original verbatim instead. if ( enforce_rev + and self.config.ccr_inject_marker and result.strategy_used in self.LOSSY_UNMARKED_STRATEGIES and not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) ): @@ -3486,6 +3647,20 @@ class ContentRouter(Transform): # Build final message list from slots transformed_messages = [m for m in result_slots if m is not None] + # Cross-turn (whole-conversation) verbatim de-dup, over the FINAL block + # forms, so it works in both modes: in lossless mode it references + # verbatim/byte-folded content; in CCR mode it references the earlier + # block's kompressed-but-CCR-recoverable form (deterministic — the CCR + # hash is content-derived — so per-block forms are stable and the rewrite + # stays prefix-monotonic → no prompt-cache bust). It never adds loss: the + # later duplicate would carry the same (recoverable) form anyway; dedup + # just points to the earlier copy instead of repeating it. Frozen + + # cache_control blocks are reference targets only (never rewritten). + if self._cross_turn_dedup_enabled: + transformed_messages = self._cross_turn_dedup_messages( + transformed_messages, frozen_message_count, transforms_applied, route_counts + ) + tokens_after = sum( tokenizer.count_text(str(m.get("content", ""))) for m in transformed_messages ) @@ -3664,6 +3839,84 @@ class ContentRouter(Transform): return 1.0 # Default: moderate + def _cross_turn_dedup_messages( + self, + messages: list[dict[str, Any]], + frozen_message_count: int, + transforms_applied: list[str], + route_counts: dict[str, int] | None, + ) -> list[dict[str, Any]]: + """Whole-conversation verbatim de-dup pass (cache-safe, information-lossless). + + Runs AFTER per-block compression, over the final message forms: a span in + a later tool output that appeared verbatim in an earlier tool output is + replaced by an in-context pointer to the original. Frozen-prefix and + cache_control blocks are reference targets only (never rewritten), so no + cached bytes change. Because per-block compression here is a pure function + of content (excluded_tool_ids is empty for bash agents, so there is no + position-dependent gate), the rewrite is prefix-monotonic → the upstream + prompt-cache prefix stays byte-stable across turns. Never raises. + """ + try: + from headroom.transforms.cross_turn_dedup import DedupBlock, dedup_blocks + + locs: list[tuple[int, int | None]] = [] + dblocks: list[DedupBlock] = [] + for i, msg in enumerate(messages): + content = msg.get("content") + frozen = i < frozen_message_count + if isinstance(content, list): + for bidx, block in enumerate(content): + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + text = block.get("content") + if not isinstance(text, str) or not text: + continue + protected = frozen or ("cache_control" in block) + locs.append((i, bidx)) + dblocks.append(DedupBlock(text=text, turn=i, protected=protected)) + elif isinstance(content, str) and msg.get("role") == "tool": + if not content: + continue + protected = frozen or ("cache_control" in msg) + locs.append((i, None)) + dblocks.append(DedupBlock(text=content, turn=i, protected=protected)) + + if len(dblocks) < 2: + return messages + deduped, stats = dedup_blocks(dblocks) + if not stats.get("spans_folded"): + return messages + + new_messages = list(messages) + touched: dict[int, dict[str, Any]] = {} + for (mi, blk_idx), od, nd in zip(locs, dblocks, deduped): + if od.protected or nd.text == od.text: + continue + if mi not in touched: + src = new_messages[mi] + copy = dict(src) + if isinstance(src.get("content"), list): + copy["content"] = [ + dict(b) if isinstance(b, dict) else b for b in src["content"] + ] + touched[mi] = copy + new_messages[mi] = copy + m = touched[mi] + if blk_idx is None: + m["content"] = nd.text + else: + m["content"][blk_idx]["content"] = nd.text + + if route_counts is not None: + route_counts["cross_turn_dedup"] = ( + route_counts.get("cross_turn_dedup", 0) + stats["spans_folded"] + ) + transforms_applied.append(f"router:cross_turn_dedup:{stats['spans_folded']}") + return new_messages + except Exception: # never break the proxy + return messages + def _process_content_blocks( self, message: dict[str, Any], @@ -3838,8 +4091,12 @@ class ContentRouter(Transform): route_counts["error_protected"] += 1 continue - # Only process string content - if isinstance(tool_content, str) and len(tool_content) > min_chars: + # Only process string content. Blocks below the lossy min_chars + # floor still pass when a byte-lossless fold shrinks them — the + # floor guards the lossy path only; lossless has no size floor. + if isinstance(tool_content, str) and ( + len(tool_content) > min_chars or self._has_lossless_fold(tool_content) + ): # Compression pinning: skip already-compressed content if ( "Retrieve more: hash=" in tool_content @@ -3885,7 +4142,9 @@ class ContentRouter(Transform): # `compress_assistant_text_blocks`). elif block_type == "text" and not protect_text_blocks: text_content = block.get("text", "") - if isinstance(text_content, str) and len(text_content) > min_chars: + if isinstance(text_content, str) and ( + len(text_content) > min_chars or self._has_lossless_fold(text_content) + ): # Pinning: skip already-compressed content if ( "Retrieve more: hash=" in text_content @@ -3972,10 +4231,16 @@ class ContentRouter(Transform): ``True`` the caller should update the block with the returned content and set ``any_compressed``. """ + # In lossless-only mode a "skip" means no byte-lossless fold exists for + # this block (e.g. source code) — it is left verbatim, which is NOT a + # rejected compression. Bucket it honestly so it doesn't masquerade as + # ratio_too_high (which properly means "a lossy attempt didn't shrink + # enough"). In CCR mode the ratio_too_high meaning is unchanged. + _noop_bucket = "lossless_noop" if self.config.lossless else "ratio_too_high" # Tier 1: skip set — instant rejection if self._cache.is_skipped(content_key): if route_counts is not None: - route_counts["ratio_too_high"] = route_counts.get("ratio_too_high", 0) + 1 + route_counts[_noop_bucket] = route_counts.get(_noop_bucket, 0) + 1 route_counts["cache_hit"] = route_counts.get("cache_hit", 0) + 1 return None, False @@ -4007,6 +4272,38 @@ class ContentRouter(Transform): if compressor_timing is not None: key = f"compressor:{result.strategy_used.value}" compressor_timing[key] = compressor_timing.get(key, 0.0) + compress_ms + # Lossless-anchored acceptance (byte-measured): a byte/data-lossless fold + # (search --heading, log run-collapse) has ZERO accuracy cost, so it must + # never be rejected by the WORD-ratio gate below — heading/indent folds + # cut tokens while word count stays flat or even rises. Accept on a real + # BYTE reduction (there is no tokenizer in scope here; byte length is a + # faithful token proxy for these folds) and store a byte-based ratio so + # the Tier-2 result cache reuses it on later turns. + # + # Two shapes take this path: + # • Pure fold (chain == [lossless_*]) — byte-exact, always safe. + # • Fold+lossy (chain == [lossless_*, kompress]) — accepted on bytes + # ONLY in no-CCR mode (config.ccr_inject_marker=False), where unmarked + # lossy is the deliberate output. The byte-exact fold is the floor, so + # the block is guaranteed to shrink and never falls below the fold. + # A pure fold bypasses the lossy-unmarked reversibility guard (it is + # recoverable); the fold+lossy tail case only reaches here when markers are + # off, where that guard is a no-op anyway. + _chain = getattr(result, "strategy_chain", None) or [] + _starts_lossless = bool(_chain) and _chain[0].startswith("lossless_") + _is_pure_lossless = _starts_lossless and all(s.startswith("lossless_") for s in _chain) + _byte_accept = _starts_lossless and (_is_pure_lossless or not self.config.ccr_inject_marker) + if _byte_accept and len(result.compressed) < len(content): + _ll_ratio = len(result.compressed) / max(1, len(content)) + _ll_label = _chain[0] if _is_pure_lossless else "+".join(_chain) + self._cache.put(content_key, result.compressed, _ll_ratio, _ll_label) + transforms_applied.append(f"router:{strategy_label}:{_ll_label}") + if compressed_details is not None: + compressed_details.append(f"{details_prefix}:{_ll_label}:{_ll_ratio:.2f}") + if route_counts is not None: + _bucket = "lossless_accept" if _is_pure_lossless else "lossless_then_lossy_accept" + route_counts[_bucket] = route_counts.get(_bucket, 0) + 1 + return result.compressed, True if result.compression_ratio < min_ratio: # Tool ground truth must stay reversible: a lossy summarizer # (kompress/text/code) that emitted no CCR retrieve marker is @@ -4014,8 +4311,16 @@ class ContentRouter(Transform): # (#1307). The string/`role=="tool"` path guards this; mirror it # here for tool_result blocks (never cached, so the Tier-2 path # above can't serve a poisoned entry). + # + # EXCEPTION: no-CCR mode (config.ccr_inject_marker=False). Here the + # operator has *deliberately* disabled retrieval markers — recovery + # is not expected, so unmarked lossy output is the intended result, + # not a bug to skip. This drops the marker-token overhead AND the + # forgone compressions the guard would otherwise skip. Only applies + # when markers are off; with markers on the guard is unchanged. if ( enforce_reversibility + and self.config.ccr_inject_marker and result.strategy_used in self.LOSSY_UNMARKED_STRATEGIES and not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) ): @@ -4038,10 +4343,12 @@ class ContentRouter(Transform): f"{details_prefix}:{result.strategy_used.value}:{result.compression_ratio:.2f}" ) return result.compressed, True - # Didn't compress enough — add to skip set + # Didn't compress enough — add to skip set. In lossless-only mode this is + # a "no fold available" passthrough (code/text left verbatim), not a + # rejected lossy compression, so bucket it as lossless_noop. self._cache.mark_skip(content_key) if route_counts is not None: - route_counts["ratio_too_high"] = route_counts.get("ratio_too_high", 0) + 1 + route_counts[_noop_bucket] = route_counts.get(_noop_bucket, 0) + 1 return None, False def _detect_analysis_intent(self, messages: list[dict[str, Any]]) -> bool: diff --git a/headroom/transforms/cross_turn_dedup.py b/headroom/transforms/cross_turn_dedup.py new file mode 100644 index 000000000..a8f46b301 --- /dev/null +++ b/headroom/transforms/cross_turn_dedup.py @@ -0,0 +1,231 @@ +"""Cross-turn (whole-conversation) verbatim de-duplication. + +Bash coding agents re-display the same file bytes many times across turns +(``cat foo.py`` -> ``sed -n 75,100p foo.py`` -> ``git diff`` -> ``cat foo.py`` +again). Every per-block compressor is blind to this: the redundancy is *across* +blocks. This transform replaces a contiguous span in a later tool output that +already appeared verbatim in an earlier tool output with a compact in-context +pointer to the original. + +Two hard invariants, both required for production use: + +1. CACHE-SAFETY via *prefix-monotonicity*. Blocks are processed in order and a + block is only ever matched against content from *strictly earlier* blocks. + Therefore the rewritten output of blocks ``0..k`` is byte-identical whether + or not block ``k+1`` exists — appending a turn never mutates an earlier turn, + so the upstream prompt-cache prefix stays byte-stable. References are + ABSOLUTE (an earlier block's ordinal), never relative, so a frozen pointer's + text never changes. :func:`is_prefix_monotonic` asserts this. + +2. ACCURACY via *no information leaves the window*. Only spans that are present + VERBATIM in an earlier block's already-emitted output are back-referenced + (the "verbatim corpus"), and the earliest occurrence is never rewritten + (keep-earliest), so the original the pointer names is always physically in + context. Only large, non-trivial contiguous spans are folded. + +Pure stdlib, deterministic, never raises (returns input unchanged on any error). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +__all__ = ["DedupBlock", "dedup_blocks", "is_prefix_monotonic"] + +# A run must be at least this many lines AND this many chars to be worth a +# pointer. Small dups are left alone (fragmenting context is not worth it) — +# and a larger floor keeps the pointer comfortably shorter than the span it +# replaces, so a fold is always a net byte win. +DEFAULT_MIN_LINES = 7 +DEFAULT_MIN_CHARS = 120 +# Cap anchor candidates examined per line so a hot line (e.g. `` return``) +# can't blow up matching. Deterministic: candidates are kept in first-seen order. +MAX_ANCHOR_CANDIDATES = 16 + + +@dataclass +class DedupBlock: + """One tool-output block. ``turn`` is a STABLE absolute ordinal used in the + pointer text (must not change as the conversation grows). ``protected`` marks + blocks that must not be rewritten (e.g. carry a cache_control breakpoint) — + they are still indexed as reference targets.""" + + text: str + turn: int + protected: bool = False + + +def _is_trivial(line: str) -> bool: + """A line too common/short to safely anchor a match on its own.""" + s = line.strip() + if len(s) < 4: + return True + return s in { + "return", + "pass", + "else:", + "try:", + "except:", + "finally:", + "break", + "continue", + "});", + "})", + "],", + "),", + '"""', + "'''", + "...", + } + + +def _pointer(span: list[str], ref_turn: int, ref_line: int) -> str: + """A one-line, obviously-a-reference marker naming the in-context original. + + Includes a first-line anchor so the model can locate the block it already + saw. Marker-free of any ``hash=`` retrieval token: recovery is in-context + (the original is physically present earlier in the same request).""" + anchor = next((ln.strip() for ln in span if ln.strip()), "") + if len(anchor) > 80: + anchor = anchor[:77] + "..." + end_line = ref_line + len(span) - 1 + return ( + f"[headroom: {len(span)} lines identical to output shown earlier " + f"(turn {ref_turn}, lines {ref_line}-{end_line}) — starts: {anchor!r}]" + ) + + +def _index_lines( + lines: list[str | None], + block_pos: int, + anchor_index: dict[str, list[tuple[int, int]]], +) -> None: + """Record each non-trivial line's (block_pos, line_idx) as a future anchor. + + Keeps first-seen order and caps the candidate list per line. Only VERBATIM + (surviving) lines should be passed here — never the lines of a span that was + replaced by a pointer.""" + for li, ln in enumerate(lines): + if ln is None or _is_trivial(ln): + continue + bucket = anchor_index.setdefault(ln, []) + if len(bucket) < MAX_ANCHOR_CANDIDATES: + bucket.append((block_pos, li)) + + +def _longest_match( + cur: list[str], + start: int, + anchor_index: dict[str, list[tuple[int, int]]], + corpus: list[list[str | None]], +) -> tuple[int, int, int] | None: + """Longest contiguous run in ``cur`` starting at ``start`` that appears + verbatim inside a single earlier block. Returns (length, block_pos, + ref_line_idx) or None. ``corpus[block_pos]`` holds that block's VERBATIM + lines (``None`` where a span was already folded, which breaks contiguity).""" + anchor = cur[start] + candidates = anchor_index.get(anchor) + if not candidates: + return None + best_len = 0 + best_bp = best_li = -1 + for bp, li in candidates: + block_lines = corpus[bp] + k = 0 + while ( + start + k < len(cur) + and li + k < len(block_lines) + and block_lines[li + k] is not None + and cur[start + k] == block_lines[li + k] + ): + k += 1 + # Deterministic tie-break: longer wins; on ties keep the earliest + # (smallest block_pos, then line) already held in best_*. + if k > best_len: + best_len, best_bp, best_li = k, bp, li + if best_len == 0: + return None + return best_len, best_bp, best_li + + +def dedup_blocks( + blocks: list[DedupBlock], + *, + min_lines: int = DEFAULT_MIN_LINES, + min_chars: int = DEFAULT_MIN_CHARS, +) -> tuple[list[DedupBlock], dict]: + """Rewrite later verbatim spans to in-context pointers. Prefix-monotonic + (cache-safe) and information-preserving (accuracy-safe). Returns + (new_blocks, stats). Never raises.""" + stats = {"spans_folded": 0, "lines_removed": 0, "chars_removed": 0, "blocks": len(blocks)} + try: + # corpus[i] = verbatim lines of block i's OUTPUT (None where folded). + corpus: list[list[str | None]] = [] + anchor_index: dict[str, list[tuple[int, int]]] = {} + out_blocks: list[DedupBlock] = [] + + for blk in blocks: + lines = blk.text.split("\n") + + if blk.protected: + # Never rewrite; still a valid verbatim reference target. + verbatim: list[str | None] = list(lines) + _index_lines(verbatim, len(corpus), anchor_index) + corpus.append(verbatim) + out_blocks.append(blk) + continue + + out: list[str] = [] + verbatim = [] + i = 0 + n = len(lines) + while i < n: + m = _longest_match(lines, i, anchor_index, corpus) + if m is not None and m[0] >= min_lines: + span = lines[i : i + m[0]] + span_text = "\n".join(span) + if len(span_text) >= min_chars: + ref_turn = blocks[m[1]].turn + ptr = _pointer(span, ref_turn, m[2]) + out.append(ptr) + # Folded span is NOT verbatim in this block's output: + # mark None so it can't seed a later contiguous match, + # and don't index it (keep-earliest). + verbatim.extend([None] * m[0]) + stats["spans_folded"] += 1 + stats["lines_removed"] += m[0] + stats["chars_removed"] += len(span_text) - len(ptr) + i += m[0] + continue + out.append(lines[i]) + verbatim.append(lines[i]) + i += 1 + + # Index only the surviving verbatim lines of THIS block (first-seen). + # None entries (folded spans) are kept in place so positions stay + # aligned with ``corpus``; _index_lines skips them. + _index_lines(verbatim, len(corpus), anchor_index) + corpus.append(verbatim) + out_blocks.append(DedupBlock(text="\n".join(out), turn=blk.turn, protected=False)) + + return out_blocks, stats + except Exception: # never break the proxy + return blocks, {"spans_folded": 0, "lines_removed": 0, "chars_removed": 0, "error": True} + + +def is_prefix_monotonic( + blocks: list[DedupBlock], + *, + min_lines: int = DEFAULT_MIN_LINES, + min_chars: int = DEFAULT_MIN_CHARS, +) -> bool: + """CACHE-SAFETY invariant: for every k, dedup(blocks[:k]) equals dedup(full) + truncated to its first k blocks. i.e. appending a later turn never changes an + earlier turn's rewritten bytes, so the prompt-cache prefix stays stable.""" + full, _ = dedup_blocks(blocks, min_lines=min_lines, min_chars=min_chars) + full_text = [b.text for b in full] + for k in range(1, len(blocks) + 1): + partial, _ = dedup_blocks(blocks[:k], min_lines=min_lines, min_chars=min_chars) + if [b.text for b in partial] != full_text[:k]: + return False + return True diff --git a/tests/test_cli_proxy_env.py b/tests/test_cli_proxy_env.py index d5443f587..b0525ca51 100644 --- a/tests/test_cli_proxy_env.py +++ b/tests/test_cli_proxy_env.py @@ -843,25 +843,25 @@ class TestCLICompressionOnlyFlags: assert cfg.ccr_inject_marker is True assert cfg.ccr_proactive_expansion is True - def test_no_ccr_inject_tool_flag(self, runner): - """--no-ccr-inject-tool disables retrieve-tool injection only.""" + def test_no_ccr_flag(self, runner): + """--no-ccr disables BOTH the retrieve-tool injection and the markers.""" captured_config = {} def mock_run_server(config, **kwargs): captured_config["config"] = config with patch("headroom.proxy.server.run_server", mock_run_server): - result = runner.invoke(main, ["proxy", "--no-ccr-inject-tool"], catch_exceptions=False) + result = runner.invoke(main, ["proxy", "--no-ccr"], catch_exceptions=False) assert result.exit_code == 0, result.output cfg = captured_config["config"] assert cfg.ccr_inject_tool is False - # Untouched flags remain on. - assert cfg.ccr_inject_marker is True + assert cfg.ccr_inject_marker is False + # Unrelated CCR knob stays on. assert cfg.ccr_proactive_expansion is True def test_compression_only_all_flags(self, runner): - """All three flags together yield a compression-only config.""" + """--no-ccr + --no-ccr-proactive-expansion yields a compression-only config.""" captured_config = {} def mock_run_server(config, **kwargs): @@ -872,8 +872,7 @@ class TestCLICompressionOnlyFlags: main, [ "proxy", - "--no-ccr-inject-tool", - "--no-ccr-marker", + "--no-ccr", "--no-ccr-proactive-expansion", ], catch_exceptions=False, @@ -885,8 +884,8 @@ class TestCLICompressionOnlyFlags: assert cfg.ccr_inject_marker is False assert cfg.ccr_proactive_expansion is False - def test_no_ccr_marker_from_env(self, runner): - """HEADROOM_NO_CCR_MARKER env var disables marker injection.""" + def test_no_ccr_from_env(self, runner): + """HEADROOM_NO_CCR env var disables both markers and tool injection.""" captured_config = {} def mock_run_server(config, **kwargs): @@ -896,16 +895,18 @@ class TestCLICompressionOnlyFlags: result = runner.invoke( main, ["proxy"], - env={"HEADROOM_NO_CCR_MARKER": "1"}, + env={"HEADROOM_NO_CCR": "1"}, catch_exceptions=False, ) assert result.exit_code == 0, result.output - assert captured_config["config"].ccr_inject_marker is False + cfg = captured_config["config"] + assert cfg.ccr_inject_marker is False + assert cfg.ccr_inject_tool is False class TestNoCcrMarkerCompressors: - """Verify --no-ccr-marker actually suppresses <> markers + """Verify --no-ccr actually suppresses <> markers from every compressor, not just SmartCrusher (#1022).""" def test_content_router_propagates_ccr_inject_marker_false_to_compressors(self): diff --git a/tests/test_cross_turn_dedup.py b/tests/test_cross_turn_dedup.py new file mode 100644 index 000000000..d4c35b7a4 --- /dev/null +++ b/tests/test_cross_turn_dedup.py @@ -0,0 +1,217 @@ +"""Cross-turn dedup: cache-safety (prefix-monotonicity) + accuracy (info-preserving).""" + +import re + +from headroom.transforms.cross_turn_dedup import ( + DedupBlock, + dedup_blocks, + is_prefix_monotonic, +) + +_PTR_RE = re.compile(r"identical to output shown earlier \(turn (\d+), lines (\d+)-(\d+)\)") + + +def _blk(text, turn, protected=False): + return DedupBlock(text=text, turn=turn, protected=protected) + + +def _code(prefix, n): + # A realistic, non-trivial multi-line source span. + return "\n".join( + f"{prefix} result_{i} = compute_overdraft(business_id={i}, amount={i * 100})" + for i in range(n) + ) + + +def _reconstruct(orig_blocks, out_blocks): + """Replace each pointer with the referenced turn's original lines and assert + it reproduces the original block — proves references are faithful & in-context.""" + by_turn = {b.turn: b.text.split("\n") for b in orig_blocks} + for orig, out in zip(orig_blocks, out_blocks): + if orig.protected: + assert out.text == orig.text + continue + rebuilt = [] + for line in out.text.split("\n"): + m = _PTR_RE.search(line) + if m and line.startswith("[headroom:"): + t, a, b = int(m.group(1)), int(m.group(2)), int(m.group(3)) + assert t < orig.turn, "reference must point to an EARLIER turn" + rebuilt.extend(by_turn[t][a : b + 1]) + else: + rebuilt.append(line) + assert "\n".join(rebuilt) == orig.text, f"turn {orig.turn} not faithfully reconstructable" + + +def test_verbatim_reread_is_folded_keep_earliest(): + span = _code("", 8) + blocks = [_blk(f"cat merge.py\n{span}\ntail", 1), _blk(f"sed run\n{span}\nmore", 5)] + out, stats = dedup_blocks(blocks) + assert out[0].text == blocks[0].text # earliest untouched + assert "[headroom:" in out[1].text # later occurrence folded + assert stats["spans_folded"] == 1 + _reconstruct(blocks, out) + + +def test_cache_safety_prefix_monotonic(): + span = _code("x", 10) + blocks = [ + _blk("intro line one\nintro line two\n" + span, 1), + _blk("unrelated diff output\n@@ -1 +1 @@\n-a\n+b", 2), + _blk("here again:\n" + span, 3), + _blk("and once more\n" + span + "\ntrailer", 4), + ] + assert is_prefix_monotonic(blocks) is True + + +def test_below_min_lines_not_folded(): + span = _code("", 3) # below min_lines (7) + blocks = [_blk(span, 1), _blk(span, 2)] + out, stats = dedup_blocks(blocks) + assert stats["spans_folded"] == 0 + assert out[1].text == blocks[1].text + + +def test_trivial_repeated_lines_not_folded(): + junk = "\n".join(["}"] * 20) # trivial lines only + blocks = [_blk(junk, 1), _blk(junk, 2)] + out, stats = dedup_blocks(blocks) + assert stats["spans_folded"] == 0 + + +def test_deterministic(): + span = _code("z", 9) + blocks = [_blk(span, 1), _blk("mid\n" + span, 2), _blk(span, 3)] + a, _ = dedup_blocks(blocks) + b, _ = dedup_blocks(blocks) + assert [x.text for x in a] == [x.text for x in b] + + +def test_protected_block_not_rewritten_but_is_reference_target(): + span = _code("", 8) + blocks = [ + _blk(span, 1, protected=True), # cache_control block — never rewritten + _blk("later:\n" + span, 2), # should still fold against the protected one + ] + out, stats = dedup_blocks(blocks) + assert out[0].text == blocks[0].text + assert "[headroom:" in out[1].text + _reconstruct(blocks, out) + + +def test_info_preserving_reconstruction_multiref(): + s1 = _code("a", 7) + s2 = _code("b", 8) + blocks = [ + _blk("h1\n" + s1, 1), + _blk("h2\n" + s2, 2), + _blk("mix\n" + s1 + "\n---\n" + s2, 3), # two folds in one block + ] + out, stats = dedup_blocks(blocks) + assert stats["spans_folded"] == 2 + _reconstruct(blocks, out) + + +# -------------------------------------------------------------------------- +# Integration: full router.apply() path (content-block tool_result format) +# -------------------------------------------------------------------------- +def _mk_tok(): + from headroom.providers import OpenAIProvider + from headroom.tokenizer import Tokenizer + + return Tokenizer(OpenAIProvider().get_token_counter("gpt-4o"), "gpt-4o") + + +def _toolmsg(text, tid): + return { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": tid, "content": text}], + } + + +def _apply_fresh(messages): + # Fresh router per call: tests the pure-function (prefix-monotonic) property, + # not cross-call cache state. + import copy + + from headroom.transforms.content_router import ContentRouter, ContentRouterConfig + + r = ContentRouter(ContentRouterConfig(lossless=True, enable_cross_turn_dedup=True)) + return r.apply(copy.deepcopy(messages), _mk_tok()).messages + + +def test_apply_dedups_reread_and_keeps_prefix_stable(): + span = "\n".join( + f" result_{i} = compute_overdraft(business_id={i}, amount={i * 100})" for i in range(12) + ) + m1 = [ + {"role": "user", "content": "fix the overdraft bug"}, + {"role": "assistant", "content": "cat merge.py"}, + _toolmsg(f"$ cat merge.py\n{span}\n# end", "t1"), + ] + m2 = m1 + [ + {"role": "assistant", "content": "sed -n range"}, + _toolmsg(f"$ sed -n 1,20p merge.py\n{span}\n# more", "t2"), + ] + out1 = _apply_fresh(m1) + out2 = _apply_fresh(m2) + + # Dedup fired on the later re-read (turn t2), earliest (t1) untouched. + later = out2[-1]["content"][0]["content"] + earlier = out2[2]["content"][0]["content"] + assert "[headroom:" in later + assert "[headroom:" not in earlier and span in earlier + + # CACHE-SAFETY at the router level: appending turn t2 did NOT change any + # earlier message's emitted bytes → the prompt-cache prefix is stable. + def _tool_texts(msgs): + return [ + b["content"] + for m in msgs + if isinstance(m.get("content"), list) + for b in m["content"] + if isinstance(b, dict) and b.get("type") == "tool_result" + ] + + assert _tool_texts(out2)[:1] == _tool_texts(out1) # t1 block byte-identical + + +def test_apply_no_dedup_when_flag_off(): + span = "\n".join(f" v_{i} = f({i})" for i in range(12)) + import copy + + from headroom.transforms.content_router import ContentRouter, ContentRouterConfig + + msgs = [ + _toolmsg(f"a\n{span}", "t1"), + _toolmsg(f"b\n{span}", "t2"), + ] + r = ContentRouter(ContentRouterConfig(lossless=True, enable_cross_turn_dedup=False)) + out = r.apply(copy.deepcopy(msgs), _mk_tok()).messages + joined = "".join(b["content"] for m in out for b in m["content"] if isinstance(b, dict)) + assert "[headroom:" not in joined + + +def test_apply_dedup_runs_in_ccr_mode_too(): + # Dedup is no longer gated to lossless mode: with lossless=False (CCR) and + # the flag on, an exact re-read still folds to an in-context pointer. + span = "\n".join(f" total_{i} = reconcile(entry_id={i}, ledger=book_{i})" for i in range(12)) + import copy + + from headroom.transforms.content_router import ContentRouter, ContentRouterConfig + + msgs = [ + _toolmsg(f"$ cat ledger.py\n{span}\n# eof", "t1"), + {"role": "assistant", "content": "re-check"}, + _toolmsg(f"$ cat ledger.py\n{span}\n# eof", "t2"), # exact re-run + ] + r = ContentRouter(ContentRouterConfig(lossless=False, enable_cross_turn_dedup=True)) + out = r.apply(copy.deepcopy(msgs), _mk_tok()).messages + joined = "".join( + b["content"] + for m in out + if isinstance(m.get("content"), list) + for b in m["content"] + if isinstance(b, dict) + ) + assert "[headroom:" in joined # dedup fired despite lossless=False diff --git a/tests/test_lossless_first_dispatch.py b/tests/test_lossless_first_dispatch.py new file mode 100644 index 000000000..04ccd729b --- /dev/null +++ b/tests/test_lossless_first_dispatch.py @@ -0,0 +1,131 @@ +"""Lossless-first dispatch (intended design). + +Lossless folds run FIRST for every tool-output block, regardless of the +``lossless`` flag, and are accepted on a real byte reduction even when the word +count is flat (the case the word-ratio gate used to reject). In lossless-only +mode (flag on, no CCR) foldable content folds and non-foldable content is left +verbatim (no lossy drop). In CCR mode (flag off) a foldable block still keeps +its byte-exact fold — the lossless floor is never discarded by a later lossy +stage. +""" + +from headroom.transforms.content_router import ContentRouter, ContentRouterConfig +from headroom.transforms.lossless_compaction import search_unheading + + +def _grep_block() -> str: + # Long, repeated path prefixes → search_heading collapses to --heading form. + # Word count stays flat/rises while bytes drop a lot (heading adds path words). + paths = [ + "src/services/wallet/overdraft/automated_overdraft_initiation.py", + "src/services/wallet/overdraft/capacity_limits.py", + ] + return ( + "\n".join( + f"{p}:{ln}: result = compute_overdraft_capacity(business_id, amount)" + for p in paths + for ln in range(1, 40) + ) + + "\n" + ) + + +def _code_block() -> str: + return ( + "\n".join( + f" def method_{i}(self, arg_{i}):\n return self.reg[{i}] + arg_{i} * {i}" + for i in range(40) + ) + + "\n" + ) + + +def _compress(content: str, *, lossless: bool): + router = ContentRouter(ContentRouterConfig(lossless=lossless)) + tr: list[str] = [] + out, was = router._compress_block_content( + content, + hash((content, lossless)), + "", + 1.0, + 1.0, + None, + tr, + {}, + [], + "tool_result", + "tool", + True, + ) + return out, was, tr + + +def test_flag_on_search_folds_lossless_byte_exact(): + block = _grep_block() + out, was, tr = _compress(block, lossless=True) + assert was is True + assert tr == ["router:tool_result:lossless_search"] + assert len(out) < len(block) + # word count is flat/higher -> the old word-ratio gate would have rejected it + assert len(out.split()) >= len(block.split()) + # fully recoverable + assert search_unheading(out) == block + + +def test_flag_on_search_fold_is_deterministic(): + block = _grep_block() + out1, _, _ = _compress(block, lossless=True) + out2, _, _ = _compress(block, lossless=True) + assert out1 == out2 # pure function of content -> prefix-cache safe + + +def test_flag_on_leaves_non_foldable_code_verbatim(): + # Lossless-only mode must never emit a lossy / marker-free drop. + out, was, tr = _compress(_code_block(), lossless=True) + assert was is False + assert tr == [] + + +def test_flag_off_still_keeps_lossless_floor_for_foldable(): + block = _grep_block() + out, was, tr = _compress(block, lossless=False) + assert was is True + assert tr == ["router:tool_result:lossless_search"] + assert search_unheading(out) == block + + +def test_has_lossless_fold_admits_small_block_below_size_floor(): + # A <500-char search block must be admitted — lossless has NO size floor + # (the min_chars floor guards the lossy path only). + router = ContentRouter(ContentRouterConfig(lossless=True)) + small = "\n".join(f"pkg/mod/long_filename.py:{n}:value = {n}" for n in range(1, 8)) + "\n" + assert len(small) < 500 + assert router._has_lossless_fold(small) is True + # non-foldable tiny code must NOT be admitted (stays "small") + assert router._has_lossless_fold("def f():\n return 1\n") is False + + +def test_lossless_mode_non_foldable_is_lossless_noop_not_ratio_too_high(): + # In lossless-only mode, code with no byte-lossless fold is left verbatim. + # That is NOT a rejected compression, so it must not be bucketed as + # ratio_too_high (which means "a lossy attempt didn't shrink enough"). + router = ContentRouter(ContentRouterConfig(lossless=True)) + code = "\n".join(f" x{i} = compute_value({i}, offset={i * 3})" for i in range(60)) + "\n" + rc: dict = {} + out, was = router._compress_block_content( + code, + hash(code), + "", + 1.0, + 1.0, + None, + [], + rc, + [], + "tool_result", + "tool", + True, + ) + assert was is False + assert rc.get("lossless_noop", 0) >= 1 + assert rc.get("ratio_too_high", 0) == 0 diff --git a/tests/test_lossless_then_lossy.py b/tests/test_lossless_then_lossy.py new file mode 100644 index 000000000..19c7276ce --- /dev/null +++ b/tests/test_lossless_then_lossy.py @@ -0,0 +1,179 @@ +"""Lossless-then-lossy dispatch. + +In lossy mode (``lossless=False``) with ``lossless_then_lossy`` on, a foldable +block is FIRST byte-folded losslessly and THEN handed to the aggressive lossy +compressor (Kompress) on the folded remainder. The lossy result is kept only +when it saves at least ``lossy_min_extra_savings`` MORE than the fold already +did; otherwise the pure byte-exact fold is kept, so it is never worse than the +plain fold. + +DIFF content is never lossy-chained (Kompressing hunks breaks ``git apply``). +Kompress is mocked so these run without the ModernBERT model. +""" + +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, +) +from headroom.transforms.lossless_compaction import search_unheading + + +def _grep_block() -> str: + # Repeated path prefixes → search_heading folds byte-exact (word count flat). + paths = [ + "src/services/wallet/overdraft/automated_overdraft_initiation.py", + "src/services/wallet/overdraft/capacity_limits.py", + ] + return ( + "\n".join( + f"{p}:{ln}: result = compute_overdraft_capacity(business_id, amount)" + for p in paths + for ln in range(1, 40) + ) + + "\n" + ) + + +def _diff_block() -> str: + files = ["foo", "bar", "baz"] + return ( + "\n".join( + f"diff --git a/{f}.py b/{f}.py\n" + f"index 1111111aaaaaaa..2222222bbbbbbb 100644\n" + f"--- a/{f}.py\n+++ b/{f}.py\n" + f"@@ -1,3 +1,3 @@\n- old_{f} = 1\n+ new_{f} = 2\n unchanged_{f}" + for f in files + ) + + "\n" + ) + + +def _router(*, lossless_then_lossy, lossless=False, ccr=False, kompress=None): + r = ContentRouter( + ContentRouterConfig( + lossless=lossless, + lossless_then_lossy=lossless_then_lossy, + ccr_inject_marker=ccr, + ) + ) + calls: list[str] = [] + + def _fake_kompress(content, context, question=None): + calls.append(content) + out = kompress(content) if kompress else content + return out, len(out.split()) + + r._try_ml_compressor = _fake_kompress # type: ignore[method-assign] + return r, calls + + +def _run(r, content): + tr, rc = [], {} + out, was = r._compress_block_content( + content, + hash(content), + "", + 1.0, + 1.0, + None, + tr, + rc, + [], + "tool_result", + "tool", + True, + ) + return out, was, tr, rc + + +def test_lossy_after_fold_chains_when_it_helps(): + block = _grep_block() + # kompress removes far more than the min-extra-savings floor → lossy kept. + r, calls = _router(lossless_then_lossy=True, kompress=lambda c: "TINY") + out, was, tr, rc = _run(r, block) + assert was is True + assert out == "TINY" + assert len(calls) == 1 # kompress ran on the folded remainder + assert rc.get("lossless_then_lossy_accept") == 1 + assert rc.get("lossless_accept", 0) == 0 + assert tr == ["router:tool_result:lossless_search+kompress"] + + +def test_keeps_pure_fold_when_lossy_marginal(): + block = _grep_block() + # kompress returns the fold unchanged (no gain) → pure byte-exact fold kept. + r, calls = _router(lossless_then_lossy=True, kompress=lambda c: c) + out, was, tr, rc = _run(r, block) + assert was is True + assert len(calls) == 1 # kompress was attempted... + assert rc.get("lossless_accept") == 1 # ...but the pure fold won + assert rc.get("lossless_then_lossy_accept", 0) == 0 + assert search_unheading(out) == block # fully recoverable (byte-exact) + + +def test_never_kompresses_diff(): + block = _diff_block() + # kompress would mangle a diff; the lossy pass must never touch diff content. + r, calls = _router(lossless_then_lossy=True, kompress=lambda c: "MANGLED") + out, was, tr, rc = _run(r, block) + assert was is True + assert calls == [] # lossy stage never touched the diff + assert "MANGLED" not in out + assert out.count("@@ ") == block.count("@@ ") # every hunk header preserved + assert "new_foo = 2" in out # hunk bodies intact → still applies + + +def test_lossy_after_fold_off_is_pure_fold(): + block = _grep_block() + r, calls = _router(lossless_then_lossy=False, kompress=lambda c: "TINY") + out, was, tr, rc = _run(r, block) + assert was is True + assert calls == [] # no lossy pass when lossless-then-lossy is disabled + assert rc.get("lossless_accept") == 1 + assert search_unheading(out) == block + + +def test_lossy_after_fold_never_worse_than_pure_fold(): + block = _grep_block() + r_fold, _ = _router(lossless_then_lossy=False, kompress=lambda c: "TINY") + r_chain, _ = _router(lossless_then_lossy=True, kompress=lambda c: "TINY") + out_fold, _, _, _ = _run(r_fold, block) + out_chain, _, _, _ = _run(r_chain, block) + assert len(out_chain) <= len(out_fold) # chaining never loses to the pure fold + + +def test_lossy_gate_boundary_default(monkeypatch): + monkeypatch.delenv("HEADROOM_LOSSY_MIN_EXTRA_SAVINGS", raising=False) + block = _grep_block() + r, _ = _router(lossless_then_lossy=True) + assert abs(r._lossy_min_extra_savings - 0.05) < 1e-9 # default: require >=5% extra + fold_tok = len(r._lossless_first(block, CompressionStrategy.SEARCH)[0].split()) + # Kompress that keeps 94% of fold tokens -> saves 6% >= the 5% floor -> chained. + keep = int(fold_tok * 0.94) + r._try_ml_compressor = lambda c, ctx, q=None: (" ".join(["w"] * keep), keep) # type: ignore + _, was, tr, rc = _run(r, block) + assert was is True and rc.get("lossless_then_lossy_accept") == 1 + + +def test_lossy_gate_env_override(monkeypatch): + monkeypatch.setenv("HEADROOM_LOSSY_MIN_EXTRA_SAVINGS", "0.20") + r, _ = _router(lossless_then_lossy=True) + assert abs(r._lossy_min_extra_savings - 0.20) < 1e-9 # env override wins + block = _grep_block() + fold_tok = len(r._lossless_first(block, CompressionStrategy.SEARCH)[0].split()) + keep = int(fold_tok * 0.90) # saves 10%: passes the 5% default but FAILS the 20% override + r._try_ml_compressor = lambda c, ctx, q=None: (" ".join(["w"] * keep), keep) # type: ignore + _, was, tr, rc = _run(r, block) + assert rc.get("lossless_accept") == 1 # kept pure fold under the stricter 20% gate + assert rc.get("lossless_then_lossy_accept", 0) == 0 + + +def test_lossy_after_fold_noop_in_lossless_only_mode(): + block = _grep_block() + # lossless-only mode never emits lossy even if lossless-then-lossy is set. + r, calls = _router(lossless_then_lossy=True, lossless=True, kompress=lambda c: "TINY") + out, was, tr, rc = _run(r, block) + assert was is True + assert calls == [] # no lossy in lossless-only mode + assert search_unheading(out) == block diff --git a/tests/test_no_ccr_lossy.py b/tests/test_no_ccr_lossy.py new file mode 100644 index 000000000..1a44699ea --- /dev/null +++ b/tests/test_no_ccr_lossy.py @@ -0,0 +1,56 @@ +"""No-CCR lossy: with markers OFF, unmarked lossy output is accepted (not skipped). + +The reversibility guard skips lossy-unmarked tool output to keep it recoverable — +but only makes sense when retrieval markers are ON. In no-CCR mode +(ccr_inject_marker=False) recovery is deliberately disabled, so the unmarked lossy +result IS the intended output. This tests both directions without needing the +ModernBERT model (self.compress is mocked to a lossy result). +""" + +from types import SimpleNamespace + +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, +) + + +def _run(marker_on): + orig = "some_identifier = compute_value(x)\n" * 60 # long, lossy-eligible + fake = SimpleNamespace( + compressed="<>", + compression_ratio=0.3, # < min_ratio → enters accept branch + strategy_used=CompressionStrategy.KOMPRESS, # lossy + unmarked + strategy_chain=["kompress"], + ) + r = ContentRouter(ContentRouterConfig(ccr_inject_marker=marker_on)) + r.compress = lambda content, context=None, bias=1.0: fake # type: ignore + tr, rc = [], {} + out, was = r._compress_block_content( + orig, + hash((orig, marker_on)), + "", + 1.0, + 1.0, + None, + tr, + rc, + [], + "tool_result", + "tool", + enforce_reversibility=True, + ) + return was, rc + + +def test_markers_on_skips_unmarked_lossy(): + was, rc = _run(marker_on=True) + assert was is False # guard skips: unrecoverable + assert rc.get("lossy_unrecoverable_skipped", 0) == 1 + + +def test_no_ccr_mode_accepts_unmarked_lossy(): + was, rc = _run(marker_on=False) + assert was is True # accepted: no-CCR mode wants unmarked lossy + assert rc.get("lossy_unrecoverable_skipped", 0) == 0 diff --git a/tests/test_relevance_split.py b/tests/test_relevance_split.py index 20782d7cd..6cc560ca9 100644 --- a/tests/test_relevance_split.py +++ b/tests/test_relevance_split.py @@ -150,13 +150,19 @@ def _router(split_on: bool, *, lossless: bool = True) -> ContentRouter: return r -def test_router_relevance_split_fires_for_search(): +def test_router_lossless_mode_folds_only_no_drop(): + # Lossless-only mode NEVER layers a lossy drop on top of the byte-exact fold: + # the fold is the whole answer (marker-free, fully recoverable). The relevance + # split — which lossy-drops the low-value tail — only rides on top in lossy/CCR + # mode (see test_router_relevance_split_fires_in_ccr_mode). So here the + # irrelevant "widget" records must be PRESERVED, not silently dropped, and the + # Kompress tail stub must never run. r = _router(split_on=True) # lossless mode out, _, chain = r._apply_strategy_to_content(_SEARCH, CompressionStrategy.SEARCH, "oauth token") - assert chain == ["lossless_search", "relevance_split"] - assert "oauth token" in out # relevant records kept verbatim - assert "widget" not in out # irrelevant tail replaced by Kompress stub - assert "[TAIL]" in out + assert chain == ["lossless_search"] + assert "oauth token" in out # relevant records kept + assert "widget" in out # irrelevant tail ALSO kept — no silent drop in lossless mode + assert "[TAIL]" not in out # the lossy Kompress stub never fired def test_router_relevance_split_fires_in_ccr_mode(): diff --git a/wiki/cli.md b/wiki/cli.md index c6fcbe201..8bdbdb098 100644 --- a/wiki/cli.md +++ b/wiki/cli.md @@ -255,8 +255,7 @@ headroom proxy --mode cache | `--no-code-aware` | off | Disable AST-aware code compression | | `--code-aware` | off | Enable code-aware compression in the proxy (env: HEADROOM_CODE_AWARE_ENABLED) | | `--no-read-lifecycle` | off | Disable stale/superseded read compression | -| `--no-ccr-inject-tool` | off | Disable injecting the `headroom_retrieve` tool | -| `--no-ccr-marker` | off | Disable adding retrieval markers to compressed output | +| `--no-ccr` | off | Disable CCR entirely — no retrieval markers in content and no injected `headroom_retrieve` tool (lossy, no recovery path) | | `--no-ccr-proactive-expansion` | off | Disable proactive CCR context expansion | | `--memory` | off | Enable persistent user memory | | `--memory-db-path` | `""` | Override memory DB path (help text: `{cwd}/.headroom/memory.db`) | diff --git a/wiki/configuration.md b/wiki/configuration.md index 274d8a15c..48b3c6964 100644 --- a/wiki/configuration.md +++ b/wiki/configuration.md @@ -53,11 +53,8 @@ headroom proxy --no-optimize # Disable semantic caching headroom proxy --no-cache -# Disable CCR tool injection -headroom proxy --no-ccr-inject-tool - -# Disable CCR retrieval markers -headroom proxy --no-ccr-marker +# Disable CCR entirely (no retrieval markers and no injected retrieve tool) +headroom proxy --no-ccr # Disable proactive CCR expansion headroom proxy --no-ccr-proactive-expansion diff --git a/wiki/proxy.md b/wiki/proxy.md index b5bd319f1..9a093720e 100644 --- a/wiki/proxy.md +++ b/wiki/proxy.md @@ -107,8 +107,7 @@ Key CCR-related proxy flags: | Option | Description | |--------|-------------| -| `--no-ccr-inject-tool` | Do not inject the `headroom_retrieve` tool into the LLM's available tools | -| `--no-ccr-marker` | Do not add retrieval markers to compressed output | +| `--no-ccr` | Disable CCR entirely — no retrieval markers in compressed output and no injected `headroom_retrieve` tool (lossy, no recovery path) | | `--no-ccr-proactive-expansion` | Disable proactive context expansion before the LLM asks | ### ML Compression — RETIRED `--llmlingua` flag