tests/test_round_trip_stability.py

"""Round-trip divergence: input -> out1 -> out2; both hops should rewrite."""

from __future__ import annotations

import pytest

from app.engine import orchestrator
from app.engine.paraphrase import paraphrase_resource_available, surface_similarity


CUSTOMER_SAMPLE = (
    "For the success of any business, providing excellent customer service is "
    "vital. Customers appreciate organizations that respond promptly to their "
    "questions, resolve issues efficiently, and treat them with respect.\n\n"
    "Employees who communicate clearly and maintain a positive attitude help "
    "create a satisfying customer experience. Areas for improvement can be "
    "identified by businesses that actively listen to customer feedback and "
    "consolidate customer loyalty.\n\n"
    "By consistently delivering high-quality service, organizations can "
    "establish a strong reputation, gain customer retention, and encourage "
    "positive word-of-mouth recommendations."
)

_BAD = (
    "decide issues",
    "define issues",
    "determine issues",
    "influence issues",
    "launch a strong reputation",
    "found a strong reputation",
    "prove a strong reputation",
    "functioning high-quality",
    "serving high-quality service",
    "presenting high-quality service",
    "answer promptly to",
    "throw a positive",
)


def _rewrite(
    text: str,
    *,
    polish: bool,
    seed: int | None = None,
    use_paraphrase: bool = False,
):
    return orchestrator.rewrite_document(
        text,
        lexical_polish=polish,
        use_lexical_refinement=True,
        use_paraphrase=use_paraphrase,
        use_minilm_safety=use_paraphrase,
        require_wording_change=True,
        variation_seed=seed,
    )


def _round_trip(
    text: str,
    *,
    polish: bool,
    use_paraphrase: bool = False,
) -> tuple[str, str, float, float]:
    first = _rewrite(text, polish=polish, use_paraphrase=use_paraphrase)
    second = _rewrite(first.text, polish=polish, use_paraphrase=use_paraphrase)
    hop1 = surface_similarity(text, first.text)
    hop2 = surface_similarity(first.text, second.text)
    return first.text, second.text, hop1, hop2


def _assert_no_bad(text: str) -> None:
    low = text.lower()
    for bad in _BAD:
        assert bad not in low, bad


def test_round_trip_polish_true_keeps_meaning_on_both_hops():
    out1, out2, hop1, hop2 = _round_trip(CUSTOMER_SAMPLE, polish=True)
    print(f"polish=true hop1(input->out1)={hop1:.4f} hop2(out1->out2)={hop2:.4f}")
    print(f"identical={out1 == out2}")
    assert out1 != CUSTOMER_SAMPLE
    # Corrected SequenceMatcher (autojunk=False) scores long prose higher.
    assert hop1 < 0.95, hop1
    for text in (out1, out2):
        _assert_no_bad(text)
        for marker in ("vital", "employee", "reputation"):
            assert marker in text.lower()


def test_round_trip_polish_false_still_rewrites():
    out1, out2, hop1, hop2 = _round_trip(CUSTOMER_SAMPLE, polish=False)
    print(f"polish=false hop1={hop1:.4f} hop2={hop2:.4f}")
    assert out1 != CUSTOMER_SAMPLE or hop1 < 1.0
    _assert_no_bad(out1)
    _assert_no_bad(out2)


def test_round_trip_reports_similarity_for_inspection():
    out1, out2, hop1, hop2 = _round_trip(CUSTOMER_SAMPLE, polish=True)
    print(f"round_trip hop1={hop1:.4f} hop2={hop2:.4f}")
    print(f"out1_words={len(out1.split())} out2_words={len(out2.split())}")
    assert hop1 < 0.95
    # Classical-only path: second hop may only flip structure lightly.
    assert hop2 > 0.70


@pytest.mark.skipif(
    not paraphrase_resource_available(),
    reason="T5 paraphraser unavailable in this environment",
)
def test_generative_round_trip_diverges_below_half():
    """Primary T5 paraphrase should drive out1→out2 well below classical levels."""
    out1, out2, hop1, hop2 = _round_trip(
        CUSTOMER_SAMPLE, polish=True, use_paraphrase=True
    )
    print(f"generative hop1={hop1:.4f} hop2={hop2:.4f}")
    print(f"out1={out1[:180]}...")
    print(f"out2={out2[:180]}...")
    assert out1 != CUSTOMER_SAMPLE
    assert out1 != out2
    assert hop2 < 0.40, hop2
    for text in (out1, out2):
        _assert_no_bad(text)
        for marker in ("vital", "employee", "reputation"):
            assert marker in text.lower()


def test_repeated_requests_pick_different_structures(structural_variation):
    """The point of rotation: the same input must not always rewrite the same."""
    outputs = {_rewrite(CUSTOMER_SAMPLE, polish=True).text for _ in range(6)}
    for text in outputs:
        _assert_no_bad(text)
        assert text != CUSTOMER_SAMPLE
        for marker in ("vital", "employee", "reputation"):
            assert marker in text.lower()


def test_variation_seed_is_reproducible(structural_variation):
    first = _rewrite(CUSTOMER_SAMPLE, polish=True, seed=1234).text
    second = _rewrite(CUSTOMER_SAMPLE, polish=True, seed=1234).text
    assert first == second


def test_seed_zero_disables_rotation(structural_variation):
    first = _rewrite(CUSTOMER_SAMPLE, polish=True, seed=0).text
    second = _rewrite(CUSTOMER_SAMPLE, polish=True, seed=0).text
    assert first == second


def test_marginal_sense_verb_swap_is_blocked():
    """settle "reside" carries `locate`; it must not reach a verb-object swap."""
    text = (
        "Customers appreciate organizations that respond promptly to their "
        "questions, settle issues efficiently, and treat them with respect."
    )
    result = _rewrite(text, polish=True)
    assert "locate issues" not in result.text.lower()
    assert "settle issues" in result.text.lower()