"""Recommendation-transparency (deploy-#2 follow-up, 2026-05-16). CONTEXT (owner Image#8 diagnosis, CONFIRMED): the recommendation-fit gate is CORRECT — when a new hard constraint appears (e.g. user says "zero co-pay, individual only" after "Royal Sundaram Multiplier" was shown), the gate rightly drops Multiplier because it carries a co-pay. The BUG is purely conversational: the assistant SILENTLY swaps the recommendation set with no explanation, so it feels "random / dropped a policy" to the user. These tests pin the FIX (single_brain.py only — the gate logic is NOT touched): 1. Pure-derivation layer (`_recommendation_change_note` / `_constraint_reason_clause`): • constraint added → previously-cited policy dropped → returns a one-line note that NAMES the real dropped policy and ties the removal to the REAL constraint the user stated (from this turn's save_profile_field updates) — nothing invented. • set unchanged / no new constraint / no prior snapshot / empty current set → returns "" (NO spurious explanation). • canonical identity: a doctype-sibling re-id of a still-cited policy is NOT mis-reported as dropped. 2. End-to-end `handle_turn` integration (Gemini + retrieve_policies mocked at their network seams): • Turn 1: Multiplier (has co-pay) is in the cited set, snapshot persisted, NO drop note. • Turn 2: user states "zero co-pay" → fit gate drops Multiplier → reply is PREPENDED with a transparent line naming Multiplier and citing the zero-co-pay constraint. • Turn where the set is unchanged → NO note prepended. Run: .venv/bin/python -m pytest -q tests/test_recommendation_transparency.py """ from __future__ import annotations import asyncio 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.single_brain import ( # noqa: E402 _constraint_reason_clause, _recommendation_change_note, ) def _run(coro): return asyncio.new_event_loop().run_until_complete(coro) def _cite(pid, name): """A cited-card dict shaped like _build_recommendation_citations output.""" return { "chunk_id": f"{pid}#c", "policy_id": pid, "policy_name": name, "insurer_slug": pid.split("__")[0], "doc_type": "policy", "source_url": f"https://example.com/{pid}.pdf", "score": 0.5, } # ════════════════════════════════════════════════════════════════════════════ # LAYER 1 — pure derivation: _constraint_reason_clause # ════════════════════════════════════════════════════════════════════════════ class TestConstraintReasonClause(unittest.TestCase): def test_zero_copay_is_the_canonical_reason(self): # The Image#8 scenario: user said "zero co-pay" → LLM persisted # copay_pct=0 this turn. self.assertEqual( _constraint_reason_clause({"copay_pct": "0"}), "you want zero co-pay", ) self.assertEqual( _constraint_reason_clause({"copay_pct": 0}), "you want zero co-pay", ) def test_nonzero_copay_does_not_claim_zero(self): # User accepted SOME co-pay — must NOT say "you want zero co-pay". clause = _constraint_reason_clause({"copay_pct": "20"}) self.assertNotIn("zero", clause) self.assertIn("co-pay", clause) def test_other_known_fields_map_to_their_phrase(self): self.assertEqual( _constraint_reason_clause({"budget_band": "under_15k"}), "you gave a budget", ) self.assertEqual( _constraint_reason_clause({"parents_to_insure": True}), "you're now insuring parents", ) def test_unknown_field_falls_back_to_generic_not_invented(self): # An unrecognised field must NOT fabricate a specific reason. clause = _constraint_reason_clause({"some_future_field": "x"}) self.assertEqual(clause, "based on the preference you just shared") def test_no_updates_is_generic(self): self.assertEqual( _constraint_reason_clause({}), "based on the preference you just shared", ) # ════════════════════════════════════════════════════════════════════════════ # LAYER 1 — pure derivation: _recommendation_change_note # ════════════════════════════════════════════════════════════════════════════ class TestRecommendationChangeNote(unittest.TestCase): def test_drop_due_to_zero_copay_is_explained_and_named(self): # Royal Sundaram Multiplier (has a co-pay) WAS shown last turn; # this turn it's gone and the user just set copay_pct=0. prev = { "royal-sundaram__multiplier": "Royal Sundaram Multiplier", "niva-bupa__reassure-3": "ReAssure 3.0", } current = [_cite("niva-bupa__reassure-3", "ReAssure 3.0")] note = _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={"copay_pct": "0"}, ) self.assertTrue(note, "a drop+constraint must produce a note") # Names the REAL dropped policy (from the snapshot, not invented). self.assertIn("Royal Sundaram Multiplier", note) # Ties it to the REAL constraint the user stated this turn. self.assertIn("zero co-pay", note) # Does NOT wrongly name a policy that is still cited. self.assertNotIn("ReAssure", note) self.assertTrue(note.rstrip().endswith(":"), "note should lead INTO the new shortlist") def test_no_note_when_set_unchanged(self): prev = {"niva-bupa__reassure-3": "ReAssure 3.0"} current = [_cite("niva-bupa__reassure-3", "ReAssure 3.0")] # Even WITH a constraint update, nothing was dropped → no note. self.assertEqual( _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={"copay_pct": "0"}, ), "", ) def test_no_note_when_set_only_grew(self): prev = {"niva-bupa__reassure-3": "ReAssure 3.0"} current = [ _cite("niva-bupa__reassure-3", "ReAssure 3.0"), _cite("care__supreme", "Care Supreme"), ] self.assertEqual( _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={"copay_pct": "0"}, ), "", ) def test_no_note_without_a_new_constraint(self): # A set change with NO constraint persisted this turn is a normal # refinement, not a silent constraint-driven drop → stay quiet. prev = {"royal-sundaram__multiplier": "Royal Sundaram Multiplier"} current = [_cite("niva-bupa__reassure-3", "ReAssure 3.0")] self.assertEqual( _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={}, ), "", ) def test_no_note_without_prior_snapshot(self): # First recommendation ever — nothing to diff against. self.assertEqual( _recommendation_change_note( prev_snapshot={}, current_citations=[_cite("niva-bupa__reassure-3", "ReAssure 3.0")], profile_updates={"copay_pct": "0"}, ), "", ) def test_no_note_when_current_set_empty(self): # Empty cited set is a separate "no plan fits" path — there is no # "and these now fit better" to lead into. self.assertEqual( _recommendation_change_note( prev_snapshot={"royal-sundaram__multiplier": "Royal Sundaram Multiplier"}, current_citations=[], profile_updates={"copay_pct": "0"}, ), "", ) def test_doctype_sibling_reid_not_misreported_as_dropped(self): # Same product, different doctype-sibling id this turn. canonical # identity must treat it as STILL cited → no false "removed" note. prev = { "national-insurance__new-national-parivar-mediclaim__brochure": "New National Parivar Mediclaim", } current = [_cite( "national-insurance__new-national-parivar-mediclaim__wordings", "New National Parivar Mediclaim")] self.assertEqual( _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={"copay_pct": "0"}, ), "", "a re-id of a STILL-cited product must not be reported dropped", ) def test_multiple_dropped_policies_all_named(self): prev = { "royal-sundaram__multiplier": "Royal Sundaram Multiplier", "star__assure": "Star Assure", "niva-bupa__reassure-3": "ReAssure 3.0", } current = [_cite("niva-bupa__reassure-3", "ReAssure 3.0")] note = _recommendation_change_note( prev_snapshot=prev, current_citations=current, profile_updates={"copay_pct": "0"}, ) self.assertIn("Royal Sundaram Multiplier", note) self.assertIn("Star Assure", note) self.assertIn("they don't", note) # ════════════════════════════════════════════════════════════════════════════ # LAYER 2 — end-to-end handle_turn (Gemini + retrieve_policies mocked) # ════════════════════════════════════════════════════════════════════════════ 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}}]} # Catalog the fake retrieve_policies serves. Multiplier carries a co-pay; # ReAssure 3.0 / Care Supreme are zero-co-pay. _MULTIPLIER = { "chunk_id": "rs1", "policy_id": "royal-sundaram__multiplier", "policy_name": "Royal Sundaram Multiplier", "insurer_slug": "royal-sundaram", "doc_type": "policy", "source_url": "https://example.com/multiplier.pdf", "score": 0.61, "uin_code": "RSAHLIP21001V010001", } _REASSURE = { "chunk_id": "nb1", "policy_id": "niva-bupa__reassure-3", "policy_name": "ReAssure 3.0", "insurer_slug": "niva-bupa", "doc_type": "policy", "source_url": "https://example.com/reassure.pdf", "score": 0.55, "uin_code": "MAXHLIP21177V032122", } _CARE = { "chunk_id": "ch1", "policy_id": "care__supreme", "policy_name": "Care Supreme", "insurer_slug": "care", "doc_type": "policy", "source_url": "https://example.com/caresupreme.pdf", "score": 0.50, "uin_code": "CARHLIP21001V010001", } class _HandleTurnHarness(unittest.TestCase): """Drives single_brain.handle_turn with the two network seams stubbed: _gemini_call (scripted per-turn payloads) and brain_tools.retrieve_policies (scripted chunk lists). The fit gate, citation builder and transparency layer all run for real.""" def setUp(self): import os 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) # mark_recommendation gates on session.last_retrieved_chunks — # mirror what the real retrieve_policies stamps. sess = _k.get("session") if sess is not None: sess.last_retrieved_chunks = list(chunks) sess.slug_to_insurer = { c["policy_id"]: c["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): """Session with the 7 required slots filled + the post-recap pricing bundle marked skipped — the realistic precondition for a RECOMMENDATION turn. Bug #107 only attaches policy citations once 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; recommendation-transparency assertions (snapshot persists, drop note fires) must reflect that real flow.""" 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 class TestHandleTurnTransparency(_HandleTurnHarness): def test_silent_swap_becomes_explicit_on_new_constraint(self): sess = self._ready_session() # ---- Turn 1: recommend Multiplier + ReAssure (no constraint yet). self._retrieve_chunks = [_MULTIPLIER, _REASSURE] self._gemini_script = [ _tool_payload([_fc_part("retrieve_policies", {"query": "comprehensive health cover"})]), _tool_payload([_fc_part("mark_recommendation", { "policy_ids": ["royal-sundaram__multiplier", "niva-bupa__reassure-3"]})]), _text_payload( "I recommend Royal Sundaram Multiplier and ReAssure 3.0 " "for comprehensive cover."), ] r1 = _run(single_brain.handle_turn(sess, "I want health insurance")) self.assertNotIn("I've removed", r1.reply_text, "no prior shortlist → no drop note on turn 1") cited1 = {c["policy_id"] for c in r1.citations} self.assertIn("royal-sundaram__multiplier", cited1) # Snapshot of THIS turn's cited set must persist for the next diff. snap = getattr(sess, "last_recommendation_snapshot", {}) self.assertEqual( snap.get("royal-sundaram__multiplier"), "Royal Sundaram Multiplier") # ---- Turn 2: user demands zero co-pay. The gate drops Multiplier # (it has a co-pay); retrieval now only surfaces zero-co-pay plans. self._retrieve_chunks = [_REASSURE, _CARE] self._gemini_script = [ _tool_payload([_fc_part("save_profile_field", {"field": "copay_pct", "value": "0"})]), _tool_payload([_fc_part("retrieve_policies", {"query": "zero co-pay comprehensive"})]), _tool_payload([_fc_part("mark_recommendation", { "policy_ids": ["niva-bupa__reassure-3", "care__supreme"]})]), _text_payload( "ReAssure 3.0 and Care Supreme are strong zero-co-pay " "options."), ] r2 = _run(single_brain.handle_turn( sess, "I want zero co-pay, individual only")) # The swap is now EXPLAINED, naming the real dropped policy and the # real stated constraint — and prepended ahead of the rec prose. self.assertIn("Royal Sundaram Multiplier", r2.reply_text) self.assertIn("zero co-pay", r2.reply_text) self.assertIn("I've removed", r2.reply_text) self.assertLess( r2.reply_text.index("Royal Sundaram Multiplier"), r2.reply_text.index("ReAssure 3.0"), "transparency line must be PREPENDED before the rec prose") # Gate behaviour unchanged: Multiplier is no longer cited. cited2 = {c["policy_id"] for c in r2.citations} self.assertNotIn("royal-sundaram__multiplier", cited2) self.assertIn("niva-bupa__reassure-3", cited2) def test_no_spurious_note_when_set_unchanged(self): sess = self._ready_session() self._retrieve_chunks = [_REASSURE, _CARE] self._gemini_script = [ _tool_payload([_fc_part("retrieve_policies", {"query": "cover"})]), _tool_payload([_fc_part("mark_recommendation", { "policy_ids": ["niva-bupa__reassure-3", "care__supreme"]})]), _text_payload("ReAssure 3.0 and Care Supreme look good."), ] _run(single_brain.handle_turn(sess, "show me plans")) # Turn 2: a constraint is stated BUT the same set still fits → no # policy dropped → must NOT fabricate a removal note. self._retrieve_chunks = [_REASSURE, _CARE] self._gemini_script = [ _tool_payload([_fc_part("save_profile_field", {"field": "copay_pct", "value": "0"})]), _tool_payload([_fc_part("retrieve_policies", {"query": "cover"})]), _tool_payload([_fc_part("mark_recommendation", { "policy_ids": ["niva-bupa__reassure-3", "care__supreme"]})]), _text_payload("Both still fit with zero co-pay."), ] r2 = _run(single_brain.handle_turn(sess, "I want zero co-pay")) self.assertNotIn("I've removed", r2.reply_text) self.assertEqual(r2.reply_text, "Both still fit with zero co-pay.") def test_no_note_on_pure_qa_turn_after_recommendation(self): sess = self._ready_session() self._retrieve_chunks = [_MULTIPLIER, _REASSURE] self._gemini_script = [ _tool_payload([_fc_part("retrieve_policies", {"query": "cover"})]), _tool_payload([_fc_part("mark_recommendation", { "policy_ids": ["royal-sundaram__multiplier", "niva-bupa__reassure-3"]})]), _text_payload("Royal Sundaram Multiplier and ReAssure 3.0."), ] _run(single_brain.handle_turn(sess, "recommend plans")) snap_before = dict( getattr(sess, "last_recommendation_snapshot", {})) # Pure QA follow-up: no shortlist named, is_recommendation False. self._gemini_script = [ _text_payload("A waiting period is the time before a benefit " "becomes claimable."), ] r2 = _run(single_brain.handle_turn( sess, "what does waiting period mean?")) self.assertNotIn("I've removed", r2.reply_text) # Active shortlist identity must NOT be wiped by a QA turn. self.assertEqual( getattr(sess, "last_recommendation_snapshot", {}), snap_before, "QA turn must not erase the active shortlist snapshot") if __name__ == "__main__": unittest.main()