"""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()