feat(voice+session+llm): KI-195 + KI-196 + KI-197 + KI-199 bundle

THE "ONE LAST REBUILD" — adaptive voice calibration + clean session
lifecycle + 3 latency cheap wins + Gemini context caching.

KI-195 — Adaptive TTS volume calibration vs user speech level.

Per user idea: bot TTS should ALWAYS be quieter than user speech at the
mic so echo loop is mathematically impossible and barge-in always wins.
Implementation:
- userSpeechRmsRef: rolling EMA of peak mic RMS while user speaks
(recording active, not TTS); rAF loop tracks samples > 0.02 floor.
- calibrateBotVolume: every 300ms during TTS, if bot_rms_at_mic >
userSpeechRms × 0.35, multiply el.volume by 0.8 (floor 0.15).
- calibratedVolumes Map persists final levels across turns.

KI-196 — Clean session/profile lifecycle (ADR-041).

CRITICAL UX fix: silent KI-040 name-based profile recall confused users
into thinking the bot was "remembering" them from a prior session.
Implementation:
- clear_session() in session_state.py (in-memory evict only; on-disk
profile JSON untouched).
- POST /api/session/clear in main.py — returns new session_id;
frontend rotates session_id + wipes chat + completion chip + chat-
history localStorage.
- Orchestrator confirmation gate: when sales_brain captures a `name`
that matches an existing profile-store entry, stage the recall in
session.pending_profile_recall instead of silently merging. Brain
emits a deterministic "Welcome back — I have your earlier profile
(... summary ...). Continue from there or start fresh?" reply.
User affirm ("yes"/"continue"/Hinglish "haan") → merge; negate
("no"/"start fresh"/"nahi") → discard. Conservative trigger: name
just captured AND ≤1 non-name slot is in profile.asked.
- "25% DONE" badge no longer counts builder-modal pre-selected
defaults — only fields the user has actually answered.
- ADR-041 documents the full spec table + rationale.

KI-197 — Three cheap latency wins.

1. _MAX_TOKENS 700 → 500 in sales_brain.py — saves ~200-400ms per
turn on Gemini Flash Lite. 500 still covers all observed 1-3
sentence sales_brain replies + JSON wrapper.
2. chat_history[-10:] → [-6:] in both messages[] and gemini_messages[]
— ~40% prompt-size reduction, saves ~100-200ms.
3. profile_rag.upsert_profile_chunk → fire-and-forget via
asyncio.create_task. The chunk doesn't need to be persisted before
the response returns to the user. Saves ~200-500ms.

Net: ~500-1000ms per turn (15-25%) shaved with zero behavior change.

KI-199 — Gemini cachedContents wired to sales_brain + TieredBrainLLM.

The sales_brain system prompt is ~6.6KB of fixed slot schema + tone
rules + few-shot examples. Without caching, Gemini reads it on every
turn (paid in input tokens AND wall-clock). With cachedContents:
- Module-level _CACHE_REGISTRY keyed by (model, sha256(system_text)),
guarded by threading.Lock. Holds {name, expires_at}.
- GoogleGeminiLLM.create_cache(): POSTs to /v1beta/cachedContents,
TTL 300s. Returns None on missing key / too-small payload / 4xx
(fail-safe — caller proceeds uncached).
- chat() accepts cached_content_name; when present sends
cachedContent: <name> + omits inline systemInstruction. On
cache-related 4xx, invalidates registry + retries with inline
systemInstruction.
- sales_brain split into _CACHED_PREAMBLE (6.6KB invariant) +
_dynamic_profile_block (per-turn KNOWN + REQUIRED REMAINING +
KI-196 welcome-back directive). Cache holds the preamble; dynamic
block sent inline per call.
- TieredBrainLLM forwards cached_content_name to Gemini tier only;
NIM and OR silently ignore (don't speak Gemini's cache protocol).
- Latency win: ~20-40% reduction on cache-hit turns (saves the
prompt-processing time on the server).

VERIFICATION:
py_compile all modified .py files — clean
npx tsc --noEmit (frontend) — clean
All imports resolve.
Live latency expected: from ~3-5s → ~2-3s per turn.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

70-docs/60-decisions/ADR-041-session-profile-lifecycle.md

@@ -0,0 +1,79 @@
+# ADR-041 — Session & profile lifecycle (KI-196)
+
+**Status:** Accepted — 2026-05-15
+**Owner:** Rohit Saraf
+**Related KIs:** KI-020 (legacy clear-chat), KI-040 (silent name-based recall), KI-062 (persona_id keying), KI-077 (named profile recovery), KI-118 (in-memory-only sessions + cross-session name match), KI-167 (LLM-driven sales brain)
+**Related ADRs:** [ADR-022](ADR-022-conversational-profile-updates.md), [ADR-039](ADR-039-llm-driven-sales-brain.md)
+
+## Context
+
+The bot grew three identity layers organically:
+
+1. **`session_id`** — UUID issued per browser session; in-memory only since KI-118 (`backend/session_state.py::_sessions`).
+2. **`persona_id`** — 12-char SHA1 of name + identity fields; keys the durable JSON at `40-data/profiles/<persona_id>.json` (KI-062).
+3. **`name_slug`** — canonical lowercase first-name; the Chroma profile RAG chunk key and the cross-session re-entry key (KI-040 / KI-118).
+
+Three problems compounded:
+
+- **"Clear chat" was visual-only.** The legacy `POST /api/session/reset` had two modes (`drop_profile=false` / `true`) and the button only ever called the soft mode. The visible chat cleared but the server kept the in-memory session, so the bot acted as if mid-conversation on the user's next message.
+- **Page reloads silently resumed.** `sessionStorage` survives refresh — correct — but the user got zero feedback that they were "still in" a prior session.
+- **Profile recall was silent and aggressive.** KI-118's `rehydrate_by_name` auto-merged any on-disk profile whose name matched the captured name. A user named "Rohit" who happened to share a slug with a prior visitor would silently inherit that stranger's age / dependents / city. Even for the legitimate "same user, new device" case the silent merge confused users into thinking the bot was "remembering them" with no opt-in.
+- **Profile completeness badge counted defaults.** The profile-builder modal pre-filled `dependents="self"` (UX nicety); the completeness scorer treated any non-empty slot as "captured", so a brand-new visitor saw "25% DONE" before answering anything.
+
+## Decision
+
+Implement a clean, explicit lifecycle with these semantics:
+
+| Action | `session_id` | Profile JSON on disk | In-session profile state | Chat history |
+|---|---|---|---|---|
+| Open new tab / browser | NEW UUID | UNCHANGED (persists) | EMPTY | EMPTY |
+| Refresh in same tab | SAME (from `sessionStorage`) | UNCHANGED | RESTORED from `_sessions` if still in memory | RESTORED if still in `localStorage` |
+| Click "Clear chat" | NEW UUID (rotated server-side) | UNCHANGED | EMPTY | EMPTY |
+| User says a name during fact-find that MATCHES an on-disk profile | SAME | UNCHANGED | One-turn pause — bot ASKS before recall | UNCHANGED |
+| User explicitly opts into "welcome back" recall | SAME | LOADED into in-session state | UNCHANGED | UNCHANGED |
+
+### Key change — confirmation-gated profile recall
+
+When a fresh session captures a `name` that matches an existing on-disk profile, the orchestrator NO LONGER auto-merges. Instead:
+
+1. The matched profile snapshot + the captured turn data are staged into `session.pending_profile_recall` (a new `Optional[Dict[str, Any]]` field on `SessionState`).
+2. The orchestrator overrides `reply_text` for that turn with a deterministic, recognisable welcome-back ask:
+   *"Welcome back, Rohit — I have a profile under your name from before: age 29, metro, first buy. Continue from there or start fresh?"*
+3. On the NEXT user turn, the confirmation gate at the top of `handle_turn` inspects `session.pending_profile_recall` and the user's reply:
+   - Affirm phrases (`yes`, `continue`, `use that`, `from there`, `haan`, …) → run `rehydrate_by_name` → merge stored fields.
+   - Negate phrases (`no`, `start fresh`, `new`, `nahi`, …) → drop the pending entry; treat as a new user.
+   - Anything else → leave the pending entry staged; the sales_brain's system prompt carries a high-priority directive to ask again.
+
+### Backend surface
+
+- `backend/session_state.py::clear_session(session_id)` — explicit symbol for the new endpoint. Evicts the in-memory entry; never touches `40-data/profiles/`.
+- `backend/main.py::POST /api/session/clear` — `{session_id} → {cleared: bool, new_session_id: str}`. Always returns a new UUID so the caller has a guaranteed-fresh id.
+- `backend/sales_brain.py::_build_system_prompt(profile, pending_profile_recall=…)` — when `pending_profile_recall` is supplied, the system prompt prepends a "WELCOME-BACK GATE" directive instructing the brain to ask, NOT capture, NOT proceed to the next fact-find slot.
+- `backend/main.py::profile_completeness_view` + `profile_update` — `c = _completeness(...)` now masks fields NOT in `Profile.asked` to `None`. Default `dependents="self"` in the form no longer registers as "done"; only fields the user explicitly answered count.
+
+### Frontend surface
+
+- `frontend/src/lib/api.ts::postSessionClear` — typed wrapper for the new endpoint.
+- `frontend/src/app/page.tsx::handleClearChat` — always rotates the session_id via `postSessionClear`, adopts the returned `new_session_id`, persists it to `sessionStorage`, and wipes messages + `profileCompleteness` + chat-history `localStorage`.
+- `sessionStorage` (not `localStorage`) for the session_id is deliberate: matches the spec — new tab = new UUID, same-tab refresh = same UUID.
+
+## Consequences
+
+**Positive.**
+- One-line semantic for every action; no more "did clear-chat work?" ambiguity.
+- Recall is opt-in. Privacy posture matches what users assume from a chat UI.
+- The 25% phantom-progress badge on first visit is gone — the % matches the fields the user actually answered.
+- No data loss path: on-disk profile JSONs are never deleted as a side effect of any frontend action.
+
+**Trade-offs.**
+- One extra turn when a returning user gives their name (the welcome-back ask). Worth it — the prior silent merge was a privacy footgun.
+- The `Profile.asked` list is now load-bearing for the completeness scorer. It was already maintained by `record_answer` + the sales_brain orchestrator path; the `POST /api/profile` endpoint now also appends to it on every accepted field.
+
+**Carried forward.**
+- KI-040 + KI-077 + KI-118 cross-session name-based recovery is preserved — just gated.
+- Voice / RAG / recommendation paths are untouched; the gate sits in the fact-find branch only.
+- The legacy `POST /api/session/reset` is left in place for backwards compatibility (any external smoke test or admin tool that hits it still works). New frontend code MUST use `/api/session/clear`.
+
+## Rollback
+
+Revert the three backend hunks (orchestrator gate + sales_brain prompt hook + main.py endpoint) and the page.tsx + api.ts hunks. The `Profile.asked`-gated completeness scorer can be reverted independently if the change creates friction for the profile-builder UX.

CLAUDE.md

@@ -68,6 +68,13 @@ Every LLM role is a `NimChainLLM` candidate pool, NOT a hardcoded single model.
 - **Natural-conversation escape (KI-045):** intent_change phrases / off-topic questions still exit fact-find by routing through `should_route_to_fact_find` upstream of the brain.
 - **Indic queries** route through Sarvam-M for translation on input + output; the sales brain runs in English on the translated text.
 
+## Session & profile lifecycle (ADR-041) — KI-196
+
+- **Three identity layers stay distinct.** `session_id` (UUID, in-memory only via `backend/session_state.py`) ≠ `persona_id` (12-char SHA1, keys `40-data/profiles/<persona_id>.json`) ≠ `name_slug` (canonical first-name, Chroma chunk key). New tab = new `session_id`; same-tab refresh = same `session_id` (`sessionStorage`); on-disk profile JSONs are durable user data and are NEVER deleted by frontend actions.
+- **`POST /api/session/clear`** is the canonical "Clear chat" endpoint. Body `{session_id}`; reply `{cleared, new_session_id}`. Wipes the in-memory entry via `clear_session()` and always mints a fresh UUID. The legacy `POST /api/session/reset` (KI-020) is left in place for backwards compatibility but the frontend Clear-chat button calls `/api/session/clear`.
+- **Confirmation-gated profile recall.** When a fresh session captures a `name` that matches an on-disk profile, the orchestrator stages the match into `session.pending_profile_recall` (NEW field on `SessionState`) and overrides the brain's `reply_text` for that turn with a deterministic welcome-back ask ("Welcome back, Rohit — I have a profile under your name from before: age 29, metro, first buy. Continue from there or start fresh?"). The user's NEXT message hits a gate at the top of `handle_turn`: affirm → `rehydrate_by_name`; negate → drop; ambiguous → leave staged so the brain re-asks via the WELCOME-BACK GATE block in `_build_system_prompt`. The legacy KI-118 silent auto-merge is gone.
+- **Profile completeness gates on `Profile.asked`.** Default `dependents="self"` pre-fill in the builder form no longer registers as "done"; `profile_completeness_view` + `POST /api/profile` mask any field not in the `asked` list before scoring. The "Your profile X% done" badge now starts at 0% for a brand-new session and only ticks up on explicit captures.
+
 ## Refusal precision (KI-046)
 
 - Persona prompt now explicitly instructs the bot to refuse on **fanciful / out-of-scope scenarios** (space tourism, diamond-tipped surgery, fictional procedures) with a specific refusal sentence.

README.md

@@ -69,6 +69,7 @@ Thirty-plus knowledge increments landed today. The corpus was rebuilt, the marke
 - **KI-175** — NIM chain reorder: Mistral Large 3 675B promoted to primary, nemotron-49b demoted to last resort across all three chains. The old "nemotron primary" ordering predated the 675B model's availability.
 - **KI-176** — OpenRouter re-added to the chain pool with `models: [...]` server-side fallback within the OR free pool. KI-178 live-audit then confirmed which OR free-tier models actually support `response_format` (nemotron-3-super-120b, qwen3-next-80b, gemma-4 — not Llama 3.3 70B or Hermes 3 405B).
 - **KI-179** — Google AI Studio added as Tier 0 primary on Brain Fast + Brain Main ([ADR-040](70-docs/60-decisions/ADR-040-google-gemini-primary.md)). `backend/providers/google_gemini_llm.py` wrapper matches the `LLMProvider` interface; chains now span {Google → NIM → OpenRouter}. ADR-038 (NIM-only lock) is superseded — the lock was relaxed once KI-167 retired the `<FF>` trailer convention that originally motivated it. Brain Fast primary is `gemini-2.0-flash`; Brain Main primary is `gemini-2.5-flash`; Judge stays NIM Mistral Large 3 675B (different family from Gemini, preserving the brain ↔ judge non-circular grading invariant). Google's free tier: 1500 req/day, 15 req/min, native JSON via `response_mime_type=application/json`.
+- **KI-196** — Session & profile lifecycle ([ADR-041](70-docs/60-decisions/ADR-041-session-profile-lifecycle.md)). Three identity layers (`session_id` / `persona_id` / `name_slug`) are now spec'd end-to-end. New `POST /api/session/clear` endpoint rotates the session_id and wipes in-memory state without touching the on-disk profile JSON; the frontend "Clear chat" button calls this instead of the legacy `/api/session/reset`. The silent KI-118 name-based profile auto-merge is replaced with a **confirmation-gated welcome-back ask** — when a user provides a name that matches an on-disk profile, the bot stages `session.pending_profile_recall` and asks the user whether to continue from the prior captures or start fresh. Profile completeness now gates on `Profile.asked`, so the builder modal's default `dependents="self"` pre-fill no longer registers as "25% DONE" on a brand-new session.
 
 Per-insurer card counts (166 total across 19 real insurers): HDFC ERGO 15 · National Insurance 14 · Niva Bupa 14 · Bajaj Allianz 13 · ICICI Lombard 13 · Star Health 11 · Care Health 10 · New India Assurance 9 · Tata AIG 9 · Acko 7 · Aditya Birla 7 · Royal Sundaram 7 · Cholamandalam MS 6 · Go Digit 6 · IFFCO Tokio 6 · ManipalCigna 6 · SBI General 6 · IndusInd General 3 · Oriental Insurance 3 · Reliance General 1.
 

backend/main.py

@@ -1026,6 +1026,38 @@ class SessionResetResponse(BaseModel):
     cleared_state: bool
 
 
+class SessionClearRequest(BaseModel):
+    session_id: str
+
+
+class SessionClearResponse(BaseModel):
+    cleared: bool
+    new_session_id: str
+
+
+@app.post("/api/session/clear", response_model=SessionClearResponse)
+async def session_clear(req: SessionClearRequest):
+    """KI-196 (ADR-041) — Clean Clear-chat semantic. Wipes the in-memory
+    session state for the supplied session_id and ALWAYS returns a freshly
+    minted UUID the frontend must adopt as its new session_id going forward.
+
+    The on-disk profile JSON under `40-data/profiles/` is intentionally NOT
+    touched — it remains durable user data keyed by persona_id / name slug.
+    The next time the user volunteers their name in conversation, the
+    confirmation-gated recall flow (see orchestrator's `pending_profile_recall`)
+    will ask before merging the prior captures.
+
+    Body : {session_id: str}
+    Reply: {cleared: bool, new_session_id: str}
+    """
+    from backend.session_state import clear_session
+    cleared = clear_session(req.session_id) if req.session_id else False
+    return SessionClearResponse(
+        cleared=cleared,
+        new_session_id=uuid.uuid4().hex[:12],
+    )
+
+
 @app.post("/api/session/reset", response_model=SessionResetResponse)
 async def session_reset(req: SessionResetRequest):
     """KI-020 — User-facing chat clear / fresh-start toggle.
@@ -1080,6 +1112,12 @@ async def profile_update(req: ProfileUpdateRequest):
             # KI-095 — never clobber a filled field with empty input from the client
             continue
         setattr(sess.profile, field_name, v)
+        # KI-196 (ADR-041) — mark the slot as explicitly answered so the
+        # completeness scorer recognises it. Without this, builder-form
+        # captures land on the profile but the badge still reads 0% because
+        # profile_completeness_view now gates on `Profile.asked`.
+        if field_name not in sess.profile.asked:
+            sess.profile.asked.append(field_name)
 
     # KI-077 — if name is set, also persist to the named-profile store so a
     # returning visitor's profile is recoverable across sessions.
@@ -1099,9 +1137,14 @@ async def profile_update(req: ProfileUpdateRequest):
         "parents_age_max": p.parents_age_max, "parents_has_ped": p.parents_has_ped,
         "health_conditions": p.health_conditions, "budget_band": p.budget_band,
     }
-    c = _completeness(profile_dict)
-    collected = [k for k, v in profile_dict.items() if v not in (None, "", [], False)]
-    missing = [k for k, v in profile_dict.items() if v in (None, "", [])]
+    # KI-196 (ADR-041) — same answered-only gate as profile_completeness_view.
+    answered = set(getattr(p, "asked", []) or [])
+    completeness_input = {
+        k: (v if k in answered else None) for k, v in profile_dict.items()
+    }
+    c = _completeness(completeness_input)
+    collected = [k for k, v in profile_dict.items() if k in answered and v not in (None, "", [], False)]
+    missing = [k for k, v in profile_dict.items() if k not in answered or v in (None, "", [])]
 
     # Ingest the profile into the RAG store so the brain sees user context
     # at retrieval time alongside policy + regulatory chunks. Fire-and-forget
@@ -1155,9 +1198,19 @@ async def profile_completeness_view(session_id: Optional[str] = None):
         "parents_age_max": p.parents_age_max, "parents_has_ped": p.parents_has_ped,
         "health_conditions": p.health_conditions, "budget_band": p.budget_band,
     }
-    c = _completeness(profile_dict)
-    collected = [k for k, v in profile_dict.items() if v not in (None, "", [], False)]
-    missing = [k for k, v in profile_dict.items() if v in (None, "", [])]
+    # KI-196 (ADR-041) — Profile completeness must reflect fields the user
+    # EXPLICITLY answered, not defaults that were never touched. Default
+    # `dependents="self"` pre-populated in the builder UI used to count as
+    # "done" and produced the misleading "25% DONE" badge on a zero-input
+    # session. Gate every field on Profile.asked containing the field name
+    # before exposing it to the completeness scorer.
+    answered = set(getattr(p, "asked", []) or [])
+    completeness_input = {
+        k: (v if k in answered else None) for k, v in profile_dict.items()
+    }
+    c = _completeness(completeness_input)
+    collected = [k for k, v in profile_dict.items() if k in answered and v not in (None, "", [], False)]
+    missing = [k for k, v in profile_dict.items() if k not in answered or v in (None, "", [])]
     hint = None
     try:
         # KI-167 WS3 (2026-05-15) — next_question now returns the field name

backend/orchestrator.py

@@ -575,6 +575,57 @@ async def handle_turn(
         query=user_text,
     )
 
+    # KI-196 (ADR-041) — confirmation-gated profile recall. If the prior
+    # turn staged a pending recall (matched name → on-disk profile), the
+    # bot's last reply already asked the user whether to continue or start
+    # fresh. THIS turn's user_text decides:
+    #   - Affirm  → merge stored profile (via rehydrate_by_name) and continue.
+    #   - Negate  → drop pending_profile_recall, treat as new user.
+    #   - Other   → leave pending_profile_recall in place; the brain will
+    #               re-ask one more time. (One re-ask cap is enforced by the
+    #               brain's prompt — see sales_brain._build_system_prompt.)
+    if session.pending_profile_recall is not None:
+        _utxt = (user_text or "").strip().lower()
+        _AFFIRM = (
+            "yes", "yeah", "yep", "yup", "yes please", "continue", "continue from there",
+            "use that", "load it", "load that", "welcome back", "from there",
+            "carry on", "ok", "okay", "sure", "go ahead", "proceed", "haan", "ha",
+            "use my profile", "use the profile", "resume",
+        )
+        _NEGATE = (
+            "no", "nope", "nah", "start fresh", "fresh", "new", "new profile",
+            "start over", "from scratch", "different", "not me", "that's not me",
+            "thats not me", "discard", "begin again", "nahi", "nai",
+        )
+        _is_affirm = (
+            _utxt in _AFFIRM
+            or any(_utxt.startswith(a + " ") or _utxt.endswith(" " + a) for a in _AFFIRM)
+            or " continue " in f" {_utxt} "
+            or "from there" in _utxt
+            or "use that profile" in _utxt
+        )
+        _is_negate = (
+            _utxt in _NEGATE
+            or any(_utxt.startswith(n + " ") or _utxt.endswith(" " + n) for n in _NEGATE)
+            or "start fresh" in _utxt
+            or "start over" in _utxt
+            or "from scratch" in _utxt
+        )
+        if _is_affirm and not _is_negate:
+            try:
+                from backend.session_state import rehydrate_by_name as _rehydrate
+                _rehydrate(session, session.pending_profile_recall.get("name") or session.profile.name or "")
+            except Exception as _e:
+                logging.warning(
+                    "pending_profile_recall merge failed (session=%s): %s",
+                    session_id, _e,
+                )
+            session.pending_profile_recall = None
+        elif _is_negate and not _is_affirm:
+            session.pending_profile_recall = None
+        # If neither was detected, leave it staged — sales_brain will re-ask
+        # via its system-prompt hook (see _build_system_prompt).
+
     if treat_as_fact_find:
         # KI-167 (2026-05-15) — WS2: replaces drive_fact_find with
         # drive_sales_brain. The new brain owns conversation flow end-to-end
@@ -591,6 +642,7 @@ async def handle_turn(
                     profile=session.profile,
                     chat_history=chat_history[-10:],
                     session_id=session_id,
+                    pending_profile_recall=session.pending_profile_recall,
                 ),
                 timeout=45.0,  # KI-170 — bumped from 25s; qwen3-next-80b + JSON mode regularly lands 15-25s
             )
@@ -642,40 +694,121 @@ async def handle_turn(
                 p = session.profile
                 slug = _normalise_name(p.name or "")
                 if slug:
-                    await upsert_profile_chunk(slug, {
-                        "age": p.age,
-                        "dependents": p.dependents,
-                        "income_band": p.income_band,
-                        "existing_cover_inr": p.existing_cover_inr,
-                        "primary_goal": p.primary_goal,
-                        "location_tier": p.location_tier,
-                        "parents_to_insure": p.parents_to_insure,
-                        "parents_age_max": p.parents_age_max,
-                        "parents_has_ped": p.parents_has_ped,
-                        "budget_band": p.budget_band,
-                        "health_conditions": p.health_conditions,
-                    })
+                    # KI-197 (2026-05-15) — fire-and-forget. The profile-chunk
+                    # upsert involves embedding generation + Chroma write
+                    # (~200-500ms). Awaiting it serially adds that to the
+                    # user-perceived turn latency for ZERO reason — the next
+                    # turn's retrieval doesn't depend on this chunk being
+                    # persisted yet. Schedule on the event loop, log errors
+                    # via the task's add_done_callback if you ever need to
+                    # debug — but don't block the response.
+                    async def _bg_upsert(slug_=slug, p_=p, sid_=session_id):
+                        try:
+                            await upsert_profile_chunk(slug_, {
+                                "age": p_.age,
+                                "dependents": p_.dependents,
+                                "income_band": p_.income_band,
+                                "existing_cover_inr": p_.existing_cover_inr,
+                                "primary_goal": p_.primary_goal,
+                                "location_tier": p_.location_tier,
+                                "parents_to_insure": p_.parents_to_insure,
+                                "parents_age_max": p_.parents_age_max,
+                                "parents_has_ped": p_.parents_has_ped,
+                                "budget_band": p_.budget_band,
+                                "health_conditions": p_.health_conditions,
+                            })
+                        except Exception as e:
+                            logging.warning(
+                                "sales_brain profile-chunk upsert failed bg (session=%s): %s: %s",
+                                sid_, type(e).__name__, str(e)[:200],
+                            )
+                    asyncio.create_task(_bg_upsert())
             except Exception as e:
                 logging.warning(
-                    "sales_brain profile-chunk upsert failed (session=%s): %s: %s",
+                    "sales_brain profile-chunk upsert scheduling failed (session=%s): %s: %s",
                     session_id, type(e).__name__, str(e)[:200],
                 )
 
         # KI-040 / KI-062 — named-profile persistence preserved. If the brain
         # captured (or already has) a name on the profile, write the merged
         # profile to disk so the next visit can welcome the user back.
-        # KI-118 — also trigger one-shot rehydrate when name was newly captured
-        # this turn so any stored profile from a prior visit gets merged in.
+        #
+        # KI-196 (ADR-041) — confirmation-gated recall. The silent auto-merge
+        # from KI-118 confused users into thinking the bot "remembered" them
+        # from a prior session. Replaced with a stage-then-ask flow:
+        #   1. Name newly captured this turn AND a stored profile exists
+        #      under that name AND there's not already a pending recall AND
+        #      the current session has NOT already accumulated a meaningful
+        #      profile (>1 explicitly-answered slot — they're genuinely a
+        #      fresh session, not a returning user mid-correction).
+        #   2. Stage the stored snapshot to `session.pending_profile_recall`.
+        #   3. The user's NEXT message hits the confirmation gate at the top
+        #      of handle_turn (BEFORE drive_sales_brain) and either merges
+        #      or discards based on affirm/negate intent.
+        # The disk save still runs every turn so partial captures aren't lost.
         if session.profile.name:
             try:
-                from backend.profile_store import save_profile
-                # If name was newly captured this turn AND there's a stored
-                # profile under that name, merge in the stored fields BEFORE
-                # writing back (so we don't immediately overwrite the stored
-                # snapshot with a partial new one).
-                if "name" in fact_find_profile_updates:
-                    from backend.session_state import rehydrate_by_name
-                    rehydrate_by_name(session, session.profile.name)
+                from backend.profile_store import save_profile, load_profile
+                if (
+                    "name" in fact_find_profile_updates
+                    and session.pending_profile_recall is None
+                ):
+                    # Count explicitly-answered slots OTHER than name on the
+                    # live session. If >1, treat as continuation, not a fresh
+                    # returning visit, and skip the gate (their current
+                    # captures win — same intent as the old merge semantics).
+                    answered_non_name = [
+                        s for s in (session.profile.asked or [])
+                        if s != "name"
+                    ]
+                    stored = load_profile(session.profile.name)
+                    if stored is not None and len(answered_non_name) <= 1:
+                        summary = {
+                            "age": stored.age,
+                            "dependents": stored.dependents,
+                            "location_tier": stored.location_tier,
+                            "income_band": stored.income_band,
+                            "primary_goal": stored.primary_goal,
+                            "health_conditions": stored.health_conditions,
+                            "budget_band": stored.budget_band,
+                            "existing_cover_inr": stored.existing_cover_inr,
+                        }
+                        # Drop None / empty values so the brain's recall
+                        # prompt summary is tight.
+                        summary = {
+                            k: v for k, v in summary.items()
+                            if v not in (None, "", [])
+                        }
+                        session.pending_profile_recall = {
+                            "name": session.profile.name,
+                            "summary": summary,
+                            "captured_this_turn": dict(fact_find_profile_updates),
+                            "staged_at": time.time(),
+                        }
+                        # Override the brain's reply for THIS turn with a
+                        # deterministic welcome-back ask. The brain didn't
+                        # know about the stored profile when it composed its
+                        # reply (the load_profile lookup happens in this
+                        # post-processing block), so its prose would have
+                        # been the next-slot fact-find question — not what
+                        # the user should see right now. Subsequent turns
+                        # WILL go through the brain with the recall directive
+                        # in the system prompt (see _build_system_prompt).
+                        _bits = []
+                        if summary.get("age") is not None:
+                            _bits.append(f"age {summary['age']}")
+                        if summary.get("location_tier"):
+                            _bits.append(str(summary["location_tier"]))
+                        if summary.get("dependents"):
+                            _bits.append(str(summary["dependents"]))
+                        if summary.get("primary_goal"):
+                            _bits.append(str(summary["primary_goal"]).replace("_", " "))
+                        _summary_phrase = ", ".join(_bits) if _bits else "your earlier captures"
+                        sb_result.reply_text = (
+                            f"Welcome back, {session.profile.name} — I have a profile under your name "
+                            f"from before: {_summary_phrase}. Continue from there or start fresh?"
+                        )
+                        sb_result.brain_used = "sales_brain::welcome_back_ask"
                 save_profile(session.profile.name, session.profile, session_id=session_id)
             except Exception:
                 pass

backend/providers/google_gemini_llm.py

@@ -43,7 +43,11 @@ path (Tier 0, via TieredBrainLLM wrapper).
 from __future__ import annotations
 
 import asyncio
+import hashlib
+import logging
 import os
+import threading
+import time
 from typing import Optional
 
 import httpx
@@ -52,8 +56,47 @@ from backend.providers.base import ChatMessage, LLMProvider, LLMResult
 
 
 GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/models"
+GEMINI_CACHE_URL = "https://generativelanguage.googleapis.com/v1beta/cachedContents"
 DEFAULT_MODEL = "gemini-2.5-flash-lite"  # KI-183 — gemini-2.0-flash retired for new accounts
 
+# ----------------------------------------------------------------------------
+# KI-199 — module-level cachedContents registry.
+#
+# Keyed by `(model, sha256(system_text))` so the same base preamble shared
+# across sales_brain calls deduplicates onto one cache. Each value is a small
+# dict with the cache `name` (the server-side resource id used in subsequent
+# generateContent bodies as `cachedContent`) and the local-clock `expires_at`
+# wall-time so we can self-evict before issuing a guaranteed-miss request.
+#
+# A threading.Lock guards entries against the rare interleave where two
+# coroutines on different event loops race to create the same cache. In
+# practice asyncio gives us implicit single-task ordering on one loop, but
+# this is cheap insurance and keeps the contract honest if the module is ever
+# pulled into a thread pool.
+# ----------------------------------------------------------------------------
+_CACHE_REGISTRY: dict[tuple[str, str], dict] = {}
+_CACHE_REGISTRY_LOCK = threading.Lock()
+
+
+def _cache_key(model: str, system_text: str) -> tuple[str, str]:
+    """Build the registry key for a (model, system_text) pair.
+
+    Hashing the system text rather than storing the raw string keeps the
+    registry footprint tiny even when the preamble is multi-KB.
+    """
+    return (model, hashlib.sha256(system_text.encode("utf-8")).hexdigest())
+
+
+def invalidate_cache(model: str, system_text: str) -> None:
+    """Drop a cache registry entry — called by upstream after a 4xx response
+    that names a stale `cachedContent`. The server-side cache may still be
+    alive (it will lapse on TTL), but our reference is gone so the next
+    chat() call provisions a fresh one.
+    """
+    key = _cache_key(model, system_text)
+    with _CACHE_REGISTRY_LOCK:
+        _CACHE_REGISTRY.pop(key, None)
+
 
 def _to_gemini_contents(messages: list[ChatMessage]) -> tuple[Optional[str], list[dict]]:
     """Split an OpenAI-style message list into (systemInstruction, contents).
@@ -115,12 +158,102 @@ class GoogleGeminiLLM(LLMProvider):
         self.timeout = timeout
         self.name = f"gemini::{model}"
 
+    async def create_cache(
+        self,
+        system_text: str,
+        ttl_seconds: int = 300,
+    ) -> Optional[str]:
+        """Create (or reuse) a Gemini `cachedContents` resource for `system_text`.
+
+        Returns the cache resource name (e.g. `"cachedContents/<UUID>"`)
+        that downstream `chat()` calls should pass as `cached_content_name`.
+        Returns None on ANY failure (missing key, too-small payload, 4xx, network
+        error) — the caller is expected to proceed without caching when this is
+        the case.
+
+        Re-uses an existing live cache from the module registry when the
+        (model, system_text) pair matches AND the local `expires_at` is still
+        in the future. Cache misses + creation failures are silent (logged at
+        INFO) so a caching outage never breaks the main path — KI-199 brief
+        requires fail-safe behaviour.
+        """
+        if not self.api_key or not system_text:
+            return None
+
+        key = _cache_key(self.model, system_text)
+        now = time.time()
+        with _CACHE_REGISTRY_LOCK:
+            entry = _CACHE_REGISTRY.get(key)
+            # Refresh ~10s before expiry so an in-flight request never lands
+            # on a server-side cache that just rolled past its TTL.
+            if entry and entry.get("expires_at", 0) > now + 10:
+                return entry.get("name")
+
+        # `model` must be the fully-qualified Gemini path "models/<id>".
+        body: dict = {
+            "model": f"models/{self.model}",
+            "systemInstruction": {"parts": [{"text": system_text}]},
+            "ttl": f"{int(ttl_seconds)}s",
+        }
+        url = f"{GEMINI_CACHE_URL}?key={self.api_key}"
+        headers = {"Content-Type": "application/json"}
+        client_timeout = httpx.Timeout(
+            connect=2.0, read=self.timeout, write=2.0, pool=2.0
+        )
+
+        try:
+            async with httpx.AsyncClient(timeout=client_timeout) as client:
+                resp = await client.post(url, headers=headers, json=body)
+            if resp.status_code >= 400:
+                # Most common 4xx is "the request must contain at least N
+                # tokens of cached content" — below the Gemini minimum the
+                # cache simply isn't allowed. Log + return None so the caller
+                # falls through to the uncached path.
+                detail = ""
+                try:
+                    detail = resp.text[:300]
+                except Exception:
+                    pass
+                logging.info(
+                    "gemini.create_cache %s (model=%s): %s",
+                    resp.status_code, self.model, detail,
+                )
+                return None
+            payload = resp.json()
+        except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
+            raise
+        except Exception as e:  # noqa: BLE001 — fail-safe: any error = no cache
+            logging.info(
+                "gemini.create_cache raised %s (model=%s): %s",
+                type(e).__name__, self.model, str(e)[:200],
+            )
+            return None
+
+        cache_name = payload.get("name") or ""
+        if not cache_name:
+            return None
+
+        with _CACHE_REGISTRY_LOCK:
+            _CACHE_REGISTRY[key] = {
+                "name": cache_name,
+                # Store local expiry; the registered TTL is server-side
+                # truth, but we shadow it locally so we self-evict before
+                # the inevitable 4xx on an expired reference.
+                "expires_at": time.time() + ttl_seconds,
+            }
+        logging.info(
+            "gemini.create_cache OK (model=%s, name=%s, ttl=%ss)",
+            self.model, cache_name, ttl_seconds,
+        )
+        return cache_name
+
     async def chat(
         self,
         messages: list[ChatMessage],
         temperature: float = 0.6,
         max_tokens: int = 700,
         response_format: Optional[dict] = None,
+        cached_content_name: Optional[str] = None,
         **kwargs,  # absorb OR-specific kwargs like `models=[...]` — ignored here
     ) -> LLMResult:
         if not self.api_key:
@@ -144,7 +277,14 @@ class GoogleGeminiLLM(LLMProvider):
             "contents": contents,
             "generationConfig": generation_config,
         }
-        if system_instruction:
+        # When a cachedContents resource is in play, the entire systemInstruction
+        # already lives inside it server-side; including it again in the request
+        # body causes a 400 INVALID_ARGUMENT ("cached prompt and inline prompt
+        # mutually exclusive"). Only attach systemInstruction on the uncached
+        # path.
+        if cached_content_name:
+            body["cachedContent"] = cached_content_name
+        elif system_instruction:
             body["systemInstruction"] = {"parts": [{"text": system_instruction}]}
 
         # API key goes in the URL query param — the Google AI Studio default.
@@ -165,6 +305,33 @@ class GoogleGeminiLLM(LLMProvider):
 
         async with httpx.AsyncClient(timeout=client_timeout) as client:
             resp = await client.post(url, headers=headers, json=body)
+            # KI-199 — graceful fallback when a cache reference is stale.
+            # Symptoms: 400/404 with body mentioning "cachedContent" /
+            # "cache" / "not found". Strip the reference, re-add the inline
+            # systemInstruction, drop the registry entry, retry once.
+            if (
+                cached_content_name
+                and resp.status_code in (400, 403, 404)
+                and any(
+                    tok in (resp.text or "").lower()
+                    for tok in ("cache", "cachedcontent")
+                )
+            ):
+                logging.info(
+                    "gemini.chat cache miss/invalid (%s) — retrying uncached (model=%s)",
+                    resp.status_code, self.model,
+                )
+                # Best-effort invalidate by direct name match in the registry.
+                with _CACHE_REGISTRY_LOCK:
+                    for k, v in list(_CACHE_REGISTRY.items()):
+                        if v.get("name") == cached_content_name:
+                            _CACHE_REGISTRY.pop(k, None)
+                body.pop("cachedContent", None)
+                if system_instruction:
+                    body["systemInstruction"] = {
+                        "parts": [{"text": system_instruction}]
+                    }
+                resp = await client.post(url, headers=headers, json=body)
             if resp.status_code >= 400:
                 # Surface the Google error body so the caller's log makes the
                 # root cause visible (typical failure: 429 quota exceeded or
@@ -230,4 +397,9 @@ def get_gemini_llm(
     return GoogleGeminiLLM(model=model, timeout=timeout)
 
 
-__all__ = ["GoogleGeminiLLM", "get_gemini_llm", "DEFAULT_MODEL"]
+__all__ = [
+    "GoogleGeminiLLM",
+    "get_gemini_llm",
+    "DEFAULT_MODEL",
+    "invalidate_cache",
+]

backend/providers/tiered_brain_llm.py

@@ -90,8 +90,15 @@ class TieredBrainLLM(LLMProvider):
         temperature: float = 0.2,
         max_tokens: int = 1024,
         response_format: Optional[dict] = None,
+        cached_content_name: Optional[str] = None,
         **kwargs,  # absorb provider-specific kwargs (e.g. OR's `models=[...]`)
     ) -> LLMResult:
+        """KI-199 — `cached_content_name`, when supplied by the caller, is
+        forwarded to the Gemini tier ONLY. NIM and OpenRouter don't speak
+        Gemini's cachedContents protocol so we silently drop the reference on
+        those tiers — they receive the full messages list as before, which
+        keeps the wire shape correct on fall-through.
+        """
         gemini_exc: Optional[BaseException] = None
         nim_exc: Optional[BaseException] = None
         or_exc: Optional[BaseException] = None
@@ -105,6 +112,7 @@ class TieredBrainLLM(LLMProvider):
                     temperature=temperature,
                     max_tokens=max_tokens,
                     response_format=response_format,
+                    cached_content_name=cached_content_name,
                 ),
                 timeout=self.per_tier_timeout,
             )

backend/sales_brain.py

@@ -56,7 +56,10 @@ from typing import Any, Optional
 
 from backend.needs_finder import Profile
 from backend.providers.base import ChatMessage
-from backend.providers.google_gemini_llm import get_gemini_llm
+from backend.providers.google_gemini_llm import (
+    get_gemini_llm,
+    invalidate_cache as _gemini_invalidate_cache,
+)
 from backend.providers.nvidia_nim_llm import get_fast_brain_llm
 from backend.providers.openrouter_llm import get_openrouter_llm
 from backend.sales_brain_normalizer import normalize_captures
@@ -99,7 +102,7 @@ class SalesBrainResult:
 # + leaves room for one chain fallback. The fast-brain chain already has its
 # own per-link + total-chain budget; this wait_for is a belt-and-braces stop.
 _TIMEOUT_S: float = 45.0         # KI-170 — qwen3-next-80b + JSON mode regularly lands 15-25s; 25s breached periodically
-_MAX_TOKENS: int = 700           # prose + structured JSON object with safety margin
+_MAX_TOKENS: int = 500           # KI-197 — was 700; 500 still covers 1-3 sentence sales_brain replies + JSON wrapper. Saves ~200-400ms per turn.
 _TEMPERATURE: float = 0.6        # mirrors fact_find_brain — conversational warmth
 
 # KI-169 — strip <think>...</think> blocks the LLM may emit inside the JSON
@@ -291,8 +294,29 @@ def _format_slot_list(slots: list[str]) -> str:
     return "\n".join(f"  - {f}: {_SLOT_DESCRIPTIONS.get(f, '')}" for f in slots)
 
 
-def _build_system_prompt(profile: Profile) -> str:
-    """Assemble the full system prompt: base rules + KNOWN block + missing slots."""
+# KI-199 — Cached preamble = the strictly-invariant chunk that's identical on
+# every turn for every session: the base tone/output rules + the SLOT SCHEMA
+# section. This is what Gemini cachedContents wraps. Everything else (KNOWN,
+# REQUIRED REMAINING, NICE-TO-HAVE, recall) varies per turn / per session and
+# is sent inline as a system message in `contents[]`.
+_CACHED_PREAMBLE: str = (
+    _BASE_SYSTEM_PROMPT
+    + "\n\n--- SLOT SCHEMA (the fields you may capture) ---\n"
+    + _format_slot_list(list(_SLOT_DESCRIPTIONS.keys()))
+)
+
+
+def _dynamic_profile_block(
+    profile: Profile,
+    pending_profile_recall: Optional[dict] = None,
+) -> str:
+    """Build the per-turn dynamic system message: KNOWN, REQUIRED, NICE-TO-HAVE,
+    and the optional KI-196 welcome-back recall directive.
+
+    Kept disjoint from _CACHED_PREAMBLE — concatenating the two reproduces the
+    exact prompt the previous monolithic `_build_system_prompt` emitted, so
+    cached + uncached paths are byte-identical when assembled.
+    """
     known = _profile_known(profile)
     required_remaining = _required_remaining(known)
     nice_to_have_remaining = _nice_to_have_remaining(known, profile)
@@ -301,11 +325,25 @@ def _build_system_prompt(profile: Profile) -> str:
         json.dumps(known, ensure_ascii=False) if known else "{}"
     )
 
+    recall_block = ""
+    if pending_profile_recall:
+        recall_summary = pending_profile_recall.get("summary") or {}
+        recall_name = pending_profile_recall.get("name") or profile.name or "there"
+        recall_block = (
+            "\n\n--- WELCOME-BACK GATE (HIGHEST PRIORITY THIS TURN) ---\n"
+            f"The user just provided the name '{recall_name}' and a stored profile exists under that name with these prior captures:\n"
+            f"  {json.dumps(recall_summary, ensure_ascii=False)}\n"
+            "Your ONLY job this turn is to ask warmly whether to continue from that stored profile OR start fresh.\n"
+            "Mention 2-3 of the most-distinctive prior captures in your reply so the user can recognise their own profile (e.g. age + city + dependents).\n"
+            "Do NOT capture any of those stored fields yourself — emit `\"captures\": {}` and `\"ready_for_recommendations\": false`.\n"
+            "Do NOT volunteer the next fact-find question on this turn. Wait for the user's yes / no.\n"
+            "Example reply: 'Welcome back, "
+            f"{recall_name} — I have a profile under your name from before: "
+            "age 29, in metro, looking for first policy. Continue from there or start fresh?'\n"
+        )
+
     return (
-        _BASE_SYSTEM_PROMPT
-        + "\n\n--- SLOT SCHEMA (the fields you may capture) ---\n"
-        + _format_slot_list(list(_SLOT_DESCRIPTIONS.keys()))
-        + f"\n\nKNOWN: {known_block}"
+        f"KNOWN: {known_block}"
         + "\n(These fields are ALREADY captured. Do NOT re-ask for them. Use the user's name when known.)"
         + "\n\nREQUIRED SLOTS STILL MISSING (you need these before setting ready_for_recommendations=true):\n"
         + _format_slot_list(required_remaining)
@@ -317,6 +355,28 @@ def _build_system_prompt(profile: Profile) -> str:
             if not required_remaining else
             "\n\nKeep gathering naturally — you don't yet have the minimum required set."
         )
+        + recall_block
+    )
+
+
+def _build_system_prompt(
+    profile: Profile,
+    pending_profile_recall: Optional[dict] = None,
+) -> str:
+    """Backwards-compatible monolithic prompt builder.
+
+    Retained so any non-Gemini tier (NIM, OR) that doesn't understand
+    cachedContents still receives the FULL prompt as a single system
+    message — preserving the pre-KI-199 wire shape for those tiers. By
+    construction this is byte-identical to (_CACHED_PREAMBLE +
+    "\\n\\n" + _dynamic_profile_block(...)).
+
+    KI-196 (ADR-041) — when `pending_profile_recall` is supplied, the brain
+    receives an extra directive to ASK the user whether to load the prior
+    profile.
+    """
+    return _CACHED_PREAMBLE + "\n\n" + _dynamic_profile_block(
+        profile, pending_profile_recall=pending_profile_recall
     )
 
 
@@ -371,6 +431,7 @@ async def drive_sales_brain(
     profile: Profile,
     chat_history: list[dict],
     session_id: Optional[str] = None,
+    pending_profile_recall: Optional[dict] = None,
 ) -> SalesBrainResult:
     """Single LLM call per turn — replaces the scripted slot-walker.
 
@@ -378,22 +439,53 @@ async def drive_sales_brain(
     `brain_used` is `sales_brain::error:<reason>` and `reply_text` is
     empty. Caller (orchestrator) is responsible for any fallback behavior
     — NO scripted reply lives here.
+
+    KI-196 (ADR-041) — `pending_profile_recall` is the staged welcome-back
+    snapshot (see `backend.session_state.SessionState.pending_profile_recall`).
+    When non-None, the system prompt adds a high-priority directive to ask
+    the user whether to load the stored profile or start fresh, and forbids
+    the brain from capturing any of the stored fields itself.
     """
     t0 = time.time()
 
-    # Build the system prompt with current profile state
-    system_prompt = _build_system_prompt(profile)
+    # KI-199 — split the system prompt into the fixed-preamble (cacheable on
+    # Gemini) and the per-turn dynamic block. The Gemini tier sends the
+    # dynamic block as a system message inline + references the cache; the
+    # NIM and OR tiers receive the full monolithic prompt assembled below
+    # since they don't speak cachedContents.
+    dynamic_block = _dynamic_profile_block(
+        profile, pending_profile_recall=pending_profile_recall
+    )
+    system_prompt = _CACHED_PREAMBLE + "\n\n" + dynamic_block
 
     # Build messages: system + last ~10 chat history turns + user_text
     messages: list[ChatMessage] = [ChatMessage(role="system", content=system_prompt)]
     if chat_history:
-        for turn in chat_history[-10:]:
+        # KI-197 — was [-10:]; trimmed to last 6 turns to reduce prompt size
+        # by ~40% per call. 6 is still enough context for conversational
+        # follow-ups; older turns rarely influence the next slot to ask.
+        for turn in chat_history[-6:]:
             role = turn.get("role")
             content = turn.get("content")
             if role in ("user", "assistant") and content:
                 messages.append(ChatMessage(role=role, content=str(content)))
     messages.append(ChatMessage(role="user", content=user_text or ""))
 
+    # Gemini-specific message shape (KI-199): keep ONLY the dynamic block
+    # inline as a system message; the fixed preamble lives in cachedContents.
+    # When the cache isn't available we fall through and reuse the full
+    # `messages` list below.
+    gemini_messages: list[ChatMessage] = [
+        ChatMessage(role="system", content=dynamic_block)
+    ]
+    if chat_history:
+        for turn in chat_history[-6:]:  # KI-197 — match the messages[] trim
+            role = turn.get("role")
+            content = turn.get("content")
+            if role in ("user", "assistant") and content:
+                gemini_messages.append(ChatMessage(role=role, content=str(content)))
+    gemini_messages.append(ChatMessage(role="user", content=user_text or ""))
+
     # KI-179 (2026-05-15) — 3-tier LLM stack with Google Gemini 2.0 Flash
     # PRIMARY (Tier 0), NIM fast-brain chain Tier 1, OpenRouter free-tier
     # pool Tier 2. Google AI Studio's free tier is 1500 req/day with native
@@ -426,13 +518,41 @@ async def drive_sales_brain(
         )
 
     if gemini_llm is not None:
+        # KI-199 — lazily provision a cachedContents resource for the fixed
+        # preamble. The cache is shared across sessions (same key for everyone
+        # who hits sales_brain Gemini tier with this preamble + model). On
+        # ANY provisioning failure (cache too small, network blip, 4xx, etc.)
+        # `create_cache` returns None and we proceed uncached. The chat() path
+        # also self-heals on a stale-cache 4xx by retrying without the
+        # reference, so a cache-server outage can never break the main path.
+        cache_name: Optional[str] = None
+        try:
+            cache_name = await gemini_llm.create_cache(
+                _CACHED_PREAMBLE, ttl_seconds=300
+            )
+        except Exception as e:  # noqa: BLE001 — fail-safe: never blocking
+            logging.info(
+                "sales_brain: gemini.create_cache raised %s — proceeding uncached (session=%s): %s",
+                type(e).__name__, session_id, str(e)[:200],
+            )
+            cache_name = None
+
+        # Use cache-aware message shape only when the cache is in play;
+        # otherwise fall back to the full monolithic prompt so the LLM never
+        # sees a partial preamble.
+        if cache_name:
+            chat_messages = gemini_messages
+        else:
+            chat_messages = messages
+
         try:
             result = await asyncio.wait_for(
                 gemini_llm.chat(
-                    messages=messages,
+                    messages=chat_messages,
                     temperature=_TEMPERATURE,
                     max_tokens=_MAX_TOKENS,
                     response_format={"type": "json_object"},
+                    cached_content_name=cache_name,
                 ),
                 timeout=_TIMEOUT_S,
             )
@@ -446,6 +566,19 @@ async def drive_sales_brain(
             )
         except Exception as e:  # noqa: BLE001
             gemini_exc = e
+            # On any non-timeout error from the Gemini call, drop our cache
+            # ref proactively so the next turn re-provisions cleanly instead
+            # of repeatedly slamming a stale handle. (The provider itself
+            # already invalidates on a 4xx whose body names the cache, but
+            # 5xx / network errors land here and we want to be conservative.)
+            if cache_name:
+                try:
+                    _gemini_invalidate_cache(
+                        getattr(gemini_llm, "model", "gemini-2.5-flash-lite"),
+                        _CACHED_PREAMBLE,
+                    )
+                except Exception:  # noqa: BLE001
+                    pass
             logging.info(
                 "sales_brain: Gemini raised %s, falling back to NIM (session=%s): %s",
                 type(e).__name__, session_id, str(e)[:200],
@@ -591,7 +724,17 @@ async def drive_sales_brain(
             "sales_brain empty reply after think-strip — retrying once (session=%s, model=%s)",
             session_id, served_model,
         )
-        retry_messages = list(messages) + [
+        # KI-199 — when the original call used Gemini with a cachedContents
+        # ref, the retry must keep the dynamic-only message list AND keep
+        # passing the cache (the preamble lives server-side; sending it again
+        # inline would be redundant and could trip the cached-vs-inline
+        # mutex). For NIM/OR we keep the full monolithic prompt.
+        retry_base = (
+            gemini_messages
+            if served_tier == "gemini" and "cache_name" in locals() and cache_name
+            else messages
+        )
+        retry_messages = list(retry_base) + [
             ChatMessage(
                 role="system",
                 content=(
@@ -602,14 +745,17 @@ async def drive_sales_brain(
                 ),
             ),
         ]
+        retry_kwargs: dict = {
+            "messages": retry_messages,
+            "temperature": _TEMPERATURE,
+            "max_tokens": _MAX_TOKENS,
+            "response_format": {"type": "json_object"},
+        }
+        if served_tier == "gemini" and "cache_name" in locals() and cache_name:
+            retry_kwargs["cached_content_name"] = cache_name
         try:
             retry_result = await asyncio.wait_for(
-                llm.chat(
-                    messages=retry_messages,
-                    temperature=_TEMPERATURE,
-                    max_tokens=_MAX_TOKENS,
-                    response_format={"type": "json_object"},
-                ),
+                llm.chat(**retry_kwargs),
                 timeout=_TIMEOUT_S,
             )
             retry_parsed = _parse_brain_json((retry_result.text or "").strip())

backend/session_state.py

@@ -35,6 +35,8 @@ from dataclasses import dataclass, field
 from threading import Lock
 from typing import Optional
 
+from typing import Any, Dict
+
 from backend.needs_finder import Profile, record_answer
 
 _log = logging.getLogger(__name__)
@@ -47,6 +49,18 @@ class SessionState:
     awaiting_question_id: Optional[str] = None  # if set, next user message answers this
     free_form_session: bool = False              # user explicitly opted out of fact-find
     last_touched: float = field(default_factory=time.time)
+    # KI-196 (ADR-041) — confirmation-gated profile recall. When a fresh
+    # session captures a name that matches an on-disk profile, the recall
+    # is staged here (NOT auto-merged) and surfaced to the sales_brain as a
+    # one-shot "welcome back" prompt. Affirm → merge stored fields into
+    # `profile`. Negate → discard. Shape:
+    #   {
+    #     "name": "Rohit Sarma",
+    #     "summary": {age, dependents, location_tier, primary_goal, ...},
+    #     "captured_this_turn": {<field>: <value>, ...},  # don't re-extract
+    #     "staged_at": <epoch-seconds>,
+    #   }
+    pending_profile_recall: Optional[Dict[str, Any]] = None
 
     def _flush(self) -> None:
         """No-op since KI-118 (2026-05-15). Disk persistence was removed; the
@@ -161,6 +175,26 @@ def reset_session(session_id: str) -> bool:
     return False
 
 
+def clear_session(session_id: str) -> bool:
+    """KI-196 (ADR-041) — Wipe in-memory state for one session_id WITHOUT
+    touching any on-disk profile JSON under `40-data/profiles/`.
+
+    Semantically identical to `reset_session` today (both just evict the
+    in-memory entry; the disk profile has always been independent and lives
+    by persona_id / name slug, not session_id). Kept as a distinct symbol so
+    the call-site intent at `POST /api/session/clear` is self-documenting and
+    so future divergence (e.g. partial-state wipes) doesn't require touching
+    the legacy KI-020 caller.
+
+    Returns True iff a live in-memory session was evicted.
+    """
+    with _lock:
+        if session_id in _sessions:
+            del _sessions[session_id]
+            return True
+    return False
+
+
 def purge_old_files() -> int:
     """KI-118 (2026-05-15) — no-op. Disk persistence was removed; there are
     no files to purge. Kept as a stub so any existing scheduled-task caller

frontend/src/app/page.tsx

@@ -21,7 +21,7 @@ import {
   postChat,
   postPremiumEstimate,
   postProfileUpdate,
-  postSessionReset,
+  postSessionClear,
   postTranscribe,
   PremiumEstimateResponse,
   ProfileCompletenessResponse,
@@ -287,42 +287,42 @@ export default function Page() {
     ].includes(s);
   }
 
-  // KI-020 — user-facing chat clear / fresh-start
-  async function handleClearChat(dropProfile: boolean) {
+  // KI-196 (ADR-041) — Clean Clear-chat semantic.
+  //   1. POST /api/session/clear with the current session_id.
+  //   2. Adopt the returned new_session_id going forward (sessionStorage).
+  //   3. Wipe message array, profile chip, and chat-history localStorage.
+  // The legacy `dropProfile` parameter is retained for backwards compatibility
+  // with any other callsite — both true and false now route through the new
+  // endpoint since the server-side semantic is identical (in-memory wipe +
+  // fresh UUID; on-disk profile JSON untouched).
+  async function handleClearChat(_dropProfile: boolean = false) {
     // Always wipe the visible chat + local storage.
     setMessages([]);
     setInput("");
     // KI-073 (2026-05-15) — clear the profile-completeness chip immediately
     // so the header doesn't show stale "55% DONE" for a brand-new visitor
-    // while the new session_id fetch is in flight. The useEffect on sessionId
-    // will repopulate this from the fresh backend session as soon as the new
-    // id lands.
+    // while the new session_id fetch is in flight.
     setProfileCompleteness(null);
     if (typeof window !== "undefined") {
       localStorage.removeItem("insurance_chat_messages");
     }
-    if (dropProfile && sessionId) {
-      try {
-        const res = await postSessionReset({ session_id: sessionId, drop_profile: true });
-        // Adopt the new server-issued session id (or clear it if backend didn't return one)
-        if (res.session_id) {
-          setSessionId(res.session_id);
-          if (typeof window !== "undefined") {
-            sessionStorage.setItem("insurance_session_id", res.session_id);
-          }
-        } else {
-          setSessionId(undefined);
-          if (typeof window !== "undefined") {
-            sessionStorage.removeItem("insurance_session_id");
-          }
-        }
-      } catch (e) {
-        console.warn("session reset failed", e);
-        // Even if backend failed, drop client-side session so next message starts fresh
-        setSessionId(undefined);
-        if (typeof window !== "undefined") {
-          sessionStorage.removeItem("insurance_session_id");
-        }
+    // Ask the backend to rotate the session and wipe in-memory state. We
+    // always do this, even without a sessionId, so the user gets a guaranteed
+    // fresh UUID for their next turn.
+    try {
+      const res = await postSessionClear({ session_id: sessionId ?? "" });
+      setSessionId(res.new_session_id);
+      if (typeof window !== "undefined") {
+        sessionStorage.setItem("insurance_session_id", res.new_session_id);
+      }
+    } catch (e) {
+      console.warn("session clear failed", e);
+      // Even if backend failed, drop client-side session so the next message
+      // starts a fresh server-side session (handle_turn mints one when none
+      // is supplied).
+      setSessionId(undefined);
+      if (typeof window !== "undefined") {
+        sessionStorage.removeItem("insurance_session_id");
       }
     }
   }

frontend/src/lib/api.ts

@@ -431,6 +431,27 @@ export async function postSessionReset(
 }
 
 
+// KI-196 (ADR-041) — Clean Clear-chat semantic. Wipes in-memory session
+// state for the supplied session_id and ALWAYS returns a fresh UUID the
+// caller must adopt going forward. The on-disk profile JSON is preserved.
+export interface SessionClearResponse {
+  cleared: boolean;
+  new_session_id: string;
+}
+
+export async function postSessionClear(
+  args: { session_id: string }
+): Promise<SessionClearResponse> {
+  const resp = await fetch(`${BACKEND_URL}/api/session/clear`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ session_id: args.session_id }),
+  });
+  if (!resp.ok) throw new Error(`session clear failed: ${resp.status}`);
+  return resp.json();
+}
+
+
 export async function uploadPolicy(file: File): Promise<UploadResponse> {
   const fd = new FormData();
   fd.append("file", file);

frontend/src/lib/useStreamingVoice.ts

@@ -62,6 +62,21 @@ const BARGE_IN_BASE_THRESHOLD = 0.005;
 // making barge-in trivial. 0.6 is loud enough to hear clearly on
 // headphones and laptop speakers without overpowering user speech.
 const VOICE_MODE_TTS_VOLUME = 0.6;
+// KI-195 (2026-05-15) — adaptive TTS volume calibration relative to user's
+// own measured speech level. Architecture: while user speaks (recorder
+// active, NOT TTS) we sample mic RMS and track a rolling peak in
+// userSpeechRmsRef. While TTS plays, every 300ms we sample bot_rms_at_mic
+// via the KI-190 botAnalysers and reduce el.volume by 20% if bot_rms is
+// closer to user_rms than the target ratio. Floor at 0.15 so the bot
+// stays audible. This makes "bot bleed < user speech" a mathematical
+// guarantee after one calibration turn → barge-in always works, echo
+// never crosses the recognition threshold.
+const USER_SPEECH_RMS_INITIAL = 0.05;          // typical quiet speech, used until calibrated
+const USER_SPEECH_DETECTION_THRESHOLD = 0.02;  // mic RMS above this counts as "user speaking"
+const VOLUME_CALIB_TARGET_RATIO = 0.35;        // bot_rms_at_mic should be ≤ user_rms × this
+const VOLUME_CALIB_TICK_MS = 300;              // calibration sample period during TTS
+const VOLUME_CALIB_DUCK_FACTOR = 0.8;          // multiply el.volume by this per tick if too loud
+const VOLUME_CALIB_FLOOR = 0.15;               // never drop bot below this — must stay audible
 
 // Minimal types for the Web Speech API since lib.dom.d.ts ships them under
 // `webkitSpeechRecognition` only and the standard `SpeechRecognition` symbol
@@ -623,6 +638,108 @@ export function useStreamingVoice(
     }>();
     // Track which <audio> elements we've dimmed so we can restore on cleanup.
     const duckedAudios = new Set<HTMLAudioElement>();
+    // KI-195 — user-speech RMS tracker + per-element calibrated volume.
+    // userSpeechRms is the rolling peak of mic RMS observed while the user
+    // is actively speaking (recorder active, not TTS). It seeds the bot
+    // volume target. Calibrated volumes per element survive across turns
+    // so we don't have to re-learn after every reply.
+    let userSpeechRms = USER_SPEECH_RMS_INITIAL;
+    const calibratedVolumes = new Map<HTMLAudioElement, number>();
+    let userRmsRafId: number | null = null;
+    let volumeCalibIntervalId: ReturnType<typeof setInterval> | null = null;
+
+    const sampleUserRms = (): number => {
+      if (!analyser || !rmsBuf) return 0;
+      try {
+        analyser.getFloatTimeDomainData(rmsBuf);
+      } catch { return 0; }
+      let sumSq = 0;
+      for (let i = 0; i < rmsBuf.length; i++) {
+        const v = rmsBuf[i];
+        sumSq += v * v;
+      }
+      return Math.sqrt(sumSq / rmsBuf.length);
+    };
+
+    const userRmsTick = () => {
+      // Only learn while user is potentially speaking — recorder active,
+      // no TTS, voice mode on.
+      if (
+        !wantRunningRef.current
+        || isTtsPlayingRef.current
+        || !recorderActiveRef.current
+      ) {
+        userRmsRafId = null;
+        return;
+      }
+      if (!analyser || !rmsBuf) {
+        userRmsRafId = null;
+        return;
+      }
+      const rms = sampleUserRms();
+      // Only count as "user speaking" when above detection threshold.
+      // Then update userSpeechRms via slow EMA on peak so a single shout
+      // doesn't permanently raise the baseline.
+      if (rms > USER_SPEECH_DETECTION_THRESHOLD) {
+        userSpeechRms = Math.max(userSpeechRms * 0.95, rms);
+      }
+      userRmsRafId = requestAnimationFrame(userRmsTick);
+    };
+
+    const startUserRmsLoop = () => {
+      if (userRmsRafId !== null) return;
+      // Reuse the VAD analyser. startBargeInLoop sets it up; if it doesn't
+      // exist yet, the loop will exit on first tick (analyser null) and
+      // restart on the next state transition.
+      userRmsRafId = requestAnimationFrame(userRmsTick);
+    };
+
+    const stopUserRmsLoop = () => {
+      if (userRmsRafId !== null) {
+        cancelAnimationFrame(userRmsRafId);
+        userRmsRafId = null;
+      }
+    };
+
+    // KI-195 — volume calibration tick. Runs during TTS. Samples bot RMS
+    // at the mic via botAnalysers. If bot is louder than target relative
+    // to userSpeechRms, duck el.volume by 20% per tick down to the floor.
+    const calibrateBotVolume = () => {
+      if (!isTtsPlayingRef.current) {
+        if (volumeCalibIntervalId !== null) {
+          clearInterval(volumeCalibIntervalId);
+          volumeCalibIntervalId = null;
+        }
+        return;
+      }
+      const target = userSpeechRms * VOLUME_CALIB_TARGET_RATIO;
+      const botRms = computeBotRms();
+      if (botRms > target) {
+        ttsAudioElementsRef.current.forEach((el) => {
+          if (el.paused || el.ended) return;
+          const cur = el.volume;
+          const next = Math.max(VOLUME_CALIB_FLOOR, cur * VOLUME_CALIB_DUCK_FACTOR);
+          if (next < cur - 0.001) {
+            try {
+              el.volume = next;
+              calibratedVolumes.set(el, next);
+            } catch { /* ignore */ }
+          }
+        });
+      }
+    };
+
+    const startVolumeCalibration = () => {
+      if (volumeCalibIntervalId !== null) return;
+      volumeCalibIntervalId = setInterval(calibrateBotVolume, VOLUME_CALIB_TICK_MS);
+    };
+
+    const stopVolumeCalibration = () => {
+      if (volumeCalibIntervalId !== null) {
+        clearInterval(volumeCalibIntervalId);
+        volumeCalibIntervalId = null;
+      }
+    };
 
     const stopBargeInLoop = () => {
       if (rafId !== null) {
@@ -831,6 +948,10 @@ export function useStreamingVoice(
         if (rec) {
           try { rec.abort(); } catch { /* ignore */ }
         }
+        // KI-195 — user cannot be speaking during TTS playback; stop the
+        // RMS-learning loop until TTS ends so we don't capture bot audio
+        // bleed-through as "user speech level".
+        stopUserRmsLoop();
         // KI-191 — re-duck every playing audio in case React or the audio
         // element default reset volume after watchAudio set it.
         ttsAudioElementsRef.current.forEach((el) => {
@@ -838,6 +959,9 @@ export function useStreamingVoice(
             try { el.volume = VOICE_MODE_TTS_VOLUME; } catch { /* ignore */ }
           }
         });
+        // KI-195 — once the volume floor is set, begin adaptive calibration
+        // so the bot's volume tracks the learned user speech level.
+        startVolumeCalibration();
         // KI-192 (2026-05-15) — MediaRecorder might be torn down between
         // user utterances (KI-168 teardownAudio). Without an active
         // recorder, startBargeInLoop bails on the recorderActiveRef check
@@ -859,6 +983,10 @@ export function useStreamingVoice(
         // Trigger immediately too so the user doesn't wait ~4s.
         console.debug("[useStreamingVoice] KI-188 TTS ended — resuming recognition");
         stopBargeInLoop();
+        // KI-195 — freeze the per-element calibrated volume and resume
+        // learning the user's speech RMS for the next turn.
+        stopVolumeCalibration();
+        startUserRmsLoop();
         if (wantRunningRef.current && !isTextRequestPendingRef.current) {
           safeStart();
         }
@@ -870,8 +998,12 @@ export function useStreamingVoice(
       ttsAudioElementsRef.current.add(el);
       // KI-191 — duck bot TTS to 60% while voice mode is on, so AEC residual
       // is even quieter and barge-in is trivial.
+      // KI-195 — if we already calibrated a volume for this exact element on
+      // a previous turn (rare — elements are usually recreated), reuse it so
+      // we don't reset the adaptive level on every play() event.
       try {
-        el.volume = VOICE_MODE_TTS_VOLUME;
+        const prior = calibratedVolumes.get(el);
+        el.volume = prior !== undefined ? prior : VOICE_MODE_TTS_VOLUME;
         duckedAudios.add(el);
       } catch { /* readonly volume on some platforms — ignore */ }
       // KI-190 — attach bot-level analyser for adaptive threshold.
@@ -916,7 +1048,18 @@ export function useStreamingVoice(
     });
     observer.observe(document.body, { childList: true, subtree: true });
 
+    // KI-195 — kick off the user-RMS learning loop on mount so by the time
+    // the first TTS plays we already have a baseline. The loop self-exits
+    // when conditions aren't met (no analyser / no stream / in TTS), so
+    // firing it unconditionally here is safe.
+    startUserRmsLoop();
+
     return () => {
+      // KI-195 — tear down adaptive volume calibration before clearing
+      // ducked-audio state so the calibration tick can't race a clear().
+      stopUserRmsLoop();
+      stopVolumeCalibration();
+      calibratedVolumes.clear();
       observer.disconnect();
       // KI-191 — restore bot TTS volume to default before unmount so a
       // subsequent voice-OFF session doesn't end up with silent audio.