hermes-agent/tests/gateway/test_compression_progress_notices.py

"""Opt-in compression progress notices on chat gateways (#52995).

Routine automatic compression is silent-by-design on human-facing chat
platforms: the gateway noise filter (`_TELEGRAM_NOISY_STATUS_RE` via
`_prepare_gateway_status_message`) swallows every ROUTINE compression status.
`compression.progress_notices: true` (default: false) opens an opt-in gate
that lets those ROUTINE compression statuses through — scoped strictly to
the #69550 compression status template constants, so unrelated operational
noise (auxiliary failures, provider retry/rate-limit chatter) stays
suppressed even when the gate is enabled.

The default-OFF path must stay byte-identical to silent-by-design main:
tests/gateway/test_telegram_noise_filter.py pins that suite unchanged.
"""

import pytest

import gateway.run as gateway_run
from agent.conversation_compression import (
    COMPACTION_DONE_STATUS,
    ROUTINE_COMPRESSION_STATUS_SAMPLES,
)
from gateway.run import _prepare_gateway_status_message

# Chat surfaces the opt-in must deliver to (subset of the noise-filter
# suite's CHAT_PLATFORMS; telegram + discord are the required anchors).
CHAT_PLATFORMS = ["telegram", "discord", "slack", "whatsapp"]

# Noisy statuses that are NOT routine compression progress — they must stay
# suppressed on chat platforms even when progress_notices is enabled.
NON_COMPRESSION_NOISE = [
    "⚠ Auxiliary title generation failed: HTTP 400: Operation contains cybersecurity risk",
    "⏳ Retrying in 4.2s (attempt 1/3)...",
    "⏱️ Rate limited. Waiting 30.0s (attempt 2/3)...",
    "⚠️ Max retries (3) exhausted — trying fallback...",
    "⚠ Compression summary failed: upstream error. Inserted a fallback context marker.",
    (
        "⚠ Configured auxiliary compression provider 'openai' is unavailable — "
        "context compression will drop middle turns without a summary. Check "
        "auxiliary.compression in config.yaml and reauthenticate that provider."
    ),
    (
        "⚠ Skipping concurrent compression — another path is already "
        "compressing this session. Will retry after it finishes."
    ),
]


@pytest.fixture
def progress_notices_enabled(monkeypatch):
    """Gateway config with compression.progress_notices: true."""
    monkeypatch.setattr(
        gateway_run,
        "_load_gateway_config",
        lambda: {"compression": {"progress_notices": True}},
    )


@pytest.fixture
def progress_notices_default(monkeypatch):
    """Gateway config without the key — the silent-by-design default."""
    monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {})


@pytest.mark.parametrize("platform", CHAT_PLATFORMS)
@pytest.mark.parametrize("message", NON_COMPRESSION_NOISE, ids=lambda m: m[:32])
def test_enabled_still_suppresses_non_compression_noise(
    progress_notices_enabled, platform, message
):
    """The gate is scoped to compression progress statuses ONLY.

    Aux-model failures, provider retry/rate-limit chatter, and other noisy
    statuses that are not #69550 compression progress templates must stay
    suppressed on chat surfaces even when progress_notices is enabled.
    """
    assert _prepare_gateway_status_message(platform, "warn", message) is None


@pytest.mark.parametrize("enabled", [True, False], ids=["enabled", "default"])
@pytest.mark.parametrize("platform", CHAT_PLATFORMS)
def test_compaction_completion_notice_reaches_chat(monkeypatch, platform, enabled):
    """The #69546 'compacted' lifecycle edge is deliverable on chat surfaces.

    COMPACTION_DONE_STATUS already flows through the status callback on
    compaction completion and is not matched by the noise regex — the opt-in
    gate must not change that in either mode, so users who enable
    progress_notices see the completion stat notice paired with the start.
    """
    monkeypatch.setattr(
        gateway_run,
        "_load_gateway_config",
        lambda: {"compression": {"progress_notices": enabled}},
    )
    assert (
        _prepare_gateway_status_message(platform, "compacted", COMPACTION_DONE_STATUS)
        == COMPACTION_DONE_STATUS
    )


def test_enabled_gate_does_not_leak_to_raw_platforms(progress_notices_enabled):
    """Programmatic surfaces keep raw text regardless of the gate."""
    message = ROUTINE_COMPRESSION_STATUS_SAMPLES[0]
    for platform in ("local", "api_server", "webhook", "msgraph_webhook"):
        assert (
            _prepare_gateway_status_message(platform, "lifecycle", message) == message
        )


def test_progress_regex_covers_every_routine_sample():
    """The template-derived membership regex matches every ROUTINE sample.

    Guards the coupling: a new #69550 template constant added to
    ROUTINE_COMPRESSION_STATUS_SAMPLES without being added to the gateway's
    _COMPRESSION_PROGRESS_STATUS_RE alternatives fails here.
    """
    for message in ROUTINE_COMPRESSION_STATUS_SAMPLES:
        assert gateway_run._COMPRESSION_PROGRESS_STATUS_RE.search(message), (
            f"routine compression sample not covered by the opt-in gate: {message!r}"
        )