"""Retrieval-side filters + guards (A3 / A6 audit fixes). This module is a SIDECAR to the retrieval path (`rag/retrieve.py` and the single-brain tool layer in `backend/brain_tools.py`). It is deliberately a separate file so retrieval-correctness work stays isolated from the turn-handling code. Public API: apply_profile_filter(chunks, profile) -> list[RetrievedChunk] Drop chunks for policies the user is demographically ineligible for. apply_eligibility_filter(chunks, profile) -> list[RetrievedChunk] Drop chunks the user is STRUCTURALLY ineligible for / that clearly contradict an explicit stated need (KI-278 / KI-279): - top-up / super-top-up when the user has no existing base cover - plans whose max sum-insured cannot meet the requested SI - high co-pay plans when the user explicitly wants zero co-pay - fixed-benefit products (hospital daily cash / personal accident / critical illness / cancer) when the profile clearly signals the user wants comprehensive INDEMNITY cover (KI-279) rank_by_profile_fit(chunks, profile) -> list[RetrievedChunk] Re-order surviving chunks so the plans that best match the stated needs (grade, co-pay, SI headroom) rank above weak-fit plans whose only advantage was raw cosine similarity (KI-278). bypass_cosine_for_exact_match(chunks, query) -> list[RetrievedChunk] | None If the query contains an IRDAI UIN or an exact policy name, return a substring-matched chunk list (caller can use this instead of cosine). Returns None if no exact-match signal in the query. empty_retrieval_guard(chunks, intent) -> dict | None Return a structured "empty_retrieval" signal when filtered chunk count is below the minimum for a recommendation intent. enforce_citation_grounding(chunks) -> list[RetrievedChunk] Reject chunks missing policy_id / policy_name / chunk_offset. dedup_by_policy(chunks) -> list[RetrievedChunk] Within a top-K, keep highest-score chunk per policy_id. All functions are pure / side-effect-free; safe to call from the single-brain tool layer or from rag/retrieve.py without import-cycle risk. """ from __future__ import annotations import re from typing import Any, Iterable, Optional # Shared canonical-identity / dedup rule. The same # UIN-primary + product_key invariant the marketplace endpoint uses # (main.py /api/policies/all). Imported (not reimplemented) so the # recommender and the marketplace agree on "the same policy". The module # is dependency-free so this stays import-cycle-safe. from backend.policy_identity import canonical_key # We avoid importing RetrievedChunk at module-import time to keep this file # import-cycle-safe. Instead we type-duck on attributes. # --------------------------------------------------------------------------- # Constants — keep tunables here so they're greppable from one place. # --------------------------------------------------------------------------- # Age tolerance for the profile pre-filter. The catalog's min_entry_age / # max_entry_age is the regulator-filed entry age, but real underwriters allow # a small grace band; we mirror that grace here so we don't accidentally # drop a perfectly fine policy because the user is one year off the edge. PROFILE_AGE_TOLERANCE = 2 # A "senior-only" policy is one whose marketing/eligibility makes it # inappropriate for adults under 50. We detect this from the policy name # AND from the min_entry_age metadata (>=60 ⇒ senior-only). SENIOR_ONLY_NAME_RE = re.compile( r"\b(senior|red\s*carpet|silver|elder|varisht|varistha|" r"sixty\s*plus|60\s*plus|seniority|golden\s*years)\b", flags=re.IGNORECASE, ) # Adult-only / young-adult-only plans typically cap at 50 or 55 and have # no senior variant. If the user is >=60 these are inappropriate # (they need a senior variant of the SAME insurer/family instead). ADULT_ONLY_NAME_RE = re.compile( r"\b(young\s*star|young\s*adult|millennial|gen\s*z|under\s*45|" r"early\s*career|first\s*time)\b", flags=re.IGNORECASE, ) # Maternity-themed policies. If the profile has no female adult AND no # maternity goal, these are noise. MATERNITY_NAME_RE = re.compile( r"\b(maternity|mother\s*&?\s*baby|mother\s*to\s*be|" r"new\s*born|joy|stork|baby\s*shield|pregnancy)\b", flags=re.IGNORECASE, ) # --------------------------------------------------------------------------- # Eligibility / profile-fit tunables # --------------------------------------------------------------------------- # A plan is a TOP-UP / SUPER-TOP-UP (only useful ALONGSIDE an existing base # policy or above a large out-of-pocket deductible) when ANY of these hold: # 1. policy_type_indemnity_or_fixed contains "top_up" / "top-up" / "super" # 2. the policy NAME matches TOP_UP_NAME_RE # 3. it carries a large aggregate `deductible_amount` (the de-facto base # cover the user must already hold) — see TOP_UP_DEDUCTIBLE_FLOOR_INR TOP_UP_NAME_RE = re.compile( r"\b(super[\s\-]?top[\s\-]?up|top[\s\-]?up|topup|super[\s\-]?top)\b", flags=re.IGNORECASE, ) # An aggregate deductible at/above this rupee floor means the plan only pays # AFTER the insured has already spent this much (from a base policy or # pocket) — i.e. it is functionally a top-up and unusable as a sole first # policy. ₹1,00,000 is conservative: real top-ups deductibles start ~₹2-3L; # a genuine non-top-up plan never carries a six-figure aggregate deductible. TOP_UP_DEDUCTIBLE_FLOOR_INR = 100_000 # When the user explicitly wants zero co-pay AND is not income-constrained # (can afford full cover), any plan whose mandatory co-payment exceeds this # percentage is a hard mismatch and is dropped. A 0 here means "drop ANY # plan with a non-zero mandatory co-pay for a strict zero-copay user". ZERO_COPAY_USER_MAX_COPAY_PCT = 0 # Even when the user did NOT explicitly demand zero co-pay, a punitive # co-pay (>= this) is dropped because no metric recovers from a 50% claim # haircut for a metro / high-income first-time buyer. PUNITIVE_COPAY_PCT = 40 # SI headroom: a plan is SI-eligible only if its largest sum-insured option # is at least the requested SI. (We do NOT require an exact tier — most # insurers interpolate; but a plan whose ceiling is below the ask cannot # deliver the cover the user said they need.) # Minimum chunk count to attempt a recommendation. Below this we ask for # one more clarifier instead of letting the brain hallucinate. MIN_CHUNKS_FOR_RECOMMENDATION = 3 # Recommendation-style intents — the empty-retrieval guard fires only on these. # Other intents (faq, regulatory, smalltalk) tolerate sparse retrieval. _RECOMMENDATION_INTENTS = { "recommend", "recommendation", "compare", "comparison", "suggest", "shortlist", "best_policy", "pick_for_me", } # IRDAI UIN pattern. Real UINs look like "IRDA/HLT/HDFC/V.I/188/14-15" # or "IRDAI/HLT/HDFC/V.I/188/14-15" — we accept both. UIN_RE = re.compile( r"\b(?:IRDAI?|UIN)[/:]\s*[A-Z0-9./\-]{6,}", flags=re.IGNORECASE, ) # --------------------------------------------------------------------------- # Fixed-benefit exclusion for comprehensive-indemnity intent # --------------------------------------------------------------------------- # A FIXED-BENEFIT product (hospital daily cash, personal accident, critical # illness, cancer / other defined-benefit) pays a fixed lump-sum / per-day # amount rather than reimbursing actual hospitalisation expenses. It is a # legitimate SUPPLEMENT but a wrong PRIMARY recommendation for a buyer who # clearly wants comprehensive INDEMNITY cover (their first health policy, # a desired sum insured, no existing base cover). This regex mirrors # scorecard._FIXED_BENEFIT_RE so classification stays consistent across the # scorecard and the retrieval filter (single behavioural contract). FIXED_BENEFIT_NAME_RE = re.compile( r"hospital[\s_-]*cash|hospi[\s_-]*cash|daily[\s_-]*cash|" r"personal[\s_-]*accident|critical[\s_-]*illness|" r"criti[\s_-]*(?:care|medicare)|\bcancer\b|wellsurance|hospi[\s_-]*care", flags=re.IGNORECASE, ) # Tokens in policy_type / policy_type_indemnity_or_fixed that mark a # fixed-benefit / defined-benefit product (NOT indemnity reimbursement). _FIXED_BENEFIT_TYPE_TOKENS = ( "fixed", "benefit", "defined", "hospital_cash", "hospital cash", "daily_cash", "daily cash", "personal_accident", "personal accident", "critical_illness", "critical illness", "cancer", ) # Goal phrases that mean the user EXPLICITLY wants a supplement / add-on / # top-up / fixed-benefit product — the comprehensive-indemnity intent must # NOT fire for these (we must keep showing them PA / CI / hospital-cash / # top-up products on purpose). _SUPPLEMENT_GOAL_TOKENS = ( "supplement", "add-on", "add on", "addon", "rider", "top-up", "top up", "topup", "super top", "critical illness", "critical-illness", "ci only", "ci cover", "personal accident", "accident cover", "accident plan", "hospital cash", "daily cash", "cash plan", "cancer cover", "cancer plan", "defined benefit", "fixed benefit", "alongside", "in addition to", "on top of", ) # Goal phrases that POSITIVELY signal a primary comprehensive health plan. _COMPREHENSIVE_GOAL_TOKENS = ( "first", "new policy", "fresh", "primary", "comprehensive", "general health", "health cover", "main cover", "base cover", "family floater", "indemnity", "hospitalisation", "hospitalization", "cover my", "protect my family", "medical cover", "buy health", ) # --------------------------------------------------------------------------- # Unified recommendation-fit gate tunables # --------------------------------------------------------------------------- # Recommendation-fit is SYSTEMIC: the cited-card list and the advisory prose # are gated by the SAME fitness logic. Two hard gates (entry-age for the # INSURED person; required feature = maternity/newborn) plus ranking signals # for grade/rank ordering and the cost-objective lead. # Entry-age grace. The catalog's max_entry_age is the regulator-filed entry # age; underwriters allow a small grace band (mirrors PROFILE_AGE_TOLERANCE). ENTRY_AGE_TOLERANCE = 2 # Goal phrases that mean COST is the dominant objective (P5). When present, # ranking must not let a pricier plan win purely on cosine — the lowest-cost # *appropriate* plan leads. We proxy "cost" by (low/zero co-pay surcharge + # scorecard Cost Predictability already folded into _overall_score); the # explicit signal here flips a stronger cosine-discount so a cheap plan is # not buried under a higher-cosine pricier one. _COST_OBJECTIVE_TOKENS = ( "cost_optimize", "cheapest", "cheap", "lowest premium", "low premium", "budget", "affordable", "tight", "money is tight", "save money", "least expensive", "minimum premium", "low cost", ) # Profile/goal phrases that say the policy INSURES senior parents (so the # entry-age gate must use the PARENTS' age, not the paying child's age). _PARENTS_COVER_TOKENS = ( "parent", "parents", "mother", "father", "mom", "dad", "senior citizen", "senior citizens", "elderly", ) # Explicit maternity / newborn requirement phrases. When the profile states # this need, plans whose facts CONFIRM maternity/newborn rank above those # that do not (and unverified ones are ranked strictly below confirmed). _MATERNITY_NEED_TOKENS = ( "maternity", "pregnan", "newborn", "new born", "new-born", "delivery cover", "childbirth", "planning a child", "planning another", "expecting", "baby cover", "having a baby", ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _meta(chunk: Any) -> dict: """Pull metadata off a chunk regardless of whether the caller hands us a RetrievedChunk dataclass or a raw dict. RetrievedChunk stores fields as top-level attributes; ingestion-side code sometimes passes a dict.""" if isinstance(chunk, dict): return chunk out = {} for k in ( "policy_id", "policy_name", "insurer_slug", "doc_type", "chunk_idx", "chunk_offset", "min_entry_age", "max_entry_age", "score", "text", ): if hasattr(chunk, k): out[k] = getattr(chunk, k) return out def _meta_full(chunk: Any) -> dict: """Like `_meta` but also surfaces the KI-278 enriched structured-fact keys (policy_type_indemnity_or_fixed / deductible_amount / co_payment_pct / sum_insured_options / _grade / _overall_score). `brain_tools.retrieve_policies` passes plain dicts, so the common path is just `return chunk`. For dataclass chunks we duck-type the union of citation + enriched fields so the eligibility/ranking rules still work if a future caller hands us a dataclass. """ if isinstance(chunk, dict): return chunk out = _meta(chunk) for k in ( "policy_type_indemnity_or_fixed", "deductible_amount", "co_payment_pct", "sum_insured_options", "_grade", "_overall_score", # KI-280 — unified-gate signals (entry-age / required-feature / # canonical dedup identity). "uin_code", "max_entry_age", "min_entry_age", "maternity_coverage", "newborn_coverage", "policy_type", ): if hasattr(chunk, k): out[k] = getattr(chunk, k) return out def _profile_get(profile: Any, key: str, default=None): """Profile may be a dataclass (needs_finder.Profile) OR a dict — accept either.""" if profile is None: return default if isinstance(profile, dict): return profile.get(key, default) return getattr(profile, key, default) def _profile_has_female_adult(profile: Any) -> bool: """Heuristic: profile covers a female adult if dependents includes spouse (the user might be the female adult, or spouse may be). We treat 'spouse' tokens as a positive signal and 'self+spouse', 'family' shapes likewise. Conservative: return True when uncertain (better to keep maternity chunks than to wrongly drop them).""" dep = _profile_get(profile, "dependents") or "" if not isinstance(dep, str): return True dep_lower = dep.lower() if any(tok in dep_lower for tok in ("spouse", "wife", "family", "kids", "child")): return True # No spouse / no family signal AND explicit self-only → no female adult if dep_lower in ("self", "self_only", "individual", "just_me"): return False return True # default permissive def _profile_has_maternity_goal(profile: Any) -> bool: goal = _profile_get(profile, "primary_goal") or "" if not isinstance(goal, str): return False return "maternity" in goal.lower() or "pregnan" in goal.lower() or "baby" in goal.lower() # --------------------------------------------------------------------------- # (1) Profile-fit pre-filter (A3 / A6) # --------------------------------------------------------------------------- def apply_profile_filter(chunks: Iterable[Any], profile: Any) -> list[Any]: """Drop chunks that are demographically inappropriate for this user. Rules: - min_entry_age > age + 2 → drop - max_entry_age < age - 2 → drop - senior-only plan AND age < 50 → drop (user is too young) - adult-only plan AND age >= 60 → drop (user needs senior variant) - maternity plan AND no female adult / no maternity goal → drop Profile chunks (doc_type == 'profile') and regulatory chunks (doc_type == 'regulatory' / 'review') are NEVER dropped here — those aren't policies and the demographic rules don't apply. Conservative on missing data: if a chunk doesn't expose min/max entry age metadata, we DO NOT drop it on those rules (we still apply the name-based senior/adult/maternity rules where the name pattern matches). """ chunks_list = list(chunks) if not chunks_list: return chunks_list # The demographic gates (numeric age-range, senior-only, adult-only) # reason about the age of the person the policy actually INSURES, not # the payer's. When the policy covers parents (~70), using the paying # child's age (e.g. 36) would wrongly DROP a senior-citizen plan # ("min_entry_age>=60 AND payer<50") and wrongly KEEP a max-entry-65 # plan. `_oldest_insured_age` returns the eldest parent's age when # insuring parents, else the profile's own age. Falls back to the raw # profile age if no resolved signal. _resolved_age = _oldest_insured_age(profile) if _resolved_age is None: _resolved_age = _profile_get(profile, "age") age = _resolved_age # If age unknown, only the maternity rule can fire — keep everything else. has_age = isinstance(age, int) has_female_adult = _profile_has_female_adult(profile) maternity_goal = _profile_has_maternity_goal(profile) kept: list[Any] = [] for ch in chunks_list: m = _meta(ch) doc_type = (m.get("doc_type") or "").lower() # Never drop non-policy chunks via demographic filter. # #52 — `user_upload` is a globally-visible uploaded marketplace # doc: a Q&A TARGET, not a demographically-ranked recommendable # corpus policy. Exempt it exactly like regulatory/review so a # question literally about the uploaded document isn't dropped # because the (often anonymous) asker's age/eligibility doesn't # match the uploaded plan. if doc_type in ("profile", "regulatory", "review", "user_upload"): kept.append(ch) continue name = (m.get("policy_name") or "").strip() min_age = m.get("min_entry_age") max_age = m.get("max_entry_age") # Numeric age-range gate if has_age: try: if isinstance(min_age, (int, float)) and min_age > age + PROFILE_AGE_TOLERANCE: continue if isinstance(max_age, (int, float)) and max_age < age - PROFILE_AGE_TOLERANCE: continue except TypeError: pass # metadata corruption — fall through to name rules # Senior-only inferred from name OR from min_entry_age >= 60 is_senior_only = bool(SENIOR_ONLY_NAME_RE.search(name)) or ( isinstance(min_age, (int, float)) and min_age >= 60 ) if is_senior_only and has_age and age < 50: continue # Adult-only inferred from name (no metadata signal exists for this). # If user is 60+ AND policy looks adult-only AND has max_age < 60, drop. is_adult_only_name = bool(ADULT_ONLY_NAME_RE.search(name)) if has_age and age >= 60 and is_adult_only_name: continue if has_age and age >= 60 and isinstance(max_age, (int, float)) and max_age < 60: continue # Maternity gate — only drop if BOTH conditions fail if MATERNITY_NAME_RE.search(name) and not maternity_goal and not has_female_adult: continue kept.append(ch) return kept # --------------------------------------------------------------------------- # (1b) Eligibility filter (KI-278) — structural mismatch hard-drop # --------------------------------------------------------------------------- def _as_int(v: Any) -> Optional[int]: """Best-effort int coercion (bool excluded — it's an int subclass).""" if v is None or isinstance(v, bool): return None if isinstance(v, (int, float)): return int(v) try: return int(str(v).strip()) except (TypeError, ValueError): return None def _max_sum_insured(chunk_meta: dict) -> Optional[int]: """Largest SI the policy can offer, from the enriched `sum_insured_options` list. None when the fact is absent (don't drop on missing data).""" opts = chunk_meta.get("sum_insured_options") if not opts: return None nums: list[int] = [] if isinstance(opts, (list, tuple)): for o in opts: n = _as_int(o) if n is not None: nums.append(n) else: n = _as_int(opts) if n is not None: nums.append(n) return max(nums) if nums else None def _is_top_up(chunk_meta: dict) -> bool: """A plan is a top-up / super-top-up when ANY structural signal fires: 1. policy_type_indemnity_or_fixed says top_up / super 2. the policy name matches TOP_UP_NAME_RE 3. it carries a large aggregate deductible (>= TOP_UP_DEDUCTIBLE_FLOOR) """ ptype = str(chunk_meta.get("policy_type_indemnity_or_fixed") or "").lower() if "top" in ptype or "super_top" in ptype or "super-top" in ptype: return True name = str(chunk_meta.get("policy_name") or "") if TOP_UP_NAME_RE.search(name): return True ded = _as_int(chunk_meta.get("deductible_amount")) if ded is not None and ded >= TOP_UP_DEDUCTIBLE_FLOOR_INR: return True return False def _user_has_base_cover(profile: Any) -> bool: """True when the user already holds base health cover. A first-time buyer (existing_cover_inr == 0 / None / falsy, OR primary_goal looks like a first purchase) has NO base cover, so a top-up is unusable.""" existing = _as_int(_profile_get(profile, "existing_cover_inr")) if existing and existing > 0: return True goal = str(_profile_get(profile, "primary_goal") or "").lower() # "first_buy" / "first policy" / "first-time" → definitively no base. if any(tok in goal for tok in ("first", "new policy", "fresh")): return False # No explicit existing cover and no first-buy signal → treat as no base # cover (conservative: a top-up only helps someone who KNOWS they have a # base; absent that signal, don't surface an unusable product). return False def _user_wants_zero_copay(profile: Any) -> bool: copay = _as_int(_profile_get(profile, "copay_pct")) return copay is not None and copay <= ZERO_COPAY_USER_MAX_COPAY_PCT # --------------------------------------------------------------------------- # KI-279 — fixed-benefit classification + comprehensive-indemnity intent # --------------------------------------------------------------------------- def _is_fixed_benefit_chunk(chunk_meta: dict) -> bool: """True when the policy is a FIXED-BENEFIT / defined-benefit product (hospital daily cash, personal accident, critical illness, cancer) rather than an indemnity hospitalisation-reimbursement plan. Detection mirrors scorecard._is_fixed_benefit so the scorecard and the retrieval filter classify identically (one behavioural contract): 1. The canonical type key `policy_type_indemnity_or_fixed` contains a fixed / benefit / defined token. 2. Fallback — the raw catalog key `policy_type` contains a fixed-benefit token (hospital_cash / personal_accident / ...). Critically, several curated files (e.g. Star Hospital Cash) carry the type ONLY here. 3. Last-resort — the policy_id / policy_name matches FIXED_BENEFIT_NAME_RE (hospital cash / daily cash / PA / CI / cancer / wellsurance). Conservative: if NO type signal exists AND the name doesn't match, the policy is treated as indemnity (not dropped). """ # Canonical curated key first. canon = str(chunk_meta.get("policy_type_indemnity_or_fixed") or "").lower() if canon and any(tok in canon for tok in _FIXED_BENEFIT_TYPE_TOKENS): return True if canon and "indemnity" in canon: return False # explicitly indemnity — never a false positive # Raw catalog type key (the Star-Hospital-Cash case). raw = str(chunk_meta.get("policy_type") or "").lower() if raw and any(tok in raw for tok in _FIXED_BENEFIT_TYPE_TOKENS): return True if raw and "indemnity" in raw: return False # Name / id fallback. blob = f"{chunk_meta.get('policy_id','')} {chunk_meta.get('policy_name','')}" return bool(FIXED_BENEFIT_NAME_RE.search(blob)) def _user_explicitly_wants_supplement(profile: Any) -> bool: """True only when primary_goal explicitly names a supplement / add-on / top-up / PA / CI / hospital-cash / cancer / defined-benefit product. Used to SUPPRESS the fixed-benefit ranking demotion (we must not bury a product the user explicitly asked for).""" goal = str(_profile_get(profile, "primary_goal") or "").lower().strip() if not goal: return False return any(tok in goal for tok in _SUPPLEMENT_GOAL_TOKENS) def _insures_parents(profile: Any) -> bool: """True when the policy is being bought to cover the user's PARENTS / seniors (so the entry-age gate must test the parents' age, not the paying child's). Signals (any): parents_to_insure flag, a parents_age_max value, dependents mentioning parents, or a parents-cover / senior-citizen primary_goal.""" if _profile_get(profile, "parents_to_insure") is True: return True if _as_int(_profile_get(profile, "parents_age_max")) is not None: return True dep = str(_profile_get(profile, "dependents") or "").lower() goal = str(_profile_get(profile, "primary_goal") or "").lower() blob = f"{dep} {goal}" return any(tok in blob for tok in _PARENTS_COVER_TOKENS) def _oldest_insured_age(profile: Any) -> Optional[int]: """The age of the OLDEST person the policy must accept at entry. When the policy insures parents, that is `parents_age_max` (the eldest parent). Otherwise it is the profile's own `age`. Returns None when no usable age signal exists (the gate then stays conservative and does not drop on missing data — same philosophy as apply_profile_filter). """ if _insures_parents(profile): pa = _as_int(_profile_get(profile, "parents_age_max")) if pa is not None: return pa # Insures parents but their age unknown — fall through to own age # only if it is itself senior; otherwise None (don't gate blind). own = _as_int(_profile_get(profile, "age")) return own if (own is not None and own >= 60) else None return _as_int(_profile_get(profile, "age")) def _profile_requires_maternity(profile: Any) -> bool: """True when the profile EXPLICITLY needs maternity / newborn cover. Checks primary_goal first (the strongest signal — single_brain folds 'maternity' into the goal), then a dedicated maternity flag if present. Conservative: only fires on an explicit token, never inferred from a female adult alone (that would over-trigger and wrongly demote good plans for users who never asked for maternity).""" if _profile_has_maternity_goal(profile): return True goal = str(_profile_get(profile, "primary_goal") or "").lower() if any(tok in goal for tok in _MATERNITY_NEED_TOKENS): return True # A dedicated boolean slot, if a future Profile grows one. for key in ("needs_maternity", "maternity_required", "wants_maternity"): if _profile_get(profile, key) is True: return True return False def _cost_is_primary_objective(profile: Any) -> bool: """True when the user's stated objective is dominated by cost (P5 — 'cheapest decent cover, lowest premium is my top priority').""" goal = str(_profile_get(profile, "primary_goal") or "").lower() band = str(_profile_get(profile, "budget_band") or "").lower() blob = f"{goal} {band}" return any(tok in blob for tok in _COST_OBJECTIVE_TOKENS) def _chunk_confirms_maternity(chunk_meta: dict) -> Optional[bool]: """Tri-state: True (facts confirm maternity OR newborn cover), False (facts explicitly say neither), None (unverified — no fact present). None must rank BELOW True but is NOT hard-dropped (the curated facts are incomplete; a hard drop would wrongly hide good plans whose maternity flag simply wasn't curated yet).""" mat = chunk_meta.get("maternity_coverage") nb = chunk_meta.get("newborn_coverage") def _tri(v): if isinstance(v, bool): return v if isinstance(v, str): s = v.strip().lower() if s in ("true", "yes", "y", "covered"): return True if s in ("false", "no", "n", "not covered", "excluded"): return False if isinstance(v, dict) and "covered" in v: return bool(v.get("covered")) return None tm, tn = _tri(mat), _tri(nb) if tm is True or tn is True: return True if tm is False or tn is False: return False return None def _wants_comprehensive_indemnity(profile: Any) -> bool: """Conservatively detect that the user wants a PRIMARY COMPREHENSIVE INDEMNITY health plan (so a fixed-benefit daily-cash / PA / CI product is the wrong top pick). Fires ONLY when the signal is clear and NO explicit supplement signal is present: • NOT an explicit supplement / add-on / top-up / PA / CI / cancer / hospital-cash goal (hard veto — these users WANT fixed-benefit). • The user has NO existing base cover (first-time buyer). • AND at least one strong positive signal: - primary_goal reads as a first / primary / comprehensive health-cover goal, OR - a desired_sum_insured_inr is present (you only state a target cover amount when you want a real indemnity plan). Returns False on ambiguity (better to keep a fixed-benefit chunk and let ranking demote it than to wrongly hard-drop for an unclear case). """ goal = str(_profile_get(profile, "primary_goal") or "").lower().strip() # Hard veto — the user explicitly asked for a supplement / fixed-benefit # style product. Never hide those from them. if goal and any(tok in goal for tok in _SUPPLEMENT_GOAL_TOKENS): return False # A user who already holds base cover is plausibly shopping for a # supplement; don't aggressively hard-drop fixed-benefit for them. if _user_has_base_cover(profile): return False desired_si = _as_int(_profile_get(profile, "desired_sum_insured_inr")) has_desired_si = desired_si is not None and desired_si > 0 goal_is_comprehensive = bool(goal) and any( tok in goal for tok in _COMPREHENSIVE_GOAL_TOKENS ) # Clear signal = (first/primary/comprehensive goal) OR (a desired SI was # stated). Either alone, combined with "no base cover" and "no # supplement veto", is a confident comprehensive-indemnity intent. return goal_is_comprehensive or has_desired_si def apply_eligibility_filter(chunks: Iterable[Any], profile: Any) -> list[Any]: """Drop chunks the user is STRUCTURALLY ineligible for, or that plainly contradict an explicit stated need (KI-278). Hard-drop rules (policy chunks only — profile/regulatory/review chunks are never policies and pass through untouched): 1. TOP-UP GATE — a top-up / super-top-up plan is dropped when the user has no existing base cover (first-time buyer). A top-up only pays above a base policy / large deductible the user does not have. 2. SI FLOOR — when the user stated a desired sum insured, drop any plan whose largest SI option is below that requirement (it cannot deliver the cover the user said they need). 3. ZERO-COPAY GATE — when the user explicitly wants zero co-pay, drop any plan with a mandatory co-payment above ZERO_COPAY_USER_MAX_COPAY_PCT. Regardless of stated preference, drop plans with a punitive co-pay (>= PUNITIVE_COPAY_PCT). 4. FIXED-BENEFIT GATE (KI-279) — when the profile clearly signals the user wants a PRIMARY COMPREHENSIVE INDEMNITY health plan (first policy / general-cover goal + a desired SI + no existing base cover, and NO explicit supplement / PA / CI / hospital-cash / top-up goal), drop fixed-benefit products (hospital daily cash, personal accident, critical illness, cancer / defined-benefit). These pay a fixed per-day / lump-sum amount, not actual medical expenses, so they are a wrong PRIMARY recommendation. They are NEVER dropped for a user who explicitly wants a supplement. 5. ENTRY-AGE GATE — drop a plan whose max entry age cannot accept the OLDEST person the policy must insure. When the profile covers parents (~70), the gate uses the parents' age, not the paying child's (the demographic pre-filter only sees the payer's age). Conservative: fires only when both the insured age AND the policy fact are present. Conservative on missing data: a rule only fires when BOTH the profile signal AND the policy fact are present. A chunk with no enriched facts is never dropped here (it just won't get a fit boost in ranking). """ chunks_list = list(chunks) if not chunks_list: return chunks_list has_base = _user_has_base_cover(profile) wants_zero_copay = _user_wants_zero_copay(profile) desired_si = _as_int(_profile_get(profile, "desired_sum_insured_inr")) wants_comprehensive = _wants_comprehensive_indemnity(profile) # KI-280 — the age of the OLDEST person the plan must accept at entry # (parents' age when insuring parents, else the user's own age). oldest_age = _oldest_insured_age(profile) kept: list[Any] = [] for ch in chunks_list: m = _meta_full(ch) doc_type = (m.get("doc_type") or "").lower() # #52 — uploaded marketplace docs are Q&A targets; never hard-drop # them on eligibility (same class as regulatory/review). if doc_type in ("profile", "regulatory", "review", "user_upload"): kept.append(ch) continue # Rule 1 — top-up unusable without a base policy if not has_base and _is_top_up(m): continue # Rule 4 (KI-279) — fixed-benefit is a wrong PRIMARY pick when the # user clearly wants comprehensive indemnity cover. Hard-drop only # for that intent; supplement/PA/CI seekers keep seeing them. if wants_comprehensive and _is_fixed_benefit_chunk(m): continue # Rule 5 — HARD ELIGIBILITY: a plan whose max entry age cannot # accept the oldest insured person is structurally unusable (e.g. a # 70yo parent cannot be enrolled in a plan whose max_entry_age is # 65). Fires only when BOTH the age and the policy fact are present # (conservative on missing data) and only when the profile actually # insures someone whose age we know (the demographic # apply_profile_filter uses the *payer's* age and never sees the # parents' age). if oldest_age is not None: maxe = _as_int(m.get("max_entry_age")) if maxe is not None and maxe + ENTRY_AGE_TOLERANCE < oldest_age: continue # Rule 2 — SI floor if desired_si: max_si = _max_sum_insured(m) if max_si is not None and max_si < desired_si: continue # Rule 3 — co-pay gate copay = _as_int(m.get("co_payment_pct")) if copay is not None: if copay >= PUNITIVE_COPAY_PCT: continue if wants_zero_copay and copay > ZERO_COPAY_USER_MAX_COPAY_PCT: continue kept.append(ch) return kept # --------------------------------------------------------------------------- # (1c) Profile-fit ranking (KI-278) — re-order by stated-need match # --------------------------------------------------------------------------- _GRADE_POINTS = {"A": 100.0, "B": 70.0, "C": 45.0, "D": 25.0, "F": 5.0} def _fit_score(chunk_meta: dict, profile: Any, wants_zero_copay: bool, desired_si: Optional[int], *, requires_maternity: bool = False, cost_objective: bool = False) -> float: """Composite profile-fit score (higher = better). Blends raw cosine (so genuinely-relevant chunks still matter) with structural fit: • scorecard grade / overall (when enriched) • co-pay penalty (steep when the user wants zero co-pay) • SI headroom (a plan that comfortably offers the requested SI beats one that barely scrapes it) • KI-280 required-feature term: when the profile explicitly needs maternity/newborn, a plan whose facts CONFIRM it ranks above one that does not, and an UNVERIFIED plan ranks strictly below a confirmed one (but is not hard-dropped — curated facts are incomplete). • KI-280 cost-objective term: when cost is the dominant stated objective, shrink cosine's pull so a pricier plan can't win purely on vector similarity over a cheaper appropriate plan. """ try: cosine = float(chunk_meta.get("score") or 0.0) except (TypeError, ValueError): cosine = 0.0 # Cosine is ~[0,1]; scale so it contributes but never dominates fit. # KI-280: when COST is the primary objective (P5), damp cosine hard so # a higher-cosine pricier plan does not out-rank a cheaper appropriate # one. The scorecard's Cost Predictability is already inside # _overall_score; this just stops raw similarity from overriding it. cosine_weight = 6.0 if cost_objective else 30.0 score = cosine * cosine_weight # Grade / overall scorecard signal (enriched by brain_tools). grade = str(chunk_meta.get("_grade") or "").strip().upper() overall = _as_int(chunk_meta.get("_overall_score")) if overall is not None: score += float(overall) # 0-100 scorecard points elif grade in _GRADE_POINTS: score += _GRADE_POINTS[grade] # Co-pay penalty. KI-280: for a COST-primary objective where the user # explicitly tolerates a co-pay (P5 — "co-pay is completely fine if it # lowers premium"), a co-pay is a PREMIUM-REDUCING feature for this # buyer, not a defect — suppress the penalty so the cost-appropriate # plan is not wrongly demoted below a pricier zero-copay plan. The # zero-copay-user penalty path is untouched (P1 still drops high-copay). copay = _as_int(chunk_meta.get("co_payment_pct")) if copay is not None and copay > 0: if cost_objective and not wants_zero_copay: penalty = 0.0 else: penalty = copay * (3.0 if wants_zero_copay else 1.0) score -= penalty # SI headroom bonus — reward a plan that offers >= the requested SI. if desired_si: max_si = _max_sum_insured(chunk_meta) if max_si is not None: if max_si >= desired_si: score += 15.0 else: score -= 25.0 # shouldn't survive eligibility, belt-and-braces # BUG #30 (B1-c) — EXISTING-COVER term. When the user already holds ANY # base cover (even a small ₹1L employer policy), a top-up / super-top-up # is a directly relevant product that the profile-neutral scorecard is # blind to. Surface it: the bonus (+22) clears roughly one letter-grade # gap so a relevant top-up lands alongside the primary indemnity picks # (which are untouched), giving a shortlist that mixes one strong primary # plan with one relevant top-up. Inert when the user holds no base cover. existing = _as_int(_profile_get(profile, "existing_cover_inr")) if existing and existing > 0 and _is_top_up(chunk_meta): score += 22.0 # KI-280 — REQUIRED-FEATURE term. When the profile explicitly needs # maternity / newborn cover (P3), a plan whose curated facts CONFIRM it # must outrank one that does not, and an UNVERIFIED plan (fact absent) # ranks strictly below a confirmed one. We do NOT hard-drop unverified # plans — the curated facts are incomplete and a hard drop would wrongly # hide good plans whose maternity flag simply wasn't curated. The # magnitudes are chosen so a confirmed plan clears a one-letter-grade # gap (≈ +25 confirmed vs −40 explicitly-absent) — enough to put a # maternity C-grade plan above a non-maternity B-grade plan when # maternity is a stated hard requirement, without disturbing ordering # for profiles that did not ask for maternity (term is inert then). if requires_maternity: conf = _chunk_confirms_maternity(chunk_meta) if conf is True: score += 25.0 elif conf is False: score -= 40.0 # facts say NO maternity — wrong for this need else: score -= 12.0 # unverified — rank below any confirmed plan # KI-279 — fixed-benefit demotion. The eligibility filter already # hard-drops fixed-benefit for a strong comprehensive-indemnity intent; # this penalty is the belt-and-braces for weaker signals where the # intent did NOT fire (so a fixed-benefit chunk survived) but the user # still has no explicit supplement goal and an indemnity plan is also # present — the indemnity plan must still lead. Suppressed only when # the user explicitly wants a supplement / PA / CI / hospital-cash # product (then fixed-benefit is exactly what they asked for). Penalty # is large enough to sink it below any comparable indemnity plan # without affecting indemnity-vs-indemnity ordering. if not _user_explicitly_wants_supplement(profile) and _is_fixed_benefit_chunk( chunk_meta ): score -= 120.0 return score def rank_by_profile_fit(chunks: Iterable[Any], profile: Any) -> list[Any]: """Stable-sort chunks by composite profile-fit (descending). Non-policy chunks (profile/regulatory/review) keep their relative position at the front so grounding context isn't reordered away. Stable: ties preserve the incoming (cosine / dedup) order, so this only *promotes* better-fit plans — it never randomly shuffles equals. """ chunks_list = list(chunks) if len(chunks_list) <= 1: return chunks_list wants_zero_copay = _user_wants_zero_copay(profile) desired_si = _as_int(_profile_get(profile, "desired_sum_insured_inr")) requires_maternity = _profile_requires_maternity(profile) cost_objective = _cost_is_primary_objective(profile) non_policy: list[Any] = [] policy: list[Any] = [] for ch in chunks_list: dt = (_meta_full(ch).get("doc_type") or "").lower() # #52 — keep uploaded marketplace docs in the non-policy lane so # profile-fit re-ranking can't bury them below recommendable corpus # policies when the user asked about the uploaded doc itself. (non_policy if dt in ("profile", "regulatory", "review", "user_upload") else policy).append(ch) # Decorate-sort-undecorate with original index as the stable tiebreaker. decorated = [ ( -_fit_score( _meta_full(ch), profile, wants_zero_copay, desired_si, requires_maternity=requires_maternity, cost_objective=cost_objective, ), i, ch, ) for i, ch in enumerate(policy) ] decorated.sort(key=lambda t: (t[0], t[1])) ranked_policy = [ch for _, _, ch in decorated] return non_policy + ranked_policy # --------------------------------------------------------------------------- # (2) Hybrid retrieval — exact-match bypass on UIN or policy name # --------------------------------------------------------------------------- def _extract_uin(query: str) -> Optional[str]: if not query: return None m = UIN_RE.search(query) return m.group(0) if m else None def _extract_quoted_policy_name(query: str) -> Optional[str]: """Pull a quoted policy name out of the query if present.""" if not query: return None # "..." or "..." or '...' for pat in (r'"([^"]{6,})"', r"“([^”]{6,})”", r"'([^']{6,})'"): m = re.search(pat, query) if m: return m.group(1).strip() return None def bypass_cosine_for_exact_match( chunks: Iterable[Any], query: str, ) -> Optional[list[Any]]: """If the query contains an exact UIN or an exact (quoted) policy name, return a substring-matched subset of `chunks`. This is the lexical "BM25-style" fallback — when the user clearly knows what they want, we should not let cosine similarity reorder away from their literal target. Returns None when no exact-match signal is present (caller falls back to normal cosine results). Returns an empty list if a signal was present but no chunk matched — that's a useful "we know what you mean, but our catalog doesn't have it" signal for the orchestrator's empty-retrieval guard. """ chunks_list = list(chunks) uin = _extract_uin(query) quoted = _extract_quoted_policy_name(query) if not uin and not quoted: return None needles: list[str] = [] if uin: needles.append(uin.lower()) if quoted: needles.append(quoted.lower()) matched: list[Any] = [] for ch in chunks_list: m = _meta(ch) haystack = " ".join( str(m.get(k, "")) for k in ("policy_id", "policy_name", "insurer_slug", "text") ).lower() if any(n in haystack for n in needles): matched.append(ch) return matched # --------------------------------------------------------------------------- # (3) Empty-retrieval guard # --------------------------------------------------------------------------- def empty_retrieval_guard( chunks: Iterable[Any], intent: Optional[str] = None, min_chunks: int = MIN_CHUNKS_FOR_RECOMMENDATION, ) -> Optional[dict]: """Return a structured signal if a recommendation intent has too few chunks to ground an answer. The orchestrator should surface this to the user as a clarifier question instead of calling the brain. Returns None when the retrieval is healthy (or the intent doesn't need much grounding). """ chunks_list = list(chunks) intent_norm = (intent or "").lower().strip() if intent_norm and intent_norm not in _RECOMMENDATION_INTENTS: return None # FAQ / regulatory / smalltalk intents are fine with sparse retrieval if len(chunks_list) >= min_chunks: return None return { "reason": "empty_retrieval", "fallback": "Ask 1 more clarifier", "chunk_count": len(chunks_list), "min_required": min_chunks, } # --------------------------------------------------------------------------- # (4) Citation grounding — require policy_id, policy_name, chunk_offset # --------------------------------------------------------------------------- def enforce_citation_grounding(chunks: Iterable[Any]) -> list[Any]: """Drop chunks missing citation-critical fields. A citable chunk MUST expose: - policy_id (non-empty str) - policy_name (non-empty str) The chunk offset field (`chunk_offset` or legacy `chunk_idx`) is INFORMATIONAL only — it is not required for citation grounding because upstream call sites (e.g. brain_tools.retrieve_policies) build pruned dicts that intentionally omit it, and the brain cites by policy identity, not by chunk offset. Requiring an offset here would drop every chunk built by those pruned-dict call sites. """ kept: list[Any] = [] for ch in chunks: m = _meta(ch) pid = m.get("policy_id") pname = m.get("policy_name") if not pid or not isinstance(pid, str): continue if not pname or not isinstance(pname, str): continue kept.append(ch) return kept # --------------------------------------------------------------------------- # (5) Dedup by policy_id — keep highest-score chunk per policy # --------------------------------------------------------------------------- def dedup_by_policy(chunks: Iterable[Any]) -> list[Any]: """Within top-K results, collapse same-product / marketing-variant duplicates to one chunk, keeping the highest-scoring chunk. Preserves the order of first appearance of each kept product. Keying is by the SHARED canonical identity (policy_identity.canonical_key) — UIN-primary, product_key fallback — NOT the raw policy_id. The same product can appear under different policy_ids: a marketing rename ("my:Optima Secure" vs "my:Optima Secure (older variant)" — same UIN) or two doctype siblings ("...__wordings" vs "...__brochure" — same product_key). policy_id-only dedup would let both through; canonical-identity keying collapses them. This reuses the exact rule the marketplace endpoint uses so the recommender and the marketplace agree on "the same policy". """ best: dict[str, Any] = {} order: list[str] = [] for ch in chunks: m = _meta_full(ch) key = canonical_key(m) score = m.get("score") if score is None: # try attribute path score = getattr(ch, "score", 0.0) try: score_f = float(score) except (TypeError, ValueError): score_f = 0.0 if key not in best: best[key] = (ch, score_f) order.append(key) continue prev_chunk, prev_score = best[key] if score_f > prev_score: best[key] = (ch, score_f) return [best[key][0] for key in order] # --------------------------------------------------------------------------- # Convenience — compose the standard retrieval-side pipeline. # --------------------------------------------------------------------------- def filter_pipeline( chunks: Iterable[Any], profile: Any = None, query: str = "", intent: Optional[str] = None, ) -> tuple[list[Any], Optional[dict]]: """Run the standard A3/A6 + KI-278 retrieval-side pipeline: 1. Citation grounding (reject malformed chunks) 2. Profile pre-filter (drop demographically inappropriate policies) 3. Eligibility filter (KI-278 — drop structural mismatches: top-up w/o base cover, SI floor, co-pay; KI-279 — drop fixed-benefit products when the user wants comprehensive indemnity) 4. Exact-match bypass (if query has a UIN / quoted name, swap in) 5. Dedup by policy_id 6. Profile-fit ranking (KI-278 — promote best-fit over raw cosine; KI-279 — demote any surviving fixed-benefit below indemnity options) 7. Empty-retrieval guard Returns (filtered_chunks, guard_signal_or_None). When guard_signal is set, the caller should NOT pass `filtered_chunks` to the brain — they should surface the clarifier instead. KI-278 ordering note: the eligibility filter runs on the demographically- fitted set and BEFORE the empty-retrieval guard, so a profile whose only matches are ineligible (e.g. a first-time buyer where cosine only found top-ups) correctly trips the clarifier guard instead of recommending an unusable product. The exact-match bypass still wins when the user names a specific policy (they may legitimately want to read about a top-up). """ grounded = enforce_citation_grounding(chunks) fitted = apply_profile_filter(grounded, profile) eligible = apply_eligibility_filter(fitted, profile) exact = bypass_cosine_for_exact_match(grounded, query) if exact is not None and exact: # Exact match wins over cosine when the user clearly named a target. # The user explicitly asked about THIS policy — show it even if it # would fail the eligibility/fit gates (they may be researching a # top-up or a high-copay plan on purpose). Demographic filter still # applies (a 25yo asking about a senior plan still gets steered). eligible = apply_profile_filter(exact, profile) deduped = dedup_by_policy(eligible) ranked = rank_by_profile_fit(deduped, profile) guard = empty_retrieval_guard(ranked, intent=intent) return ranked, guard