diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 1460b9db2..72593b4f3 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -973,6 +973,10 @@ def _add_timing(name: str, started_at: float) -> None: extraction_started = time.perf_counter() candidates: list[tuple[int, tuple[str, int | None], str]] = [] + # Excluded-tool outputs that are losslessly foldable (grep/log/json): + # (item_index, slot_ref, folded_text, original_text). Spliced after the + # normal candidate compression — no ML, byte/data-lossless only. + lossless_excluded: list[tuple[int, tuple[str, int | None], str, str]] = [] extraction_debug: list[dict[str, Any]] = [] for idx, item in enumerate(items): if not isinstance(item, dict): @@ -1004,12 +1008,27 @@ def _add_timing(name: str, started_at: float) -> None: ) continue if isinstance(call_id, str) and call_id in excluded_call_ids: + # Protected from lossy compression — but grep/log/json output + # can still be losslessly compacted. Reuse the router helper + # so the Responses path matches the chat/Anthropic behavior. + excl_out = item.get("output") + fold = ( + router._lossless_compact_excluded(excl_out) + if isinstance(excl_out, str) + else None + ) + if fold is not None: + lossless_excluded.append((idx, ("output", None), fold[0], excl_out)) if debug_enabled: extraction_debug.append( { "index": idx, "eligible": False, - "reason": "exclude_tools_protected", + "reason": ( + "exclude_tools_lossless_fold" + if fold is not None + else "exclude_tools_protected" + ), "item_type": item_type, "call_id": call_id, "tool_name": function_name_by_call_id.get(call_id), @@ -1068,7 +1087,7 @@ def _add_timing(name: str, started_at: float) -> None: payload=payload, extraction=extraction_debug, ) - if not candidates: + if not candidates and not lossless_excluded: _log( "codex_compression_payload_result", modified=False, @@ -1345,6 +1364,23 @@ def _record_routed_result( transforms.append(transform) _add_timing("compression_unit_apply_results", apply_started) + # Splice byte/data-lossless folds of excluded tool outputs (grep/log/ + # json). These skip the ML compressor entirely — the fold is already + # information-preserving — so "excluded = no lossy" still holds. + for e_idx, e_slot, e_folded, e_orig in lossless_excluded: + e_target = updated_items[e_idx] if e_idx < len(updated_items) else None + if not isinstance(e_target, dict): + continue + _set_slot_text(e_target, e_slot, e_folded) + modified = True + e_before = tokenizer.count_text(e_orig) + e_saved = e_before - tokenizer.count_text(e_folded) + if e_saved > 0: + tokens_saved_total += e_saved + attempted_input_tokens += e_before + if "router:excluded:lossless" not in transforms: + transforms.append("router:excluded:lossless") + _log( "codex_compression_payload_result", modified=modified, diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 6423c0386..7bba5346a 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -93,6 +93,92 @@ def _tool_call_args_text(raw: Any) -> str: return " ".join(text.split())[:300] +def _tool_call_command_text(raw: Any) -> str: + """Extract the raw shell command from a tool call's args, if present. + + Anthropic ``input`` is a dict ({"command": "grep …"}); OpenAI ``arguments`` + is a JSON string; Codex's shell uses a ``command`` list. Returns "" when + there is no command field (non-shell tools). + """ + if isinstance(raw, str): + try: + raw = json.loads(raw) + except (ValueError, TypeError): + return "" + if not isinstance(raw, dict): + return "" + cmd = raw.get("command", raw.get("cmd", "")) + if isinstance(cmd, list): + cmd = " ".join(str(c) for c in cmd) + return cmd if isinstance(cmd, str) else "" + + +# Shell wrappers that prefix the real program — peeled to find it. Shell +# grammar, not tunable policy: rtk (the user's token proxy), sudo/env/timeout/… +_SHELL_WRAPPERS = frozenset( + { + "rtk", + "sudo", + "env", + "time", + "nice", + "ionice", + "nohup", + "stdbuf", + "command", + "timeout", + "xargs", + } +) + + +def _bash_program(command: str) -> tuple[str, list[str]]: + """Return ``(program_basename_lower, trailing_tokens)`` for a shell command. + + Peels leading wrappers (``rtk grep`` -> ``grep``, ``timeout 30 rg`` -> ``rg``) + and env assignments (``FOO=1 grep`` -> ``grep``). Empty program when it can't + be determined. Whitespace-split is deliberately simple — the reversibility + guard downstream makes a parse miss harmless. + """ + toks = command.strip().split() + i = 0 + while i < len(toks): + tok = toks[i] + if "=" in tok and not tok.startswith("-"): # VAR=val env assignment + i += 1 + continue + base = tok.rsplit("/", 1)[-1].lower() # /usr/bin/grep -> grep + if base in _SHELL_WRAPPERS: + i += 1 + # Skip this wrapper's own option/numeric args (timeout 30, nice -n 5). + while i < len(toks) and ( + toks[i].startswith("-") or toks[i].replace(".", "", 1).isdigit() + ): + i += 1 + continue + return base, toks[i + 1 :] + return "", [] + + +def _bash_command_is_search(command: str, search_commands: frozenset[str]) -> bool: + """True when ``command`` is a read-only search whose output folds byte- + losslessly (grep/rg/git grep/…). Peels wrappers and recurses into ``sh -c``. + """ + prog, rest = _bash_program(command) + if not prog: + return False + if prog in {"sh", "bash", "zsh", "dash"} and rest: + # `bash -lc "grep …"` (Codex): the real command is the -c argument. + for j, tok in enumerate(rest): + if tok in {"-c", "-lc", "-lic", "-ic"} and j + 1 < len(rest): + inner = " ".join(rest[j + 1 :]).strip("'\"") + return _bash_command_is_search(inner, search_commands) + return False + if prog == "git" and rest and rest[0].lower() == "grep": + return True + return prog in search_commands + + def _log_router_debug(event: str, **payload: Any) -> None: if not logger.isEnabledFor(logging.DEBUG): return @@ -889,6 +975,22 @@ class ContentRouterConfig: # Set to None to use DEFAULT_EXCLUDE_TOOLS, or provide custom set exclude_tools: set[str] | None = None + # Excluded tools are protected only from *lossy* compression. Their output + # is still given information-preserving compaction by detected shape (grep + # -> ripgrep --heading fold; logs -> ANSI strip + run-collapse; JSON -> + # whitespace-minify, data-lossless), in every path — see + # ``_lossless_compact_excluded``. Always recoverable, so no config gate. + + # Shell tool names (case-insensitive). Their output is non-excluded/lossy, + # BUT a read-only *search* run through them (grep/rg/git grep) yields byte- + # losslessly foldable output — folded instead of lossy-compressed. See + # ``_bash_search_fold``. Config so new harness tool names / search programs + # can be added without code changes. + bash_tool_names: frozenset[str] = frozenset({"bash", "shell", "local_shell"}) + bash_search_commands: frozenset[str] = frozenset( + {"grep", "egrep", "fgrep", "rg", "ripgrep", "ag", "ack"} + ) + # Read lifecycle management (stale/superseded detection) read_lifecycle: ReadLifecycleConfig = field(default_factory=ReadLifecycleConfig) @@ -1192,6 +1294,8 @@ def __init__( self._relevance_prewarm_started: bool = False # tool_call_id → compact args text, populated by _build_tool_name_map. self._tool_call_args: dict[str, str] = {} + # tool_call_id → raw shell command (bash-search fold), same population. + self._tool_call_commands: dict[str, str] = {} # Phase 0 (#1171): cap the input size handed to kompress (ModernBERT # ONNX). Its inference scales O(tokens) and runs synchronously on the @@ -2611,6 +2715,7 @@ def _build_tool_name_map(self, messages: list[dict[str, Any]]) -> dict[str, str] """ mapping: dict[str, str] = {} args_map: dict[str, str] = {} + commands_map: dict[str, str] = {} for msg in messages: if msg.get("role") != "assistant": @@ -2627,6 +2732,9 @@ def _build_tool_name_map(self, messages: list[dict[str, Any]]) -> dict[str, str] args = _tool_call_args_text(fn.get("arguments")) if args: args_map[tc_id] = args + command = _tool_call_command_text(fn.get("arguments")) + if command: + commands_map[tc_id] = command # Anthropic format: content blocks with type=tool_use content = msg.get("content", []) @@ -2640,8 +2748,12 @@ def _build_tool_name_map(self, messages: list[dict[str, Any]]) -> dict[str, str] args = _tool_call_args_text(block.get("input")) if args: args_map[tc_id] = args + command = _tool_call_command_text(block.get("input")) + if command: + commands_map[tc_id] = command self._tool_call_args = args_map + self._tool_call_commands = commands_map return mapping def _net_cost_allows( @@ -3080,6 +3192,17 @@ def apply( tool_call_id = message.get("tool_call_id", "") if tool_call_id in excluded_tool_ids: if messages_from_end <= read_protection_window: + # Protected from lossy compression — but grep/log/json + # output can still be losslessly compacted. + compacted = self._lossless_compact_excluded(content) + if compacted is not None: + folded, kind = compacted + result_slots[i] = {**message, "content": folded} + transforms_applied.append(f"router:excluded:lossless_{kind}") + route_counts["excluded_tool_lossless"] = ( + route_counts.get("excluded_tool_lossless", 0) + 1 + ) + continue # Recent — protect as before result_slots[i] = message transforms_applied.append("router:excluded:tool") @@ -3092,6 +3215,18 @@ def apply( tool_name = tool_name_map.get(tool_call_id, "") bias = self._get_tool_bias(tool_name) if tool_name else 1.0 + # Bash-search lossless pre-empt: a read-only search (grep/rg/git + # grep) run via a shell tool yields byte-losslessly foldable + # output. Fold it instead of the lossy strategy path. + bash_folded = self._bash_search_fold(tool_name, tool_call_id, content) + if bash_folded is not None: + result_slots[i] = {**message, "content": bash_folded} + transforms_applied.append("router:bash:lossless_search") + route_counts["bash_lossless_search"] = ( + route_counts.get("bash_lossless_search", 0) + 1 + ) + continue + # Protection 1: Never compress user messages (unless overridden) if skip_user and role == "user": result_slots[i] = message @@ -3406,6 +3541,90 @@ def apply( timing=compressor_timing, ) + def _lossless_compact_excluded(self, content: Any) -> tuple[str, str] | None: + """Information-preserving compaction for a protected (excluded) tool output. + + Excluded tools are kept out of *lossy* compression for accuracy. This + applies only reversible/data-preserving transforms, dispatched by shape: + + * SEARCH (grep ``path:line:content``) -> ripgrep --heading fold. + Byte-recoverable (``search_unheading`` reproduces the original). Gated + on the dedicated ``_try_detect_search`` — the general classifier calls + grep-over-code SOURCE_CODE and would wrongly reject it. + * LOG (build/test/app logs) -> ANSI strip + run-collapse. Recoverable + modulo non-semantic ANSI color (``expand_runs`` restores the lines). + * JSON -> whitespace-minify. **Data-lossless** (``json.loads`` equals the + original object) — same information, fewer tokens. NOT byte-exact, so a + read-then-``Edit(old_string=…)`` on the *same* JSON file could miss; the + data is fully preserved. + + Returns ``(compacted, kind)`` when a recognized shape actually shrinks, + else ``None``. Source code and glob path-lists match nothing -> verbatim. + Always safe to run (information-preserving) so there is no feature gate. + Never raises. + """ + if not isinstance(content, str) or len(content) < 200: + return None + try: + from .lossless_compaction import compact_lossless + + det = _try_detect_search(content) + if det is not None and det.content_type is ContentType.SEARCH_RESULTS: + out = compact_lossless(content, "search") + return (out, "search") if len(out) < len(content) else None + if _try_detect_log(content) is not None: + out = compact_lossless(content, "log") + return (out, "log") if len(out) < len(content) else None + minified = self._minify_json_data_lossless(content) + return (minified, "json") if minified is not None else None + except Exception: # noqa: BLE001 + return None + + @staticmethod + def _minify_json_data_lossless(content: str) -> str | None: + """Whitespace-minify a complete JSON value: data-preserving, not byte-exact. + + The ``json.loads`` parse is the data-equality guarantee (identical + object). Returns the minified form only when the content is a JSON + object/array and the result is smaller; ``None`` otherwise (source code, + partial/non-JSON). + """ + stripped = content.strip() + if not stripped or stripped[0] not in "{[": + return None + obj = json.loads(stripped) + minified = json.dumps(obj, separators=(",", ":"), ensure_ascii=False) + return minified if len(minified) < len(content) else None + + def _bash_search_fold(self, tool_name: str, tool_id: str, content: Any) -> str | None: + """Byte-lossless fold for a read-only search run through a shell tool. + + ``bash`` is not excluded, so its output normally takes the lossy strategy + path. But when the command is a read-only search (grep/rg/git grep/…), + its output is byte-losslessly foldable — so fold it (the same guarantee + excluded Grep gets) instead of lossy-compressing. The command whitelist + is only a *gate to attempt*: ``compact_lossless`` verifies reversibility + and returns the input unchanged when it can't safely shrink, so a mis- + gated command (``grep -l`` path-lists, ``grep -c`` counts) simply falls + through to the normal path with no accuracy risk. + + Returns the folded text (smaller, recoverable) or ``None`` to fall through. + """ + if not isinstance(content, str) or len(content) < 200: + return None + if tool_name.lower() not in self.config.bash_tool_names: + return None + command = self._tool_call_commands.get(tool_id, "") + if not command or not _bash_command_is_search(command, self.config.bash_search_commands): + return None + try: + from .lossless_compaction import compact_lossless + + folded = compact_lossless(content, "search") + except Exception: # noqa: BLE001 + return None + return folded if len(folded) < len(content) else None + def _get_tool_bias(self, tool_name: str) -> float: """Look up compression bias for a tool name. @@ -3528,6 +3747,19 @@ def _process_content_blocks( tool_use_id = block.get("tool_use_id", "") if tool_use_id in excluded_tool_ids: if messages_from_end <= read_protection_window: + # Protected from lossy compression — but grep/log/json + # output can still be losslessly compacted. + compacted = self._lossless_compact_excluded(block.get("content")) + if compacted is not None: + folded, kind = compacted + new_blocks.append({**block, "content": folded}) + transforms_applied.append(f"router:excluded:lossless_{kind}") + if route_counts is not None: + route_counts["excluded_tool_lossless"] = ( + route_counts.get("excluded_tool_lossless", 0) + 1 + ) + any_compressed = True + continue # Recent — protect as before new_blocks.append(block) transforms_applied.append("router:excluded:tool") @@ -3551,6 +3783,20 @@ def _process_content_blocks( tool_content = block.get("content", "") + # Bash-search lossless pre-empt (twin of the string-form path): + # fold read-only search output (grep/rg/git grep) byte-losslessly + # instead of taking the lossy strategy path. + bash_folded = self._bash_search_fold(tool_name, tool_use_id, tool_content) + if bash_folded is not None: + new_blocks.append({**block, "content": bash_folded}) + transforms_applied.append("router:bash:lossless_search") + if route_counts is not None: + route_counts["bash_lossless_search"] = ( + route_counts.get("bash_lossless_search", 0) + 1 + ) + any_compressed = True + continue + # Protection: failed tool calls / error outputs stay verbatim # (issue #847). `is_error` is Anthropic's explicit failure # flag and suffices alone; the indicator scan catches error diff --git a/tests/test_bash_search_lossless_fold.py b/tests/test_bash_search_lossless_fold.py new file mode 100644 index 000000000..687b94d25 --- /dev/null +++ b/tests/test_bash_search_lossless_fold.py @@ -0,0 +1,150 @@ +"""Bash-search lossless fold. + +`bash` is not an excluded tool, so its output normally takes the lossy strategy +path. But a read-only search run through it (grep/rg/git grep) produces byte- +losslessly foldable output — the router detects the *command* and folds it with +the same ripgrep --heading transform excluded Grep gets, instead of lossy +compression. Non-search bash commands (cat/build/mutate) are untouched. +""" + +from __future__ import annotations + +import json + +import pytest + +from headroom.providers import OpenAIProvider +from headroom.tokenizer import Tokenizer +from headroom.transforms.content_router import ( + ContentRouter, + ContentRouterConfig, + _bash_command_is_search, + _bash_program, +) +from headroom.transforms.lossless_compaction import search_unheading + +SEARCH = frozenset({"grep", "egrep", "fgrep", "rg", "ripgrep", "ag", "ack"}) +GREP = "".join( + f"src/module_{f}.py:{ln * 3}:matched occurrence with some real content here\n" + for f in range(6) + for ln in range(15) +) +CODE = "def foo(x):\n return x + 1\n\nclass Bar:\n value = 42\n" * 30 + + +@pytest.fixture +def tokenizer(): + provider = OpenAIProvider() + return Tokenizer(provider.get_token_counter("gpt-4o"), "gpt-4o") + + +# --- command parsing: peel wrappers, detect search programs --- + + +@pytest.mark.parametrize( + "command", + [ + "grep -rn foo .", + "rtk grep def headroom/transforms", # the user's token-proxy wrapper + "rg --heading pattern src/", + "git grep -n TODO", + "sudo grep root /etc/passwd", + "timeout 30 rg foo", # wrapper takes a numeric arg + "FOO=1 BAR=2 grep foo", # env assignments + "/usr/bin/grep -rn foo .", # absolute path + 'bash -lc "grep -rn foo ."', # Codex-style shell -c + "nice -n 5 ack pattern", # wrapper with option arg + ], +) +def test_detects_search_commands(command): + assert _bash_command_is_search(command, SEARCH) is True + + +@pytest.mark.parametrize( + "command", + [ + "cat headroom/server.py", + "cargo test", + "pytest tests/ -x", + "git diff HEAD~1", # diff, NOT search + "echo grep", # echo, not a real grep + "python script.py", + "rm -rf build", + "ls -la", + ], +) +def test_ignores_non_search_commands(command): + assert _bash_command_is_search(command, SEARCH) is False + + +def test_bash_program_peels_wrappers(): + assert _bash_program("rtk grep foo")[0] == "grep" + assert _bash_program("timeout 30 rg x")[0] == "rg" + assert _bash_program("FOO=1 /usr/bin/grep y")[0] == "grep" + assert _bash_program("")[0] == "" + + +# --- end-to-end through the router (both wire formats) --- + + +def _openai(command: str, content: str, tokenizer): + router = ContentRouter(ContentRouterConfig()) + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "c1", + "function": {"name": "bash", "arguments": json.dumps({"command": command})}, + } + ], + }, + {"role": "tool", "tool_call_id": "c1", "content": content}, + ] + result = router.apply(messages, tokenizer, compress_user_messages=True) + return result.messages[1]["content"], result.transforms_applied + + +def _anthropic(command: str, content: str, tokenizer): + router = ContentRouter(ContentRouterConfig()) + messages = [ + { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "t1", "name": "bash", "input": {"command": command}} + ], + }, + { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": "t1", "content": content}], + }, + ] + result = router.apply(messages, tokenizer, compress_user_messages=True) + return result.messages[1]["content"][0]["content"], result.transforms_applied + + +def test_openai_bash_grep_folds_and_recovers(tokenizer): + out, transforms = _openai("grep -rn foo .", GREP, tokenizer) + assert "router:bash:lossless_search" in transforms + assert len(out) < len(GREP) + assert search_unheading(out) == GREP # byte-exact + + +def test_anthropic_bash_rtk_grep_folds_and_recovers(tokenizer): + out, transforms = _anthropic("rtk grep foo headroom/", GREP, tokenizer) + assert "router:bash:lossless_search" in transforms + assert search_unheading(out) == GREP + + +def test_non_search_bash_command_not_folded(tokenizer): + # `cat` is not a search — must NOT take the bash-search fold. + _out, transforms = _openai("cat headroom/server.py", GREP, tokenizer) + assert "router:bash:lossless_search" not in transforms + + +def test_source_output_from_search_command_untouched(tokenizer): + # Command is a search, but the output isn't path:line:content — the + # reversibility guard makes compact_lossless return it unchanged. + out, transforms = _openai("grep -l foo", CODE, tokenizer) + assert "router:bash:lossless_search" not in transforms + assert out == CODE diff --git a/tests/test_lossless_excluded_compaction.py b/tests/test_lossless_excluded_compaction.py new file mode 100644 index 000000000..c8a8f8c76 --- /dev/null +++ b/tests/test_lossless_excluded_compaction.py @@ -0,0 +1,119 @@ +"""Information-preserving compaction for EXCLUDED tool output. + +Excluded tools (Read/Grep/Glob/Write/Edit) are protected from *lossy* +compression for accuracy. This feature still compacts them by detected shape, +using only reversible / data-preserving transforms: + +* SEARCH (grep) -> ripgrep --heading fold [byte-lossless] +* LOG -> ANSI strip + run-collapse [byte-lossless modulo ANSI color] +* JSON -> whitespace-minify [data-lossless; same object, NOT byte-exact] + +Source code and glob path-lists match nothing -> untouched. Always on +(information-preserving, so it needs no feature gate) in every path. +""" + +from __future__ import annotations + +import json + +import pytest + +from headroom.providers import OpenAIProvider +from headroom.tokenizer import Tokenizer +from headroom.transforms.content_router import ContentRouter, ContentRouterConfig +from headroom.transforms.lossless_compaction import expand_runs, search_unheading, strip_ansi + +GREP = "".join( + f"src/module_{f}.py:{ln * 3}:matched occurrence with some real content here\n" + for f in range(6) + for ln in range(15) +) +LOG = "".join( + f"\x1b[32m2026-07-03 INFO worker {i % 3} processing job batch\x1b[0m\n" for i in range(40) +) +LOG += "".join("2026-07-03 WARN transient retry, backing off\n" for _ in range(25)) +JSON = json.dumps( + {"users": [{"id": i, "name": f"user{i}", "active": i % 2 == 0} for i in range(40)]}, + indent=2, +) +CODE = "def foo(x):\n return x + 1\n\nclass Bar:\n value = 42\n" * 30 +GLOB = "\n".join(f"src/module_{i}.py" for i in range(60)) + "\n" + + +@pytest.fixture +def tokenizer(): + provider = OpenAIProvider() + return Tokenizer(provider.get_token_counter("gpt-4o"), "gpt-4o") + + +def _compact(content: str): + router = ContentRouter(ContentRouterConfig()) + return router._lossless_compact_excluded(content) + + +# --- helper: right transform per shape, right guarantee --- + + +def test_grep_search_fold_is_byte_lossless(): + out, kind = _compact(GREP) + assert kind == "search" + assert len(out) < len(GREP) + assert search_unheading(out) == GREP # byte-exact + + +def test_log_compaction_recovers_modulo_ansi(): + out, kind = _compact(LOG) + assert kind == "log" + assert len(out) < len(LOG) + assert expand_runs(out) == strip_ansi(LOG) # recover the lines (ANSI dropped) + + +def test_json_minify_is_data_lossless(): + out, kind = _compact(JSON) + assert kind == "json" + assert len(out) < len(JSON) + assert json.loads(out) == json.loads(JSON) # same object; NOT byte-exact + + +def test_source_and_glob_untouched(): + assert _compact(CODE) is None + assert _compact(GLOB) is None + + +# --- end-to-end through the router pipeline (excluded tools) --- + + +def _run(content: str, tool: str, tokenizer): + router = ContentRouter(ContentRouterConfig()) + messages = [ + { + "role": "assistant", + "tool_calls": [{"id": "c1", "function": {"name": tool, "arguments": "{}"}}], + }, + {"role": "tool", "tool_call_id": "c1", "content": content}, + ] + result = router.apply(messages, tokenizer, compress_user_messages=True) + return result.messages[1]["content"], result.transforms_applied + + +def test_pipeline_folds_grep_and_recovers(tokenizer): + out, transforms = _run(GREP, "grep", tokenizer) + assert "router:excluded:lossless_search" in transforms + assert search_unheading(out) == GREP + + +def test_pipeline_compacts_log_read(tokenizer): + out, transforms = _run(LOG, "read", tokenizer) + assert "router:excluded:lossless_log" in transforms + assert expand_runs(out) == strip_ansi(LOG) + + +def test_pipeline_minifies_json_read(tokenizer): + out, transforms = _run(JSON, "read", tokenizer) + assert "router:excluded:lossless_json" in transforms + assert json.loads(out) == json.loads(JSON) # data-lossless (same object) + + +def test_pipeline_leaves_source_read_untouched(tokenizer): + out, _ = _run(CODE, "read", tokenizer) + assert out == CODE diff --git a/tests/test_openai_responses_compression_units.py b/tests/test_openai_responses_compression_units.py index be323c47e..e16dbcb4a 100644 --- a/tests/test_openai_responses_compression_units.py +++ b/tests/test_openai_responses_compression_units.py @@ -452,6 +452,44 @@ def compress(self, content: str, **_kwargs): assert strategy_chain == [] +def test_openai_responses_adapter_losslessly_folds_excluded_grep_output(): + """Excluded tools skip *lossy* compression, but grep/log/json output is still + byte/data-losslessly compacted on the Responses path (matches chat/Anthropic). + """ + from headroom.transforms.lossless_compaction import search_unheading + + router = ContentRouter() + router.config.exclude_tools = {"grep"} + handler = _handler_with_router(router) + grep_out = "".join( + f"src/mod_{f}.py:{ln}:some matching content on this line here\n" + for f in range(8) + for ln in range(6) + ) + payload = { + "model": "gpt-5", + "input": [ + {"type": "function_call", "call_id": "call_1", "name": "grep", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_1", "output": grep_out}, + ], + } + + new_payload, modified, saved, transforms, _units, _chain, _attempted = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5", + request_id="req_test", + ) + ) + + assert modified is True + assert saved >= 0 # token accounting never goes negative + assert "router:excluded:lossless" in transforms + folded = new_payload["input"][1]["output"] + assert len(folded) < len(grep_out) # byte-smaller (real guarantee) + assert search_unheading(folded) == grep_out # byte-exact recovery + + def test_openai_responses_adapter_excludes_tool_case_insensitively_with_debug(monkeypatch): """Excluded match is case-insensitive, and the debug path stays exercised. diff --git a/tests/test_transforms/test_content_router.py b/tests/test_transforms/test_content_router.py index a336a1b6a..7d53e584a 100644 --- a/tests/test_transforms/test_content_router.py +++ b/tests/test_transforms/test_content_router.py @@ -8,6 +8,7 @@ - Transform interface: apply(), should_apply() methods """ +import json from types import SimpleNamespace import pytest @@ -20,6 +21,7 @@ RouterCompressionResult, RoutingDecision, ) +from headroom.transforms.lossless_compaction import search_unheading # ============================================================================= # Test Fixtures @@ -734,9 +736,10 @@ def test_custom_exclude_tools(self, tokenizer): result = router.apply(messages, tokenizer) - # Content should be unchanged - assert result.messages[1]["content"] == messages[1]["content"] - assert "router:excluded:tool" in result.transforms_applied + # Excluded from *lossy* compression, but JSON still gets a data-lossless + # minify (same object, fewer tokens). Assert recovery, not byte-identity. + assert json.loads(result.messages[1]["content"]) == json.loads(messages[1]["content"]) + assert "router:excluded:lossless_json" in result.transforms_applied def test_glob_exclude_tools(self, tokenizer): """Glob patterns in exclude_tools match by prefix (issue #870).""" @@ -770,9 +773,10 @@ def test_glob_exclude_tools(self, tokenizer): result = router.apply(messages, tokenizer) - # The MCP tool result matched the glob and was left unchanged. - assert result.messages[1]["content"] == messages[1]["content"] - assert "router:excluded:tool" in result.transforms_applied + # The MCP tool result matched the glob → excluded from lossy compression. + # Its JSON still gets a data-lossless minify; assert recovery. + assert json.loads(result.messages[1]["content"]) == json.loads(messages[1]["content"]) + assert "router:excluded:lossless_json" in result.transforms_applied def test_is_tool_excluded_helper(self): """is_tool_excluded: exact (case-insensitive) and glob matching.""" @@ -896,9 +900,11 @@ def test_anthropic_format_tool_result_exclusion(self, tokenizer): (b for b in user_msg["content"] if b.get("type") == "tool_result"), None ) assert tool_result_block is not None - assert tool_result_block["content"] == messages[1]["content"][0]["content"] - # Verify exclusion was tracked (consistent with OpenAI format) - assert "router:excluded:tool" in result.transforms_applied + # Excluded from lossy compression; search results get a byte-lossless + # heading fold. Verify byte-exact recovery (Anthropic block format). + original = messages[1]["content"][0]["content"] + assert search_unheading(tool_result_block["content"]) == original + assert "router:excluded:lossless_search" in result.transforms_applied def test_anthropic_tool_result_runtime_window_allows_old_excluded_tools(self, tokenizer): """Agent profiles can shrink the protected window for Claude tool results."""