headroomlabs-ai--headroom/tests/test_cache_aligner_detector_only.py

"""Detector-only tests for ``CacheAligner`` (PR-A2 / P2-23).

After PR-A2, ``CacheAligner`` is a detector-only transform: it never
mutates messages, only emits warnings. These tests pin that contract.
"""

from __future__ import annotations

import logging
from collections.abc import Iterator
from contextlib import contextmanager

import pytest

from headroom import OpenAIProvider, Tokenizer
from headroom.config import CacheAlignerConfig
from headroom.transforms.cache_aligner import (
    CacheAligner,
    detect_volatile_content,
)

_provider = OpenAIProvider()


def _tokenizer() -> Tokenizer:
    counter = _provider.get_token_counter("gpt-4o")
    return Tokenizer(counter, "gpt-4o")


@pytest.fixture
def tokenizer() -> Tokenizer:
    return _tokenizer()


def _system_user_messages(system_text: str) -> list[dict[str, object]]:
    return [
        {"role": "system", "content": system_text},
        {"role": "user", "content": "hello"},
    ]


@contextmanager
def _capture_warnings() -> Iterator[list[str]]:
    """Capture warning messages from the cache_aligner logger.

    Bypasses pytest's ``caplog`` so logger-propagation tweaks elsewhere in
    the test suite don't break detection assertions.
    """
    captured: list[str] = []

    class _ListHandler(logging.Handler):
        def emit(self, record: logging.LogRecord) -> None:
            captured.append(record.getMessage())

    target_logger = logging.getLogger("headroom.transforms.cache_aligner")
    prev_level = target_logger.level
    target_logger.setLevel(logging.WARNING)
    handler = _ListHandler(level=logging.WARNING)
    target_logger.addHandler(handler)
    try:
        yield captured
    finally:
        target_logger.removeHandler(handler)
        target_logger.setLevel(prev_level)


# ---------------------------------------------------------------------------
# Detector tests on the ``apply`` surface
# ---------------------------------------------------------------------------


def test_volatile_uuid_detected_warned_not_rewritten(tokenizer: Tokenizer) -> None:
    system_text = (
        "You are a helpful assistant.\n"
        "Session ID: 550e8400-e29b-41d4-a716-446655440000\n"
        "Help the user."
    )
    messages = _system_user_messages(system_text)

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    with _capture_warnings() as records:
        result = aligner.apply(messages, tokenizer)

    # Bytes preserved — never rewritten.
    assert result.messages[0]["content"] == system_text
    assert result.messages[1]["content"] == "hello"

    # Warning surfaced (in result + via the logger).
    assert result.warnings, "expected a warning for detected UUID"
    assert any("uuid" in w.lower() for w in result.warnings)
    assert any("uuid" in rec.lower() for rec in records)


def test_volatile_iso8601_detected_warned(tokenizer: Tokenizer) -> None:
    system_text = "You are a logging assistant.\nCurrent time: 2024-01-15T10:30:00\nHelp the user."
    messages = _system_user_messages(system_text)

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    with _capture_warnings() as records:
        result = aligner.apply(messages, tokenizer)

    assert result.messages[0]["content"] == system_text
    assert any("iso8601" in w.lower() for w in result.warnings)
    assert any("iso8601" in rec.lower() for rec in records)


def test_volatile_jwt_detected_warned(tokenizer: Tokenizer) -> None:
    jwt = (
        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
        "eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ."
        "SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"
    )
    system_text = f"You are an assistant.\nAuth Token: {jwt}\nHelp the user."
    messages = _system_user_messages(system_text)

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    with _capture_warnings() as records:
        result = aligner.apply(messages, tokenizer)

    assert result.messages[0]["content"] == system_text
    assert any("jwt" in w.lower() for w in result.warnings)
    assert any("jwt" in rec.lower() for rec in records)


def test_volatile_hex_hash_detected_warned(tokenizer: Tokenizer) -> None:
    sha256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    system_text = f"You are a security assistant.\nHash: {sha256}\nHelp."
    messages = _system_user_messages(system_text)

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    with _capture_warnings() as records:
        result = aligner.apply(messages, tokenizer)

    assert result.messages[0]["content"] == system_text
    assert any("hex_hash" in w.lower() for w in result.warnings)
    assert any("hex_hash" in rec.lower() for rec in records)


def test_no_false_positives_on_normal_prose(tokenizer: Tokenizer) -> None:
    system_text = (
        "You are a helpful assistant. Answer the user's questions clearly. "
        "Be polite, concise, and accurate. If you do not know the answer, "
        "say so plainly."
    )
    messages = _system_user_messages(system_text)

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    with _capture_warnings() as records:
        result = aligner.apply(messages, tokenizer)

    assert result.messages[0]["content"] == system_text
    assert result.warnings == [], f"unexpected warnings: {result.warnings}"
    assert records == [], f"unexpected log records: {records}"


def test_apply_never_mutates_input(tokenizer: Tokenizer) -> None:
    """The detector never mutates the caller's input list."""
    system_text = "Session: 550e8400-e29b-41d4-a716-446655440000"
    messages = _system_user_messages(system_text)
    snapshot = [dict(m) for m in messages]

    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    aligner.apply(messages, tokenizer)

    for original, current in zip(snapshot, messages, strict=True):
        assert original == current


def test_apply_does_not_attempt_rewrite_on_dynamic_content(
    tokenizer: Tokenizer,
) -> None:
    """``transforms_applied`` must be empty even when warnings fire."""
    system_text = (
        "Session: 550e8400-e29b-41d4-a716-446655440000\n"
        "Hash: e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    )
    messages = _system_user_messages(system_text)
    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    result = aligner.apply(messages, tokenizer)
    assert result.transforms_applied == []
    # System bytes must equal input bytes (no rewrite).
    assert result.messages[0]["content"] == system_text


def test_should_apply_false_when_disabled(tokenizer: Tokenizer) -> None:
    messages = _system_user_messages("Session: 550e8400-e29b-41d4-a716-446655440000")
    aligner = CacheAligner(CacheAlignerConfig(enabled=False))
    assert not aligner.should_apply(messages, tokenizer)


def test_should_apply_false_without_system_message(tokenizer: Tokenizer) -> None:
    messages = [{"role": "user", "content": "hello"}]
    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    assert not aligner.should_apply(messages, tokenizer)


def test_should_apply_false_when_policy_disables_aligner(tokenizer: Tokenizer) -> None:
    """F2.1 c4/5: ``compression_policy.cache_aligner_enabled=False``
    must disable the detector even when content + config would
    otherwise opt in.

    This is the canary that the per-auth-mode plumbing actually
    reaches CacheAligner. A regression here means subscription
    requests would silently keep updating
    ``self._previous_prefix_hash`` and emitting volatility warnings,
    which are the exact log lines #327/#388 reporters complained
    about.

    F2.2 added three per-mode tuning fields to CompressionPolicy
    (``volatile_token_threshold``, ``max_lossy_ratio``,
    ``toin_read_only``); the policy here uses the Subscription
    defaults from ``policy_for_mode(AuthMode.SUBSCRIPTION)`` so the
    fixture mirrors a real subscription request.
    """
    from headroom.transforms.compression_policy import CompressionPolicy

    messages = _system_user_messages("Session: 550e8400-e29b-41d4-a716-446655440000")
    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    # Sanity: without a policy, the detector opts in.
    assert aligner.should_apply(messages, tokenizer)
    # F2.1 gate: with the subscription policy, the detector opts out.
    sub_policy = CompressionPolicy(
        live_zone_only=True,
        cache_aligner_enabled=False,
        volatile_token_threshold=32,
        max_lossy_ratio=0.25,
        toin_read_only=True,
    )
    assert not aligner.should_apply(messages, tokenizer, compression_policy=sub_policy)


def test_should_apply_true_when_policy_enables_aligner(tokenizer: Tokenizer) -> None:
    """F2.1 c4/5: ``compression_policy.cache_aligner_enabled=True``
    must NOT short-circuit. PAYG/OAuth keep current behaviour.

    F2.2 added three per-mode tuning fields; the policy here uses the
    PAYG defaults from ``policy_for_mode(AuthMode.PAYG)`` so the
    fixture mirrors a real PAYG request.
    """
    from headroom.transforms.compression_policy import CompressionPolicy

    messages = _system_user_messages("Session: 550e8400-e29b-41d4-a716-446655440000")
    aligner = CacheAligner(CacheAlignerConfig(enabled=True))
    payg_policy = CompressionPolicy(
        live_zone_only=False,
        cache_aligner_enabled=True,
        volatile_token_threshold=128,
        max_lossy_ratio=0.45,
        toin_read_only=False,
    )
    assert aligner.should_apply(messages, tokenizer, compression_policy=payg_policy)


# ---------------------------------------------------------------------------
# Pure-function detection tests
# ---------------------------------------------------------------------------


def test_detect_volatile_content_uuid() -> None:
    findings = detect_volatile_content("Session: 550e8400-e29b-41d4-a716-446655440000")
    labels = [f.label for f in findings]
    assert "uuid" in labels


def test_detect_volatile_content_iso_date() -> None:
    findings = detect_volatile_content("Now: 2024-01-15T10:30:00")
    labels = [f.label for f in findings]
    assert "iso8601" in labels


def test_detect_volatile_content_jwt() -> None:
    # ggignore: canonical fake JWT, payload {"sub":"1"}, no privilege.
    jwt = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"  # noqa: S105
    findings = detect_volatile_content(f"Token: {jwt}")
    labels = [f.label for f in findings]
    assert "jwt" in labels


def test_detect_volatile_content_md5() -> None:
    findings = detect_volatile_content("Hash: d41d8cd98f00b204e9800998ecf8427e")
    labels = [f.label for f in findings]
    assert "hex_hash" in labels


def test_detect_volatile_content_sha1() -> None:
    findings = detect_volatile_content("Hash: da39a3ee5e6b4b0d3255bfef95601890afd80709")
    labels = [f.label for f in findings]
    assert "hex_hash" in labels


def test_detect_volatile_content_returns_truncated_sample() -> None:
    long_token = "0123456789abcdef0123456789abcdef"  # 32-char hex (MD5-shaped)
    findings = detect_volatile_content(f"hash: {long_token}")
    assert findings
    # Sample is truncated for any token longer than 16 chars.
    assert "..." in findings[0].sample


def test_detect_volatile_content_empty_string_no_findings() -> None:
    assert detect_volatile_content("") == []


def test_detect_volatile_content_normal_text_no_findings() -> None:
    findings = detect_volatile_content("You are a helpful assistant. Be polite.")
    assert findings == []


def test_alignment_score_perfect_when_no_dynamic_content() -> None:
    aligner = CacheAligner()
    score = aligner.get_alignment_score([{"role": "system", "content": "You are helpful."}])
    assert score == 100.0


def test_alignment_score_decreases_with_dynamic_content() -> None:
    aligner = CacheAligner()
    score = aligner.get_alignment_score(
        [
            {
                "role": "system",
                "content": (
                    "Session: 550e8400-e29b-41d4-a716-446655440000\n"
                    "Hash: d41d8cd98f00b204e9800998ecf8427e"
                ),
            }
        ]
    )
    assert score < 100.0


def test_no_regex_imported() -> None:
    """The cache_aligner module must not import ``re`` (build constraint)."""
    import headroom.transforms.cache_aligner as mod

    # ``re`` should not be in the module's globals.
    assert "re" not in mod.__dict__, (
        "headroom.transforms.cache_aligner must not depend on the regex module"
    )