diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py index 1d479d106..6377407bb 100644 --- a/headroom/cli/proxy.py +++ b/headroom/cli/proxy.py @@ -297,6 +297,16 @@ def dashboard(port: int, no_open: bool) -> None: 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, + envvar="HEADROOM_LOSSLESS", + help=( + "No-CCR lossless mode: compress tool outputs with format-native lossless " + "compaction (and marker-free SmartCrusher) without emitting any CCR " + "retrieval marker, so no MCP retrieve tool is needed. Env: HEADROOM_LOSSLESS=1." + ), +) @click.option( "--no-ccr-proactive-expansion", is_flag=True, @@ -844,6 +854,7 @@ def proxy( tpm: int | None, no_ccr_inject_tool: bool, no_ccr_marker: bool, + lossless: bool, no_ccr_proactive_expansion: bool, proxy_extension: tuple[str, ...], no_subscription_tracking: bool, @@ -1086,6 +1097,7 @@ def proxy( # 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, + lossless=lossless, ccr_proactive_expansion=not no_ccr_proactive_expansion, # Flatten repeat-flag tuple AND any comma-separated values inside it. # `--proxy-extension a,b --proxy-extension c` and `HEADROOM_PROXY_EXTENSIONS=a,b,c` diff --git a/headroom/proxy/models.py b/headroom/proxy/models.py index 34b6e8e25..b1b9fd0a8 100644 --- a/headroom/proxy/models.py +++ b/headroom/proxy/models.py @@ -180,6 +180,8 @@ class ProxyConfig: # CLI: --force-kompress-all; env: HEADROOM_FORCE_KOMPRESS_ALL=1. force_kompress_all: bool = False + lossless: bool = False # CLI: --lossless; env: HEADROOM_LOSSLESS=1. No-CCR mode: compress without any retrieval marker. + # Code graph live watcher (triggers incremental reindex on file changes) code_graph_watcher: bool = False diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index 5d861e6d3..bab13668d 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -674,7 +674,18 @@ class HeadroomProxy( ), ccr_inject_marker=config.ccr_inject_marker, force_kompress_all=config.force_kompress_all, + lossless=config.lossless, ) + # No-CCR lossless mode: compress tool outputs with format-native + # lossless compaction and marker-free SmartCrusher, and suppress every + # retrieval marker + the retrieve-tool injection so no MCP round-trip is + # needed. Mirrors the force_kompress_all wiring precedent. + if config.lossless: + router_config.lossless = True + router_config.smart_crusher_lossless_only = True + router_config.ccr_inject_marker = False + if hasattr(config, "ccr_inject_tool"): + config.ccr_inject_tool = False if config.disable_kompress: router_config.enable_kompress = False # Opt-in restore of the legacy behaviour: send fall-through content @@ -4080,6 +4091,7 @@ def _proxy_config_from_env() -> ProxyConfig: disable_kompress_anthropic=_get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_ANTHROPIC"), disable_kompress_openai=_get_env_optional_bool("HEADROOM_DISABLE_KOMPRESS_OPENAI"), force_kompress_all=_get_env_bool("HEADROOM_FORCE_KOMPRESS_ALL", False), + lossless=_get_env_bool("HEADROOM_LOSSLESS", False), max_connections=_get_env_int("HEADROOM_MAX_CONNECTIONS", 500), max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", 100), keepalive_expiry=_get_env_float("HEADROOM_KEEPALIVE_EXPIRY", 90.0), @@ -4574,6 +4586,16 @@ if __name__ == "__main__": "Also settable via HEADROOM_FORCE_KOMPRESS_ALL=1." ), ) + parser.add_argument( + "--lossless", + action="store_true", + help=( + "No-CCR lossless mode: compress LOG/SEARCH/DIFF tool outputs with " + "format-native lossless compaction (and marker-free SmartCrusher) " + "without emitting any CCR retrieval marker, so no MCP retrieve tool " + "is needed. Also settable via HEADROOM_LOSSLESS=1." + ), + ) parser.add_argument( "--exclude-tools", default=None, @@ -4655,6 +4677,7 @@ if __name__ == "__main__": force_kompress_all = args.force_kompress_all or _get_env_bool( "HEADROOM_FORCE_KOMPRESS_ALL", False ) + lossless = getattr(args, "lossless", False) or _get_env_bool("HEADROOM_LOSSLESS", False) # Set OpenRouter API key from CLI if provided if hasattr(args, "openrouter_api_key") and args.openrouter_api_key: @@ -4710,6 +4733,7 @@ if __name__ == "__main__": disable_kompress_anthropic=disable_kompress_anthropic, disable_kompress_openai=disable_kompress_openai, force_kompress_all=force_kompress_all, + lossless=lossless, # Connection pool settings max_connections=_get_env_int("HEADROOM_MAX_CONNECTIONS", args.max_connections), max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", args.max_keepalive), diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 191b1ec1b..fa13c044d 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -762,6 +762,13 @@ class ContentRouterConfig: # Route ALL compressible content to Kompress, skipping per-type selection. # Tool exclusion (Read/Glob/...) and reversibility gates still apply. force_kompress_all: bool = False + + # No-CCR lossless mode. When True the router compresses LOG/SEARCH/DIFF + # content with format-native lossless compaction (headroom.transforms. + # lossless_compaction) instead of the lossy Rust drop path, and never + # emits a `<>` / `Retrieve …` retrieval marker. SmartCrusher is + # additionally forced marker-free via smart_crusher_lossless_only. + lossless: bool = False mixed_content_threshold: int = 2 # Min types to consider mixed min_section_tokens: int = 20 # Min tokens to compress a section @@ -1115,6 +1122,13 @@ class ContentRouter(Transform): rule in the audit doc. """ self.config = config or ContentRouterConfig() + # No-CCR lossless mode is self-consistent regardless of how the config + # was built: force marker-free output and marker-free SmartCrusher so + # the invariant (no `<>` / `Retrieve …`) holds even when a caller + # constructs ContentRouterConfig(lossless=True) directly. + if self.config.lossless: + self.config.ccr_inject_marker = False + self.config.smart_crusher_lossless_only = True self._observer = observer # Lazy-loaded compressors @@ -1616,6 +1630,31 @@ class ContentRouter(Transform): strategy_chain: list[str] = [strategy.value] error: str | None = None + # 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 + + 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}"] + try: if strategy == CompressionStrategy.CODE_AWARE: if self.config.enable_code_aware: @@ -2276,7 +2315,11 @@ class ContentRouter(Transform): ) if is_kompress_available(): - return KompressCompressor(config=KompressConfig(model_id=model_id)) + return KompressCompressor( + config=KompressConfig( + model_id=model_id, enable_ccr=self.config.ccr_inject_marker + ) + ) except ImportError: pass return None @@ -2284,10 +2327,21 @@ class ContentRouter(Transform): # Default path — exactly as before, cached on self if self._kompress is None: try: - from .kompress_compressor import KompressCompressor, is_kompress_available + from .kompress_compressor import ( + KompressCompressor, + KompressConfig, + is_kompress_available, + ) if is_kompress_available(): - self._kompress = KompressCompressor() + # Honor the router's marker policy. In no-CCR / lossless mode + # (ccr_inject_marker=False) Kompress still compresses (lossy), + # but must NOT append a `Retrieve more: hash=` marker or write + # to the CCR store — otherwise the no-MCP guarantee breaks. + # Matches how search/log/diff/code receive enable_ccr. + self._kompress = KompressCompressor( + config=KompressConfig(enable_ccr=self.config.ccr_inject_marker) + ) except ImportError: logger.debug("Kompress dependencies not available") return self._kompress diff --git a/headroom/transforms/lossless_compaction.py b/headroom/transforms/lossless_compaction.py new file mode 100644 index 000000000..4cdcfac98 --- /dev/null +++ b/headroom/transforms/lossless_compaction.py @@ -0,0 +1,251 @@ +"""Format-native, reversible lossless compaction for no-CCR proxy mode. + +Every helper here is pure stdlib and keeps its output *looking like its own +type* — grep stays grep, logs stay logs, diffs stay diffs. No retrieval +marker (``<>`` / ``Retrieve …``) is ever emitted, so the proxy needs +no MCP retrieve round-trip to stay recoverable. + +The reversible transforms ship with exact inverses and are self-checked at +runtime by :func:`compact_lossless`: if a round-trip does not reproduce the +original (modulo intentionally-dropped non-semantic bits such as ANSI color) +or the result is not actually smaller, the original content is returned +unchanged. Nothing here raises. +""" + +from __future__ import annotations + +import re + +__all__ = [ + "strip_ansi", + "collapse_runs", + "expand_runs", + "is_run_collapsed", + "search_heading", + "search_unheading", + "diff_strip_index", + "compact_lossless", +] + +# ANSI CSI SGR (color/style) escape sequences: ESC [ ... m. Color is +# non-semantic, so stripping it is a safe (one-way) lossless-of-meaning op. +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + +# syslog-style run-collapse marker. The count is captured for exact inversion. +_RUN_MARKER_RE = re.compile(r"^\.\.\. \(repeated (\d+) times\)$") + +# grep/ripgrep default row shape: ``path:line:content``. ``line`` is digits; +# ``path`` must not itself look like ``line:content`` (i.e. not start with a +# bare number) so we don't mis-split a heading-form ``line:content`` row. +_GREP_ROW_RE = re.compile(r"^(?P[^\n:]+):(?P\d+):(?P.*)$") +# heading-form data row (``line:content``) produced by search_heading. +_HEADING_ROW_RE = re.compile(r"^(?P\d+):(?P.*)$") + +# unified-diff ``index .. `` line. The diff still applies +# without it (git only uses it for rename/blob bookkeeping). +_DIFF_INDEX_RE = re.compile(r"^index [0-9a-fA-F]+\.\.[0-9a-fA-F]+( [0-7]+)?$") + + +def strip_ansi(text: str) -> str: + """Remove ANSI CSI/SGR (color) escape sequences. Color is non-semantic.""" + return _ANSI_RE.sub("", text) + + +def _split_keep_trailing(text: str) -> tuple[list[str], bool]: + """Split into lines, remembering whether a trailing newline was present. + + Returns (lines, had_trailing_newline). This lets the run helpers rejoin + byte-exactly instead of always appending or always dropping a newline. + """ + if text == "": + return [], False + had_trailing = text.endswith("\n") + body = text[:-1] if had_trailing else text + return body.split("\n"), had_trailing + + +def _join(lines: list[str], had_trailing: bool) -> str: + out = "\n".join(lines) + if had_trailing: + out += "\n" + return out + + +def collapse_runs(text: str) -> str: + """Collapse runs of >=2 identical consecutive lines (syslog convention). + + A run of N (N>=2) identical lines becomes the line once followed by + ``... (repeated N times)``. Exact inverse: :func:`expand_runs`. + """ + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out: list[str] = [] + i = 0 + n = len(lines) + while i < n: + j = i + while j + 1 < n and lines[j + 1] == lines[i]: + j += 1 + run_len = j - i + 1 + if run_len >= 2: + out.append(lines[i]) + out.append(f"... (repeated {run_len} times)") + else: + out.append(lines[i]) + i = j + 1 + return _join(out, had_trailing) + + +def expand_runs(text: str) -> str: + """Exact inverse of :func:`collapse_runs`.""" + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out: list[str] = [] + i = 0 + n = len(lines) + while i < n: + line = lines[i] + if i + 1 < n: + m = _RUN_MARKER_RE.match(lines[i + 1]) + if m: + count = int(m.group(1)) + out.extend([line] * count) + i += 2 + continue + out.append(line) + i += 1 + return _join(out, had_trailing) + + +def is_run_collapsed(text: str) -> bool: + """True if any run-collapse marker line is present.""" + for line in text.split("\n"): + if _RUN_MARKER_RE.match(line): + return True + return False + + +def search_heading(text: str) -> str: + """Convert grep ``path:line:content`` rows into ripgrep --heading form. + + Consecutive rows sharing a path collapse to the path once on its own line + (a *header* line), then ``line:content`` rows beneath it. Lines that don't + match the ``path:line:content`` shape are passed through untouched. No + blank separators are inserted (they would be ambiguous with passthrough + content), keeping the transform exactly reversible via + :func:`search_unheading`. + """ + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out: list[str] = [] + current_path: str | None = None + for line in lines: + m = _GREP_ROW_RE.match(line) + if m: + path = m.group("path") + if path != current_path: + out.append(path) + current_path = path + out.append(f"{m.group('line')}:{m.group('content')}") + else: + # Any non-grep-row line ends the current file grouping. + out.append(line) + current_path = None + return _join(out, had_trailing) + + +def search_unheading(text: str) -> str: + """Exact inverse of :func:`search_heading`. + + A *header* line is any line that is not itself a ``line:content`` data row + and is immediately followed by at least one ``line:content`` data row; it + is consumed (not re-emitted) and its text becomes the ``path`` prefix for + the data rows that follow, until a non-data line appears. + """ + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out: list[str] = [] + current_path: str | None = None + n = len(lines) + i = 0 + while i < n: + line = lines[i] + data = _HEADING_ROW_RE.match(line) + if current_path is not None and data: + out.append(f"{current_path}:{data.group('line')}:{data.group('content')}") + i += 1 + continue + # Not a data row under an active header. Decide if THIS line is a new + # header: it must not be a data row itself and must be followed by a + # data row. If so, consume it as the path prefix (do not emit). + if not data and i + 1 < n and _HEADING_ROW_RE.match(lines[i + 1]): + current_path = line + i += 1 + continue + # Plain passthrough line (or a stray data row with no header): emit it + # verbatim and clear any active grouping. + current_path = None + out.append(line) + i += 1 + return _join(out, had_trailing) + + +def diff_strip_index(text: str) -> str: + """Drop ``index ..`` lines from a unified diff (still applies).""" + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out = [line for line in lines if not _DIFF_INDEX_RE.match(line)] + return _join(out, had_trailing) + + +def _smaller(candidate: str, original: str) -> bool: + return len(candidate) < len(original) + + +def compact_lossless(content: str, kind: str) -> str: + """Dispatch format-native lossless compaction by ``kind``. + + ``kind`` in {'log', 'search', 'diff', 'text'}. For reversible kinds the + round-trip is verified internally (modulo the intentionally-dropped + non-semantic bits, e.g. ANSI color for logs); if verification fails or the + result is not smaller, the original content is returned unchanged. Never + raises; unknown kinds pass through. + """ + if not content: + return content + try: + if kind == "log": + # ANSI is non-semantic and dropped one-way; run-collapse must be + # exactly reversible against the de-ANSI'd baseline. + baseline = strip_ansi(content) + candidate = collapse_runs(baseline) + if expand_runs(candidate) != baseline: + return content + return candidate if _smaller(candidate, content) else content + + if kind == "search": + candidate = search_heading(content) + if search_unheading(candidate) != content: + return content + return candidate if _smaller(candidate, content) else content + + if kind == "diff": + # Purely subtractive of non-semantic bookkeeping lines; the + # remaining hunks still apply. No exact inverse needed. + candidate = diff_strip_index(content) + return candidate if _smaller(candidate, content) else content + + if kind == "text": + # Collapse blank-line runs; reversible against itself. + candidate = collapse_runs(content) + if expand_runs(candidate) != content: + return content + return candidate if _smaller(candidate, content) else content + except Exception: + return content + return content diff --git a/tests/test_lossless_mode.py b/tests/test_lossless_mode.py new file mode 100644 index 000000000..65bddd325 --- /dev/null +++ b/tests/test_lossless_mode.py @@ -0,0 +1,345 @@ +"""Tests for no-CCR --lossless proxy mode (Stage A). + +Covers: + * flag plumbing (ProxyConfig field, CLI option parses), + * reversibility of each format-native lossless compaction, + * end-to-end ContentRouter(lossless=True) invariant: smaller output, NO + ``< None: + assert ProxyConfig().lossless is False + assert ProxyConfig(lossless=True).lossless is True + + +def test_content_router_config_has_lossless_field() -> None: + assert ContentRouterConfig().lossless is False + cfg = ContentRouterConfig(lossless=True) + assert cfg.lossless is True + + +def test_router_lossless_forces_marker_free_config() -> None: + # Building the router normalizes the config so the no-CCR invariant holds + # regardless of how ContentRouterConfig was constructed. + router = ContentRouter(ContentRouterConfig(lossless=True)) + assert router.config.ccr_inject_marker is False + assert router.config.smart_crusher_lossless_only is True + + +def test_cli_proxy_option_parses_lossless() -> None: + from click.testing import CliRunner + + from headroom.cli.proxy import proxy + + runner = CliRunner() + result = runner.invoke(proxy, ["--help"]) + assert result.exit_code == 0 + assert "--lossless" in result.output + + +# -------------------------------------------------------------------------- +# Reversibility: collapse_runs / expand_runs +# -------------------------------------------------------------------------- +def test_collapse_expand_runs_byte_roundtrip() -> None: + log = ( + "starting worker\n" + "connection refused\n" + "connection refused\n" + "connection refused\n" + "connection refused\n" + "connection refused\n" + "retrying\n" + "retrying\n" + "done\n" + ) + collapsed = collapse_runs(log) + assert is_run_collapsed(collapsed) + assert len(collapsed) < len(log) + assert expand_runs(collapsed) == log + + +def test_collapse_runs_no_trailing_newline_roundtrip() -> None: + log = "a\na\na\nb" + assert expand_runs(collapse_runs(log)) == log + + +def test_collapse_runs_singletons_untouched() -> None: + log = "one\ntwo\nthree\n" + assert collapse_runs(log) == log + assert not is_run_collapsed(log) + + +def test_collapse_runs_empty() -> None: + assert collapse_runs("") == "" + assert expand_runs("") == "" + + +# -------------------------------------------------------------------------- +# Reversibility: search_heading / search_unheading +# -------------------------------------------------------------------------- +def test_search_heading_unheading_roundtrip() -> None: + grep = ( + "src/app.py:10:def main():\n" + "src/app.py:11: run()\n" + "src/app.py:42: return 0\n" + "src/util.py:3:import os\n" + "src/util.py:9:import sys\n" + ) + headed = search_heading(grep) + # heading form: each path appears once as its own header line + assert headed.count("src/app.py") == 1 + assert headed.count("src/util.py") == 1 + assert search_unheading(headed) == grep + + +def test_search_heading_smaller_for_repeated_paths() -> None: + grep = "\n".join(f"a/very/long/path/module.py:{i}:line{i}" for i in range(1, 30)) + "\n" + headed = search_heading(grep) + assert len(headed) < len(grep) + assert search_unheading(headed) == grep + + +def test_search_heading_leaves_non_matching_lines() -> None: + text = "just some prose\nnot a grep row at all\n" + assert search_heading(text) == text + assert search_unheading(text) == text + + +def test_search_heading_mixed_content_roundtrip() -> None: + grep = "banner line\nsrc/a.py:1:x\nsrc/a.py:2:y\nmiddle prose\nsrc/b.py:5:z\n" + headed = search_heading(grep) + assert search_unheading(headed) == grep + + +# -------------------------------------------------------------------------- +# strip_ansi +# -------------------------------------------------------------------------- +def test_strip_ansi_removes_only_escapes() -> None: + colored = "\x1b[31mERROR\x1b[0m: boom \x1b[1mbold\x1b[0m end" + assert strip_ansi(colored) == "ERROR: boom bold end" + plain = "no escapes here : 1:2:3" + assert strip_ansi(plain) == plain + + +# -------------------------------------------------------------------------- +# diff_strip_index +# -------------------------------------------------------------------------- +def test_diff_strip_index_keeps_plus_minus() -> None: + diff = ( + "diff --git a/f.py b/f.py\n" + "index 0123abc..def4567 100644\n" + "--- a/f.py\n" + "+++ b/f.py\n" + "@@ -1,3 +1,3 @@\n" + " context\n" + "-old line\n" + "+new line\n" + ) + stripped = diff_strip_index(diff) + assert "index 0123abc..def4567" not in stripped + assert "-old line" in stripped + assert "+new line" in stripped + assert "@@ -1,3 +1,3 @@" in stripped + assert "--- a/f.py" in stripped + assert len(stripped) < len(diff) + + +# -------------------------------------------------------------------------- +# compact_lossless dispatch + safety gates +# -------------------------------------------------------------------------- +def test_compact_lossless_log_roundtrips_modulo_ansi() -> None: + log = "\x1b[31mfail\x1b[0m\n" + "fail\n" * 5 + out = compact_lossless(log, "log") + # recoverable modulo ANSI: expand back == de-ANSI'd original + assert expand_runs(out) == strip_ansi(log) + assert len(out) < len(log) + + +def test_compact_lossless_returns_original_when_not_smaller() -> None: + # No repeats, no ANSI -> nothing to gain; returns original unchanged. + log = "line one\nline two\nline three\n" + assert compact_lossless(log, "log") == log + + +def test_compact_lossless_search() -> None: + grep = "\n".join(f"pkg/mod.py:{i}:code{i}" for i in range(1, 20)) + "\n" + out = compact_lossless(grep, "search") + assert len(out) < len(grep) + assert search_unheading(out) == grep + + +def test_compact_lossless_unknown_kind_passthrough() -> None: + assert compact_lossless("whatever", "mystery") == "whatever" + + +def test_compact_lossless_never_raises_on_empty() -> None: + for kind in ("log", "search", "diff", "text"): + assert compact_lossless("", kind) == "" + + +# -------------------------------------------------------------------------- +# End-to-end: ContentRouter(lossless=True) invariant +# -------------------------------------------------------------------------- +def _assert_no_marker(text: str) -> None: + assert "< None: + # Drive the LOG strategy directly: detection is content-dependent, but the + # router's lossless disposition for an explicit LOG strategy must apply + # format-native lossless compaction with no marker and full recovery. + router = ContentRouter(ContentRouterConfig(lossless=True)) + log = ( + "[info] boot sequence start\n" + + "connection refused by upstream service alpha\n" * 40 + + "[info] boot sequence complete\n" + ) + out, _tokens, chain = router._apply_strategy_to_content( + log, CompressionStrategy.LOG, context="" + ) + _assert_no_marker(out) + assert len(out) < len(log) + assert chain == ["lossless_log"] + # every original line recoverable (no ANSI here, so exact) + assert expand_runs(out) == log + + +def test_router_lossless_diff_strategy_no_marker() -> None: + router = ContentRouter(ContentRouterConfig(lossless=True)) + diff = ( + "diff --git a/f.py b/f.py\n" + "index 0123abc..def4567 100644\n" + "--- a/f.py\n" + "+++ b/f.py\n" + "@@ -1,2 +1,2 @@\n" + "-old\n" + "+new\n" + ) + out, _tokens, chain = router._apply_strategy_to_content( + diff, CompressionStrategy.DIFF, context="" + ) + _assert_no_marker(out) + assert chain == ["lossless_diff"] + assert "index 0123abc..def4567" not in out + assert "-old" in out and "+new" in out + + +def test_router_lossless_search_no_marker_and_recoverable() -> None: + router = ContentRouter(ContentRouterConfig(lossless=True)) + grep = ( + "\n".join( + f"headroom/transforms/content_router.py:{i}: identifier_{i} = compute()" + for i in range(1, 60) + ) + + "\n" + ) + result = router.compress(grep, context="") + out = result.compressed + _assert_no_marker(out) + # identifiers still present / recoverable + if result.strategy_used == CompressionStrategy.SEARCH: + assert search_unheading(out) == grep + # at minimum, no data lost and no marker + for i in (1, 30, 59): + assert f"identifier_{i}" in out + + +def test_router_lossless_never_emits_marker_various_inputs() -> None: + router = ContentRouter(ContentRouterConfig(lossless=True)) + samples = [ + "err\n" * 100, + "\n".join(f"a/b/c.py:{i}:x{i}" for i in range(50)), + "plain prose line\n" * 30, + ] + for s in samples: + _assert_no_marker(router.compress(s, context="").compressed) + + +# -------------------------------------------------------------------------- +# Token-delta measurement (informational) +# -------------------------------------------------------------------------- +def test_measure_token_deltas(capsys) -> None: # type: ignore[no-untyped-def] + log = ( + "[warn] disk usage high\n" + + "\x1b[33mretrying connection to db-primary\x1b[0m\n" * 50 + + "[info] recovered\n" + ) + grep = ( + "\n".join( + f"headroom/proxy/server.py:{i}: router_config.value_{i} = {i}" for i in range(1, 80) + ) + + "\n" + ) + log_out = compact_lossless(log, "log") + grep_out = compact_lossless(grep, "search") + # chars are what the byte-size gate optimizes and what subword tokenizers + # track closely; whitespace word-count is reported for direction only. + log_c0, log_c1 = len(log), len(log_out) + grep_c0, grep_c1 = len(grep), len(grep_out) + log_t0, log_t1 = len(log.split()), len(log_out.split()) + grep_t0, grep_t1 = len(grep.split()), len(grep_out.split()) + print( + f"\n[lossless deltas] " + f"log: {log_c0}->{log_c1} chars " + f"({100 * (log_c0 - log_c1) / log_c0:.1f}% saved), {log_t0}->{log_t1} words; " + f"grep: {grep_c0}->{grep_c1} chars " + f"({100 * (grep_c0 - grep_c1) / grep_c0:.1f}% saved), {grep_t0}->{grep_t1} words" + ) + assert log_c1 < log_c0 + assert grep_c1 < grep_c0 + + +def test_lossless_mode_builds_kompress_marker_free(monkeypatch) -> None: + """In lossless (no-CCR) mode Kompress must be built with enable_ccr=False so + it never appends a `Retrieve more: hash=` marker or writes the CCR store — + the agent has no MCP tool to redeem it. Kompress still runs (lossy); only + the marker/store is suppressed. Verified WITHOUT loading the model. + """ + import headroom.transforms.kompress_compressor as kc + from headroom.transforms.content_router import ContentRouter, ContentRouterConfig + + captured: dict[str, object] = {} + + class _FakeKompress: + def __init__(self, config=None) -> None: + captured["enable_ccr"] = getattr(config, "enable_ccr", None) + + def is_ready(self) -> bool: + return False + + def ensure_background_load(self) -> None: + pass + + monkeypatch.setattr(kc, "is_kompress_available", lambda: True) + monkeypatch.setattr(kc, "KompressCompressor", _FakeKompress) + + ContentRouter(ContentRouterConfig(lossless=True))._get_kompress() + assert captured["enable_ccr"] is False # marker suppressed in lossless mode + + captured.clear() + ContentRouter(ContentRouterConfig(lossless=False))._get_kompress() + assert captured["enable_ccr"] is True # normal mode: unchanged (marker on)