feat(content-router): lossless-first dispatch, cross-turn dedup, and A7 lossy-after-fold (#1818)

## Description

<!-- Briefly explain the change and why it is needed. -->

Closes #

## Type of Change

- [ ] 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

- 

## Testing

<!-- Check what you actually ran, then paste the real command output
below. -->

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

### Test Output

```text
# Paste relevant command output or artifact links here
```

## Real Behavior Proof

- Environment:
- Exact command / steps:
- Observed result:
- Not tested:

## Review Readiness

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

## Checklist

- [ ] My code follows the project's style guidelines
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] 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)

Add screenshots to help explain your changes.

## Additional Notes

<!-- Mention any N/A checklist items, tradeoffs, follow-ups, or
maintainer context. -->

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

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.

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

headroom/transforms/content_router.py

@@ -873,6 +873,22 @@ class ContentRouterConfig:
     # emits a `<<ccr:…>>` / `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_<kind>")`` 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 `<<ccr:…>>` / `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:

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

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 <<ccr:...>> markers
+    """Verify --no-ccr actually suppresses <<ccr:...>> markers
     from every compressor, not just SmartCrusher (#1022)."""
 
     def test_content_router_propagates_ccr_inject_marker_false_to_compressors(self):

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

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

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

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="<<lossy summary, marker-free>>",
+        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

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():

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`) |

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

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