refactor(session): KI-118 — rip out session_id concept, name-based profiles only

- drop disk persistence in session_state.py (40-data/sessions/<id>.json no longer written)
- profile_rag chunks keyed by name_slug instead of session_id
- retrieve() takes profile_name_slug not session_id
- frontend uses sessionStorage not localStorage so token clears on tab close
- anonymous sessions never write to Chroma — corruption surface eliminated
- KI-102/107/112/117 isolation scaffolding still in place but its threat model is now moot
- added rehydrate_by_name() for cross-session re-entry via name lookup
- tests flipped: profile-isolation now pins anonymous-no-write + name-A-vs-name-B
- added tests/test_session_no_disk_persistence.py to lock the no-disk contract

Rationale: insurance shoppers don't multi-session; cross-session memory
is name-based (returning user provides name → pull saved profile). The
session_id machinery was the root of the Chroma corruption fought
today (profile_anonymous dangling row).

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

backend/main.py

@@ -250,8 +250,16 @@ async def _startup_purge_dangling_profile_chunks():
         metas = res.get("metadatas") or []
         bad_ids: list[str] = []
         for cid, meta in zip(ids, metas):
+            # KI-118 (2026-05-15) — profile chunks are now keyed by name_slug;
+            # accept EITHER a non-empty name_slug (new) OR a non-empty
+            # session_id (legacy KI-102 row) as proof-of-ownership. A profile
+            # chunk with neither key is the dangling-row corruption case and
+            # must be purged.
+            slug = (meta or {}).get("name_slug")
             sid = (meta or {}).get("session_id")
-            if not (isinstance(sid, str) and sid.strip()):
+            slug_ok = isinstance(slug, str) and slug.strip()
+            sid_ok = isinstance(sid, str) and sid.strip()
+            if not (slug_ok or sid_ok):
                 bad_ids.append(cid)
 
         if bad_ids:
@@ -849,8 +857,15 @@ async def profile_update(req: ProfileUpdateRequest):
     # Ingest the profile into the RAG store so the brain sees user context
     # at retrieval time alongside policy + regulatory chunks. Fire-and-forget
     # — a profile upsert failure shouldn't block the API response.
+    # KI-118 (2026-05-15) — gated on a known name; anonymous saves don't
+    # write to Chroma. The chunk is keyed by canonical name slug, not the
+    # session_id which is now opaque/in-memory.
     try:
-        await upsert_profile_chunk(req.session_id, profile_dict)
+        if p.name:
+            from backend.profile_store import _normalise_name
+            name_slug = _normalise_name(p.name)
+            if name_slug:
+                await upsert_profile_chunk(name_slug, profile_dict)
     except Exception as e:
         print(f"[profile_rag] upsert failed for {req.session_id}: {type(e).__name__}: {e}")
 

backend/orchestrator.py

@@ -548,26 +548,30 @@ async def handle_turn(
             session.free_form_session = True
             session._flush()
 
-        # KI-063 — opportunistic profile-chunk re-upsert if any field changed
-        # so this turn's downstream retrieval (when complete=true triggers a
-        # follow-up recommendation) sees the latest profile.
-        if fact_find_profile_updates:
+        # KI-118 (2026-05-15) — profile-chunk upsert is gated on a known
+        # name. Anonymous sessions never write to Chroma; the corruption
+        # surface (profile_anonymous dangling row) is eliminated. Named
+        # users get their chunk keyed by canonical name slug, not session_id.
+        if fact_find_profile_updates and session.profile.name:
             try:
                 from backend.profile_rag import upsert_profile_chunk
+                from backend.profile_store import _normalise_name
                 p = session.profile
-                await upsert_profile_chunk(session_id or "anonymous", {
-                    "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,
-                })
+                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,
+                    })
             except Exception as e:
                 logging.warning(
                     "fact_find_brain profile-chunk upsert failed (session=%s): %s: %s",
@@ -577,9 +581,18 @@ async def handle_turn(
         # 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.
         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)
                 save_profile(session.profile.name, session.profile, session_id=session_id)
             except Exception:
                 pass
@@ -707,23 +720,29 @@ async def handle_turn(
                 else:
                     session.update_profile_field(field_name, new_value)
                     profile_updates_applied[field_name] = new_value
-            # Re-upsert profile chunk so retrieval sees fresh profile THIS turn
+            # Re-upsert profile chunk so retrieval sees fresh profile THIS turn.
+            # KI-118 (2026-05-15) — gated on a known name; anonymous sessions
+            # never write to Chroma.
             try:
-                from backend.profile_rag import upsert_profile_chunk
-                profile_dict_for_chunk = {
-                    "age": session.profile.age,
-                    "dependents": session.profile.dependents,
-                    "income_band": session.profile.income_band,
-                    "existing_cover_inr": session.profile.existing_cover_inr,
-                    "primary_goal": session.profile.primary_goal,
-                    "location_tier": session.profile.location_tier,
-                    "parents_to_insure": session.profile.parents_to_insure,
-                    "parents_age_max": session.profile.parents_age_max,
-                    "parents_has_ped": session.profile.parents_has_ped,
-                    "budget_band": session.profile.budget_band,
-                    "health_conditions": session.profile.health_conditions,
-                }
-                await upsert_profile_chunk(session_id or "anonymous", profile_dict_for_chunk)
+                if session.profile.name:
+                    from backend.profile_rag import upsert_profile_chunk
+                    from backend.profile_store import _normalise_name
+                    slug = _normalise_name(session.profile.name)
+                    if slug:
+                        profile_dict_for_chunk = {
+                            "age": session.profile.age,
+                            "dependents": session.profile.dependents,
+                            "income_band": session.profile.income_band,
+                            "existing_cover_inr": session.profile.existing_cover_inr,
+                            "primary_goal": session.profile.primary_goal,
+                            "location_tier": session.profile.location_tier,
+                            "parents_to_insure": session.profile.parents_to_insure,
+                            "parents_age_max": session.profile.parents_age_max,
+                            "parents_has_ped": session.profile.parents_has_ped,
+                            "budget_band": session.profile.budget_band,
+                            "health_conditions": session.profile.health_conditions,
+                        }
+                        await upsert_profile_chunk(slug, profile_dict_for_chunk)
             except Exception as e:
                 # KI-005 — log profile-chunk upsert failures so we can see
                 # when Chroma is locking or schema-drifting. The chat still
@@ -764,10 +783,22 @@ async def handle_turn(
     if intent in ("recommendation", "comparison") and not policy_filter_ids:
         effective_top_k = max(top_k, 12)
 
+    # KI-118 (2026-05-15) — profile chunks are keyed by name_slug, not
+    # session_id. Pass the slug only when the live session has a captured
+    # name; anonymous sessions get retrieval without a profile-boost pass.
+    profile_slug_for_retrieve: Optional[str] = None
+    if session.profile.name:
+        try:
+            from backend.profile_store import _normalise_name
+            profile_slug_for_retrieve = _normalise_name(session.profile.name) or None
+        except Exception:
+            profile_slug_for_retrieve = None
+
     chunks: list[RetrievedChunk] = await retrieve(
         query=user_text,
         top_k=effective_top_k,
         policy_ids=policy_filter_ids,
+        profile_name_slug=profile_slug_for_retrieve,
         session_id=session_id,
     )
     context_str = format_for_llm_context(chunks)

backend/profile_rag.py

@@ -1,27 +1,26 @@
 """Customer-profile-as-RAG layer.
 
-When a user saves their profile (POST /api/profile), the profile dict is
-serialised into a natural-language paragraph and ingested into the same
-Chroma collection that holds policy + regulatory chunks. Metadata fields
-`doc_type='profile'` and `policy_id='profile_<session_id>'` distinguish it.
-
-At retrieval time, `rag/retrieve.py::retrieve(..., session_id=...)` can
-preferentially boost the matching profile chunk so the LLM sees the user's
-context inline with the retrieved policy/regulatory text — answers become
-personalised at the BRAIN level, not just at scorecard re-weighting.
-
-This is what the user meant by "we need an architecture to store customer
-profiles for RAG, complementing the brain alongside policy + regulation".
+KI-118 (2026-05-15) — profile chunks are now keyed by `name_slug` (the
+canonicalised user name), NOT by session_id. Only NAMED users ever get
+embedded; anonymous sessions never write to Chroma. This eliminates the
+corruption surface that the session_id-keyed chunks introduced (a missing
+session_id metadata field on the legacy `profile_anonymous` row poisoned
+every subsequent retrieval query — see KI-117 boot cleanup).
+
+At retrieval time, `rag/retrieve.py::retrieve(..., profile_name_slug=...)`
+boosts the user's profile chunk so the LLM sees the user's context inline
+with the retrieved policy/regulatory text — answers become personalised at
+the BRAIN level, not just at scorecard re-weighting.
 
 Public API:
     profile_to_chunk_text(profile_dict) -> str
         Render the structured profile as a single English paragraph.
-    upsert_profile_chunk(session_id, profile_dict, embedder) -> None
-        Ingest / update the chunk for this session in Chroma.
-    remove_profile_chunk(session_id) -> None
-        Optional cleanup on session expiry.
+    upsert_profile_chunk(name_slug, profile_dict, embedder) -> None
+        Ingest / update the chunk for this named user in Chroma.
+    remove_profile_chunk(name_slug) -> None
+        Optional cleanup.
 
-Storage model — one chunk per session_id. Replaced on each profile update.
+Storage model — one chunk per name_slug. Replaced on each profile update.
 Profile chunks live in the SAME collection as policies so retrieval can
 naturally surface them when scoring policies for the user.
 """
@@ -122,11 +121,16 @@ def _get_collection():
     )
 
 
-async def upsert_profile_chunk(session_id: str, profile_dict: dict) -> None:
+async def upsert_profile_chunk(name_slug: str, profile_dict: dict) -> None:
     """Embed the profile paragraph and store as a single chunk in Chroma.
 
+    KI-118 (2026-05-15) — keyed by `name_slug` (canonical user name) instead
+    of `session_id`. Only NAMED users ever get embedded — anonymous chats
+    never write to Chroma, which eliminates the corruption surface that the
+    session_id keying introduced.
+
     Idempotent — calling this on every profile update is safe; existing
-    chunks for the same session_id get replaced.
+    chunks for the same name_slug get replaced.
     """
     from backend.providers.local_embeddings import LocalEmbeddings
 
@@ -134,20 +138,19 @@ async def upsert_profile_chunk(session_id: str, profile_dict: dict) -> None:
     if not text or len(text) < 30:
         return
 
-    # KI-112 (2026-05-15) — input guard 1: session_id must be a non-empty str.
-    # Pre-fix, a missing/empty session_id caused upsert under id "profile_"
-    # with `policy_id` colliding across all anonymous sessions — and the
-    # initial KI-102 deploy wrote a `profile_anonymous` chunk WITHOUT a
-    # session_id metadata field. That legacy chunk poisoned every subsequent
-    # query whose `where` clause referenced session_id ($ne / $eq) — Chroma's
-    # HNSW + metadata-filter plan executor raised "Error finding id" against
-    # the dangling row. Reject early with a noisy log so the bad write never
-    # reaches Chroma.
-    if not isinstance(session_id, str) or not session_id.strip():
+    # KI-112 (2026-05-15) / KI-118 — input guard 1: name_slug must be a
+    # non-empty str. Pre-fix, a missing/empty key caused upsert under id
+    # "profile_" with `policy_id` colliding across all anonymous sessions;
+    # the initial KI-102 deploy wrote a `profile_anonymous` chunk WITHOUT a
+    # session_id metadata field that poisoned every subsequent query whose
+    # `where` clause referenced session_id. Anonymous users no longer reach
+    # this function at all (the orchestrator gates on
+    # `session.profile.name` before calling) — this guard is belt-and-braces.
+    if not isinstance(name_slug, str) or not name_slug.strip():
         _log.warning(
-            "profile_rag.upsert_profile_chunk: refusing to write — session_id "
+            "profile_rag.upsert_profile_chunk: refusing to write — name_slug "
             "must be a non-empty str, got %r. Profile not persisted.",
-            session_id,
+            name_slug,
         )
         return
 
@@ -167,17 +170,17 @@ async def upsert_profile_chunk(session_id: str, profile_dict: dict) -> None:
     ):
         _log.warning(
             "profile_rag.upsert_profile_chunk: refusing to write — embedding "
-            "shape invalid for session_id=%s (expected %d-dim list of floats, "
+            "shape invalid for name_slug=%s (expected %d-dim list of floats, "
             "got type=%s len=%s). Profile not persisted.",
-            session_id, expected_dim, type(vec).__name__,
+            name_slug, expected_dim, type(vec).__name__,
             (len(vec) if hasattr(vec, "__len__") else "?"),
         )
         return
 
     coll = _get_collection()
-    chunk_id = f"profile_{session_id}"
+    chunk_id = f"profile_{name_slug}"
 
-    # Replace any existing chunk for this session
+    # Replace any existing chunk for this name
     try:
         coll.delete(where={"policy_id": chunk_id})
     except Exception as e:
@@ -203,49 +206,46 @@ async def upsert_profile_chunk(session_id: str, profile_dict: dict) -> None:
             metadatas=[{
                 "policy_id": chunk_id,
                 "insurer_slug": "profile",
-                "policy_name": f"User profile (session {session_id[:8]})",
+                "policy_name": f"User profile ({name_slug[:16]})",
                 "doc_type": "profile",
-                # KI-102 (2026-05-15) — privacy P0. Stamp the owning session_id so
-                # the retrieve path can hard-exclude any OTHER session's profile
-                # chunk via where={"session_id": current}. Pre-fix, profile chunks
-                # had no session_id metadata and the main retrieval pass had no
-                # doc_type filter, so chunks from sessions smokeA_1, ki100_ve, etc.
-                # surfaced as cosine matches in session smokeB_B4's context —
-                # leaking one user's profile facts into another user's reply.
-                "session_id": session_id,
+                # KI-118 (2026-05-15) — stamp name_slug instead of session_id.
+                # The retrieve path filters profile chunks via this field.
+                "name_slug": name_slug,
                 "source_url": "",
                 "page_start": 0,
                 "page_end": 0,
                 "chunk_idx": 0,
-                "local_path": "in-memory session profile",
+                "local_path": "in-memory named-profile chunk",
             }],
         )
     except Exception as e:
         _log.warning(
             "profile_rag.upsert_profile_chunk: add(id=%s) failed: %s: %s — "
-            "user reply will proceed without per-session profile context; "
+            "user reply will proceed without per-user profile context; "
             "next profile change will retry.",
             chunk_id, type(e).__name__, str(e)[:200],
         )
 
 
-def remove_profile_chunk(session_id: str) -> None:
-    """Optional cleanup. Called on session expiry (1h TTL in session_state)."""
+def remove_profile_chunk(name_slug: str) -> None:
+    """Optional cleanup. KI-118 — keyed by name_slug."""
+    if not name_slug:
+        return
     try:
         coll = _get_collection()
-        coll.delete(where={"policy_id": f"profile_{session_id}"})
+        coll.delete(where={"policy_id": f"profile_{name_slug}"})
     except Exception:
         pass
 
 
-def upsert_profile_chunk_sync(session_id: str, profile_dict: dict) -> None:
+def upsert_profile_chunk_sync(name_slug: str, profile_dict: dict) -> None:
     """Sync wrapper for callers that aren't async — schedules + waits."""
     try:
         loop = asyncio.get_event_loop()
         if loop.is_running():
             # Already inside an async context — schedule on the loop
-            asyncio.ensure_future(upsert_profile_chunk(session_id, profile_dict))
+            asyncio.ensure_future(upsert_profile_chunk(name_slug, profile_dict))
             return
     except RuntimeError:
         pass
-    asyncio.run(upsert_profile_chunk(session_id, profile_dict))
+    asyncio.run(upsert_profile_chunk(name_slug, profile_dict))

backend/session_state.py

@@ -1,44 +1,43 @@
-"""Per-session state for multi-turn fact-find continuity, persisted to disk.
+"""Per-session state for multi-turn fact-find continuity (in-memory only).
 
 The orchestrator was originally stateless — each user turn re-classified
 intent from scratch. That broke fact-find: after the bot asked "what's
 your age?", the user's "39 years old" wasn't matched by intent_classifier
 and got routed to RAG retrieval (which then refused). This module fixes that.
 
-Persistence model (changed 2026-05-14):
-  - In-memory dict for hot reads (avoids hitting disk every turn).
-  - JSON file per session at 40-data/sessions/<session_id>.json — survives
-    Space restarts so a user returning after HF hibernation finds their
-    profile intact.
-  - Loaded lazily on first get_session(); flushed on every state mutation.
-  - Same 1h idle TTL applies but ONLY garbage-collects the in-memory cache.
-    The on-disk file lives until session_state.purge_old_files() runs (called
-    daily by a cron, or on startup). 30-day disk TTL.
+Persistence model (KI-118, 2026-05-15):
+  - In-memory dict ONLY. No disk persistence.
+  - Sessions are evicted from memory after `_TTL_SECONDS = 60 * 60` idle.
+  - Cross-session memory is name-based: when the user provides a name,
+    `rehydrate_by_name(session, name)` pulls the named profile from
+    `backend.profile_store.load_profile(name)` (canonical JSON at
+    `40-data/profiles/<persona_id>.json`).
+  - Anonymous sessions live only in-memory and never leave a trace on disk.
 
-The previous architecture (in-memory only) was lost on every Space cold-
-start. With HF Spaces hibernating after ~30 min of idleness, every returning
-user was getting a fresh blank profile.
+Rationale: insurance shoppers don't multi-session within a browsing window.
+Cross-session memory is name-based. The previous disk-write side
+(`40-data/sessions/<session_id>.json`) was the root of the Chroma
+corruption fought 2026-05-14/15 (profile_anonymous dangling row).
 
 Public API:
     get_session(session_id) -> SessionState
+    rehydrate_by_name(session, name) -> bool   # KI-118 cross-session re-entry
     SessionState.profile, .asked, .awaiting (question id pending answer)
     SessionState.set_awaiting(qid)
-    SessionState.record_answer(qid, raw_answer) → also clears awaiting + flushes
+    SessionState.record_answer(qid, raw_answer) → also clears awaiting
 """
 
 from __future__ import annotations
 
-import json
+import logging
 import time
-from dataclasses import dataclass, field, asdict
-from pathlib import Path
+from dataclasses import dataclass, field
 from threading import Lock
 from typing import Optional
 
 from backend.needs_finder import Profile, record_answer
 
-# On-disk storage root. Created on first write.
-_DATA_ROOT = Path(__file__).resolve().parent.parent / "40-data" / "sessions"
+_log = logging.getLogger(__name__)
 
 
 @dataclass
@@ -50,33 +49,16 @@ class SessionState:
     last_touched: float = field(default_factory=time.time)
 
     def _flush(self) -> None:
-        """Atomic write to 40-data/sessions/<id>.json so a restart doesn't lose state."""
-        try:
-            _DATA_ROOT.mkdir(parents=True, exist_ok=True)
-            target = _DATA_ROOT / f"{self.session_id}.json"
-            payload = {
-                "session_id": self.session_id,
-                "profile": asdict(self.profile),
-                "awaiting_question_id": self.awaiting_question_id,
-                "free_form_session": self.free_form_session,
-                "last_touched": self.last_touched,
-            }
-            tmp = target.with_suffix(".json.tmp")
-            tmp.write_text(json.dumps(payload, indent=2))
-            tmp.replace(target)
-        except Exception as e:
-            # KI-002 — Log silent failures so HF Space logs reveal them.
-            # Don't crash the request — disk hiccups shouldn't kill chat.
-            import logging
-            logging.warning(
-                "session_state flush failed for %s: %s: %s",
-                self.session_id, type(e).__name__, str(e)[:200],
-            )
+        """No-op since KI-118 (2026-05-15). Disk persistence was removed; the
+        in-memory dict is now the only store. We keep the method so existing
+        callers (orchestrator + fact_find_brain + tests) don't have to change
+        their write paths.
+        """
+        return None
 
     def set_awaiting(self, question_id: Optional[str]) -> None:
         self.awaiting_question_id = question_id
         self.last_touched = time.time()
-        self._flush()
 
     def record_user_answer(self, raw_answer: str) -> Optional[str]:
         """If we're awaiting an answer, parse + store it. Returns the answered question_id."""
@@ -86,122 +68,102 @@ class SessionState:
         record_answer(self.profile, qid, raw_answer)
         self.awaiting_question_id = None
         self.last_touched = time.time()
-        self._flush()
         return qid
 
     def update_profile_field(self, name: str, value) -> None:
-        """Set a Profile attribute + flush. Used by /api/profile."""
+        """Set a Profile attribute. Used by /api/profile."""
         if hasattr(self.profile, name):
             setattr(self.profile, name, value)
             self.last_touched = time.time()
-            self._flush()
-
-
-def _load_from_disk(session_id: str) -> Optional[SessionState]:
-    """Rehydrate from 40-data/sessions/<id>.json if it exists."""
-    target = _DATA_ROOT / f"{session_id}.json"
-    if not target.exists():
-        return None
-    try:
-        raw = json.loads(target.read_text())
-        raw_profile_dict = raw.get("profile", {}) or {}
-        # Profile may have new fields added since this file was written —
-        # filter to only what the current Profile dataclass accepts so we
-        # never crash on schema drift.
-        valid_fields = {f for f in Profile.__dataclass_fields__.keys()}
-        dropped = set(raw_profile_dict.keys()) - valid_fields
-        if dropped:
-            # KI-095 — log schema-drift drops so silent data loss is visible.
-            import logging
-            logging.warning(
-                "session_state load_from_disk dropped %d unknown profile keys for %s: %s",
-                len(dropped), session_id, sorted(dropped),
-            )
-        prof_dict = {k: v for k, v in raw_profile_dict.items() if k in valid_fields}
-        return SessionState(
-            session_id=raw["session_id"],
-            profile=Profile(**prof_dict),
-            awaiting_question_id=raw.get("awaiting_question_id"),
-            free_form_session=bool(raw.get("free_form_session", False)),
-            last_touched=float(raw.get("last_touched", time.time())),
-        )
-    except Exception as e:
-        # KI-003 — Log schema-drift / corrupt-JSON failures. The user will
-        # get a fresh session either way, but the log lets us detect when
-        # the Profile dataclass evolves in a way that breaks old sessions.
-        import logging
-        logging.warning(
-            "session_state load_from_disk failed for %s: %s: %s",
-            session_id, type(e).__name__, str(e)[:200],
-        )
-        return None
 
 
 _sessions: dict[str, SessionState] = {}
 _lock = Lock()
-_TTL_SECONDS = 60 * 60         # 1h idle → evict from in-memory cache (still on disk)
-_DISK_TTL_SECONDS = 30 * 86400  # 30 days → delete the JSON file
+_TTL_SECONDS = 60 * 60         # 1h idle → evict from in-memory cache
 
 
 def get_session(session_id: str) -> SessionState:
     with _lock:
         now = time.time()
-        # Evict idle entries from the hot cache (file on disk survives)
+        # Evict idle entries from the hot cache
         to_kill = [k for k, v in _sessions.items() if now - v.last_touched > _TTL_SECONDS]
         for k in to_kill:
             del _sessions[k]
         if session_id in _sessions:
             return _sessions[session_id]
-        # Try disk first — survives Space restarts
-        rehydrated = _load_from_disk(session_id)
-        if rehydrated is None:
-            rehydrated = SessionState(session_id=session_id)
-        _sessions[session_id] = rehydrated
+        # KI-118 — no disk lookup; fresh sessions start blank. Cross-session
+        # rehydration happens via rehydrate_by_name() when the user provides
+        # their name to the fact_find brain.
+        _sessions[session_id] = SessionState(session_id=session_id)
         return _sessions[session_id]
 
 
+def rehydrate_by_name(session: SessionState, name: str) -> bool:
+    """KI-118 (2026-05-15) — cross-session re-entry point.
+
+    When a user provides their name (captured in the fact_find brain), look
+    up the named profile via `backend.profile_store.load_profile(name)` and
+    populate the in-memory session with the stored profile data.
+
+    Returns True if a stored profile was found and merged; False otherwise.
+    Failures are logged but never raise — a fresh chat must always proceed.
+
+    Merge semantics: stored fields override current session-state fields IF
+    the current field is empty. Already-captured fields on the live session
+    win (the user may have corrected themselves mid-conversation).
+    """
+    if not name or not name.strip():
+        return False
+    try:
+        from backend.profile_store import load_profile
+        stored = load_profile(name)
+        if stored is None:
+            return False
+        # Merge: stored value fills any empty slot on the live profile.
+        for fld in Profile.__dataclass_fields__.keys():
+            try:
+                cur = getattr(session.profile, fld, None)
+                if cur in (None, "", []):
+                    new = getattr(stored, fld, None)
+                    if new not in (None, "", []):
+                        setattr(session.profile, fld, new)
+            except Exception:
+                continue
+        session.last_touched = time.time()
+        return True
+    except Exception as e:
+        _log.warning(
+            "rehydrate_by_name failed (name=%r): %s: %s",
+            name, type(e).__name__, str(e)[:200],
+        )
+        return False
+
+
 def set_free_form(session_id: str, free_form: bool = True) -> None:
     s = get_session(session_id)
     s.free_form_session = free_form
     s.awaiting_question_id = None
     s.last_touched = time.time()
-    s._flush()
 
 
 def reset_session(session_id: str) -> bool:
-    """Delete a session — evict from in-memory cache and remove the disk file.
+    """Delete a session — evict from in-memory cache.
     Returns True if anything was actually deleted.
-    KI-020 (2026-05-14) — backs the user-facing "Clear chat / start fresh" toggle."""
-    deleted_any = False
+
+    KI-020 (2026-05-14) — backs the user-facing "Clear chat / start fresh" toggle.
+    KI-118 (2026-05-15) — no disk file to remove anymore; in-memory eviction
+    is the only side effect. Returns True iff the session id was live.
+    """
     with _lock:
         if session_id in _sessions:
             del _sessions[session_id]
-            deleted_any = True
-    target = _DATA_ROOT / f"{session_id}.json"
-    if target.exists():
-        try:
-            target.unlink()
-            deleted_any = True
-        except Exception as e:
-            import logging
-            logging.warning(
-                "reset_session unlink failed for %s: %s: %s",
-                session_id, type(e).__name__, str(e)[:200],
-            )
-    return deleted_any
+            return True
+    return False
 
 
 def purge_old_files() -> int:
-    """Delete on-disk session files older than _DISK_TTL_SECONDS. Returns count."""
-    if not _DATA_ROOT.exists():
-        return 0
-    now = time.time()
-    purged = 0
-    for f in _DATA_ROOT.glob("*.json"):
-        try:
-            if (now - f.stat().st_mtime) > _DISK_TTL_SECONDS:
-                f.unlink()
-                purged += 1
-        except Exception:
-            continue
-    return purged
+    """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
+    (cron / startup hook) doesn't crash on attribute miss.
+    """
+    return 0

frontend/src/app/page.tsx

@@ -83,7 +83,9 @@ export default function Page() {
   // turn) — drives the score-gate on marketplace cards + detail modal.
   useEffect(() => {
     if (typeof window !== "undefined" && sessionId) {
-      localStorage.setItem("insurance_session_id", sessionId);
+      // KI-118 (2026-05-15) — sessionStorage clears on tab close so no
+      // persistent ghost session. Within-tab refresh still rehydrates.
+      sessionStorage.setItem("insurance_session_id", sessionId);
       getProfileCompleteness(sessionId)
         .then(setProfileCompleteness)
         .catch(() => setProfileCompleteness(null));
@@ -104,7 +106,10 @@ export default function Page() {
         localStorage.removeItem("insurance_chat_messages");
       }
     }
-    const savedSession = localStorage.getItem("insurance_session_id");
+    // KI-118 (2026-05-15) — sessionStorage clears on tab close. Cross-tab
+    // re-entry is name-based: when the user provides their name in chat,
+    // the backend pulls the named profile via profile_store.load_profile().
+    const savedSession = sessionStorage.getItem("insurance_session_id");
     if (savedSession) setSessionId(savedSession);
   }, []);
 
@@ -250,12 +255,12 @@ export default function Page() {
         if (res.session_id) {
           setSessionId(res.session_id);
           if (typeof window !== "undefined") {
-            localStorage.setItem("insurance_session_id", res.session_id);
+            sessionStorage.setItem("insurance_session_id", res.session_id);
           }
         } else {
           setSessionId(undefined);
           if (typeof window !== "undefined") {
-            localStorage.removeItem("insurance_session_id");
+            sessionStorage.removeItem("insurance_session_id");
           }
         }
       } catch (e) {
@@ -263,7 +268,7 @@ export default function Page() {
         // Even if backend failed, drop client-side session so next message starts fresh
         setSessionId(undefined);
         if (typeof window !== "undefined") {
-          localStorage.removeItem("insurance_session_id");
+          sessionStorage.removeItem("insurance_session_id");
         }
       }
     }
@@ -971,7 +976,7 @@ function ProfileBuilderPanel({
     if (!sid) {
       sid = `s_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
       setSessionId(sid);
-      if (typeof window !== "undefined") localStorage.setItem("insurance_session_id", sid);
+      if (typeof window !== "undefined") sessionStorage.setItem("insurance_session_id", sid);
     }
     try {
       const resp = await postProfileUpdate({
@@ -2619,7 +2624,7 @@ function PolicyDetailModal({ policy, onClose }: { policy: MarketplacePolicy; onC
     // Profile completeness gates whether we render the per-user grade.
     // Below threshold: show universal grade only (insurer-quality-led) with a
     // CTA to complete the profile.
-    const sid = typeof window !== "undefined" ? localStorage.getItem("insurance_session_id") || undefined : undefined;
+    const sid = typeof window !== "undefined" ? sessionStorage.getItem("insurance_session_id") || undefined : undefined;
     getProfileCompleteness(sid).then(setCompleteness).catch(() => setCompleteness(null));
   }, [policy.policy_id, policy.insurer_slug]);
   const isPersonalized = completeness?.is_personalized === true;

rag/retrieve.py

@@ -198,6 +198,7 @@ async def retrieve(
     policy_ids: Optional[list[str]] = None,
     insurer_slugs: Optional[list[str]] = None,
     embedder: Optional[VoyageEmbeddings] = None,
+    profile_name_slug: Optional[str] = None,
     session_id: Optional[str] = None,
 ) -> list[RetrievedChunk]:
     """Embed the query and return top-k most similar chunks.
@@ -210,6 +211,12 @@ async def retrieve(
     merges the top 3 of those (score-boosted ×1.2) into the result set.
     This ensures the brain sees regulatory ceilings even when policy
     chunks dominate raw cosine.
+
+    KI-118 (2026-05-15) — profile chunks are keyed by `profile_name_slug`
+    (canonical user name) instead of session_id. Only named users have a
+    profile chunk to boost. `session_id` is retained for the per-session
+    quarantine (user-uploaded PDF) lookup but is no longer used for
+    profile boosting.
     """
     # KI-034 — short-circuit identical-query re-asks via the LRU cache.
     cache_key = _cache_key(query, top_k, policy_ids, insurer_slugs)
@@ -295,29 +302,28 @@ async def retrieve(
             effective_top_k, where, type(e).__name__, str(e)[:300],
         )
 
-    # Profile boost pass — when the orchestrator passes a session_id, look
-    # up THAT user's profile chunk in Chroma. Inject it at the top of the
-    # context (always, not just on high cosine) so the LLM sees the user
-    # context block before any policy text. Mirrors the regulatory-boost
-    # pattern below.
-    if session_id:
-        profile_chunk_id = f"profile_{session_id}"
-        # KI-102 — defence-in-depth. Filter by BOTH id AND session_id
-        # metadata so any future ID collision (or migration-era chunk
-        # written under a shared id) still can't leak another user's
-        # profile into this session's context.
+    # Profile boost pass — KI-118 (2026-05-15). Profile chunks are now
+    # keyed by `profile_name_slug` (canonical user name), not session_id.
+    # The orchestrator passes the slug only when the live session has a
+    # known name; anonymous chats skip this branch entirely.
+    if profile_name_slug:
+        profile_chunk_id = f"profile_{profile_name_slug}"
+        # KI-118 — filter by BOTH id AND name_slug metadata so any future
+        # ID collision (or migration-era chunk written under a shared id)
+        # still can't leak the wrong profile into this user's context.
         #
-        # KI-107 (2026-05-15) — route through _safe_collection_get because new
-        # sessions (no profile saved yet) and certain transient Chroma sqlite
-        # states cause collection.get(ids=[missing_id]) to raise instead of
-        # returning empty lists. Before this fix the bare `except: pass` here
-        # masked the failure, the orchestrator's plan-executor still saw an
-        # AttributeError on the swallowed-None path, and the user got HTTP
-        # 500 "Error executing plan: Internal error: Error finding id".
+        # KI-107 (2026-05-15) — route through _safe_collection_get because
+        # first-time named users (no profile saved yet) and certain transient
+        # Chroma sqlite states cause collection.get(ids=[missing_id]) to
+        # raise instead of returning empty lists. Before this fix the bare
+        # `except: pass` masked the failure, the orchestrator's plan-executor
+        # still saw an AttributeError on the swallowed-None path, and the
+        # user got HTTP 500 "Error executing plan: Internal error: Error
+        # finding id".
         prof_res = _safe_collection_get(
             collection,
             ids=[profile_chunk_id],
-            where={"session_id": session_id},
+            where={"name_slug": profile_name_slug},
             include=["documents", "metadatas"],
         )
         # prof_res is None on exception, a dict (possibly with empty lists) on success.
@@ -326,9 +332,9 @@ async def retrieve(
             metadatas = prof_res.get("metadatas") or []
             p_doc = documents[0] if documents else ""
             p_meta = metadatas[0] if metadatas else {}
-            # Triple-check: even if Chroma returns a row, refuse it unless
-            # metadata.session_id matches. Belt + suspenders + parachute.
-            if p_doc and p_meta.get("session_id") == session_id:
+            # Triple-check: refuse the row unless its metadata.name_slug
+            # matches the caller's slug. Belt + suspenders + parachute.
+            if p_doc and p_meta.get("name_slug") == profile_name_slug:
                 # Profile gets max score (1.0) so it always tops the context
                 profile_chunk = _build_chunk(profile_chunk_id, p_doc, p_meta, 1.0)
                 # Prepend; trim to top_k so we keep budget

tests/test_name_persistence.py

@@ -73,7 +73,7 @@ def _stub_brain_pick(intent: str, language: str):
     return BrainPick(_StubProvider(), f"stub::{intent}")
 
 
-async def _stub_retrieve(query, top_k=5, policy_ids=None, session_id=None):
+async def _stub_retrieve(query, top_k=5, policy_ids=None, profile_name_slug=None, session_id=None):
     return []  # empty context — perfectly valid for these tests
 
 

tests/test_profile_rag_isolation.py

@@ -1,31 +1,29 @@
-"""Regression tests for KI-102 — profile-RAG cross-session privacy leak.
-
-Pre-fix bug (caught by live 15-persona smoke test 2026-05-15):
-    1. User A saves profile → upsert_profile_chunk(session_id=A) writes a
-       chunk to the shared 'policies' Chroma collection with metadata
-       {policy_id: 'profile_A', doc_type: 'profile'} — NO session_id field.
-    2. User B sends a chat → retrieve(query, session_id=B) runs the main
-       cosine pass with no doc_type filter, so user A's profile chunk is
-       a candidate for top-k by raw cosine.
-    3. If user B's query is profile-shaped (age / dependents / health),
-       user A's profile chunk surfaces in B's context and the LLM cites
-       it as B's "User profile" — leaking A's facts into B's reply.
-
-Three fixes ship together:
-    KI-102.a — upsert_profile_chunk stamps session_id into Chroma metadata,
-               so the retrieve path can filter by it.
-    KI-102.b — retrieve()'s main cosine pass now passes
-               where={'doc_type': {'$ne': 'profile'}} so NO profile chunk
-               can ever surface via the cosine path. Profile chunks are
-               exclusively surfaced via the explicit per-session
-               collection.get(ids=[f'profile_{session_id}']) lookup.
-    KI-102.c — that per-session lookup gates on metadata.session_id ==
-               current session_id (triple-check) so even an ID collision
-               or legacy chunk without session_id metadata cannot leak.
-
-These tests run WITHOUT touching a real LLM / network. We stub the
-embedder and use chromadb's in-memory ephemeral client to verify the
-retrieve path's filter behaviour end-to-end.
+"""Regression tests for KI-102 / KI-112 / KI-117 / KI-118 — profile-RAG safety.
+
+KI-118 (2026-05-15) rewrites the threat model. Profile chunks are NO LONGER
+keyed by session_id; they are keyed by `name_slug` (canonical user name)
+and only NAMED users ever get embedded. Anonymous chats never write to
+Chroma — the corruption surface that the session_id keying introduced
+(the legacy `profile_anonymous` dangling row that poisoned every query
+with a `where` clause referencing session_id) is now structurally
+unreachable.
+
+The remaining safety guarantees these tests pin:
+
+  KI-102 / KI-118 — main retrieval cosine pass MUST NOT return profile
+                    chunks (`where={'doc_type': {'$ne': 'profile'}}`).
+                    Profile chunks are exclusively surfaced via the
+                    explicit per-name lookup
+                    `collection.get(ids=[f'profile_{name_slug}'])`.
+
+  KI-118.a — upsert_profile_chunk stamps `name_slug` into Chroma metadata.
+  KI-118.b — retrieve()'s per-name lookup gates on metadata.name_slug ==
+             caller's slug (triple-check), so cross-name leakage is blocked.
+
+  KI-112 — input guards: empty/None name_slug refused; mis-shaped embeddings
+           refused. These remain in place.
+
+  KI-107 — retrieve() with a missing/non-existent profile must NEVER raise.
 
 Run:
     cd /Users/rohitsar/Developer/Insurance\\ Sales\\ Bot
@@ -68,7 +66,7 @@ def _make_ephemeral_collection():
 
 class _StubEmbedder:
     """Deterministic 8-dim embedder so semantically-similar text gets
-    semantically-similar vectors. Two profile chunks (one for each session)
+    semantically-similar vectors. Two profile chunks (one per named user)
     will end up with near-identical embeddings, which is exactly what
     triggers the pre-fix leak in the wild."""
 
@@ -92,23 +90,24 @@ class _StubEmbedder:
 
 
 # ---------------------------------------------------------------------------
-# Test cases
+# Test cases — KI-118 threat model. Profile chunks keyed by name_slug.
 # ---------------------------------------------------------------------------
 
 
 class TestProfileIsolation(unittest.TestCase):
-    """KI-102 — session A's profile must NEVER surface in session B's retrieve."""
+    """KI-118 — anonymous sessions never write to Chroma; named user A's
+    profile must NEVER surface in named user B's retrieve."""
 
     def setUp(self):
         self.coll = _make_ephemeral_collection()
-        self.session_a = f"sessA_{uuid.uuid4().hex[:6]}"
-        self.session_b = f"sessB_{uuid.uuid4().hex[:6]}"
+        self.name_a = f"alice_{uuid.uuid4().hex[:4]}"
+        self.name_b = f"bob_{uuid.uuid4().hex[:4]}"
 
-    def _seed_profile(self, session_id: str, text: str) -> None:
-        """Write a profile chunk for `session_id` directly to the test
+    def _seed_profile(self, name_slug: str, text: str) -> None:
+        """Write a profile chunk for `name_slug` directly to the test
         collection, mirroring what upsert_profile_chunk does in prod."""
         vec = asyncio.run(_StubEmbedder().embed([text]))[0]
-        chunk_id = f"profile_{session_id}"
+        chunk_id = f"profile_{name_slug}"
         self.coll.add(
             ids=[chunk_id],
             documents=[text],
@@ -116,9 +115,9 @@ class TestProfileIsolation(unittest.TestCase):
             metadatas=[{
                 "policy_id": chunk_id,
                 "insurer_slug": "profile",
-                "policy_name": f"User profile (session {session_id[:8]})",
+                "policy_name": f"User profile ({name_slug[:16]})",
                 "doc_type": "profile",
-                "session_id": session_id,  # KI-102.a — stamped at write time
+                "name_slug": name_slug,  # KI-118.a — stamped at write time
                 "source_url": "",
                 "page_start": 0,
                 "page_end": 0,
@@ -147,7 +146,7 @@ class TestProfileIsolation(unittest.TestCase):
             }],
         )
 
-    def _run_retrieve(self, query: str, session_id: str, top_k: int = 5):
+    def _run_retrieve(self, query: str, profile_name_slug: str, top_k: int = 5):
         """Invoke rag.retrieve.retrieve() with the test collection +
         stub embedder patched in."""
         from rag import retrieve as retrieve_mod
@@ -158,46 +157,46 @@ class TestProfileIsolation(unittest.TestCase):
                 query=query,
                 top_k=top_k,
                 embedder=_StubEmbedder(),
-                session_id=session_id,
+                profile_name_slug=profile_name_slug,
             ))
 
     # -----------------------------------------------------------------
-    # CASE 1 — pre-fix leak repro: session A's profile must NOT show up
-    # in session B's retrieved context.
+    # CASE 1 — pre-fix leak repro: named user A's profile must NOT show
+    # up in named user B's retrieved context.
     # -----------------------------------------------------------------
-    def test_session_a_profile_never_leaks_into_session_b(self):
+    def test_name_a_profile_never_leaks_into_name_b(self):
         self._seed_profile(
-            self.session_a,
+            self.name_a,
             "USER CONTEXT — facts about the person asking this question:\n"
             "- Age: 45 years.\n- User's own pre-existing conditions: diabetes, hypertension.",
         )
         self._seed_profile(
-            self.session_b,
+            self.name_b,
             "USER CONTEXT — facts about the person asking this question:\n"
             "- Age: 28 years.\n- First-time buyer; no existing health insurance.",
         )
         # Add a generic policy chunk so there's something to retrieve.
         self._seed_policy("hdfc_ergo_optima_secure_v1", "Standard health policy text about waiting periods.")
 
-        # Session B asks a profile-flavoured query
+        # User B asks a profile-flavoured query
         chunks = self._run_retrieve(
             query="what plan suits my age and dependents",
-            session_id=self.session_b,
+            profile_name_slug=self.name_b,
         )
 
-        leaked = [c for c in chunks if c.policy_id == f"profile_{self.session_a}"]
+        leaked = [c for c in chunks if c.policy_id == f"profile_{self.name_a}"]
         self.assertEqual(
             leaked, [],
-            f"PRIVACY LEAK: session A's profile chunk surfaced in session B's "
+            f"PRIVACY LEAK: user A's profile chunk surfaced in user B's "
             f"retrieval. Found: {[c.policy_id for c in chunks]}",
         )
 
     # -----------------------------------------------------------------
-    # CASE 2 — session B's OWN profile must still surface (positive path).
+    # CASE 2 — named user B's OWN profile must still surface (positive path).
     # -----------------------------------------------------------------
-    def test_session_b_own_profile_is_surfaced(self):
+    def test_name_b_own_profile_is_surfaced(self):
         self._seed_profile(
-            self.session_b,
+            self.name_b,
             "USER CONTEXT — facts about the person asking this question:\n"
             "- Age: 28 years.",
         )
@@ -205,62 +204,62 @@ class TestProfileIsolation(unittest.TestCase):
 
         chunks = self._run_retrieve(
             query="recommend a plan for me",
-            session_id=self.session_b,
+            profile_name_slug=self.name_b,
         )
-        own = [c for c in chunks if c.policy_id == f"profile_{self.session_b}"]
+        own = [c for c in chunks if c.policy_id == f"profile_{self.name_b}"]
         self.assertEqual(
             len(own), 1,
-            f"Session B should see its OWN profile chunk. Got: {[c.policy_id for c in chunks]}",
+            f"User B should see its OWN profile chunk. Got: {[c.policy_id for c in chunks]}",
         )
 
     # -----------------------------------------------------------------
     # CASE 3 — multiple foreign profiles + one own profile. Only the
-    # current session's chunk may be present.
+    # current user's chunk may be present.
     # -----------------------------------------------------------------
     def test_three_foreign_profiles_none_leak(self):
-        for sid in ["smokeA_1", "ki100_ve", "smokeB_B2"]:
+        for name in ["alice_1", "carol_2", "dave_3"]:
             self._seed_profile(
-                sid,
+                name,
                 f"USER CONTEXT — facts about the person asking this question:\n"
-                f"- Age: {30 + len(sid)} years.\n- Health conditions: PII for {sid}.",
+                f"- Age: {30 + len(name)} years.\n- Health conditions: PII for {name}.",
             )
         self._seed_profile(
-            self.session_b,
+            self.name_b,
             "USER CONTEXT — facts about the person asking this question:\n- Age: 28 years.",
         )
         self._seed_policy("test_policy_2", "Generic policy text.")
 
         chunks = self._run_retrieve(
             query="my age health conditions dependents",
-            session_id=self.session_b,
+            profile_name_slug=self.name_b,
             top_k=10,
         )
         profile_pids = [c.policy_id for c in chunks if c.doc_type == "profile"]
-        # Only ONE profile chunk may appear, and it must be session_b's
+        # Only ONE profile chunk may appear, and it must be name_b's
         self.assertEqual(
-            profile_pids, [f"profile_{self.session_b}"],
+            profile_pids, [f"profile_{self.name_b}"],
             f"Foreign profile leaked. profile chunks in result: {profile_pids}",
         )
 
     # -----------------------------------------------------------------
-    # CASE 4 — legacy chunk without session_id metadata is refused even
-    # if its id happens to match (defence-in-depth from KI-102.c).
+    # CASE 4 — legacy chunk without name_slug metadata is refused even
+    # if its id happens to match (defence-in-depth from KI-118.b).
     # -----------------------------------------------------------------
-    def test_legacy_chunk_without_session_id_metadata_is_refused(self):
-        # Write a chunk under id 'profile_<session_b>' but with NO
-        # session_id field (simulating a pre-fix legacy row).
+    def test_legacy_chunk_without_name_slug_metadata_is_refused(self):
+        # Write a chunk under id 'profile_<name_b>' but with NO
+        # name_slug field (simulating a pre-KI-118 legacy row).
         vec = asyncio.run(_StubEmbedder().embed(["USER CONTEXT — legacy"]))[0]
-        chunk_id = f"profile_{self.session_b}"
+        chunk_id = f"profile_{self.name_b}"
         self.coll.add(
             ids=[chunk_id],
-            documents=["USER CONTEXT — legacy row from before KI-102 deploy"],
+            documents=["USER CONTEXT — legacy row from before KI-118 deploy"],
             embeddings=[vec],
             metadatas=[{
                 "policy_id": chunk_id,
                 "insurer_slug": "profile",
                 "policy_name": "legacy profile",
                 "doc_type": "profile",
-                # No 'session_id' — simulating pre-fix state
+                # No 'name_slug' — simulating pre-fix state
                 "source_url": "",
                 "page_start": 0,
                 "page_end": 0,
@@ -271,14 +270,42 @@ class TestProfileIsolation(unittest.TestCase):
 
         chunks = self._run_retrieve(
             query="anything",
-            session_id=self.session_b,
+            profile_name_slug=self.name_b,
         )
         # Legacy chunk must be refused — the triple-check at retrieve's
-        # per-session lookup gates on metadata.session_id match.
+        # per-name lookup gates on metadata.name_slug match.
         legacy_hits = [c for c in chunks if c.policy_id == chunk_id]
         self.assertEqual(
             legacy_hits, [],
-            "Legacy profile chunk without session_id metadata must be refused. "
+            "Legacy profile chunk without name_slug metadata must be refused. "
+            f"Got: {[c.policy_id for c in chunks]}",
+        )
+
+    # -----------------------------------------------------------------
+    # CASE 5 — KI-118 core invariant: anonymous calls (no profile_name_slug)
+    # produce NO profile chunks at all, even if the collection contains
+    # foreign profile rows that match the query.
+    # -----------------------------------------------------------------
+    def test_anonymous_retrieve_never_surfaces_any_profile_chunk(self):
+        # Seed two named-user profiles
+        self._seed_profile(self.name_a, "USER CONTEXT — Age: 45 years.")
+        self._seed_profile(self.name_b, "USER CONTEXT — Age: 28 years.")
+        self._seed_policy("test_policy_anon", "Generic policy text.")
+
+        # Anonymous call — no profile_name_slug
+        from rag import retrieve as retrieve_mod
+        retrieve_mod._RETRIEVAL_CACHE.clear()
+        with mock.patch.object(retrieve_mod, "get_collection", return_value=self.coll):
+            chunks = asyncio.run(retrieve_mod.retrieve(
+                query="my age health conditions",
+                top_k=5,
+                embedder=_StubEmbedder(),
+                profile_name_slug=None,
+            ))
+        profile_chunks = [c for c in chunks if c.doc_type == "profile"]
+        self.assertEqual(
+            profile_chunks, [],
+            "PRIVACY LEAK: anonymous retrieve surfaced a profile chunk. "
             f"Got: {[c.policy_id for c in chunks]}",
         )
 
@@ -287,19 +314,18 @@ class TestProfileIsolation(unittest.TestCase):
 # KI-107 (2026-05-15) — graceful handling of Chroma get(ids=[missing]).
 # C5 port-in persona saw 3× HTTP 500 "Error executing plan: Internal error:
 # Error finding id". After KI-102 added the per-session profile-chunk
-# lookup, retrieve() runs collection.get(ids=[f"profile_{sid}"]) on EVERY
-# query — and for new sessions (no profile saved yet) and certain Chroma
-# sqlite states, that call can raise. The bare `except: pass` in the
-# pre-KI-107 code masked the failure into the orchestrator's plan executor.
-# These tests pin the contract: retrieve() with a never-existed session_id
-# must NEVER raise and must NEVER return a profile chunk.
+# lookup, retrieve() runs collection.get(ids=[f"profile_{slug}"]) on EVERY
+# named query — and for new users (no profile saved yet) and certain
+# Chroma sqlite states, that call can raise. These tests pin the contract:
+# retrieve() with a never-existed name_slug must NEVER raise and must
+# NEVER return a profile chunk.
 # ---------------------------------------------------------------------------
 
 
 class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
-    """KI-107 — retrieve(session_id=...) must be exception-safe across:
-    (1) brand-new sessions with no profile chunk yet,
-    (2) Chroma.get raising on the per-session lookup,
+    """KI-107 — retrieve(profile_name_slug=...) must be exception-safe across:
+    (1) first-time named users with no profile chunk yet,
+    (2) Chroma.get raising on the per-name lookup,
     (3) Chroma.get returning empty lists for missing ids."""
 
     def setUp(self):
@@ -324,7 +350,7 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
             }],
         )
 
-    def _run_retrieve(self, query: str, session_id: str, top_k: int = 5):
+    def _run_retrieve(self, query: str, profile_name_slug: str, top_k: int = 5):
         from rag import retrieve as retrieve_mod
         retrieve_mod._RETRIEVAL_CACHE.clear()
         with mock.patch.object(retrieve_mod, "get_collection", return_value=self.coll):
@@ -332,12 +358,12 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
                 query=query,
                 top_k=top_k,
                 embedder=_StubEmbedder(),
-                session_id=session_id,
+                profile_name_slug=profile_name_slug,
             ))
 
-    def test_retrieve_with_never_existed_session_does_not_raise(self):
-        """New session, no profile saved yet — must return policy chunks
-        without raising. Pre-KI-107 this surfaced as HTTP 500 "Error
+    def test_retrieve_with_never_existed_name_does_not_raise(self):
+        """First-time named user, no profile saved yet — must return policy
+        chunks without raising. Pre-KI-107 this surfaced as HTTP 500 "Error
         finding id" because get(ids=[missing]) raised + bare except: pass
         let downstream code index into a None result."""
         self._seed_one_policy()
@@ -345,14 +371,14 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
         # Should not raise
         chunks = self._run_retrieve(
             query="what is the waiting period for cataract",
-            session_id="never_existed_session_xyz",
+            profile_name_slug="never_existed_name_xyz",
         )
 
         # No profile chunk should appear (none was ever written)
         profile_chunks = [c for c in chunks if c.doc_type == "profile"]
         self.assertEqual(
             profile_chunks, [],
-            f"never-existed session should not produce profile chunks. "
+            f"never-existed name should not produce profile chunks. "
             f"Got: {[(c.chunk_id, c.doc_type) for c in chunks]}",
         )
         # But main cosine retrieval must still work
@@ -361,10 +387,10 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
             "main cosine pass should still return the seeded policy chunk.",
         )
 
-    def test_retrieve_handles_chroma_get_raising_on_per_session_lookup(self):
-        """Simulate the worst case: Chroma raises on the per-session
-        profile lookup (e.g. transient sqlite lock during compaction).
-        retrieve() must still return main cosine results, not 500."""
+    def test_retrieve_handles_chroma_get_raising_on_per_name_lookup(self):
+        """Simulate the worst case: Chroma raises on the per-name profile
+        lookup (e.g. transient sqlite lock during compaction). retrieve()
+        must still return main cosine results, not 500."""
         self._seed_one_policy()
 
         # Wrap the real collection so .get() raises but .query() works
@@ -390,14 +416,14 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
                 query="what is the waiting period",
                 top_k=5,
                 embedder=_StubEmbedder(),
-                session_id="some_session_id",
+                profile_name_slug="some_name_slug",
             ))
 
         # Main cosine still works → at least the seeded policy chunk returns
         self.assertGreater(
             len(chunks), 0,
             "retrieve() should fall back to main cosine results when "
-            "the per-session profile lookup raises.",
+            "the per-name profile lookup raises.",
         )
         # No profile chunk surfaced
         self.assertEqual(
@@ -437,16 +463,16 @@ class TestRetrieveSurvivesMissingProfileId(unittest.TestCase):
 
 # ---------------------------------------------------------------------------
 # Standalone upsert metadata test — no Chroma client; just verify the
-# upsert builds metadata containing session_id.
+# upsert builds metadata containing name_slug.
 # ---------------------------------------------------------------------------
 
 
-class TestUpsertStampsSessionId(unittest.TestCase):
-    """KI-102.a — upsert_profile_chunk MUST write session_id into the
+class TestUpsertStampsNameSlug(unittest.TestCase):
+    """KI-118.a — upsert_profile_chunk MUST write name_slug into the
     chunk's Chroma metadata. Without it, the retrieve filter can't
-    distinguish session A's profile from session B's."""
+    distinguish user A's profile from user B's."""
 
-    def test_upsert_writes_session_id_to_metadata(self):
+    def test_upsert_writes_name_slug_to_metadata(self):
         from backend import profile_rag
 
         captured: dict = {}
@@ -471,7 +497,7 @@ class TestUpsertStampsSessionId(unittest.TestCase):
                 return [[0.1] * 384 for _ in texts]
 
         fake_coll = _FakeColl()
-        sid = f"test_{uuid.uuid4().hex[:6]}"
+        slug = f"alice_{uuid.uuid4().hex[:6]}"
         profile = {
             "age": 32,
             "dependents": "self_spouse",
@@ -481,45 +507,37 @@ class TestUpsertStampsSessionId(unittest.TestCase):
 
         with mock.patch.object(profile_rag, "_get_collection", return_value=fake_coll), \
              mock.patch("backend.providers.local_embeddings.LocalEmbeddings", _FakeEmbedder):
-            asyncio.run(profile_rag.upsert_profile_chunk(sid, profile))
+            asyncio.run(profile_rag.upsert_profile_chunk(slug, profile))
 
         self.assertIn("metadatas", captured, "upsert never called coll.add")
         meta = captured["metadatas"][0]
         self.assertEqual(
-            meta.get("session_id"), sid,
-            f"profile chunk metadata missing session_id. Got: {meta}",
+            meta.get("name_slug"), slug,
+            f"profile chunk metadata missing name_slug. Got: {meta}",
         )
         self.assertEqual(meta.get("doc_type"), "profile")
-        self.assertEqual(captured["ids"], [f"profile_{sid}"])
+        self.assertEqual(captured["ids"], [f"profile_{slug}"])
 
 
 # ---------------------------------------------------------------------------
-# KI-112 (2026-05-15) — input validation hardening.
-#
-# Root cause of the HNSW corruption: KI-102's initial deploy wrote a profile
-# chunk under id "profile_anonymous" with NO session_id metadata. That legacy
-# chunk poisoned every subsequent collection.query() that referenced
-# session_id or doc_type$ne in the where clause — Chroma's plan executor
-# raised "Error finding id" against the dangling row's HNSW pointer, and
-# corruption propagated across HNSW segments (1580 / 7356 chunks turned
-# unfetchable on full collection rebuild).
+# KI-112 (2026-05-15) — input validation hardening (still applies post-KI-118).
 #
-# Two guards added in upsert_profile_chunk:
-#   1) session_id must be a non-empty str (rejects "", None, falsy values
-#      that would write under "profile_" / "profile_anonymous" etc).
-#   2) embedding shape must match the embedder's declared dimension and
-#      contain no None values (rejects mis-shaped vectors that would
-#      corrupt HNSW segment files).
+# Root cause of the historical HNSW corruption: KI-102's initial deploy wrote
+# a profile chunk under id "profile_anonymous" with NO session_id metadata.
+# That legacy chunk poisoned every subsequent collection.query() that
+# referenced session_id or doc_type$ne in the where clause — Chroma's plan
+# executor raised "Error finding id" against the dangling row's HNSW pointer.
 #
+# KI-118 moved the key from session_id to name_slug; the guards are the same.
 # These tests pin the contract: bad inputs MUST be rejected at write time,
-# not silently corrupt the index for future sessions.
+# not silently corrupt the index for future users.
 # ---------------------------------------------------------------------------
 
 
 class TestUpsertRejectsBadInputs(unittest.TestCase):
-    """KI-112 — bad session_id or embedding shape must NOT reach Chroma."""
+    """KI-112 / KI-118 — bad name_slug or embedding shape must NOT reach Chroma."""
 
-    def test_upsert_rejects_empty_session_id(self):
+    def test_upsert_rejects_empty_name_slug(self):
         from backend import profile_rag
 
         captured: dict = {"add_called": False}
@@ -537,14 +555,14 @@ class TestUpsertRejectsBadInputs(unittest.TestCase):
             async def embed(self, texts, input_type="document"):
                 return [[0.1] * 384 for _ in texts]
 
-        for bad_sid in ["", "   ", None]:
+        for bad_slug in ["", "   ", None]:
             with mock.patch.object(profile_rag, "_get_collection", return_value=_FakeColl()), \
                  mock.patch("backend.providers.local_embeddings.LocalEmbeddings", _FakeEmbedder):
                 captured["add_called"] = False
-                asyncio.run(profile_rag.upsert_profile_chunk(bad_sid, {"age": 30}))
+                asyncio.run(profile_rag.upsert_profile_chunk(bad_slug, {"age": 30}))
             self.assertFalse(
                 captured["add_called"],
-                f"upsert MUST refuse session_id={bad_sid!r} — bad write would "
+                f"upsert MUST refuse name_slug={bad_slug!r} — bad write would "
                 "corrupt the policies collection.",
             )
 
@@ -573,7 +591,7 @@ class TestUpsertRejectsBadInputs(unittest.TestCase):
         with mock.patch.object(profile_rag, "_get_collection", return_value=_FakeColl()), \
              mock.patch("backend.providers.local_embeddings.LocalEmbeddings", _BadDimEmbedder):
             asyncio.run(profile_rag.upsert_profile_chunk(
-                "valid_session_xyz", {"age": 30, "dependents": "self"},
+                "valid_slug_xyz", {"age": 30, "dependents": "self"},
             ))
 
         self.assertFalse(
@@ -605,7 +623,7 @@ class TestUpsertRejectsBadInputs(unittest.TestCase):
         with mock.patch.object(profile_rag, "_get_collection", return_value=_FakeColl()), \
              mock.patch("backend.providers.local_embeddings.LocalEmbeddings", _NoneVecEmbedder):
             asyncio.run(profile_rag.upsert_profile_chunk(
-                "valid_session_xyz", {"age": 30, "dependents": "self"},
+                "valid_slug_xyz", {"age": 30, "dependents": "self"},
             ))
 
         self.assertFalse(
@@ -636,10 +654,10 @@ class TestUpsertRejectsBadInputs(unittest.TestCase):
         with mock.patch.object(profile_rag, "_get_collection", return_value=_FakeColl()), \
              mock.patch("backend.providers.local_embeddings.LocalEmbeddings", _GoodEmbedder):
             asyncio.run(profile_rag.upsert_profile_chunk(
-                "good_session", {"age": 30, "dependents": "self"},
+                "good_slug", {"age": 30, "dependents": "self"},
             ))
 
-        self.assertEqual(captured.get("ids"), ["profile_good_session"])
+        self.assertEqual(captured.get("ids"), ["profile_good_slug"])
         self.assertEqual(len(captured["embeddings"][0]), 384)
 
 

tests/test_session_no_disk_persistence.py

@@ -0,0 +1,128 @@
+"""Regression test for KI-118 (2026-05-15) — session_state has NO disk persistence.
+
+The pre-KI-118 architecture wrote `40-data/sessions/<session_id>.json` on
+every state mutation so a Space restart wouldn't lose state. That disk-write
+side was:
+
+  1. The root of the Chroma corruption incident (the legacy
+     `profile_anonymous` chunk that poisoned every subsequent query).
+  2. A privacy and operational liability (orphan files accumulate, schema
+     drift hits old files, cross-session memory leak surface).
+
+KI-118 rip-out: session_state is in-memory only. Cross-session memory is
+strictly name-based (returning user provides their name → fact_find brain
+captures it → `rehydrate_by_name` pulls the named profile from
+`40-data/profiles/`).
+
+This test pins the contract: a session lifecycle (get, mutate, set_awaiting,
+record_answer, update_profile_field, reset) MUST NOT create ANY file under
+`40-data/sessions/`. If a future change reintroduces disk persistence, this
+test fires.
+
+Run:
+    cd /Users/rohitsar/Developer/Insurance\\ Sales\\ Bot
+    PYTHONPATH=$PWD .venv/bin/python -m pytest tests/test_session_no_disk_persistence.py -v
+"""
+
+from __future__ import annotations
+
+import sys
+import unittest
+import uuid
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+
+
+_SESSIONS_DIR = _REPO_ROOT / "40-data" / "sessions"
+
+
+def _file_count() -> int:
+    if not _SESSIONS_DIR.exists():
+        return 0
+    return sum(1 for _ in _SESSIONS_DIR.glob("*.json"))
+
+
+class TestSessionNoDiskPersistence(unittest.TestCase):
+    """KI-118 — session lifecycle must not touch 40-data/sessions/."""
+
+    def setUp(self):
+        # Capture the pre-test file count so we only assert on the delta.
+        # Other tests / contributors may legitimately have files in this
+        # dir; we only care that this test's session id doesn't write one.
+        self.before_count = _file_count()
+        self.session_id = f"ki118_no_disk_{uuid.uuid4().hex[:10]}"
+
+    def tearDown(self):
+        # Defensive cleanup: if some other code path DID create our file,
+        # remove it so we don't pollute the working dir.
+        target = _SESSIONS_DIR / f"{self.session_id}.json"
+        if target.exists():
+            try:
+                target.unlink()
+            except OSError:
+                pass
+
+    def test_session_lifecycle_creates_no_disk_file(self):
+        from backend.session_state import (
+            get_session,
+            reset_session,
+        )
+
+        # 1. get_session → in-memory
+        sess = get_session(self.session_id)
+        self.assertEqual(sess.session_id, self.session_id)
+
+        # 2. mutate every public path that USED to flush to disk pre-KI-118
+        sess.profile.age = 35
+        sess.profile.name = "Rohit"
+        sess.set_awaiting("dependents")
+        sess.update_profile_field("income_band", "10-25L")
+        sess.free_form_session = True
+        sess._flush()  # legacy callers still invoke this — must be a no-op
+
+        # 3. Reset
+        reset_session(self.session_id)
+
+        # 4. Assert: no new file in 40-data/sessions/ AND specifically no
+        #    file under our session id.
+        target = _SESSIONS_DIR / f"{self.session_id}.json"
+        self.assertFalse(
+            target.exists(),
+            f"REGRESSION (KI-118): session lifecycle created {target}. "
+            "session_state.py was refactored to be in-memory only; a disk "
+            "write was reintroduced.",
+        )
+
+        # The directory may still exist if it was already there pre-test,
+        # but our file count delta must be 0.
+        after = _file_count()
+        self.assertEqual(
+            after, self.before_count,
+            f"REGRESSION (KI-118): session lifecycle created {after - self.before_count} "
+            f"new file(s) under {_SESSIONS_DIR}. Expected zero.",
+        )
+
+    def test_flush_is_a_noop(self):
+        """SessionState._flush() is kept for backwards-compat with existing
+        callers (orchestrator + fact_find_brain). It must do nothing — no
+        I/O, no exceptions on missing dir, no return value."""
+        from backend.session_state import get_session
+        sess = get_session(self.session_id)
+        sess.profile.age = 35
+        # Must not raise even if the parent dir doesn't exist.
+        result = sess._flush()
+        self.assertIsNone(result, "_flush() must be a no-op returning None")
+
+    def test_purge_old_files_is_noop(self):
+        """KI-118 — purge_old_files is a stub now; must return 0 and never
+        crash, even if 40-data/sessions/ doesn't exist."""
+        from backend.session_state import purge_old_files
+        result = purge_old_files()
+        self.assertEqual(result, 0)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)