fix(brain): wire returning-user-by-name recall into single_brain (ADR-041/KI-196)

Root cause: the confirmation-gated profile recall was ORPHANED by the
orchestrator->single-LLM rewrite. extract_potential_name /
try_recall_by_name / apply_pending_recall existed and were unit-tested,
but NOTHING on the live path (single_brain.handle_turn) called them, and
TurnResult.returning_user_recalled was hard-coded False. A same-name
revisit ('Hi, I'm Rohit') therefore never triggered 'are you the same
Rohit?' and never reused the stored profile.

Fix (single_brain.py, fits current architecture; no orchestrator revival):
- turn-1: extract_potential_name -> try_recall_by_name STAGES the match
on session.pending_profile_recall (privacy-safe; never auto-merged).
- _system_instruction injects a RETURNING-USER CHECK block so the LLM's
entire turn is just the 'Welcome back - are you the same <name>?'
confirm (no tools / no fact-find that turn).
- next turn: _affirm_or_deny (new, conservative WORD-BOUNDARY yes/no;
deny wins ties = privacy fail-closed) -> explicit yes calls
apply_pending_recall(confirmed=True) -> merges stored slots into empty
fields + flips returning_user_recalled (frontend Welcome-back banner);
explicit no discards; ambiguous re-asks once.

Verified: 7/7 new tests in tests/test_returning_user_recall_singlebrain.py
(incl. the handle_turn integration that was the untested gap); real
backend probe — turn1 exact confirm prompt, turn2 'yes' recalled the
full profile and did not re-ask; full pytest suite exit 0 (no regression).

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

backend/single_brain.py

@@ -656,8 +656,63 @@ def _build_contents(
     return out
 
 
+# WHOLE-WORD tokens (matched against the tokenised message, NOT as
+# substrings — "no" must not match "k(no)ws", "ya" must not match "Ma(ya)").
+_RECALL_DENY_TOKENS = {
+    "no", "nope", "nah", "naah", "nahi", "wrong", "never", "neither",
+    "nopes", "nahin",
+}
+_RECALL_AFFIRM_TOKENS = {
+    "yes", "yeah", "yep", "yup", "ya", "yaa", "yaah", "yess", "haan",
+    "han", "haa", "correct", "right", "sahi", "bilkul", "sure", "indeed",
+    "absolutely", "exactly", "true", "yup",
+}
+# Multi-word phrases — safe to match as substrings.
+_RECALL_DENY_PHRASES = (
+    "not me", "isn't me", "isnt me", "not the same", "start fresh",
+    "start over", "different person", "new user", "someone else",
+    "not rohit", "first time", "never been", "fresh start", "not him",
+    "not her", "i'm new", "im new", "not that person", "don't know",
+    "dont know", "different one",
+)
+_RECALL_AFFIRM_PHRASES = (
+    "that's me", "thats me", "that is me", "it's me", "its me", "i am",
+    "pick up", "go ahead", "that's right", "thats right", "that's correct",
+    "thats correct", "yes please", "continue where", "same person",
+    "carry on", "of course",
+)
+_RECALL_TOKEN_RE = __import__("re").compile(r"[a-z']+")
+
+
+def _affirm_or_deny(text: str):
+    """Conservative yes/no for the returning-user confirm gate.
+
+    Returns True (affirm), False (deny), or None (ambiguous → re-ask).
+    Deny wins ties: privacy is fail-closed — an ambiguous "no, but…" must
+    NEVER merge a stranger's stored profile (ADR-041 / KI-196). Short
+    tokens are matched whole-word (tokenised), not as substrings, so
+    "who knows" / "i don't know" / "now" are NOT read as "no".
+    """
+    t = (text or "").strip().lower()
+    if not t:
+        return None
+    toks = set(_RECALL_TOKEN_RE.findall(t))
+    deny = bool(toks & _RECALL_DENY_TOKENS) or any(
+        p in t for p in _RECALL_DENY_PHRASES
+    )
+    if deny:
+        return False
+    affirm = bool(toks & _RECALL_AFFIRM_TOKENS) or any(
+        p in t for p in _RECALL_AFFIRM_PHRASES
+    )
+    if affirm:
+        return True
+    return None
+
+
 def _system_instruction(
-    profile, is_returning_user: bool = False, shortlist_block: str = ""
+    profile, is_returning_user: bool = False, shortlist_block: str = "",
+    pending_recall: "Optional[dict]" = None,
 ) -> dict:
     """Bake the profile snapshot into the system prompt so each turn the
     LLM knows what's already captured. Returned in Gemini's expected
@@ -694,7 +749,35 @@ def _system_instruction(
                 "say 'Welcome back'):\n"
                 + json.dumps(snapshot, ensure_ascii=False, sort_keys=True)
             )
-    text = SYSTEM_PROMPT + extra + (shortlist_block or "")
+    recall_block = ""
+    if pending_recall:
+        _nm = (pending_recall.get("name") or "there").strip()
+        _sm = pending_recall.get("summary") or {}
+        _bits = []
+        for _k in ("age", "location_tier", "dependents", "primary_goal",
+                   "health_conditions"):
+            _v = _sm.get(_k)
+            if _v not in (None, "", []):
+                _bits.append(f"{_k.replace('_', ' ')}: {_v}")
+        _summ = "; ".join(_bits) if _bits else "a saved profile"
+        recall_block = (
+            "\n\n═══════════════════════════════════\n"
+            "RETURNING-USER CHECK — HIGHEST PRIORITY THIS TURN "
+            "(overrides RULE 1 / fact-find for this one turn)\n"
+            "═══════════════════════════════════\n"
+            f"A stored profile already exists under the name the user just "
+            f"gave (\"{_nm}\"). Known hints — {_summ}.\n"
+            "Your ENTIRE reply this turn MUST be ONLY the confirmation "
+            "question below. Do NOT call any tool, do NOT save_profile_field, "
+            "do NOT run the 7-question fact-find, do NOT recommend:\n"
+            f"  \"Welcome back — are you the same {_nm} who spoke with us "
+            f"before ({_summ})? If yes, I'll pick up right where we left "
+            f"off. If not, no problem — just say so and we'll start fresh.\"\n"
+            "Then wait for their yes/no on the NEXT turn. The system "
+            "applies or discards the saved profile from their answer — you "
+            "never merge anything yourself."
+        )
+    text = SYSTEM_PROMPT + extra + recall_block + (shortlist_block or "")
     return {"parts": [{"text": text}]}
 
 
@@ -1656,11 +1739,53 @@ async def handle_turn(
     # conversation — not a returning user.
     _current_turn = int(getattr(session, "turn_idx", 1) or 1)
 
-    # The single LLM (RULE 1 → save_profile_field) is the sole fact-find
-    # driver. Explicit returning-user recall is the separate
-    # POST /api/profile/recall-by-name endpoint, so handle_turn never sets
-    # returning_user_recalled True.
+    # ── Returning-user recall (ADR-041 / KI-196), wired into single_brain
+    # 2026-05-19. Previously ORPHANED by the orchestrator→single-LLM
+    # rewrite: extract_potential_name / try_recall_by_name /
+    # apply_pending_recall existed and were unit-tested, but NOTHING on the
+    # live path called them — so a same-name revisit ("Hi, I'm Rohit") was
+    # never recognised and the "are you the same Rohit?" prompt never fired.
+    # Privacy-safe by construction: a name match is only STAGED on
+    # session.pending_profile_recall (never auto-merged); only an explicit
+    # "yes" merges the stored slots, an explicit "no" discards, anything
+    # ambiguous leaves it staged so the LLM re-asks the confirm once.
     _did_recall_this_turn = False
+    try:
+        from backend.profile_persistence import (
+            extract_potential_name,
+            try_recall_by_name,
+        )
+        from backend.session_state import apply_pending_recall
+
+        _pending_recall = getattr(session, "pending_profile_recall", None)
+        if _pending_recall:
+            _ans = _affirm_or_deny(user_text)
+            if _ans is True:
+                _did_recall_this_turn = bool(
+                    apply_pending_recall(session, confirmed=True)
+                )
+                _pending_recall = None
+            elif _ans is False:
+                apply_pending_recall(session, confirmed=False)
+                _pending_recall = None
+            # ambiguous → leave staged; the confirm block is re-injected
+            # below and the LLM re-asks the "are you <name>?" question.
+        elif _current_turn == 1:
+            _nm = extract_potential_name(user_text or "")
+            if _nm:
+                # Stages session.pending_profile_recall iff a stored
+                # profile for this name exists (no match ⇒ no-op, normal
+                # fresh-user flow continues — no false confirm prompt).
+                try_recall_by_name(session, _nm)
+                _pending_recall = getattr(
+                    session, "pending_profile_recall", None
+                )
+    except Exception as _re:  # noqa: BLE001 — recall must never break a turn
+        _log.warning(
+            "returning-user recall wiring failed: %s: %s",
+            type(_re).__name__, str(_re)[:200],
+        )
+        _pending_recall = getattr(session, "pending_profile_recall", None)
 
     _has_prior_profile = any(
         getattr(session.profile, fld, None) not in (None, "", [])
@@ -1714,6 +1839,7 @@ async def handle_turn(
         session.profile,
         is_returning_user=is_returning_user,
         shortlist_block=_shortlist_block,
+        pending_recall=_pending_recall,
     )
 
     # Bug #108 + #110 — if the user explicitly declines the pricing /

tests/test_returning_user_recall_singlebrain.py

@@ -0,0 +1,195 @@
+"""Regression test for the returning-user-by-name recall fix (2026-05-19).
+
+Bug: every visit, the user gave the same name ("Rohit") and the bot NEVER
+asked "are you the same Rohit?" / never recalled the stored profile. The
+confirmation-gated recall (ADR-041/KI-196) helpers existed and were
+unit-tested, but the orchestrator→single-LLM rewrite left them ORPHANED —
+nothing on the live path (single_brain.handle_turn) called them, and
+`returning_user_recalled` was a hard-coded False. The integration boundary
+was exactly what had no test (that's how it shipped).
+
+These tests pin the now-wired chain end-to-end in single_brain:
+  turn-1 name sniff → stage (privacy-safe, never auto-merge) → confirm
+  prompt injected → explicit "yes" merges + flips returning_user_recalled
+  → explicit "no" discards → unknown name is a no-op (no false prompt).
+"""
+import asyncio
+import os
+import random
+import string
+import unittest
+import uuid
+from unittest import mock
+
+from backend import single_brain
+from backend import brain_tools
+from backend.session_state import SessionState, apply_pending_recall
+from backend.profile_persistence import extract_potential_name, try_recall_by_name
+from backend.profile_store import save_profile, _normalise_name, _PROFILES_DIR
+from backend.needs_finder import Profile
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _text_payload(text):
+    return {"candidates": [{"content": {"parts": [{"text": text}]}}]}
+
+
+class TestAffirmOrDeny(unittest.TestCase):
+    """Word-boundary yes/no — must NOT read 'no' inside 'knows'/'now', and
+    deny must win ties (privacy fail-closed)."""
+
+    def test_table(self):
+        cases = [
+            ("yes that is me", True),
+            ("yeah, same Rohit", True),
+            ("haan bilkul", True),
+            ("that is me, carry on", True),
+            ("no, I'm a new user", False),
+            ("not me, someone else", False),
+            ("no, but yes", False),          # deny wins
+            ("i don't know", False),         # fail-closed
+            ("maybe, who knows", None),      # 'knows' must NOT be 'no'
+            ("now what", None),              # 'now' must NOT be 'no'
+            ("hmm", None),
+            ("", None),
+        ]
+        for txt, exp in cases:
+            self.assertEqual(single_brain._affirm_or_deny(txt), exp, txt)
+
+
+class _StoredProfileFixture(unittest.TestCase):
+    """Creates a deterministic, uniquely-named stored profile so the test
+    never couples to the mutable 40-data/profiles/rohit.json."""
+
+    def setUp(self):
+        # Alphabetic only — extract_potential_name correctly rejects
+        # digit-bearing tokens, so a hex suffix would never be sniffed.
+        self.name = "Recalltester" + "".join(
+            random.choices(string.ascii_lowercase, k=7))
+        self.slug = _normalise_name(self.name)
+        p = Profile()
+        p.name = self.name
+        p.age = 41
+        p.dependents = "self+spouse+1 kid"
+        p.location_tier = "metro"
+        p.income_band = "10L-25L"
+        p.primary_goal = "first_buy"
+        p.health_conditions = ["none"]
+        self.assertTrue(save_profile(self.name, p),
+                        "fixture save_profile failed")
+
+    def tearDown(self):
+        try:
+            for fp in _PROFILES_DIR.glob("*.json"):
+                try:
+                    import json
+                    d = json.loads(fp.read_text())
+                except Exception:
+                    continue
+                if d.get("name_slug") == self.slug or \
+                   (d.get("profile") or {}).get("name") == self.name:
+                    fp.unlink(missing_ok=True)
+        except Exception:
+            pass
+
+
+class TestRecallChain(_StoredProfileFixture):
+    def test_stage_inject_merge(self):
+        nm = extract_potential_name(f"Hi, I'm {self.name}")
+        self.assertTrue(nm and nm.lower().startswith("recalltester"))
+        s = SessionState(session_id="rc1")
+        try_recall_by_name(s, nm)
+        pr = getattr(s, "pending_profile_recall", None)
+        self.assertTrue(pr, "name match was not STAGED")
+        self.assertEqual(pr["name"], self.name)
+        si = single_brain._system_instruction(
+            s.profile, pending_recall=pr)["parts"][0]["text"]
+        self.assertIn("RETURNING-USER CHECK", si)
+        self.assertIn("Welcome back", si)
+        self.assertIn(self.name, si)
+        # confirm=True merges into empty slots
+        self.assertTrue(apply_pending_recall(s, confirmed=True))
+        self.assertEqual(s.profile.age, 41)
+        self.assertIsNone(getattr(s, "pending_profile_recall", None))
+
+    def test_deny_discards(self):
+        s = SessionState(session_id="rc2")
+        try_recall_by_name(s, self.name)
+        self.assertTrue(s.pending_profile_recall)
+        self.assertFalse(apply_pending_recall(s, confirmed=False))
+        self.assertIn(getattr(s.profile, "age", None), (None, "", 0))
+        self.assertIsNone(s.pending_profile_recall)
+
+    def test_unknown_name_no_stage(self):
+        s = SessionState(session_id="rc3")
+        try_recall_by_name(s, "Zzqxnobodyhasthisname")
+        self.assertIsNone(getattr(s, "pending_profile_recall", None))
+
+
+class TestHandleTurnIntegration(_StoredProfileFixture):
+    """The exact gap that let the bug ship: single_brain.handle_turn
+    integration with the recall chain."""
+
+    def setUp(self):
+        super().setUp()
+        self._env = mock.patch.dict(os.environ,
+                                    {"GOOGLE_API_KEY": "test-key"})
+        self._env.start()
+        self.sys_prompts = []
+
+        async def _fake_gemini(*_a, **_k):
+            self.sys_prompts.append(
+                (_k.get("system_instruction") or {})
+                .get("parts", [{}])[0].get("text", ""))
+            return _text_payload("Welcome back — are you the same person?")
+
+        self._gp = mock.patch.object(single_brain, "_gemini_call",
+                                     _fake_gemini)
+        self._gp.start()
+
+    def tearDown(self):
+        self._gp.stop()
+        self._env.stop()
+        super().tearDown()
+
+    def test_turn1_stages_and_prompts_then_yes_recalls(self):
+        sess = SessionState(session_id=f"hti_{uuid.uuid4().hex[:8]}")
+        # Turn 1: user states their (returning) name.
+        r1 = _run(single_brain.handle_turn(sess, f"Hi, I'm {self.name}"))
+        self.assertTrue(getattr(sess, "pending_profile_recall", None),
+                        "turn-1 name sniff did not STAGE the recall")
+        self.assertFalse(r1.returning_user_recalled,
+                         "must NOT flag recall before the user confirms")
+        self.assertIn("RETURNING-USER CHECK", self.sys_prompts[-1],
+                      "confirm block not injected into the system prompt")
+        # Turn 2: user confirms.
+        r2 = _run(single_brain.handle_turn(sess, "yes, that's me"))
+        self.assertTrue(r2.returning_user_recalled,
+                        "explicit 'yes' must flip returning_user_recalled")
+        self.assertEqual(sess.profile.age, 41,
+                         "stored profile was not merged on confirm")
+        self.assertIsNone(getattr(sess, "pending_profile_recall", None))
+
+    def test_no_keeps_session_blank(self):
+        sess = SessionState(session_id=f"hti_{uuid.uuid4().hex[:8]}")
+        _run(single_brain.handle_turn(sess, f"Hello, I am {self.name}"))
+        self.assertTrue(sess.pending_profile_recall)
+        r2 = _run(single_brain.handle_turn(sess, "no, I'm a new user"))
+        self.assertFalse(r2.returning_user_recalled)
+        self.assertIn(getattr(sess.profile, "age", None), (None, "", 0))
+        self.assertIsNone(sess.pending_profile_recall)
+
+    def test_unknown_name_normal_flow(self):
+        sess = SessionState(session_id=f"hti_{uuid.uuid4().hex[:8]}")
+        r1 = _run(single_brain.handle_turn(
+            sess, "Hi, I'm Zzqxnobodyhasthisname"))
+        self.assertIsNone(getattr(sess, "pending_profile_recall", None))
+        self.assertFalse(r1.returning_user_recalled)
+        self.assertNotIn("RETURNING-USER CHECK", self.sys_prompts[-1])
+
+
+if __name__ == "__main__":
+    unittest.main()