chore(transforms): remove twotrim files\n\n- Remove two_trim.py implementation file\n- Remove test_two_trim.py test file\n- Remove bench_two_trim.py benchmark file\n\nGenerated with [Claude Code](https://claude.ai/code)\nvia [Happy](https://happy.engineering)\n\nCo-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>\nCo-Authored-By: Happy <yesreply@happy.engineering>

benchmarks/bench_two_trim.py

@@ -1,572 +0,0 @@
-"""TwoTrim Benchmark: compression impact across text types and modes.
-
-Compares:
-    - Baseline: no compression
-    - TwoTrim: conservative / balanced / aggressive
-    - Full pipeline (CacheAligner → TwoTrim → RollingWindow)
-
-Measures:
-    - Token savings (absolute + percentage)
-    - Latency (apply time in ms)
-    - Sentence retention (fraction kept)
-    - Edge: does rolling window need fewer drops when TwoTrim runs first?
-
-Run as pytest:
-    pytest benchmarks/bench_two_trim.py -v
-
-Run standalone (prints summary table):
-    python benchmarks/bench_two_trim.py
-"""
-
-from __future__ import annotations
-
-import json
-import random
-import statistics
-import sys
-import time
-from copy import deepcopy
-from dataclasses import dataclass
-
-import pytest
-
-from headroom.config import HeadroomConfig, TwoTrimConfig
-from headroom.tokenizer import Tokenizer
-from headroom.transforms.pipeline import TransformPipeline
-from headroom.transforms.two_trim import TwoTrim
-
-random.seed(42)
-
-
-# ---------------------------------------------------------------------------
-# Tokenizer / fixtures (real tiktoken for accurate measurement)
-# ---------------------------------------------------------------------------
-
-
-def _build_real_tokenizer() -> tuple[Tokenizer, object]:
-    """Build a Tokenizer backed by tiktoken (or a close fallback)."""
-    try:
-        from headroom.tokenizers import TiktokenCounter
-
-        counter = TiktokenCounter(model="gpt-4o")
-    except Exception:
-        from headroom.tokenizers import EstimatingTokenCounter
-
-        counter = EstimatingTokenCounter()
-    return Tokenizer(token_counter=counter, model="gpt-4o"), counter
-
-
-real_tokenizer, _tok_counter = _build_real_tokenizer()
-
-
-# Minimal provider stub that returns the real token counter for any model.
-# Used wherever TransformPipeline needs a provider. Avoids re-instantiating
-# the anonymous type at every callsite.
-class _MockProvider:
-    def get_token_counter(self, model: str):
-        return _tok_counter
-
-
-mock_provider = _MockProvider()
-
-
-# ---------------------------------------------------------------------------
-# Synthetic plain-text scenarios
-# ---------------------------------------------------------------------------
-
-def _filler_sentences(n: int, seed: int = 0) -> list[str]:
-    """Return *n* plain-English sentences with controlled relevance variance.
-
-    Some sentences contain query-relevant keywords; the rest are filler.
-    This ensures BM25/semantic scoring has signal to work with.
-    """
-    random.seed(seed)
-    relevant = [
-        "The database migration requires a rollback strategy before deployment.",
-        "Connection pool exhaustion is the root cause of the timeout errors.",
-        "Index optimization reduced query latency from 800ms to 45ms.",
-        "The primary key constraint was violated during the bulk insert operation.",
-        "Read replicas are falling behind the write primary by several seconds.",
-        "Deadlock detection triggered and rolled back transaction XZ-7291.",
-        "The backup verification confirmed all snapshots are restorable.",
-        "Memory pressure caused the OOM killer to terminate the worker process.",
-        "Thread pool saturation prevented new requests from being accepted.",
-        "The connection string rotated but the old credentials are still cached.",
-    ]
-    filler = [
-        "The weather today is unusually mild for this time of year in the valley.",
-        "Many people enjoy walking in the park during autumn afternoons.",
-        "The annual report from 2023 shows a modest increase in global revenue.",
-        "Birds migrate south when the temperature starts to drop significantly.",
-        "Did you remember to renew your gym membership this month?",
-        "Tomorrow's forecast predicts scattered showers across the entire region.",
-        "The library closes early on Sundays for scheduled maintenance work.",
-        "Some researchers prefer quantitative methods over qualitative approaches.",
-        "Pizza is a popular dish that originated in Italy many centuries ago.",
-        "The committee meets every other Thursday to discuss new policy changes.",
-        "Quantum entanglement puzzled physicists for decades before being resolved.",
-        "Backups should run nightly to prevent catastrophic data loss incidents.",
-        "The new highway interchange opens next month after years of construction.",
-        "Reading books is one of the best ways to expand your vocabulary.",
-        "The coffee shop on the corner serves excellent pastries every morning.",
-        "Regular exercise has been shown to improve cardiovascular health greatly.",
-        "The stock market experienced a minor correction during last week's trading.",
-        "Learning a second language opens up many professional opportunities.",
-        "The local museum is hosting an exhibition on ancient Roman artifacts.",
-        "Volunteering at the shelter is a rewarding way to spend weekend mornings.",
-    ]
-    result: list[str] = []
-    while len(result) < n:
-        pick = random.choice(relevant if random.random() < 0.25 else filler)
-        result.append(pick)
-    return result
-
-
-def _make_user_messages(
-    text_chunks: list[str],
-    system: str = "You are a database reliability engineer.",
-    query: str = "What is causing the database timeout errors?",
-) -> list[dict]:
-    return [
-        {"role": "system", "content": system},
-        {"role": "user", "content": query},
-        *(
-            {"role": "user", "content": chunk}
-            for chunk in text_chunks
-        ),
-    ]
-
-
-def _count_tokens(msgs: list[dict]) -> int:
-    return real_tokenizer.count_messages(msgs)
-
-
-# ---------------------------------------------------------------------------
-# Scenarios (name → (messages, description))
-# ---------------------------------------------------------------------------
-
-SCENARIOS: list[tuple[str, str, list[dict]]] = [
-    (
-        "short-plain",
-        "~2K tokens single user message (50 sentences)",
-        [
-            {"role": "system", "content": "You are a database reliability engineer."},
-            {"role": "user", "content": "What is causing the database timeout errors?"},
-            {"role": "user", "content": " ".join(_filler_sentences(50, seed=10))},
-        ],
-    ),
-    (
-        "medium-plain",
-        "~5K tokens single user message (150 sentences)",
-        [
-            {"role": "system", "content": "You are a database reliability engineer."},
-            {"role": "user", "content": "What is causing the database timeout errors?"},
-            {"role": "user", "content": " ".join(_filler_sentences(150, seed=20))},
-        ],
-    ),
-    (
-        "long-plain",
-        "~15K tokens single user message (450 sentences)",
-        [
-            {"role": "system", "content": "You are a database reliability engineer."},
-            {"role": "user", "content": "What is causing the database timeout errors?"},
-            {"role": "user", "content": " ".join(_filler_sentences(450, seed=30))},
-        ],
-    ),
-    (
-        "long-system-heavy",
-        "Long system prompt + 150-sentence user message",
-        [
-            {
-                "role": "system",
-                "content": (
-                    "You are an expert database reliability engineer with 20 years "
-                    "of experience managing high-traffic PostgreSQL clusters at scale. "
-                    "You specialize in performance tuning, failover strategies, backup "
-                    "planning, and disaster recovery. Always provide actionable advice. "
-                    "Reference specific configuration parameters when possible. "
-                    "Include rollback steps when recommending changes. "
-                    "Flag risks before suggesting production modifications."
-                ),
-            },
-            {"role": "user", "content": "What is causing the database timeout errors?"},
-            {"role": "user", "content": " ".join(_filler_sentences(150, seed=40))},
-        ],
-    ),
-    (
-        "multi-turn",
-        "3-turn conversation with interleaved large user messages",
-        [
-            {"role": "system", "content": "You are a helpful SRE."},
-            {"role": "user", "content": "Check the database health."},
-            {"role": "assistant", "content": "I'll check the connection pool status now."},
-            {"role": "user", "content": " ".join(_filler_sentences(80, seed=50))},
-            {"role": "assistant", "content": "The connection pool shows 45 active connections."},
-            {"role": "user", "content": " ".join(_filler_sentences(80, seed=60))},
-        ],
-    ),
-    (
-        "many-small-messages",
-        "30 small user messages (floor-effect stress test)",
-        [
-            {"role": "system", "content": "You are a helpful assistant."},
-            {"role": "user", "content": "What is causing the database timeout errors?"},
-            *(
-                {"role": "user", "content": s}
-                for s in _filler_sentences(30, seed=80)
-            ),
-        ],
-    ),
-    (
-        "json-mixed",
-        "User text mixed with JSON blocks (should skip JSON)",
-        [
-            {"role": "system", "content": "You are a helpful assistant."},
-            {
-                "role": "user",
-                "content": (
-                    "Here is the analysis:\n```json\n"
-                    + json.dumps(
-                        [
-                            {"id": i, "metric": round(i * 0.01, 2)}
-                            for i in range(100)
-                        ]
-                    )
-                    + "\n```\n"
-                    + " ".join(_filler_sentences(100, seed=70))
-                ),
-            },
-        ],
-    ),
-]
-
-
-# ---------------------------------------------------------------------------
-# Benchmark runner (standalone mode)
-# ---------------------------------------------------------------------------
-
-@dataclass
-class RunResult:
-    name: str
-    tokens_before: int
-    tokens_after: int
-    tokens_saved: int
-    savings_pct: float
-    latency_ms: float
-    transforms_applied: list[str]
-    rolling_drops: int = 0  # messages dropped by RollingWindow (parsed from window_cap:N, pipeline mode only)
-
-
-def _parse_window_cap_drops(transform_str: str) -> int:
-    """Extract drop count from a 'window_cap:N' transform entry."""
-    if transform_str.startswith("window_cap:"):
-        try:
-            return int(transform_str.split(":", 1)[1])
-        except ValueError:
-            return 0
-    return 0
-
-
-def _run_two_trim(
-    messages: list[dict],
-    mode: str,
-    *,
-    runs: int = 5,
-) -> RunResult:
-    if runs < 1:
-        raise ValueError(f"runs must be >= 1, got {runs}")
-    cfg = TwoTrimConfig(
-        enabled=True,
-        mode=mode,  # type: ignore[arg-type]
-        min_tokens_to_compress=50,
-        use_semantic_scoring=False,
-    )
-    t = TwoTrim(cfg)
-    baseline_tokens = _count_tokens(messages)
-
-    latencies: list[float] = []
-    last_result = None
-    for _ in range(runs):
-        t0 = time.perf_counter()
-        last_result = t.apply(deepcopy(messages), real_tokenizer)
-        latencies.append((time.perf_counter() - t0) * 1000)
-
-    avg_tokens_after = last_result.tokens_after  # type: ignore[union-attr]
-    return RunResult(
-        name=f"two_trim:{mode}",
-        tokens_before=baseline_tokens,
-        tokens_after=avg_tokens_after,
-        tokens_saved=max(0, baseline_tokens - avg_tokens_after),
-        savings_pct=(
-            (baseline_tokens - avg_tokens_after) / baseline_tokens * 100
-            if baseline_tokens > 0
-            else 0.0
-        ),
-        latency_ms=statistics.mean(latencies),
-        transforms_applied=last_result.transforms_applied,  # type: ignore[union-attr]
-    )
-
-
-def _run_pipeline_without_two_trim(
-    messages: list[dict],
-    *,
-    runs: int = 5,
-) -> RunResult:
-    """Run the pipeline with TwoTrim *disabled* (baseline pipeline)."""
-    if runs < 1:
-        raise ValueError(f"runs must be >= 1, got {runs}")
-    cfg = HeadroomConfig()
-    cfg.two_trim.enabled = False
-
-    pipeline = TransformPipeline(config=cfg, provider=mock_provider)  # type: ignore[arg-type]
-    baseline_tokens = _count_tokens(messages)
-
-    model_limit = max(baseline_tokens - 100, 1000)
-
-    latencies: list[float] = []
-    last_result = None
-    for _ in range(runs):
-        t0 = time.perf_counter()
-        last_result = pipeline.apply(
-            deepcopy(messages), "gpt-4o", model_limit=model_limit, output_buffer=500,
-        )
-        latencies.append((time.perf_counter() - t0) * 1000)
-
-    tokens_after = last_result.tokens_after  # type: ignore[union-attr]
-    rw_drops = sum(_parse_window_cap_drops(t) for t in last_result.transforms_applied)  # type: ignore[union-attr]
-    return RunResult(
-        name="pipeline (no TwoTrim)",
-        tokens_before=baseline_tokens,
-        tokens_after=tokens_after,
-        tokens_saved=max(0, baseline_tokens - tokens_after),
-        savings_pct=(
-            (baseline_tokens - tokens_after) / baseline_tokens * 100
-            if baseline_tokens > 0
-            else 0.0
-        ),
-        latency_ms=statistics.mean(latencies),
-        transforms_applied=last_result.transforms_applied,  # type: ignore[union-attr]
-        rolling_drops=rw_drops,
-    )
-
-
-def _run_pipeline_with_two_trim(
-    messages: list[dict],
-    mode: str,
-    *,
-    runs: int = 5,
-) -> RunResult:
-    if runs < 1:
-        raise ValueError(f"runs must be >= 1, got {runs}")
-    cfg = HeadroomConfig()
-    cfg.two_trim.enabled = True
-    cfg.two_trim.mode = mode  # type: ignore[assignment]
-    cfg.two_trim.min_tokens_to_compress = 50
-    cfg.two_trim.use_semantic_scoring = False
-
-    pipeline = TransformPipeline(config=cfg, provider=mock_provider)  # type: ignore[arg-type]
-    baseline_tokens = _count_tokens(messages)
-
-    # Force RollingWindow to drop to measure interaction
-    # Use a model_limit that's tight enough for RW to activate
-    model_limit = max(baseline_tokens - 100, 1000)
-
-    latencies: list[float] = []
-    last_result = None
-    for _ in range(runs):
-        t0 = time.perf_counter()
-        last_result = pipeline.apply(
-            deepcopy(messages), "gpt-4o", model_limit=model_limit, output_buffer=500,
-        )
-        latencies.append((time.perf_counter() - t0) * 1000)
-
-    tokens_after = last_result.tokens_after  # type: ignore[union-attr]
-    # Check if RollingWindow ran
-    rw_drops = sum(_parse_window_cap_drops(t) for t in last_result.transforms_applied)  # type: ignore[union-attr]
-    return RunResult(
-        name=f"pipeline({mode})",
-        tokens_before=baseline_tokens,
-        tokens_after=tokens_after,
-        tokens_saved=max(0, baseline_tokens - tokens_after),
-        savings_pct=(
-            (baseline_tokens - tokens_after) / baseline_tokens * 100
-            if baseline_tokens > 0
-            else 0.0
-        ),
-        latency_ms=statistics.mean(latencies),
-        transforms_applied=last_result.transforms_applied,  # type: ignore[union-attr]
-        rolling_drops=rw_drops,
-    )
-
-
-def run_all() -> list[tuple[str, list[RunResult]]]:
-    """Run all benchmarks, return scenario → results."""
-    all_results: list[tuple[str, list[RunResult]]] = []
-    modes = ["conservative", "balanced", "aggressive"]
-
-    for scenario_name, description, messages in SCENARIOS:
-        baseline_tokens = _count_tokens(messages)
-        results: list[RunResult] = [
-            RunResult(
-                name="baseline (no compression)",
-                tokens_before=baseline_tokens,
-                tokens_after=baseline_tokens,
-                tokens_saved=0,
-                savings_pct=0.0,
-                latency_ms=0.0,
-                transforms_applied=[],
-            )
-        ]
-        for mode in modes:
-            results.append(_run_two_trim(messages, mode))
-        results.append(_run_pipeline_without_two_trim(messages))
-        results.append(_run_pipeline_with_two_trim(messages, "balanced"))
-        all_results.append((f"{scenario_name}: {description}", results))
-
-    return all_results
-
-
-def print_table(all_results: list[tuple[str, list[RunResult]]]) -> None:
-    """Pretty-print results as a table."""
-    hdr = (
-        f"{'Scenario':<42} {'Approach':<28} {'Before':>8} {'After':>8} "
-        f"{'Saved':>8} {'%':>7} {'lat ms':>8} {'transforms'}"
-    )
-    sep = "-" * len(hdr)
-    print("\n" + sep)
-    print("  TWO-TRIM BENCHMARK RESULTS")
-    print(sep)
-    print(hdr)
-    print(sep)
-
-    for scenario_label, results in all_results:
-        first = True
-        for r in results:
-            scenario_col = scenario_label if first else ""
-            first = False
-            # Collapse consecutive duplicates to "name xN" form
-            compact: list[str] = []
-            i = 0
-            run = 1
-            transforms_applied = r.transforms_applied
-            while i < len(transforms_applied):
-                if i + 1 < len(transforms_applied) and transforms_applied[i + 1] == transforms_applied[i]:
-                    run += 1
-                    i += 1
-                    continue
-                compact.append(f"{transforms_applied[i]} x{run}" if run > 1 else transforms_applied[i])
-                run = 1
-                i += 1
-            transforms = ", ".join(compact) if compact else "-"
-            if len(transforms) > 60:
-                transforms = transforms[:57] + "..."
-            print(
-                f"{scenario_col:<42} {r.name:<28} {r.tokens_before:>8,} "
-                f"{r.tokens_after:>8,} {r.tokens_saved:>8,} {r.savings_pct:>6.1f}% "
-                f"{r.latency_ms:>7.1f} {transforms}"
-            )
-        print(sep)
-
-
-# ---------------------------------------------------------------------------
-# pytest API
-# ---------------------------------------------------------------------------
-
-class TestTwoTrimBenchmarks:
-    """Benchmark TwoTrim compression across scenarios (pytest-benchmark)."""
-
-    def test_short_plain_baseline(self, benchmark):
-        msgs = SCENARIOS[0][2]
-        t = TwoTrim(TwoTrimConfig(enabled=False))
-        benchmark(t.apply, deepcopy(msgs), real_tokenizer)
-
-    @pytest.mark.parametrize("mode", ["conservative", "balanced", "aggressive"])
-    def test_short_plain_two_trim(self, benchmark, mode: str):
-        msgs = SCENARIOS[0][2]
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode=mode,  # type: ignore[arg-type]
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        result = benchmark(t.apply, deepcopy(msgs), real_tokenizer)
-        assert result.tokens_saved > 0
-
-    @pytest.mark.parametrize("mode", ["conservative", "balanced", "aggressive"])
-    def test_long_plain_two_trim(self, benchmark, mode: str):
-        msgs = SCENARIOS[2][2]
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode=mode,  # type: ignore[arg-type]
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        result = benchmark(t.apply, deepcopy(msgs), real_tokenizer)
-        assert result.tokens_saved > 0
-
-    def test_long_system_heavy_two_trim(self, benchmark):
-        msgs = SCENARIOS[3][2]
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode="balanced",
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        result = benchmark(t.apply, deepcopy(msgs), real_tokenizer)
-        assert result.tokens_saved > 0
-
-    def test_json_mixed_skips_json(self, benchmark):
-        msgs = SCENARIOS[6][2]
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode="aggressive",
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        baseline = _count_tokens(msgs)
-        result = benchmark(t.apply, deepcopy(msgs), real_tokenizer)
-        # JSON portion should not be compressed — savings < 50% of baseline
-        # (the JSON block is large and skipped entirely)
-        assert result.tokens_saved < baseline * 0.5
-
-    def test_pipeline_with_two_trim(self, benchmark, long_conversation_messages):
-        msgs = long_conversation_messages
-        cfg = HeadroomConfig()
-        cfg.two_trim.enabled = True
-        cfg.two_trim.mode = "balanced"  # type: ignore[assignment]
-        cfg.two_trim.min_tokens_to_compress = 50
-        cfg.two_trim.use_semantic_scoring = False
-
-        pipeline = TransformPipeline(config=cfg, provider=mock_provider)  # type: ignore[arg-type]
-        baseline = _count_tokens(msgs)
-
-        result = benchmark(
-            pipeline.apply,
-            deepcopy(msgs),
-            "gpt-4o",
-            model_limit=100_000,
-            output_buffer=2000,
-        )
-        assert result.tokens_before == baseline
-        assert result.tokens_after < result.tokens_before
-        savings_pct = (baseline - result.tokens_after) / baseline * 100
-        assert savings_pct >= 10.0, f"expected >=10% savings, got {savings_pct:.1f}%"
-
-
-# ---------------------------------------------------------------------------
-# Standalone entry point
-# ---------------------------------------------------------------------------
-
-if __name__ == "__main__":
-    print("Running TwoTrim benchmarks (standalone mode)...")
-    results = run_all()
-    print_table(results)
-    print(
-        "\nFor more accurate latency numbers, run via pytest-benchmark:\n"
-        "  pytest benchmarks/bench_two_trim.py -v"
-    )
-    sys.exit(0)

headroom/transforms/two_trim.py

@@ -1,439 +0,0 @@
-"""TwoTrim extractive compression transform.
-
-LongLLMLingua-style text compression:
-1. Split eligible message text into sentences
-2. Score each sentence against the conversational query (semantic or BM25)
-3. Drop low-scoring filler (caps tied to mode + safety floors)
-4. Optionally reorder survivors so the highest-scoring sit at the prompt
-   start and end ("lost-in-the-middle" mitigation)
-
-Complements SmartCrusher/ContentRouter — runs on plain-text user/system
-messages only. Structured content (JSON, code blocks) is skipped so the
-dedicated compressors handle it upstream.
-
-Reference: https://github.com/overseek944/twotrim
-"""
-
-from __future__ import annotations
-
-import logging
-import re
-import time
-from functools import lru_cache
-from typing import Any
-
-from ..config import TransformResult, TwoTrimConfig
-from ..tokenizer import Tokenizer
-from ..utils import deep_copy_messages
-from .base import Transform
-from .observability import current_request_segments
-
-logger = logging.getLogger(__name__)
-
-_SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+(?=[A-Z0-9\"'\(])")
-_CODE_FENCE_RE = re.compile(r"```")
-_JSON_HINT_RE = re.compile(r"^\s*[\{\[]", re.MULTILINE)
-
-# Mode presets — target sentence keep ratio (fraction of sentences retained).
-# The floor (config.min_sentence_keep_ratio) is always respected.
-_MODE_KEEP_RATIO = {
-    "conservative": 0.80,
-    "balanced": 0.60,
-    "aggressive": 0.35,
-}
-
-
-@lru_cache(maxsize=4)
-def _load_sentence_model(model_name: str) -> Any:
-    """Lazy-load sentence-transformers model (~80MB on first call).
-
-    Cached so repeated pipeline invocations share the same model. Returns
-    None if sentence-transformers not installed — caller falls back to BM25.
-    """
-    try:
-        from sentence_transformers import SentenceTransformer  # type: ignore[import-not-found]
-    except ImportError:
-        logger.warning(
-            "sentence-transformers not installed — TwoTrim falling back to BM25. "
-            "Install with: pip install 'headroom[relevance]'"
-        )
-        return None
-    try:
-        return SentenceTransformer(model_name)
-    except Exception as exc:
-        logger.warning("TwoTrim failed to load %s: %s — using BM25", model_name, exc)
-        return None
-
-
-def _split_sentences(text: str) -> list[str]:
-    """Split text into sentences. Conservative regex — no NLP dep.
-
-    Preserves sentence-internal punctuation. Empty/whitespace-only results
-    are filtered out.
-    """
-    if not text or not text.strip():
-        return []
-    parts = _SENTENCE_SPLIT_RE.split(text)
-    return [p.strip() for p in parts if p and p.strip()]
-
-
-def _looks_structured(text: str) -> bool:
-    """True if content looks like JSON, array, or contains code fences."""
-    if not text:
-        return False
-    stripped = text.lstrip()
-    if stripped and stripped[0] in "[{":
-        return True
-    if _CODE_FENCE_RE.search(text):
-        return True
-    # Heuristic: many lines starting with JSON-ish chars
-    matches = _JSON_HINT_RE.findall(text)
-    if len(matches) >= 3:
-        return True
-    return False
-
-
-def _extract_query(messages: list[dict[str, Any]]) -> str:
-    """Pull a scoring query from the conversation — last user message wins.
-
-    Falls back to concatenation of all user messages if only system prompts
-    are present. Returns empty string if no usable query (all-assistant
-    conversation — should not happen in practice).
-    """
-    last_user = ""
-    all_user = []
-    for msg in messages:
-        if msg.get("role") != "user":
-            continue
-        content = msg.get("content")
-        if isinstance(content, str):
-            last_user = content
-            all_user.append(content)
-        elif isinstance(content, list):
-            # OpenAI-style multimodal — concatenate text blocks
-            parts = [
-                b.get("text", "")
-                for b in content
-                if isinstance(b, dict) and b.get("type") in ("text", "input_text")
-            ]
-            joined = " ".join(p for p in parts if p)
-            if joined:
-                last_user = joined
-                all_user.append(joined)
-    if last_user:
-        return last_user
-    return " ".join(all_user) if all_user else ""
-
-
-def _score_bm25(sentences: list[str], query: str) -> list[float]:
-    """Lightweight BM25-ish score using existing headroom BM25Scorer."""
-    if not sentences or not query.strip():
-        return [1.0] * len(sentences)
-    try:
-        from ..relevance.bm25 import BM25Scorer
-
-        scorer = BM25Scorer()
-        scores = scorer.score_batch(sentences, query)
-        return [s.score for s in scores]
-    except Exception as exc:
-        logger.warning("TwoTrim BM25 scoring failed: %s — using word overlap fallback", exc)
-        # Last-resort fallback: word overlap count
-        query_tokens = set(query.lower().split())
-        return [
-            float(sum(1 for w in s.lower().split() if w in query_tokens))
-            for s in sentences
-        ]
-
-
-def _score_semantic(
-    sentences: list[str], query: str, model_name: str
-) -> tuple[list[float], bool]:
-    """Score via cosine similarity of sentence embeddings to query embedding.
-
-    Returns (scores, used_semantic). If model unavailable, caller falls back.
-    """
-    model = _load_sentence_model(model_name)
-    if model is None or not sentences:
-        return [], False
-    try:
-        import numpy as np  # type: ignore[import-not-found]
-
-        all_vecs = model.encode([query] + sentences, normalize_embeddings=True)
-        query_vec = np.asarray(all_vecs[0])
-        sent_vecs = np.asarray(all_vecs[1:])
-        # cosine sim (already L2-normalized)
-        sims = (sent_vecs @ query_vec).tolist()
-        return sims, True
-    except Exception as exc:
-        logger.warning("TwoTrim semantic scoring failed: %s — using BM25", exc)
-        return [], False
-
-
-def _select_and_reorder(
-    sentences: list[str],
-    scores: list[float],
-    keep_ratio: float,
-    min_keep_ratio: float,
-    max_drop_ratio: float,
-    reorder: bool,
-) -> list[str]:
-    """Select top sentences by score, optionally reorder for attention.
-
-    Guarantees:
-    - Always keep at least ceil(N * min_keep_ratio) sentences
-    - Never drop more than floor(N * max_drop_ratio) sentences
-    """
-    n = len(sentences)
-    if n <= 1:
-        return sentences
-
-    # Target count to keep
-    target_keep = max(1, round(n * keep_ratio))
-    floor_keep = max(1, int(n * min_keep_ratio + 0.999))  # ceil
-    max_droppable = int(n * max_drop_ratio)
-    min_keep_from_cap = n - max_droppable
-
-    keep_count = max(target_keep, floor_keep, min_keep_from_cap)
-    keep_count = min(keep_count, n)
-
-    if keep_count >= n:
-        return sentences
-
-    # Rank by score (stable — ties keep original order)
-    ranked = sorted(
-        enumerate(scores), key=lambda iv: (-iv[1], iv[0])
-    )
-    keep_indices = {i for i, _ in ranked[:keep_count]}
-
-    # Preserve original ordering (don't shuffle) unless reorder=True
-    kept = [s for i, s in enumerate(sentences) if i in keep_indices]
-
-    if not reorder or len(kept) < 3:
-        return kept
-
-    # Lost-in-the-middle reorder: place highest-scoring at prompt edges
-    # (position 0 and position -1), where LLM attention is strongest.
-    kept_with_scores = [
-        (s, scores[i]) for i, s in enumerate(sentences) if i in keep_indices
-    ]
-    # Sort by score desc (stable on ties via original index order)
-    kept_with_scores.sort(key=lambda sv: (-sv[1], 0))
-
-    # Interleave from ranked list: #1 → pos 0, #2 → pos -1, #3 → pos 1, #4 → pos -2, ...
-    n_kept = len(kept_with_scores)
-    reordered: list[str | None] = [None] * n_kept
-    left, right = 0, n_kept - 1
-    for i, (sent, _) in enumerate(kept_with_scores):
-        if i % 2 == 0:
-            reordered[left] = sent
-            left += 1
-        else:
-            reordered[right] = sent
-            right -= 1
-    return [s for s in reordered if s is not None]
-
-
-class TwoTrim(Transform):
-    """Extractive text compression via semantic scoring + attention reordering.
-
-    See module docstring for algorithm overview.
-    """
-
-    name = "two_trim"
-
-    def __init__(self, config: TwoTrimConfig | None = None):
-        self.config = config or TwoTrimConfig()
-
-    def should_apply(
-        self,
-        messages: list[dict[str, Any]],
-        tokenizer: Tokenizer,
-        **kwargs: Any,
-    ) -> bool:
-        if not self.config.enabled:
-            return False
-        # Need at least one target-role message with text
-        for msg in messages:
-            if msg.get("role") not in self.config.target_roles:
-                continue
-            text = self._msg_text(msg)
-            # Approximate tokens as words * 1.3 (same ratio as _compress_text
-            # uses) so the pre-filter cannot admit messages the compress step
-            # would immediately bail on.
-            if text and len(text.split()) * 1.3 >= self.config.min_tokens_to_compress:
-                return True
-        return False
-
-    def apply(
-        self,
-        messages: list[dict[str, Any]],
-        tokenizer: Tokenizer,
-        **kwargs: Any,
-    ) -> TransformResult:
-        t0 = time.perf_counter()
-        tokens_before = tokenizer.count_messages(messages)
-
-        if not self.config.enabled:
-            # Direct-call safety: pipeline checks should_apply first, but users
-            # may call apply() directly. Respect the disabled flag.
-            return TransformResult(
-                messages=messages,
-                tokens_before=tokens_before,
-                tokens_after=tokens_before,
-                tokens_saved=0,
-                transforms_applied=[],
-                markers_inserted=[],
-                warnings=[],
-                timing={"two_trim": 0.0},
-                skip_reason="two_trim: disabled",
-            )
-        frozen_message_count = int(kwargs.get("frozen_message_count", 0) or 0)
-        messages = deep_copy_messages(messages)
-
-        query = _extract_query(messages)
-
-        warnings: list[str] = []
-        transforms_applied: list[str] = []
-        markers_inserted: list[str] = []
-        total_dropped = 0
-        total_kept = 0
-        used_semantic = False
-
-        for idx in range(frozen_message_count, len(messages)):
-            msg = messages[idx]
-            if msg.get("role") not in self.config.target_roles:
-                continue
-
-            content = msg.get("content")
-            if isinstance(content, str):
-                new_text, info = self._compress_text(content, query)
-                msg["content"] = new_text
-                if info.get("applied"):
-                    used_semantic = used_semantic or info.get("semantic", False)
-                    total_dropped += info.get("dropped", 0)
-                    total_kept += info.get("kept", 0)
-            elif isinstance(content, list):
-                for block in content:
-                    if not isinstance(block, dict):
-                        continue
-                    btype = block.get("type")
-                    if btype not in ("text", "input_text"):
-                        continue
-                    text = block.get("text") or ""
-                    new_text, info = self._compress_text(text, query)
-                    block["text"] = new_text
-                    if info.get("applied"):
-                        used_semantic = used_semantic or info.get("semantic", False)
-                        total_dropped += info.get("dropped", 0)
-                        total_kept += info.get("kept", 0)
-
-        if total_dropped > 0:
-            tag = "semantic" if used_semantic else "bm25"
-            transforms_applied.append(f"two_trim:{self.config.mode}:{tag}")
-            markers_inserted.append(
-                f"twotrim:drops={total_dropped}:kept={total_kept}"
-            )
-
-        tokens_after = tokenizer.count_messages(messages)
-        elapsed_ms = (time.perf_counter() - t0) * 1000.0
-        tokens_saved = max(0, tokens_before - tokens_after)
-
-        if transforms_applied and tokens_saved > 0:
-            segments = current_request_segments()
-            if segments is not None:
-                segments.append(
-                    {
-                        "strategy": "two_trim",
-                        "original_tokens": tokens_before,
-                        "compressed_tokens": tokens_after,
-                    }
-                )
-
-        if not transforms_applied:
-            return TransformResult(
-                messages=messages,
-                tokens_before=tokens_before,
-                tokens_after=tokens_after,
-                tokens_saved=0,
-                transforms_applied=[],
-                markers_inserted=[],
-                warnings=warnings,
-                timing={"two_trim": elapsed_ms},
-                skip_reason="two_trim: no eligible text compressed",
-            )
-
-        return TransformResult(
-            messages=messages,
-            tokens_before=tokens_before,
-            tokens_after=tokens_after,
-            tokens_saved=tokens_saved,
-            transforms_applied=transforms_applied,
-            markers_inserted=markers_inserted,
-            warnings=warnings,
-            timing={"two_trim": elapsed_ms},
-        )
-
-    def _msg_text(self, msg: dict[str, Any]) -> str:
-        content = msg.get("content")
-        if isinstance(content, str):
-            return content
-        if isinstance(content, list):
-            parts = []
-            for b in content:
-                if isinstance(b, dict) and b.get("type") in ("text", "input_text"):
-                    parts.append(b.get("text") or "")
-            return " ".join(parts)
-        return ""
-
-    def _compress_text(self, text: str, query: str) -> tuple[str, dict[str, Any]]:
-        """Returns (new_text, info_dict). info_dict has: applied, dropped, kept, semantic."""
-        no_op = {"applied": False, "dropped": 0, "kept": 0, "semantic": False}
-        if not text or not text.strip():
-            return text, no_op
-
-        # Skip structured content (handled by upstream compressors)
-        if self.config.skip_structured_content and _looks_structured(text):
-            return text, no_op
-
-        sentences = _split_sentences(text)
-        if len(sentences) < 3:
-            # Too few to meaningfully compress — preserve verbatim
-            return text, no_op
-
-        # Token floor check
-        approx_tokens = sum(len(s.split()) for s in sentences) * 1.3
-        if approx_tokens < self.config.min_tokens_to_compress:
-            return text, no_op
-
-        # Score
-        scores: list[float] = []
-        used_semantic = False
-        if self.config.use_semantic_scoring:
-            scores, used_semantic = _score_semantic(
-                sentences, query, self.config.sentence_model
-            )
-        if not scores:
-            scores = _score_bm25(sentences, query)
-            used_semantic = False
-
-        keep_ratio = _MODE_KEEP_RATIO.get(self.config.mode, 0.60)
-        kept = _select_and_reorder(
-            sentences,
-            scores,
-            keep_ratio=keep_ratio,
-            min_keep_ratio=self.config.min_sentence_keep_ratio,
-            max_drop_ratio=self.config.max_compression_ratio,
-            reorder=self.config.reorder_for_attention,
-        )
-
-        if len(kept) == len(sentences):
-            return text, no_op
-
-        new_text = " ".join(kept)
-        info = {
-            "applied": True,
-            "dropped": len(sentences) - len(kept),
-            "kept": len(kept),
-            "semantic": used_semantic,
-        }
-        return new_text, info

tests/test_two_trim.py

@@ -1,425 +0,0 @@
-"""Tests for TwoTrim extractive compression transform.
-
-These tests exercise the pure-Python path (BM25 fallback) by default
-so they run without optional deps. Semantic-scoring path is tested
-behind an import-skip gate.
-"""
-
-from __future__ import annotations
-
-import os
-
-os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
-
-import pytest
-
-from headroom.config import HeadroomConfig, TwoTrimConfig
-from headroom.tokenizer import Tokenizer
-from headroom.tokenizers import EstimatingTokenCounter
-from headroom.transforms.two_trim import (
-    TwoTrim,
-    _extract_query,
-    _looks_structured,
-    _select_and_reorder,
-    _split_sentences,
-)
-
-# =============================================================================
-# Fixtures
-# =============================================================================
-
-
-@pytest.fixture
-def tokenizer():
-    return Tokenizer(EstimatingTokenCounter(), model="gpt-4o")
-
-
-@pytest.fixture
-def long_user_text():
-    """Plain-text user message with enough sentences + tokens to compress."""
-    filler = (
-        "The quick brown fox jumps over the lazy dog near the riverbank. "
-        "Meanwhile, the weather today is unusually mild for this time of year. "
-        "Many people enjoy walking in the park during autumn afternoons. "
-        "There are several coffee shops on the corner of main street. "
-        "The annual report from 2023 shows a modest increase in revenue. "
-        "Birds migrate south when the temperature starts to drop significantly. "
-        "Did you remember to renew your gym membership this month? "
-        "Tomorrow's forecast predicts scattered showers across the region. "
-        "The library closes early on Sundays for maintenance work. "
-        "Some researchers prefer quantitative methods while others choose qualitative. "
-        "The conference keynote speaker will talk about neural network optimization. "
-        "Pizza is a popular dish that originated in Italy centuries ago. "
-        "The committee meets every other Thursday to discuss policy changes. "
-        "Quantum entanglement puzzled physicists for decades before being resolved. "
-        "Backups should run nightly to prevent catastrophic data loss incidents. "
-        "The new highway interchange opens next month after years of construction. "
-    )
-    # Replicate to push token count well past min_tokens_to_compress
-    return filler * 3
-
-
-# =============================================================================
-# Unit tests: helpers
-# =============================================================================
-
-
-class TestSplitSentences:
-    def test_basic_split(self):
-        text = "Hello world. How are you? I am fine!"
-        sents = _split_sentences(text)
-        assert len(sents) == 3
-        assert sents[0].startswith("Hello")
-
-    def test_empty(self):
-        assert _split_sentences("") == []
-        assert _split_sentences("   ") == []
-
-    def test_no_terminator(self):
-        # Single sentence with no period — returns single element
-        sents = _split_sentences("just words no terminator here")
-        assert len(sents) == 1
-
-    def test_preserves_internal_punctuation(self):
-        text = "Visit Dr. Smith at 10 a.m. for the appointment."
-        sents = _split_sentences(text)
-        # The regex is conservative — should produce >=1 sentences, not crash
-        assert len(sents) >= 1
-        # Original words all present
-        joined = " ".join(sents)
-        for word in ("Visit", "Smith", "appointment"):
-            assert word in joined
-
-
-class TestLooksStructured:
-    def test_json_object(self):
-        assert _looks_structured('{"key": "value"}') is True
-
-    def test_json_array(self):
-        assert _looks_structured("[1, 2, 3]") is True
-
-    def test_code_fence(self):
-        assert _looks_structured("```python\nprint('hi')\n```") is True
-
-    def test_plain_text(self):
-        assert _looks_structured("This is a normal sentence about cats.") is False
-
-    def test_empty(self):
-        assert _looks_structured("") is False
-
-
-class TestExtractQuery:
-    def test_last_user_wins(self):
-        msgs = [
-            {"role": "user", "content": "old question"},
-            {"role": "assistant", "content": "response"},
-            {"role": "user", "content": "new question"},
-        ]
-        assert _extract_query(msgs) == "new question"
-
-    def test_no_user_returns_empty(self):
-        msgs = [{"role": "system", "content": "sys"}, {"role": "assistant", "content": "a"}]
-        assert _extract_query(msgs) == ""
-
-    def test_list_content_text_blocks(self):
-        msgs = [
-            {
-                "role": "user",
-                "content": [
-                    {"type": "text", "text": "hello "},
-                    {"type": "text", "text": "world"},
-                ],
-            }
-        ]
-        assert "hello" in _extract_query(msgs)
-        assert "world" in _extract_query(msgs)
-
-
-class TestSelectAndReorder:
-    def test_keep_all_when_keep_ratio_one(self):
-        sents = ["a", "b", "c", "d"]
-        scores = [0.1, 0.9, 0.5, 0.2]
-        kept = _select_and_reorder(sents, scores, 1.0, 0.25, 0.75, reorder=False)
-        assert len(kept) == 4
-
-    def test_drops_lowest(self):
-        sents = ["a", "b", "c", "d", "e"]
-        scores = [0.1, 0.9, 0.5, 0.8, 0.2]
-        # 60% keep = 3 sentences
-        kept = _select_and_reorder(sents, scores, 0.6, 0.2, 0.7, reorder=False)
-        assert len(kept) == 3
-        # b, c, d kept (scores 0.9, 0.5, 0.8) — a and e dropped
-        kept_set = set(kept)
-        assert "b" in kept_set
-        assert "d" in kept_set
-        assert "a" not in kept_set
-        assert "e" not in kept_set
-
-    def test_floor_respected(self):
-        sents = ["a", "b", "c", "d"]
-        scores = [0.1, 0.1, 0.1, 0.1]
-        # Even with keep_ratio=0.1, floor of 0.5 keeps >=2
-        kept = _select_and_reorder(sents, scores, 0.1, 0.5, 0.7, reorder=False)
-        assert len(kept) >= 2
-
-    def test_reorder_puts_best_at_edges(self):
-        sents = ["low1", "HIGH1", "mid1", "mid2", "HIGH2", "low2"]
-        scores = [0.1, 0.99, 0.5, 0.5, 0.95, 0.1]
-        kept = _select_and_reorder(sents, scores, 0.5, 0.2, 0.7, reorder=True)
-        # Top 50% = 3 sentences: HIGH1 (0.99), HIGH2 (0.95), mid1 (0.5)
-        assert len(kept) == 3
-        # Best should be at start or end, not middle
-        assert kept[0] == "HIGH1" or kept[-1] == "HIGH1"
-        assert kept[0] == "HIGH2" or kept[-1] == "HIGH2"
-
-    def test_single_sentence_passthrough(self):
-        assert _select_and_reorder(["only"], [0.5], 0.5, 0.25, 0.75, False) == ["only"]
-
-
-# =============================================================================
-# Integration tests: Transform.apply
-# =============================================================================
-
-
-class TestTwoTrimApply:
-    def test_disabled_no_op(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(enabled=False)
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": long_user_text}]
-        result = t.apply(msgs, tokenizer)
-        # No-op path: skip_reason set, no transforms_applied
-        assert result.tokens_saved == 0
-        assert not result.transforms_applied
-
-    def test_should_apply_false_when_disabled(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(enabled=False)
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": long_user_text}]
-        assert t.should_apply(msgs, tokenizer) is False
-
-    def test_should_apply_true_when_enabled_with_text(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(enabled=True, min_tokens_to_compress=50)
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": long_user_text}]
-        assert t.should_apply(msgs, tokenizer) is True
-
-    def test_should_apply_false_for_short_text(self, tokenizer):
-        cfg = TwoTrimConfig(enabled=True, min_tokens_to_compress=2000)
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": "Too short to bother."}]
-        assert t.should_apply(msgs, tokenizer) is False
-
-    def test_compresses_long_text(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode="balanced",
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,  # Force BM25 for CI
-        )
-        t = TwoTrim(cfg)
-        msgs = [
-            {"role": "system", "content": "You are helpful."},
-            {"role": "user", "content": long_user_text},
-        ]
-        before = tokenizer.count_messages(msgs)
-        result = t.apply(msgs, tokenizer)
-        after = tokenizer.count_messages(result.messages)
-        # Should have compressed something (BM25 path, query has some overlap)
-        assert after < before
-        assert result.tokens_saved > 0
-        assert len(result.transforms_applied) == 1
-        assert result.transforms_applied[0].startswith("two_trim:balanced:bm25")
-
-    def test_skips_structured_content(self, tokenizer):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode="aggressive",
-            min_tokens_to_compress=10,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        big_json = '{"data": [' + ", ".join(f'{{"x": {i}}}' for i in range(500)) + "]}"
-        msgs = [{"role": "user", "content": big_json}]
-        result = t.apply(msgs, tokenizer)
-        # JSON should be left untouched
-        assert result.tokens_saved == 0
-        assert not result.transforms_applied
-
-    def test_preserves_uncompressed_roles(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode="balanced",
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        assistant_text = "I should not be modified by TwoTrim. " * 100
-        msgs = [
-            {"role": "user", "content": long_user_text},
-            {"role": "assistant", "content": assistant_text},
-        ]
-        result = t.apply(msgs, tokenizer)
-        # Assistant content should be unchanged (not in target_roles by default)
-        # Find assistant message — it should still contain all the original text
-        assistant_msg = next(m for m in result.messages if m["role"] == "assistant")
-        assert assistant_msg["content"] == assistant_text
-
-    def test_frozen_message_count_respected(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        msgs = [
-            {"role": "user", "content": long_user_text},  # would normally compress
-            {"role": "user", "content": "second message"},
-        ]
-        result = t.apply(msgs, tokenizer, frozen_message_count=1)
-        # First message should be untouched
-        assert result.messages[0]["content"] == long_user_text
-
-    def test_list_content_blocks(self, tokenizer, long_user_text):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        msgs = [
-            {
-                "role": "user",
-                "content": [
-                    {"type": "text", "text": long_user_text},
-                    {"type": "image_url", "image_url": {"url": "data:..."}},
-                ],
-            }
-        ]
-        result = t.apply(msgs, tokenizer)
-        # text block compressed, image block untouched
-        text_block = result.messages[0]["content"][0]
-        image_block = result.messages[0]["content"][1]
-        assert image_block["type"] == "image_url"
-        # Text should be shorter than original
-        assert len(text_block["text"]) < len(long_user_text)
-
-    def test_warnings_clean_when_no_op(self, tokenizer):
-        cfg = TwoTrimConfig(enabled=True, min_tokens_to_compress=10_000)
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": "short text"}]
-        result = t.apply(msgs, tokenizer)
-        assert result.tokens_saved == 0
-        assert result.warnings == []
-
-
-# =============================================================================
-# Integration: pipeline wiring
-# =============================================================================
-
-
-class TestPipelineIntegration:
-    def test_pipeline_picks_up_when_enabled(self, tokenizer, long_user_text):
-        from headroom.transforms.pipeline import TransformPipeline
-
-        cfg = HeadroomConfig()
-        cfg.two_trim.enabled = True
-        cfg.two_trim.min_tokens_to_compress = 50
-        cfg.two_trim.use_semantic_scoring = False
-
-        pipeline = TransformPipeline(config=cfg)
-        # Find the TwoTrim transform in the pipeline
-        names = [t.name for t in pipeline.transforms]
-        assert "two_trim" in names
-
-    def test_pipeline_skips_when_disabled(self, tokenizer):
-        from headroom.transforms.pipeline import TransformPipeline
-
-        cfg = HeadroomConfig()
-        cfg.two_trim.enabled = False
-
-        pipeline = TransformPipeline(config=cfg)
-        names = [t.name for t in pipeline.transforms]
-        assert "two_trim" not in names
-
-    def test_end_to_end_via_pipeline(self, long_user_text):
-        from headroom.transforms.pipeline import TransformPipeline
-
-        cfg = HeadroomConfig()
-        cfg.two_trim.enabled = True
-        cfg.two_trim.min_tokens_to_compress = 50
-        cfg.two_trim.use_semantic_scoring = False
-
-        pipeline = TransformPipeline(config=cfg)
-        msgs = [
-            {"role": "system", "content": "You are a helpful assistant."},
-            {"role": "user", "content": long_user_text},
-        ]
-        result = pipeline.apply(msgs, model="gpt-4o", model_limit=128000)
-        # TwoTrim should appear in transforms_applied if it saved anything
-        tt_applied = [t for t in result.transforms_applied if t.startswith("two_trim")]
-        # If tokens saved, two_trim should be recorded
-        if result.tokens_saved > 0:
-            assert len(tt_applied) >= 1
-
-    def test_pipeline_records_two_trim_compression_segment(self, long_user_text):
-        from headroom.transforms.pipeline import TransformPipeline
-
-        cfg = HeadroomConfig()
-        cfg.two_trim.enabled = True
-        cfg.two_trim.min_tokens_to_compress = 50
-        cfg.two_trim.use_semantic_scoring = False
-
-        pipeline = TransformPipeline(config=cfg)
-        msgs = [{"role": "user", "content": long_user_text}]
-        result = pipeline.apply(msgs, model="gpt-4o", model_limit=128000)
-
-        assert result.compression_segments is not None
-        two_trim_segments = [
-            s for s in result.compression_segments if s["strategy"] == "two_trim"
-        ]
-        assert two_trim_segments
-        segment = two_trim_segments[0]
-        assert segment["original_tokens"] == result.tokens_before
-        assert segment["compressed_tokens"] == result.tokens_after
-        assert segment["original_tokens"] > segment["compressed_tokens"]
-
-
-# =============================================================================
-# Mode behavior
-# =============================================================================
-
-
-class TestCompressionModes:
-    @pytest.mark.parametrize("mode", ["conservative", "balanced", "aggressive"])
-    def test_each_mode_runs(self, tokenizer, long_user_text, mode):
-        cfg = TwoTrimConfig(
-            enabled=True,
-            mode=mode,
-            min_tokens_to_compress=50,
-            use_semantic_scoring=False,
-        )
-        t = TwoTrim(cfg)
-        msgs = [{"role": "user", "content": long_user_text}]
-        result = t.apply(msgs, tokenizer)
-        assert result.tokens_saved > 0
-        assert f"two_trim:{mode}:bm25" in result.transforms_applied
-
-    def test_aggressive_saves_more_than_conservative(self, tokenizer, long_user_text):
-        # On the fixture text, aggressive should drop >= balanced >= conservative
-        def run(mode: str) -> int:
-            cfg = TwoTrimConfig(
-                enabled=True,
-                mode=mode,
-                min_tokens_to_compress=50,
-                use_semantic_scoring=False,
-            )
-            t = TwoTrim(cfg)
-            msgs = [{"role": "user", "content": long_user_text}]
-            return t.apply(msgs, tokenizer).tokens_saved
-
-        agg = run("aggressive")
-        bal = run("balanced")
-        con = run("conservative")
-        # Monotonic on this fixture (not a hard invariant in general, but
-        # the fixture is designed so it holds)
-        assert agg >= bal >= con