tests/test_bug69_71_conversation_logic.py

"""Bug #69 + Bug #71 (2026-05-16) — conversation-logic regressions found
live by the user.

Bug #69 — NAME gated last + misleading "no policies match".
  Live transcript: user gave age/city/budget/sum-insured etc.; the advisor
  replied "I couldn't find any policies matching all your criteria. It
  seems I'm missing a few more details to proceed. Could you please tell
  me your name?" then recommended the moment the name was given. Two
  defects:
    1. NAME (a required slot that GATES recommendations) was asked LAST.
       It must be asked EARLY (ideally first).
    2. A missing required slot was surfaced as a retrieval/no-match
       failure — false (name has nothing to do with policy matching) and
       self-defeating.

Bug #71 — weak-grade policies presented as recommendations.
  Live transcript: the advisor surfaced as "recommendations" HDFC ERGO
  my:Optima Secure (B, 75/100) AND Star Family Health Optima (C, 64/100).
  A C-graded 64/100 plan for the user's OWN profile is not a
  recommendation. The recommended set must rank strictly best-first and
  gate to a sensible minimum fit; present fewer (or none) honestly rather
  than pad with a weak plan.

Run:
    cd /Users/rohitsar/Developer/Insurance\\ Sales\\ Bot
    PYTHONPATH=$PWD .venv/bin/python -m pytest -q \\
        tests/test_bug69_71_conversation_logic.py
"""

from __future__ import annotations

import asyncio
import os
import sys
import unittest
import uuid
from pathlib import Path
from unittest import mock

_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from backend import brain_tools, single_brain  # noqa: E402
from backend.needs_finder import (  # noqa: E402
    Profile,
    _SLOT_ORDER,
    next_question,
)
from backend.single_brain import (  # noqa: E402
    SYSTEM_PROMPT,
    _HONEST_EMPTY_REPLY,
    _build_recommendation_citations,
    _recommendation_fit,
)

# Phrases the advisor must NEVER use to surface a merely-missing slot.
_BANNED_NO_MATCH = (
    "couldn't find any policies matching all your criteria",
    "couldnt find any policies matching all your criteria",
    "no policies match",
    "nothing matches your criteria",
    "i'm missing a few more details to proceed",
    "im missing a few more details to proceed",
)


def _assert_not_misleading(case: unittest.TestCase, text: str) -> None:
    low = (text or "").lower()
    for bad in _BANNED_NO_MATCH:
        case.assertNotIn(
            bad, low,
            f"missing-slot turn must not imply a search failed: {text!r}",
        )


def _run(coro):
    return asyncio.new_event_loop().run_until_complete(coro)


# ════════════════════════════════════════════════════════════════════════════
# BUG #69 — corrected design: canonical slot order (needs_finder UI hint) +
# no RULE-0 determinism + honest empty-LLM error (no fabricated slot-question)
# ════════════════════════════════════════════════════════════════════════════
class TestBug69NameFirstDeterministic(unittest.TestCase):
    def test_canonical_slot_order_puts_name_first(self):
        self.assertEqual(
            _SLOT_ORDER[0], "name",
            "name is a recommendation-gating required slot — it must be "
            "first in the canonical fact-find order")
        self.assertEqual(next_question(Profile()), "name")

    def test_system_prompt_has_no_rule0_determinism_keeps_rule1_capture(self):
        # CORRECTED DESIGN (2026-05-17): the RULE-0 "ASK FOR THE NAME FIRST"
        # overlay was a contradiction with RULE 1 ("save the name the user
        # gave") and is REMOVED. Regression guard: the prompt must NOT
        # reintroduce the RULE-0 determinism…
        self.assertNotIn("RULE 0", SYSTEM_PROMPT)
        self.assertNotIn("ASK FOR THE NAME FIRST", SYSTEM_PROMPT)
        self.assertNotIn("NAME IS COLLECTED FIRST", SYSTEM_PROMPT)
        # …and the single-LLM capture contract (RULE 1) must still mandate
        # saving a stated name via the tool — the SOLE fact-find mechanism.
        self.assertIn("RULE 1", SYSTEM_PROMPT)
        self.assertIn('save_profile_field(field="name"', SYSTEM_PROMPT)

    # The 3 `_synthesise_fallback` unit tests were removed with the
    # deterministic slot-question generator (2026-05-17). Empty-LLM
    # behaviour is now an HONEST error — covered by
    # TestBug69EndToEnd.test_empty_llm_yields_honest_error.


# ════════════════════════════════════════════════════════════════════════════
# Shared end-to-end harness (Gemini + retrieve_policies network seams stubbed)
# ════════════════════════════════════════════════════════════════════════════
def _fc_part(name, args):
    return {"functionCall": {"name": name, "args": args}}


def _text_payload(text):
    return {"candidates": [{"content": {"parts": [{"text": text}]}}]}


def _tool_payload(parts):
    return {"candidates": [{"content": {"parts": parts}}]}


class _HandleTurnHarness(unittest.TestCase):
    """Drives single_brain.handle_turn for real with the two network seams
    scripted. The fit gate + citation builder run for real;
    only Gemini and the vector store are stubbed."""

    def setUp(self):
        self._env = mock.patch.dict(os.environ,
                                    {"GOOGLE_API_KEY": "test-key"})
        self._env.start()
        self._gemini_script: list = []
        self._retrieve_chunks: list = []

        async def _fake_gemini(*_a, **_k):
            if not self._gemini_script:
                return _text_payload("(no more scripted turns)")
            return self._gemini_script.pop(0)

        async def _fake_retrieve(*_a, **_k):
            chunks = list(self._retrieve_chunks)
            sess = _k.get("session")
            if sess is not None:
                sess.last_retrieved_chunks = list(chunks)
                sess.slug_to_insurer = {
                    c["policy_id"]: c.get("insurer_slug", "") for c in chunks
                }
            return {"chunks": chunks, "count": len(chunks)}

        self._gp = mock.patch.object(single_brain, "_gemini_call",
                                     _fake_gemini)
        self._rp = mock.patch.object(brain_tools, "retrieve_policies",
                                     _fake_retrieve)
        self._gp.start()
        self._rp.start()

    def tearDown(self):
        self._rp.stop()
        self._gp.stop()
        self._env.stop()

    def _fresh_session(self):
        from backend.session_state import SessionState
        return SessionState(session_id=f"t_{uuid.uuid4().hex[:8]}")

    def _ready_session(self):
        """A session whose 7 required slots are filled AND the post-recap
        pricing bundle is marked skipped — the realistic precondition for a
        RECOMMENDATION turn. Bug #107 only attaches citations once the
        profile gate (brain_tools._profile_complete) is satisfied, and Bug
        #108's one-shot bundle re-ask gate is bypassed when the user skipped
        the pricing inputs. End-to-end recommendation tests must reflect
        that real flow rather than recommend off an empty profile."""
        sess = self._fresh_session()
        sess.profile.name = "Asha"
        sess.profile.age = 35
        sess.profile.dependents = "self+spouse"
        sess.profile.location_tier = "metro"
        sess.profile.income_band = "10L-25L"
        sess.profile.primary_goal = "first_buy"
        sess.profile.health_conditions = ["none"]
        sess.pricing_bundle_skipped = True
        return sess


# ════════════════════════════════════════════════════════════════════════════
# BUG #69 — end-to-end: all details EXCEPT name early; name turn not a no-match
# ════════════════════════════════════════════════════════════════════════════
class TestBug69EndToEnd(_HandleTurnHarness):
    def test_name_requested_early_and_missing_name_not_a_no_match(self):
        sess = self._fresh_session()

        # Turn 1 — the user dumps every substantive detail BUT their name,
        # exactly like the live transcript. The brain captures the facts;
        # the ONLY required slot still missing is `name`. The honest,
        # RULE-0-compliant reply asks for the name and does NOT claim a
        # search failed (no retrieve_policies has even run).
        self._gemini_script = [
            _tool_payload([
                _fc_part("save_profile_field", {"field": "age", "value": "39"}),
                _fc_part("save_profile_field",
                         {"field": "location_tier", "value": "metro"}),
                _fc_part("save_profile_field",
                         {"field": "dependents", "value": "self+spouse"}),
                _fc_part("save_profile_field",
                         {"field": "income_band", "value": "10L-25L"}),
                _fc_part("save_profile_field",
                         {"field": "primary_goal", "value": "first_buy"}),
                _fc_part("save_profile_field",
                         {"field": "health_conditions", "value": "none"}),
            ]),
            _text_payload(
                "I've got everything I need except your name — what should "
                "I call you? Then I'll pull your matches."),
        ]
        r1 = _run(single_brain.handle_turn(
            sess,
            "I'm 39, in Bangalore, with my wife, income about 18 lakh, "
            "first family policy, no health issues, budget ~30k, want "
            "10 lakh cover.",
        ))

        # (a) The missing-name turn must NOT imply a failed search / that
        #     it can't help. retrieve_policies never ran this turn.
        _assert_not_misleading(self, r1.reply_text)
        self.assertNotIn(
            "retrieve_policies",
            [c.get("policy_id", "") for c in r1.citations])
        # (b) It must ask for the name (the only remaining required slot).
        self.assertIn("name", r1.reply_text.lower())
        self.assertEqual(getattr(sess.profile, "name", None), None)
        # Sanity: the substantive facts WERE captured this turn (so the
        # only thing gating a recommendation really is the name).
        self.assertEqual(sess.profile.age, 39)
        self.assertEqual(sess.profile.income_band, "10L-25L")

        # Turn 2 — user gives the name; flow proceeds normally.
        self._gemini_script = [
            _tool_payload([_fc_part("save_profile_field",
                                    {"field": "name", "value": "Rohit"})]),
            _text_payload(
                "Thanks Rohit — pulling plans that fit your profile now."),
        ]
        r2 = _run(single_brain.handle_turn(sess, "Rohit"))
        self.assertEqual(sess.profile.name, "Rohit")
        _assert_not_misleading(self, r2.reply_text)

    def test_no_in_chat_recall_confirm_short_circuit(self):
        # CORRECTED DESIGN (2026-05-17): the deterministic in-conversation
        # recall scaffolding (turn-1 name-sniff → "Welcome back — are you
        # the same X? (yes/no)" short-circuit, intent="recall_confirm") was
        # REMOVED. A self-introduction must flow straight through the single
        # LLM — never a canned recall-confirm short-circuit. (Explicit
        # returning-user recall now lives only in the separate
        # POST /api/profile/recall-by-name endpoint.) Replaces the deleted
        # tests/test_returning_user_recall.py as the regression guard.
        sess = self._fresh_session()
        self._gemini_script = [
            _tool_payload([_fc_part("save_profile_field",
                                    {"field": "name", "value": "Rohit"})]),
            _text_payload("Hi Rohit! How can I help with health cover today?"),
        ]
        r = _run(single_brain.handle_turn(sess, "Hi, I'm Rohit"))
        self.assertNotEqual(r.intent, "recall_confirm")
        self.assertNotIn("recall_confirm", r.brain_used)
        self.assertNotIn("are you the same", r.reply_text.lower())
        self.assertNotIn("welcome back —", r.reply_text.lower())
        # The single LLM drove the turn and captured the name via the tool.
        self.assertEqual(sess.profile.name, "Rohit")
        self.assertFalse(r.returning_user_recalled)

    def test_empty_llm_yields_honest_error_not_fabricated_slot_question(self):
        # CORRECTED DESIGN (2026-05-17): a genuine empty-LLM turn (Gemini
        # emits no text and no tools across all iterations) must surface an
        # HONEST retry message — NEVER the old deterministic
        # `_synthesise_fallback` that fabricated a "What's your name?" /
        # slot-question and simulated a fact-find that did not happen.
        sess = self._fresh_session()
        sess.profile = Profile(
            age=39, dependents="self+spouse", location_tier="metro",
            income_band="10L-25L", primary_goal="first_buy",
            health_conditions=["none"],
        )
        self._gemini_script = [_text_payload("") for _ in range(12)]
        r = _run(single_brain.handle_turn(sess, "what next?"))
        self.assertEqual(r.reply_text, _HONEST_EMPTY_REPLY)
        _assert_not_misleading(self, r.reply_text)
        # It must NOT fabricate a slot-question (the removed determinism).
        self.assertNotIn("what's your name", r.reply_text.lower())
        self.assertNotIn("your annual household income", r.reply_text.lower())


# ════════════════════════════════════════════════════════════════════════════
# BUG #71 — recommendation fit gate (unit + live citation path)
# ════════════════════════════════════════════════════════════════════════════
def _enriched(pid, name, slug, score, cid, *, grade=None, overall=None):
    """Chunk shaped like brain_tools.retrieve_policies output AFTER the
    scorecard enrichment step (carries _grade / _overall_score)."""
    return {
        "chunk_id": cid, "policy_id": pid, "policy_name": name,
        "insurer_slug": slug, "doc_type": "policy",
        "source_url": f"https://example.com/{pid}.pdf", "score": score,
        "_grade": grade, "_overall_score": overall,
    }


class TestBug71RecommendationFitUnit(unittest.TestCase):
    def test_recommendation_fit_classification(self):
        # Strong: overall >= 70, or A/B with no numeric.
        self.assertTrue(_recommendation_fit({"_overall_score": 75})[0])
        self.assertTrue(_recommendation_fit({"_overall_score": 70})[0])
        self.assertTrue(_recommendation_fit({"_grade": "A"})[0])
        self.assertTrue(_recommendation_fit({"_grade": "B"})[0])
        # Weak with positive evidence: dropped.
        self.assertFalse(_recommendation_fit({"_overall_score": 64})[0])
        self.assertFalse(_recommendation_fit({"_grade": "C"})[0])
        self.assertFalse(_recommendation_fit({"_grade": "D"})[0])
        # A numeric overall is authoritative even if a stale letter says C.
        self.assertTrue(
            _recommendation_fit({"_grade": "C", "_overall_score": 82})[0])
        # NO evidence at all → fail OPEN (pipeline already vetted fit;
        # don't nuke the shortlist when the optional scorecard is down).
        self.assertTrue(_recommendation_fit({})[0])
        self.assertTrue(_recommendation_fit({"_grade": ""})[0])

    def test_live_report_scenario_c_grade_dropped(self):
        # The exact live report: B/75 + C/64 both cited as recs.
        chunks = [
            _enriched("hdfc-ergo__optima-secure", "my:Optima Secure",
                      "hdfc-ergo", 0.71, "c1", grade="B", overall=75),
            _enriched("star-health__family-health-optima",
                      "Star Family Health Optima", "star-health",
                      0.68, "c2", grade="C", overall=64),
        ]
        reply = ("Top recommendations:\n1. my:Optima Secure (HDFC ERGO)\n"
                 "2. Star Family Health Optima (Star Health)")
        for marks in (["hdfc-ergo__optima-secure",
                       "star-health__family-health-optima"], []):
            cites, is_rec = _build_recommendation_citations(
                reply_text=reply, retrieved_chunks_all=chunks,
                marked_policy_ids=marks)
            ids = [c["policy_id"] for c in cites]
            self.assertTrue(is_rec)
            self.assertNotIn(
                "star-health__family-health-optima", ids,
                "Bug #71: a C/64 plan must not be presented as a "
                "recommendation")
            self.assertEqual(
                ids, ["hdfc-ergo__optima-secure"],
                "only the genuinely-strong B/75 survives — present fewer, "
                "honestly, never pad with the weak one")
            # Grade is preserved on the card (was stripped before).
            self.assertEqual(cites[0]["_grade"], "B")
            self.assertEqual(cites[0]["_overall_score"], 75.0)

    def test_only_weak_matches_yields_empty_rec_set(self):
        chunks = [
            _enriched("a__x", "Plan X", "a", 0.7, "w1", grade="C",
                      overall=64),
            _enriched("b__y", "Plan Y", "b", 0.6, "w2", grade="D",
                      overall=40),
        ]
        cites, is_rec = _build_recommendation_citations(
            reply_text="Top picks: 1. Plan X 2. Plan Y",
            retrieved_chunks_all=chunks,
            marked_policy_ids=["a__x", "b__y"])
        self.assertTrue(is_rec)
        self.assertEqual(
            cites, [],
            "no strong matches ⇒ empty rec set; caller must NOT resurrect "
            "the recall dump with weak plans")

    def test_ranked_strictly_best_first_by_overall(self):
        chunks = [
            _enriched("a__c", "Plan C", "a", 0.95, "o1", grade="B",
                      overall=72),
            _enriched("b__a", "Plan A", "b", 0.40, "o2", grade="A",
                      overall=88),
        ]
        cites, _ = _build_recommendation_citations(
            reply_text="1. Plan C 2. Plan A",
            retrieved_chunks_all=chunks,
            marked_policy_ids=["a__c", "b__a"])
        self.assertEqual(
            [c["policy_id"] for c in cites], ["b__a", "a__c"],
            "strongest fit (A/88) must be #1, NOT the LLM/cosine order")

    def test_scorecard_unavailable_fails_open(self):
        # No _grade / _overall_score at all (scorecard module down). The
        # pipeline already vetted these — keep them, don't wipe the list.
        chunks = [
            _enriched("a__p1", "Alpha One", "a", 0.5, "n1"),
            _enriched("b__p2", "Beta Two", "b", 0.4, "n2"),
        ]
        cites, is_rec = _build_recommendation_citations(
            reply_text="Options: 1. Alpha One 2. Beta Two",
            retrieved_chunks_all=chunks, marked_policy_ids=[])
        self.assertTrue(is_rec)
        self.assertEqual([c["policy_id"] for c in cites],
                         ["a__p1", "b__p2"])


class TestBug71EndToEnd(_HandleTurnHarness):
    def test_handle_turn_does_not_present_c_grade_as_recommendation(self):
        # Retrieval surfaces a strong B/75 and a weak C/64 for this user's
        # own profile. Even if the LLM names both, the cited recommendation
        # set must exclude the C/64.
        sess = self._ready_session()
        self._retrieve_chunks = [
            _enriched("hdfc-ergo__optima-secure", "my:Optima Secure",
                      "hdfc-ergo", 0.71, "c1", grade="B", overall=75),
            _enriched("star-health__family-health-optima",
                      "Star Family Health Optima", "star-health",
                      0.68, "c2", grade="C", overall=64),
        ]
        self._gemini_script = [
            _tool_payload([_fc_part("retrieve_policies",
                                    {"query": "metro family 10 lakh"})]),
            _tool_payload([_fc_part("mark_recommendation", {
                "policy_ids": ["hdfc-ergo__optima-secure",
                               "star-health__family-health-optima"]})]),
            _text_payload(
                "Here are my recommendations: my:Optima Secure and "
                "Star Family Health Optima."),
        ]
        r = _run(single_brain.handle_turn(sess, "recommend plans for me"))
        ids = [c["policy_id"] for c in r.citations]
        self.assertIn("hdfc-ergo__optima-secure", ids)
        self.assertNotIn(
            "star-health__family-health-optima", ids,
            "Bug #71 end-to-end: the C/64 plan must not reach the user's "
            "cited recommendation set")
        self.assertEqual(len(ids), 1,
                         "present fewer (the one strong plan), not padded")

    def test_handle_turn_only_weak_matches_no_padded_recommendation(self):
        sess = self._ready_session()
        self._retrieve_chunks = [
            _enriched("a__x", "Plan Xray", "a", 0.7, "w1", grade="C",
                      overall=64),
            _enriched("b__y", "Plan Yankee", "b", 0.6, "w2", grade="D",
                      overall=40),
        ]
        self._gemini_script = [
            _tool_payload([_fc_part("retrieve_policies",
                                    {"query": "plans"})]),
            _tool_payload([_fc_part("mark_recommendation", {
                "policy_ids": ["a__x", "b__y"]})]),
            _text_payload("Plan Xray and Plan Yankee are options."),
        ]
        r = _run(single_brain.handle_turn(sess, "show me recommendations"))
        self.assertEqual(
            r.citations, [],
            "only weak matches ⇒ no policy presented as a recommendation")


if __name__ == "__main__":
    unittest.main(verbosity=2)