diff --git a/headroom/transforms/config_compressor.py b/headroom/transforms/config_compressor.py new file mode 100644 index 000000000..92e623478 --- /dev/null +++ b/headroom/transforms/config_compressor.py @@ -0,0 +1,283 @@ +"""Structured-config compressor for YAML/TOML/INI tool output. + +Config files (k8s manifests, CI pipelines, pyproject/Cargo manifests, INI +files) are high-frequency agent payloads with heavy structural repetition, +but they have no native compressor — magika tags them SOURCE_CODE and they +fall through to the lossy prose path. This module compresses them in two +lossless-first tiers: + +- Tier 1 (reversible, allowed in no-CCR lossless mode): format-agnostic + text-level compaction via `lossless_compaction` — identical-line runs and + repeated multi-line stanzas collapse behind exact-inverse markers, with the + round-trip self-verified before the result is adopted. +- Tier 2 (CCR-recoverable, default mode only): whole-line comments and blank + lines are elided behind a summary line carrying a ``Retrieve original: + hash=…`` marker; the full original is persisted to the CompressionStore + first, so nothing is lost. +- Tier 3 (CCR-recoverable, default mode only): TOML files with an + ``[[array-of-tables]]`` are parsed with the stdlib ``tomllib`` reference + parser and bridged to SmartCrusher's lossless ``csv-schema`` renderer, + which folds the repeated per-record keys into a single schema. Only TOML + is bridged — its stdlib parser makes the records ground-truth; YAML/INI + need non-stdlib or bespoke parsing and stay out of scope. Recovery again + rides a stored-original + ``Retrieve original: hash=…`` marker. + +No new compression algorithm and no new CCR plumbing live here — Tier 1 rides +`compact_lossless`, Tier 2/Tier 3 ride the production `CompressionStore`, and +Tier 3's fold rides `SmartCrusher`. +""" + +from __future__ import annotations + +import datetime as dt +import json +import logging +import re +from dataclasses import dataclass +from typing import Any, cast + +from .content_detector import ContentType, detect_content_type +from .lossless_compaction import compact_lossless + +logger = logging.getLogger(__name__) + +# Whole-line comment prefixes per flavor. INI values can span indented +# continuation lines, so INI only elides column-0 comment lines and keeps +# blanks (configparser keeps blank lines inside multi-line values). +_COMMENT_RES = { + "yaml": re.compile(r"^\s*#"), + "toml": re.compile(r"^\s*#"), + "ini": re.compile(r"^[#;]"), +} + +# Content where whole-line elision is unsafe: a '#' line inside a YAML block +# scalar or a TOML multi-line string is data, not a comment. Detection is +# deliberately over-broad — when in doubt, Tier 2 stays off. +_YAML_BLOCK_SCALAR_RE = re.compile(r":\s*[|>][+-]?\d*\s*$", re.MULTILINE) +_TOML_MULTILINE_RE = re.compile(r'"""|\'\'\'') + + +def _load_toml(content: str) -> dict[str, Any] | None: + """Parse TOML with the stdlib parser (or the tomli backport); None on error.""" + try: + import tomllib + except ModuleNotFoundError: # pragma: no cover - Python < 3.11 only + try: + import tomli as tomllib # type: ignore[no-redef] + except ModuleNotFoundError: + return None + try: + return cast("dict[str, Any]", tomllib.loads(content)) + except (tomllib.TOMLDecodeError, ValueError): + return None + + +def _json_default(value: Any) -> str: + """Render TOML date/time values as ISO strings; bail on anything else.""" + if isinstance(value, dt.datetime | dt.date | dt.time): + return value.isoformat() + raise TypeError(f"unserializable config value: {type(value).__name__}") + + +@dataclass +class ConfigCompressorConfig: + """Configuration for structured-config compression.""" + + # Emit the CCR-marked comment/blank elision tier. The router wires this + # to its ccr_inject_marker setting; lossless mode turns it off. + enable_ccr: bool = True + # Bridge TOML array-of-tables to SmartCrusher csv-schema (Tier 3). Rides + # CCR for recovery, so it only runs when enable_ccr is also on. + enable_schema_fold: bool = True + # Only adopt a result that is strictly smaller than the original. + min_savings_chars: int = 1 + + +@dataclass +class ConfigCompressionResult: + """Result of structured-config compression.""" + + compressed: str + original: str + was_modified: bool + flavor: str # "yaml" | "toml" | "ini" | "unknown" + lines_elided: int = 0 + ccr_hash: str | None = None + strategy: str = "config" + + @property + def compression_ratio(self) -> float: + if not self.original: + return 0.0 + return len(self.compressed) / len(self.original) + + +class ConfigCompressor: + """Compresses YAML/TOML/INI text via reversible + CCR-recoverable tiers. + + Public surface mirrors the other content-type compressors so the router + and tests treat it uniformly. + """ + + def __init__(self, config: ConfigCompressorConfig | None = None) -> None: + self.config = config or ConfigCompressorConfig() + + def compress( + self, + content: str, + context: str = "", + bias: float = 1.0, + ) -> ConfigCompressionResult: + detection = detect_content_type(content) + if detection.content_type is not ContentType.STRUCTURED_CONFIG: + return ConfigCompressionResult( + compressed=content, + original=content, + was_modified=False, + flavor="unknown", + ) + flavor = detection.metadata.get("flavor", "yaml") + + # Tier 3: schema fold (TOML array-of-tables → csv-schema). Computed + # first so it can compete with the text tiers; it wins on lockfiles + # and override-lists where repeated keys dominate. + schema_fold: tuple[str, str] | None = None + if self.config.enable_ccr and self.config.enable_schema_fold: + schema_fold = self._schema_fold(content, flavor, context, bias) + + working = content + lines_elided = 0 + ccr_hash: str | None = None + + # Tier 2: comment/blank elision behind a CCR marker. Persist first — + # the elided lines are only droppable because the original is stored. + if self.config.enable_ccr and self._elision_safe(content, flavor): + stripped, elided = self._strip_comment_lines(content, flavor) + if elided > 0: + ccr_hash = self._store_original(content, stripped) + if ccr_hash is not None: + marker = ( + f"[{elided} comment/blank lines elided. Retrieve original: hash={ccr_hash}]" + ) + working = stripped + ("" if stripped.endswith("\n") else "\n") + marker + lines_elided = elided + + # Tier 1: reversible run/stanza folding; self-verified round-trip. + compressed = compact_lossless(working, "config") + + # Prefer the schema fold when it beats the text tiers. + if schema_fold is not None and len(schema_fold[0]) < len(compressed): + return ConfigCompressionResult( + compressed=schema_fold[0], + original=content, + was_modified=True, + flavor=flavor, + ccr_hash=schema_fold[1], + strategy="config_schema_fold", + ) + + savings = len(content) - len(compressed) + if savings < self.config.min_savings_chars: + return ConfigCompressionResult( + compressed=content, + original=content, + was_modified=False, + flavor=flavor, + ) + + return ConfigCompressionResult( + compressed=compressed, + original=content, + was_modified=True, + flavor=flavor, + lines_elided=lines_elided, + ccr_hash=ccr_hash, + ) + + def _schema_fold( + self, content: str, flavor: str, context: str, bias: float + ) -> tuple[str, str] | None: + """Fold a TOML array-of-tables into SmartCrusher csv-schema. + + Returns ``(folded_text_with_marker, ccr_hash)`` when the fold is a + strictly-smaller, faithful representation whose original is safely + stored for CCR retrieval; otherwise ``None`` so the caller keeps the + text tiers. Only TOML is bridged: ``tomllib`` is the reference parser, + so the extracted records are ground-truth and the csv-schema renderer + is itself lossless — the model reads a faithful, reformatted view. + """ + if flavor != "toml" or "[[" not in content: + return None + data = _load_toml(content) + if data is None: + return None + try: + json_str = json.dumps(data, ensure_ascii=False, default=_json_default) + except TypeError: + return None # a value we can't represent faithfully → don't fold + + from .smart_crusher import SmartCrusher + + crusher = SmartCrusher(with_compaction=True, compaction_format="csv-schema") + result = crusher.crush(json_str, context, bias) + # `passthrough` means SmartCrusher only re-canonicalized the JSON and + # applied no schema fold, so there is nothing worth adopting. + if not result.was_modified or result.strategy == "passthrough": + return None + + ccr_hash = self._store_original(content, result.compressed) + if ccr_hash is None: + return None # can't recover the original → never emit a lossy form + marker = f"[config folded to schema. Retrieve original: hash={ccr_hash}]" + folded = result.compressed + "\n" + marker + if len(content) - len(folded) < self.config.min_savings_chars: + return None + return folded, ccr_hash + + @staticmethod + def _elision_safe(content: str, flavor: str) -> bool: + """False when a '#' line could be data (block scalars, multi-line strings).""" + if flavor == "yaml": + return not _YAML_BLOCK_SCALAR_RE.search(content) + if flavor == "toml": + return not _TOML_MULTILINE_RE.search(content) + return True + + @staticmethod + def _strip_comment_lines(content: str, flavor: str) -> tuple[str, int]: + """Drop whole-line comments (and, outside INI, blank lines).""" + comment_re = _COMMENT_RES.get(flavor, _COMMENT_RES["yaml"]) + keep_blanks = flavor == "ini" + had_trailing = content.endswith("\n") + lines = (content[:-1] if had_trailing else content).split("\n") + kept: list[str] = [] + elided = 0 + for line in lines: + if comment_re.match(line) or (not keep_blanks and not line.strip()): + elided += 1 + else: + kept.append(line) + return "\n".join(kept) + ("\n" if had_trailing else ""), elided + + @staticmethod + def _store_original(original: str, compressed: str) -> str | None: + """Persist the original to the CompressionStore; returns its hash.""" + try: + from ..cache.compression_store import get_compression_store + except ImportError as e: # pragma: no cover - store ships with headroom + logger.warning("CCR store import failed; config elision skipped: %s", e) + return None + try: + store: Any = get_compression_store() + stored = store.store(original, compressed, compression_strategy="config") + return str(stored) if stored else None + except Exception as e: + logger.warning("CCR store write failed; config elision skipped: %s", e) + return None + + +__all__ = [ + "ConfigCompressor", + "ConfigCompressorConfig", + "ConfigCompressionResult", +] diff --git a/headroom/transforms/content_detector.py b/headroom/transforms/content_detector.py index 850413d7e..01412dde0 100644 --- a/headroom/transforms/content_detector.py +++ b/headroom/transforms/content_detector.py @@ -10,6 +10,7 @@ Supported content types: - SEARCH_RESULTS: grep/ripgrep output (file:line:content) - BUILD_OUTPUT: Compiler, test, lint logs - GIT_DIFF: Unified diff format +- STRUCTURED_CONFIG: YAML/TOML/INI config files - PLAIN_TEXT: Generic text (fallback) """ @@ -31,6 +32,7 @@ class ContentType(Enum): GIT_DIFF = "diff" # Unified diff format HTML = "html" # Web pages (needs content extraction, not compression) TABULAR = "tabular" # CSV/TSV, markdown tables, fixed-width tables + STRUCTURED_CONFIG = "structured_config" # YAML/TOML/INI config files PLAIN_TEXT = "text" # Fallback @@ -111,6 +113,18 @@ _CODE_PATTERNS = { ], } +# Structured-config (YAML/TOML/INI) patterns. TOML and INI share the +# `[section]` header shape; the stdlib parsers disambiguate. YAML is +# heuristic-only (PyYAML is not a dependency): key/list/document-marker +# line share plus structure signals, with prose and front-matter guards. +_CONFIG_SECTION_RE = re.compile(r"^\s*\[\[?[\w.\-\"' ]+\]\]?\s*$") +_TOML_ASSIGN_RE = re.compile(r"""^\s*(?:[\w.\-]+|"[^"]+"|'[^']+')\s*=\s*\S""") +_INI_ASSIGN_RE = re.compile(r"^\s*[\w.\-@ ]+?\s*[=:]\s*") +_YAML_KEY_RE = re.compile(r"""^\s*(?:-\s+)?(?:[\w.\-/]+|"[^"]+"|'[^']+')\s*:(?:\s|$)""") +_YAML_LIST_RE = re.compile(r"^\s*-\s+\S") +_YAML_DOC_RE = re.compile(r"^---\s*$|^\.\.\.\s*$") +_CONFIG_COMMENT_RE = re.compile(r"^\s*[#;]") + # Log/build output patterns _LOG_PATTERNS = [ re.compile(r"\b(ERROR|FAIL|FAILED|FATAL|CRITICAL)\b", re.IGNORECASE), @@ -181,12 +195,19 @@ def detect_content_type(content: str) -> DetectionResult: if tabular_result and tabular_result.confidence >= 0.6: return tabular_result - # 7. Check for source code + # 7. Check for structured config (YAML/TOML/INI). Runs after tabular so + # delimited data keeps its claim, and before code so config files with + # code-ish lines route to the structure-aware config compressor. + config_result = _try_detect_structured_config(content) + if config_result and config_result.confidence >= 0.6: + return config_result + + # 8. Check for source code code_result = _try_detect_code(content) if code_result and code_result.confidence >= 0.5: return code_result - # 8. Fallback to plain text + # 9. Fallback to plain text return DetectionResult(ContentType.PLAIN_TEXT, 0.5, {}) @@ -581,6 +602,128 @@ def _try_detect_tabular(content: str) -> DetectionResult | None: return _try_detect_delimited(lines) +def _try_parse_toml(content: str) -> bool: + """True if `content` parses as TOML (stdlib tomllib, or the tomli backport).""" + try: + import tomllib + except ModuleNotFoundError: # pragma: no cover - Python < 3.11 + try: + import tomli as tomllib # type: ignore[no-redef] + except ModuleNotFoundError: + return False + try: + tomllib.loads(content) + return True + except Exception: + return False + + +def _parse_config_flavor(content: str) -> str | None: + """Disambiguate `[section]`-shaped config: TOML first, then INI. + + Both flavors share the section-header line shape; only the stdlib parsers + can tell them apart reliably. Returns "toml", "ini", or None when neither + parser accepts the content (then it is not claimed as config at all). + """ + if len(content) > 1_000_000: + return None + if _try_parse_toml(content): + return "toml" + import configparser + + parser = configparser.ConfigParser(interpolation=None, strict=False) + try: + parser.read_string(content) + except Exception: + return None + return "ini" if parser.sections() else None + + +def _try_detect_structured_config(content: str) -> DetectionResult | None: + """Try to detect structured config content (YAML, TOML, INI). + + TOML/INI claims are parser-confirmed (stdlib), so they carry high + confidence. YAML has no stdlib parser, so its claim is heuristic: + key/list/document-marker line share plus a structure signal, guarded + against prose and markdown front-matter. + """ + head = content.lstrip()[:1] + if not head or head in "{<": + # JSON objects and markup are never config; JSON arrays and real + # TOML/INI `[section]` headers disambiguate below. + return None + + lines = content.split("\n")[:200] + non_empty = [ln for ln in lines if ln.strip()] + if len(non_empty) < 3: + return None + # Comment lines are neutral: excluded from the line-share ratio so + # comment-heavy configs and #-heading markdown don't skew it either way. + body = [ln for ln in non_empty if not _CONFIG_COMMENT_RE.match(ln)] + if len(body) < 3: + return None + + # TOML / INI: require a section header plus assignment-dominant body, + # then let the stdlib parsers confirm and disambiguate. + sections = sum(1 for ln in body if _CONFIG_SECTION_RE.match(ln)) + if sections >= 1: + assigns = sum(1 for ln in body if _TOML_ASSIGN_RE.match(ln) or _INI_ASSIGN_RE.match(ln)) + if assigns >= 2 and (sections + assigns) / len(body) >= 0.6: + flavor = _parse_config_flavor(content) + if flavor is not None: + share = (sections + assigns) / len(body) + return DetectionResult( + ContentType.STRUCTURED_CONFIG, + min(0.95, 0.7 + share * 0.25), + {"flavor": flavor, "sections": sections, "assignments": assigns}, + ) + + # Markdown front-matter guard: a `---` fence closed within 60 lines and + # followed by non-YAML content is a markdown document, not standalone YAML. + if lines and lines[0].strip() == "---": + for idx in range(1, min(len(lines), 60)): + if lines[idx].strip() in ("---", "..."): + tail = [ln for ln in lines[idx + 1 :] if ln.strip()] + tail_yaml = sum( + 1 for ln in tail if _YAML_KEY_RE.match(ln) or _YAML_LIST_RE.match(ln) + ) + if tail and tail_yaml / len(tail) < 0.3: + return None + break + + # YAML heuristic. + yaml_keys = sum(1 for ln in body if _YAML_KEY_RE.match(ln)) + yaml_lists = sum(1 for ln in body if _YAML_LIST_RE.match(ln) and not _YAML_KEY_RE.match(ln)) + doc_marks = sum(1 for ln in body if _YAML_DOC_RE.match(ln.strip())) + if yaml_keys < 3: + return None + share = (yaml_keys + yaml_lists + doc_marks) / len(body) + if share < 0.6: + return None + # Prose guards: config lines are short field-ish tuples, prose reads like + # sentences (mirrors _looks_like_prose for delimited data). + enders = sum(1 for ln in body if ln.rstrip().endswith((".", "!", "?"))) + if enders / len(body) >= 0.5: + return None + avg_words = sum(len(ln.split()) for ln in body) / len(body) + if avg_words > 8: + return None + # Structure signal: nested indentation, a document marker, or a real list. + indents = { + len(ln) - len(ln.lstrip(" ")) + for ln in body + if _YAML_KEY_RE.match(ln) or _YAML_LIST_RE.match(ln) + } + if len(indents) < 2 and doc_marks == 0 and yaml_lists < 3: + return None + + return DetectionResult( + ContentType.STRUCTURED_CONFIG, + min(0.9, 0.55 + share * 0.35), + {"flavor": "yaml", "keys": yaml_keys, "list_items": yaml_lists}, + ) + + def _try_detect_code(content: str) -> DetectionResult | None: """Try to detect source code and identify language.""" lines = content.split("\n")[:100] # Check first 100 lines diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 3f2da8023..965183656 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -62,7 +62,13 @@ from ..tokenizer import Tokenizer from ..tokenizers.estimator import EstimatingTokenCounter from . import mixed_content as _mixed_content from .base import Transform -from .content_detector import ContentType, DetectionResult, _try_detect_log, _try_detect_search +from .content_detector import ( + ContentType, + DetectionResult, + _try_detect_log, + _try_detect_search, + _try_detect_structured_config, +) from .content_detector import detect_content_type as _regex_detect_content_type from .error_detection import content_has_strong_error_indicators from .mixed_content import ContentSection, mixed_content_indicators @@ -565,6 +571,15 @@ def _detect_content(content: str) -> DetectionResult: if override is not None: return override + # Config misroute guard (native/magika path): magika classifies YAML/TOML/ + # INI as SourceCode, which routes to the (default-disabled) code path and + # degrades to prose compression. When the structural config detector + # positively claims the payload, trust it over the SourceCode verdict. + if content_type is ContentType.SOURCE_CODE: + config_override = _try_detect_structured_config(content) + if config_override is not None and config_override.confidence >= 0.7: + return config_override + if content_type is ContentType.PLAIN_TEXT: regex_result = _regex_detect_content_type(content) if regex_result.content_type is not ContentType.PLAIN_TEXT: @@ -923,6 +938,7 @@ class CompressionStrategy(Enum): DIFF = "diff" HTML = "html" TABULAR = "tabular" + CONFIG = "config" MIXED = "mixed" PASSTHROUGH = "passthrough" @@ -1032,6 +1048,7 @@ class ContentRouterConfig: enable_search_compressor: Enable search result compression. enable_log_compressor: Enable build/test log compression. enable_tabular_compressor: Enable CSV/TSV/markdown-table compression. + enable_config_compressor: Enable YAML/TOML/INI config compression. enable_image_optimizer: Enable image token optimization. prefer_code_aware_for_code: Use CodeAware over Kompress for code. min_section_tokens: Minimum tokens for a section to compress. @@ -1048,6 +1065,7 @@ class ContentRouterConfig: enable_search_compressor: bool = True enable_log_compressor: bool = True enable_tabular_compressor: bool = True # CSV/TSV/markdown tables via SmartCrusher + enable_config_compressor: bool = True # YAML/TOML/INI structural compression enable_html_extractor: bool = True # HTML content extraction enable_image_optimizer: bool = True # Image token optimization @@ -1320,6 +1338,7 @@ class ContentRouter(Transform): self._diff_compressor: Any = None self._html_extractor: Any = None self._tabular_compressor: Any = None + self._config_compressor: Any = None self._kompress: Any = None # Stage B relevance split (lazy; None until first use, sentinel-checked # via _relevance_scorer_tried so a failed load isn't retried per call). @@ -1833,6 +1852,7 @@ class ContentRouter(Transform): ContentType.GIT_DIFF: CompressionStrategy.DIFF, ContentType.HTML: CompressionStrategy.HTML, ContentType.TABULAR: CompressionStrategy.TABULAR, + ContentType.STRUCTURED_CONFIG: CompressionStrategy.CONFIG, ContentType.PLAIN_TEXT: CompressionStrategy.TEXT, } @@ -1991,6 +2011,7 @@ class ContentRouter(Transform): CompressionStrategy.SEARCH: "search", CompressionStrategy.LOG: "log", CompressionStrategy.DIFF: "diff", + CompressionStrategy.CONFIG: "config", }.get(strategy) order = ([primary] if primary else []) + [ k for k in ("search", "paths", "log", "diff", "text") if k != primary @@ -2274,6 +2295,18 @@ class ContentRouter(Transform): ) decision_reason = "tabular_compressor" + elif strategy == CompressionStrategy.CONFIG: + if self.config.enable_config_compressor: + compressor = self._get_config_compressor() + if compressor: + compressor_name = type(compressor).__name__ + result = compressor.compress(content, context=context, bias=bias) + compressed, compressed_tokens = ( + result.compressed, + len(result.compressed.split()), + ) + decision_reason = "config_compressor" + elif strategy == CompressionStrategy.DIFF: compressor = self._get_diff_compressor() if compressor: @@ -2325,6 +2358,7 @@ class ContentRouter(Transform): CompressionStrategy.SMART_CRUSHER, CompressionStrategy.CODE_AWARE, CompressionStrategy.TABULAR, + CompressionStrategy.CONFIG, } fallback_no_savings = compressed == content or compressed_tokens >= original_tokens if fallback_eligible_strategy and fallback_no_savings: @@ -2655,6 +2689,7 @@ class ContentRouter(Transform): ContentType.GIT_DIFF: CompressionStrategy.DIFF, ContentType.HTML: CompressionStrategy.HTML, ContentType.TABULAR: CompressionStrategy.TABULAR, + ContentType.STRUCTURED_CONFIG: CompressionStrategy.CONFIG, ContentType.PLAIN_TEXT: CompressionStrategy.TEXT, } return mapping.get(content_type, self.config.fallback_strategy) @@ -2669,6 +2704,7 @@ class ContentRouter(Transform): CompressionStrategy.DIFF: ContentType.GIT_DIFF, CompressionStrategy.HTML: ContentType.HTML, CompressionStrategy.TABULAR: ContentType.TABULAR, + CompressionStrategy.CONFIG: ContentType.STRUCTURED_CONFIG, CompressionStrategy.TEXT: ContentType.PLAIN_TEXT, CompressionStrategy.KOMPRESS: ContentType.PLAIN_TEXT, CompressionStrategy.PASSTHROUGH: ContentType.PLAIN_TEXT, @@ -2893,6 +2929,19 @@ class ContentRouter(Transform): logger.debug("TabularCompressor not available") return self._tabular_compressor + def _get_config_compressor(self) -> Any: + """Get ConfigCompressor (lazy load).""" + if self._config_compressor is None: + try: + from .config_compressor import ConfigCompressor, ConfigCompressorConfig + + self._config_compressor = ConfigCompressor( + ConfigCompressorConfig(enable_ccr=self.config.ccr_inject_marker) + ) + except ImportError: # pragma: no cover - defensive; module is pure stdlib + logger.debug("ConfigCompressor not available") + return self._config_compressor + def _get_diff_compressor(self) -> Any: """Get DiffCompressor (lazy load). Rust-only — Python implementation retired in Stage 3b. The wheel (`headroom._core`) is a hard import. diff --git a/headroom/transforms/lossless_compaction.py b/headroom/transforms/lossless_compaction.py index a8f60c364..573f09aee 100644 --- a/headroom/transforms/lossless_compaction.py +++ b/headroom/transforms/lossless_compaction.py @@ -21,6 +21,8 @@ __all__ = [ "collapse_runs", "expand_runs", "is_run_collapsed", + "fold_repeated_blocks", + "unfold_repeated_blocks", "search_heading", "search_unheading", "diff_strip_index", @@ -34,6 +36,20 @@ _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\)$") +# multi-line block back-reference marker. Length and distance (both in lines, +# in ORIGINAL coordinates) are captured for exact inversion: everything before +# a marker expands to the exact original prefix, so `distance` lines back in +# the expanded output is the block's first occurrence. +_BLOCK_MARKER_RE = re.compile(r"^\.\.\. \(repeats (\d+) lines from (\d+) lines back\)$") + +# fold_repeated_blocks search bounds: minimum/maximum block length worth a +# marker, candidate anchors per line, and an input size cap so the scan stays +# negligible on huge payloads. +_FOLD_MIN_BLOCK = 3 +_FOLD_MAX_BLOCK = 64 +_FOLD_MAX_CANDIDATES = 8 +_FOLD_MAX_LINES = 20_000 + # 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. @@ -127,6 +143,79 @@ def is_run_collapsed(text: str) -> bool: return False +def fold_repeated_blocks(text: str) -> str: + """Collapse multi-line blocks that repeat earlier content into back-refs. + + The block-level generalization of :func:`collapse_runs`: a run of K + consecutive lines (K >= 3) that exactly reproduces K lines seen D lines + earlier becomes ``... (repeats K lines from D lines back)``. The repeats + need not be adjacent, which is what config payloads actually look like — + k8s container stanzas repeat with only the ``name:`` line differing, so + their identical tails fold even though no two whole stanzas are + consecutive. Coordinates are in original lines: the fold is only taken + when the block does not overlap its anchor (K <= D), so on expansion the + referenced region is always already reconstructed. + Exact inverse: :func:`unfold_repeated_blocks`. + """ + lines, had_trailing = _split_keep_trailing(text) + n = len(lines) + if n < _FOLD_MIN_BLOCK * 2 or n > _FOLD_MAX_LINES: + return text + positions: dict[str, list[int]] = {} + out: list[str] = [] + i = 0 + while i < n: + best_len = 0 + best_dist = 0 + for q in reversed(positions.get(lines[i], ())): + max_len = min(_FOLD_MAX_BLOCK, n - i, i - q) + length = 0 + while length < max_len and lines[q + length] == lines[i + length]: + length += 1 + if length > best_len: + best_len = length + best_dist = i - q + if best_len >= _FOLD_MIN_BLOCK: + marker = f"... (repeats {best_len} lines from {best_dist} lines back)" + block_chars = sum(len(lines[i + k]) + 1 for k in range(best_len)) + if block_chars > len(marker) + 1: + out.append(marker) + for k in range(best_len): + _remember(positions, lines[i + k], i + k) + i += best_len + continue + _remember(positions, lines[i], i) + out.append(lines[i]) + i += 1 + return _join(out, had_trailing) + + +def _remember(positions: dict[str, list[int]], line: str, index: int) -> None: + """Track recent original positions of `line`, bounded per distinct line.""" + bucket = positions.setdefault(line, []) + bucket.append(index) + if len(bucket) > _FOLD_MAX_CANDIDATES: + del bucket[0] + + +def unfold_repeated_blocks(text: str) -> str: + """Exact inverse of :func:`fold_repeated_blocks`.""" + lines, had_trailing = _split_keep_trailing(text) + if not lines: + return text + out: list[str] = [] + for line in lines: + m = _BLOCK_MARKER_RE.match(line) + if m: + length, dist = int(m.group(1)), int(m.group(2)) + start = len(out) - dist + if start >= 0 and length <= dist: + out.extend(out[start : start + length]) + continue + out.append(line) + return _join(out, had_trailing) + + def search_heading(text: str) -> str: """Convert grep ``path:line:content`` rows into ripgrep --heading form. @@ -279,7 +368,7 @@ def _smaller(candidate: str, original: str) -> bool: 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 + ``kind`` in {'log', 'search', 'diff', 'text', 'config'}. 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 @@ -322,6 +411,14 @@ def compact_lossless(content: str, kind: str) -> str: if expand_runs(candidate) != content: return content return candidate if _smaller(candidate, content) else content + + if kind == "config": + # Structured config (YAML/TOML/INI): single-line runs first, then + # repeated multi-line stanzas. Inverse applies in reverse order. + candidate = fold_repeated_blocks(collapse_runs(content)) + if expand_runs(unfold_repeated_blocks(candidate)) != content: + return content + return candidate if _smaller(candidate, content) else content except Exception: return content return content diff --git a/tests/test_transforms_config_compressor.py b/tests/test_transforms_config_compressor.py new file mode 100644 index 000000000..96d778c6a --- /dev/null +++ b/tests/test_transforms_config_compressor.py @@ -0,0 +1,590 @@ +"""Tests for structured-config (YAML/TOML/INI) detection and compression. + +Covers the detection heuristics (including prose / markdown-front-matter +non-claims), the reversible block-fold primitives in lossless_compaction, +the two-tier ConfigCompressor, and the router wiring. +""" + +from __future__ import annotations + +import pytest + +from headroom.parser import CCR_RETRIEVAL_MARKER_RE +from headroom.transforms.config_compressor import ( + ConfigCompressor, + ConfigCompressorConfig, +) +from headroom.transforms.content_detector import ( + ContentType, + _try_detect_structured_config, + detect_content_type, +) +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, +) +from headroom.transforms.lossless_compaction import ( + compact_lossless, + expand_runs, + fold_repeated_blocks, + unfold_repeated_blocks, +) + +# Reusable fixtures ---------------------------------------------------------- + +K8S_MANIFEST = """apiVersion: apps/v1 +kind: Deployment +metadata: + name: web + labels: + app: web +spec: + replicas: 3 + template: + spec: + containers: + - name: web-1 + image: nginx:1.25 + resources: + limits: + cpu: 500m + memory: 512Mi + requests: + cpu: 250m + memory: 256Mi + - name: web-2 + image: nginx:1.25 + resources: + limits: + cpu: 500m + memory: 512Mi + requests: + cpu: 250m + memory: 256Mi + - name: web-3 + image: nginx:1.25 + resources: + limits: + cpu: 500m + memory: 512Mi + requests: + cpu: 250m + memory: 256Mi +""" + +PYPROJECT_TOML = """# Build configuration +[package] +name = "demo" +version = "0.1.0" +edition = "2021" + +[dependencies] +serde = "1.0" +tokio = "1.38" + +[profile.release] +opt-level = 3 +lto = true +""" + +INI_CONFIG = """[server] +host = 127.0.0.1 +port = 8080 + +; connection tuning +[logging] +level = INFO +file = /var/log/app.log + +[auth] +enabled = true +provider = ldap +""" + +PROSE = """Note: this document describes the deployment process. +First, you should review the configuration carefully before applying it. +The rollout takes several minutes to complete in most environments. +If anything goes wrong, roll back to the previous release immediately. +Contact the on-call engineer when the dashboard shows sustained errors. +""" + +FRONT_MATTER_DOC = """--- +title: My Post +tags: + - a + - b +--- + +# Heading + +This is a markdown document body with prose that goes on and on. +More prose here, explaining things in complete English sentences. +And a final line of text to round out the document nicely today. +""" + + +# Detection ------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "content,flavor", + [ + (K8S_MANIFEST, "yaml"), + (PYPROJECT_TOML, "toml"), + (INI_CONFIG, "ini"), + ], +) +def test_detects_structured_config(content: str, flavor: str) -> None: + result = detect_content_type(content) + assert result.content_type is ContentType.STRUCTURED_CONFIG + assert result.metadata["flavor"] == flavor + assert result.confidence >= 0.7 + + +@pytest.mark.parametrize( + "content", + [ + PROSE, + FRONT_MATTER_DOC, + # grep output: colon shapes must stay SEARCH_RESULTS + "src/main.py:42:def process():\nsrc/util.py:10:import os\nsrc/x.py:5:return 1", + # CSV keeps its tabular claim + "name,age,city\nalice,30,nyc\nbob,25,sf\ncarol,35,la", + # JSON array of dicts stays with SmartCrusher + '[{"id": 1, "name": "a"}, {"id": 2, "name": "b"}]', + # JSON object bodies are never claimed as config + '{\n "key": "value",\n "other": "thing",\n "third": "entry"\n}', + # Python code with colon-ended keywords + "import os\n\n\ndef main():\n if True:\n return os.name\n", + # too short + "key: value", + ], +) +def test_does_not_claim_non_config(content: str) -> None: + result = detect_content_type(content) + assert result.content_type is not ContentType.STRUCTURED_CONFIG + + +def test_multi_document_yaml_stream_is_claimed() -> None: + doc = "---\nname: a\nimage: x:1\nports:\n - 80\n---\nname: b\nimage: y:2\nports:\n - 443\n" + result = _try_detect_structured_config(doc) + assert result is not None + assert result.metadata["flavor"] == "yaml" + + +def test_toml_beats_ini_when_both_parse() -> None: + # Valid TOML with quoted strings parses under tomllib and wins. + result = _try_detect_structured_config('[a]\nx = "1"\ny = "2"\n[b]\nz = "3"\n') + assert result is not None + assert result.metadata["flavor"] == "toml" + + +def test_ini_when_toml_rejects_bare_values() -> None: + # Unquoted strings are invalid TOML but fine for configparser. + result = _try_detect_structured_config("[a]\nx = hello\ny = world\n[b]\nz = there\n") + assert result is not None + assert result.metadata["flavor"] == "ini" + + +def test_flat_yaml_without_structure_not_claimed() -> None: + # Flat key: value lines with one indent level, no docs, no lists: too + # ambiguous with prose-ish "Key: value" notes to claim. + result = _try_detect_structured_config("alpha: 1\nbeta: 2\ngamma: 3\n") + assert result is None + + +# fold_repeated_blocks / unfold_repeated_blocks ------------------------------- + + +def test_fold_round_trip_k8s() -> None: + folded = fold_repeated_blocks(K8S_MANIFEST) + assert len(folded) < len(K8S_MANIFEST) + assert "lines back)" in folded + assert unfold_repeated_blocks(folded) == K8S_MANIFEST + + +def test_fold_handles_consecutive_identical_blocks() -> None: + block = "alpha: value-one\nbeta: value-two\ngamma: value-three\n" + text = block * 4 + folded = fold_repeated_blocks(text) + assert unfold_repeated_blocks(folded) == text + assert len(folded) < len(text) + + +def test_fold_skips_short_blocks() -> None: + text = "x: 1\ny: 2\nx: 1\ny: 2\n" # repeats are only 2 lines long + assert fold_repeated_blocks(text) == text + + +def test_fold_skips_when_marker_not_smaller() -> None: + # Three repeated single-char lines: folding would cost more than it saves. + text = "a\nb\nc\na\nb\nc\n" + assert fold_repeated_blocks(text) == text + + +def test_unfold_leaves_invalid_marker_untouched() -> None: + text = "line one\n... (repeats 5 lines from 2 lines back)\n" + # length > distance: not a marker fold_repeated_blocks could have emitted. + assert unfold_repeated_blocks(text) == text + + +def test_compact_lossless_config_verifies_round_trip() -> None: + compacted = compact_lossless(K8S_MANIFEST, "config") + assert len(compacted) < len(K8S_MANIFEST) + assert expand_runs(unfold_repeated_blocks(compacted)) == K8S_MANIFEST + + +def test_compact_lossless_config_bails_on_marker_collision() -> None: + # Content that already contains a marker-shaped line cannot round-trip, + # so the self-check must return the original unchanged. + lines = [f"item: {i}" for i in range(3)] + block = "\n".join(lines) + text = block + "\n... (repeats 3 lines from 3 lines back)\n" + block + "\n" + block + "\n" + assert compact_lossless(text, "config") == text + + +# ConfigCompressor ------------------------------------------------------------ + + +def test_tier1_lossless_only_no_marker() -> None: + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=False)) + result = comp.compress(K8S_MANIFEST) + assert result.was_modified + assert len(result.compressed) < len(K8S_MANIFEST) + assert not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + assert result.flavor == "yaml" + assert result.ccr_hash is None + + +def test_tier2_elides_comments_behind_ccr_marker(monkeypatch) -> None: + from headroom.cache.compression_store import CompressionStore + + store = CompressionStore() + monkeypatch.setattr("headroom.cache.compression_store.get_compression_store", lambda: store) + commented = PYPROJECT_TOML.replace( + 'serde = "1.0"', + "# pinned for CVE-2024-0001; do not bump without checking the advisory\n" + "# see https://example.com/advisories/CVE-2024-0001 for details\n" + "# owner: platform-team, revisit after the 2.x migration lands\n" + 'serde = "1.0"', + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(commented) + assert result.was_modified + assert result.lines_elided > 0 + assert result.ccr_hash is not None + assert CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + assert "CVE-2024-0001" not in result.compressed + # The marker hash must resolve to the byte-exact original. + assert store.retrieve(result.ccr_hash).original_content == commented + + +def test_tier2_skipped_for_yaml_block_scalars(monkeypatch) -> None: + from headroom.cache.compression_store import CompressionStore + + monkeypatch.setattr( + "headroom.cache.compression_store.get_compression_store", + lambda: CompressionStore(), + ) + doc = ( + "config:\n" + " script: |\n" + " # this hash line is DATA inside a block scalar\n" + " echo hi\n" + " replicas: 3\n" + " image: nginx\n" + " ports:\n" + " - 80\n" + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(doc) + # The '#' line inside the block scalar must survive verbatim. + assert "# this hash line is DATA" in result.compressed or not result.was_modified + + +def test_tier2_skipped_for_toml_multiline_strings(monkeypatch) -> None: + from headroom.cache.compression_store import CompressionStore + + monkeypatch.setattr( + "headroom.cache.compression_store.get_compression_store", + lambda: CompressionStore(), + ) + doc = '[a]\ntext = """\n# not a comment\n"""\nx = "1"\ny = "2"\n' + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(doc) + assert "# not a comment" in result.compressed or not result.was_modified + + +def test_tier2_ini_keeps_blank_lines(monkeypatch) -> None: + from headroom.cache.compression_store import CompressionStore + + monkeypatch.setattr( + "headroom.cache.compression_store.get_compression_store", + lambda: CompressionStore(), + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(INI_CONFIG) + if result.was_modified: + # Column-0 ';' comment goes; blank section separators stay. + assert "; connection tuning" not in result.compressed + assert "\n\n" in result.compressed + + +def test_store_failure_degrades_to_tier1(monkeypatch) -> None: + class _BrokenStore: + def store(self, *a, **kw): # noqa: ANN002, ANN003 + raise RuntimeError("disk full") + + monkeypatch.setattr( + "headroom.cache.compression_store.get_compression_store", + lambda: _BrokenStore(), + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(K8S_MANIFEST) + # Tier 1 still folds the manifest; nothing was elided. + assert result.was_modified + assert result.ccr_hash is None + assert result.lines_elided == 0 + assert not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + + +def test_non_config_content_passes_through() -> None: + comp = ConfigCompressor() + result = comp.compress(PROSE) + assert not result.was_modified + assert result.compressed == PROSE + assert result.flavor == "unknown" + + +def test_no_savings_returns_original() -> None: + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=False)) + result = comp.compress(PYPROJECT_TOML) # compact already; nothing to fold + assert not result.was_modified + assert result.compressed == PYPROJECT_TOML + + +def test_compression_ratio_zero_for_empty_original() -> None: + from headroom.transforms.config_compressor import ConfigCompressionResult + + result = ConfigCompressionResult( + compressed="", original="", was_modified=False, flavor="unknown" + ) + assert result.compression_ratio == 0.0 + + +# Tier 3: TOML array-of-tables → SmartCrusher csv-schema ----------------------- + + +def _mypy_overrides_toml(n: int) -> str: + """A pyproject-style TOML whose ``[[overrides]]`` array dominates the file.""" + records = "\n\n".join( + f"[[tool.mypy.overrides]]\n" + f'module = "pkg.sub{i}.mod"\n' + f"ignore_missing_imports = true\n" + f"disallow_untyped_defs = false" + for i in range(n) + ) + return "[tool.mypy]\nstrict = true\n\n" + records + + +def _use_fresh_store(monkeypatch): # noqa: ANN001, ANN201 + from headroom.cache.compression_store import CompressionStore + + store = CompressionStore() + monkeypatch.setattr("headroom.cache.compression_store.get_compression_store", lambda: store) + return store + + +def test_tier3_folds_toml_array_of_tables(monkeypatch) -> None: + store = _use_fresh_store(monkeypatch) + toml = _mypy_overrides_toml(25) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(toml) + + assert result.was_modified + assert result.strategy == "config_schema_fold" + assert result.flavor == "toml" + assert len(result.compressed) < len(toml) // 2 # keys folded → big win + assert 0.0 < result.compression_ratio < 0.5 + # The repeated per-record key appears once in the schema, not 25 times. + assert result.compressed.count("ignore_missing_imports") == 1 + assert CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + # The marker hash resolves to the byte-exact original. + assert result.ccr_hash is not None + assert store.retrieve(result.ccr_hash).original_content == toml + + +def test_tier3_disabled_in_lossless_mode() -> None: + # enable_ccr off (lossless) → no fold, no marker, only reversible tiers. + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=False)) + result = comp.compress(_mypy_overrides_toml(25)) + assert result.strategy != "config_schema_fold" + assert not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + + +def test_tier3_flag_off_keeps_text_tiers(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True, enable_schema_fold=False)) + result = comp.compress(_mypy_overrides_toml(25)) + assert result.strategy != "config_schema_fold" + + +def test_tier3_skips_non_toml_flavor(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + # A YAML list-of-mappings is structurally similar but must not be bridged + # (no stdlib YAML parser is a dependency). + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(K8S_MANIFEST) + assert result.strategy != "config_schema_fold" + + +def test_tier3_skips_toml_without_array_of_tables(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + # Valid TOML flavor, but no `[[ ]]` → SmartCrusher has no array to fold. + result = comp.compress(PYPROJECT_TOML) + assert result.strategy != "config_schema_fold" + + +def test_tier3_declines_small_array_as_passthrough(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + # Three long-valued records: SmartCrusher returns `passthrough` (the folded + # keys don't outweigh the unique values), so Tier 3 declines. + toml = "version = 3\n\n" + "\n\n".join( + f'[[package]]\nname = "crate-{i}"\nversion = "1.2.{i}"\nchecksum = "{i:064x}"' + for i in range(3) + ) + result = comp.compress(toml) + assert result.strategy != "config_schema_fold" + + +def test_tier3_store_failure_keeps_text_tiers(monkeypatch) -> None: + class _BrokenStore: + def store(self, *a, **kw): # noqa: ANN002, ANN003 + raise RuntimeError("disk full") + + monkeypatch.setattr( + "headroom.cache.compression_store.get_compression_store", + lambda: _BrokenStore(), + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(_mypy_overrides_toml(25)) + # Can't store the original → never emit the lossy fold. + assert result.strategy != "config_schema_fold" + assert not CCR_RETRIEVAL_MARKER_RE.search(result.compressed) + + +def test_tier3_rejected_when_marker_overhead_exceeds_savings(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + # An unreachable-in-practice savings floor forces the final gate to reject + # even a genuine fold, exercising the guard. + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True, min_savings_chars=10**6)) + result = comp.compress(_mypy_overrides_toml(25)) + assert result.strategy != "config_schema_fold" + + +def test_schema_fold_bails_on_unparseable_toml(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + import headroom.transforms.config_compressor as mod + + monkeypatch.setattr(mod, "_load_toml", lambda _content: None) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + # Detector still flags it TOML, but the fold parser bails → text tiers. + assert comp._schema_fold(_mypy_overrides_toml(25), "toml", "", 1.0) is None + + +def test_schema_fold_bails_on_non_serializable_value(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + import headroom.transforms.config_compressor as mod + + monkeypatch.setattr(mod, "_load_toml", lambda _content: {"t": [{"v": {1, 2}}, {"v": {3, 4}}]}) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + # A value tomllib could never emit but that we can't represent faithfully. + assert comp._schema_fold("[[t]]\n", "toml", "", 1.0) is None + + +def test_load_toml_returns_none_on_invalid() -> None: + from headroom.transforms.config_compressor import _load_toml + + assert _load_toml('[[x]]\nk = "unterminated') is None + assert _load_toml('[[x]]\nk = "ok"') == {"x": [{"k": "ok"}]} + + +def test_json_default_renders_dates_and_rejects_others() -> None: + import datetime as dt + + from headroom.transforms.config_compressor import _json_default + + assert _json_default(dt.date(2026, 7, 4)) == "2026-07-04" + assert _json_default(dt.datetime(2026, 7, 4, 9, 30)) == "2026-07-04T09:30:00" + with pytest.raises(TypeError): + _json_default(object()) + + +def test_tier3_toml_with_datetime_folds(monkeypatch) -> None: + _use_fresh_store(monkeypatch) + # TOML datetimes serialize through _json_default; the fold still applies. + toml = "\n\n".join( + f'[[event]]\nname = "e{i}"\nwhen = 2026-07-04T09:30:00\nactive = true' for i in range(20) + ) + comp = ConfigCompressor(ConfigCompressorConfig(enable_ccr=True)) + result = comp.compress(toml) + assert result.strategy == "config_schema_fold" + assert "2026-07-04T09:30:00" in result.compressed + + +# Router wiring --------------------------------------------------------------- + + +def test_router_maps_structured_config_to_config_strategy() -> None: + router = ContentRouter() + assert ( + router._strategy_from_detection_type(ContentType.STRUCTURED_CONFIG) + is CompressionStrategy.CONFIG + ) + assert ( + router._content_type_from_strategy(CompressionStrategy.CONFIG) + is ContentType.STRUCTURED_CONFIG + ) + + +def test_router_lazy_getter_mirrors_ccr_setting() -> None: + router = ContentRouter(ContentRouterConfig(lossless=True)) + compressor = router._get_config_compressor() + assert compressor is not None + assert compressor.config.enable_ccr is False # lossless forces markers off + + +def test_router_compresses_k8s_manifest_end_to_end() -> None: + router = ContentRouter() + result = router.compress(K8S_MANIFEST) + assert result.compressed != K8S_MANIFEST or result.strategy_used in ( + CompressionStrategy.CONFIG, + CompressionStrategy.PASSTHROUGH, + ) + # Whatever the gates decide, the output must never be a lossy mangle: + # either untouched or a reversible fold of the manifest. + if result.compressed != K8S_MANIFEST: + assert expand_runs(unfold_repeated_blocks(result.compressed)) == K8S_MANIFEST + + +def test_router_lossless_mode_uses_lossless_config_label() -> None: + router = ContentRouter(ContentRouterConfig(lossless=True)) + compressed, _tokens, chain = router._apply_strategy_to_content( + K8S_MANIFEST, CompressionStrategy.CONFIG, context="" + ) + assert chain == ["lossless_config"] + assert expand_runs(unfold_repeated_blocks(compressed)) == K8S_MANIFEST + assert not CCR_RETRIEVAL_MARKER_RE.search(compressed) + + +def test_router_disabled_flag_skips_config_compressor() -> None: + router = ContentRouter(ContentRouterConfig(enable_config_compressor=False)) + compressed, _tokens, _chain = router._apply_strategy_to_content( + K8S_MANIFEST, CompressionStrategy.CONFIG, context="" + ) + # Falls through to the unified fallback path; must not crash and must + # not claim CONFIG did work it didn't do. + assert isinstance(compressed, str)