diff --git a/agent/title_generator.py b/agent/title_generator.py
index 5323f8bd047a8..89ddb077f2e84 100644
--- a/agent/title_generator.py
+++ b/agent/title_generator.py
@@ -152,6 +152,45 @@
     "[System: The active model for this chat has changed to ",
 )
 
+# Re-title prompt. Unlike the first-exchange prompts above, this one is shown
+# the WHOLE conversation plus the current title, and is biased to KEEP the
+# existing title unless the durable topic has clearly drifted. This is what
+# stops a localized detour (e.g. "write the PDF" inside a long USCIS-RFE
+# conversation) from clobbering a title that describes the real subject.
+_RETITLE_PROMPT = (
+    "You are maintaining the title of an ongoing conversation. Below is the conversation "
+    "so far (it may be condensed to its opening and most recent turns) and its CURRENT title.\n\n"
+    "Assess whether the current title still captures the conversation's main topic and intent "
+    "considered as a WHOLE — not just the most recent message.\n\n"
+    "Rules:\n"
+    "- If the current title is still accurate, return it UNCHANGED, verbatim.\n"
+    "- Strongly prefer keeping or only lightly adjusting the current title. Most of the time it "
+    "does not need to change.\n"
+    "- Only write a substantially different title if the conversation's main subject has clearly "
+    "and durably moved away from what the current title describes.\n"
+    "- Do NOT retitle based on the latest message alone; a brief detour or sub-task is not a "
+    "topic change.\n"
+    "- Keep it short (3-7 words). Write the title in the same language the user is writing in.\n"
+    "Return ONLY the title text, nothing else. No quotes, no punctuation at the end, no prefixes."
+)
+
+_RETITLE_PROMPT_PINNED_LANGUAGE = (
+    "You are maintaining the title of an ongoing conversation. Below is the conversation "
+    "so far (it may be condensed to its opening and most recent turns) and its CURRENT title.\n\n"
+    "Assess whether the current title still captures the conversation's main topic and intent "
+    "considered as a WHOLE — not just the most recent message.\n\n"
+    "Rules:\n"
+    "- If the current title is still accurate, return it UNCHANGED, verbatim.\n"
+    "- Strongly prefer keeping or only lightly adjusting the current title. Most of the time it "
+    "does not need to change.\n"
+    "- Only write a substantially different title if the conversation's main subject has clearly "
+    "and durably moved away from what the current title describes.\n"
+    "- Do NOT retitle based on the latest message alone; a brief detour or sub-task is not a "
+    "topic change.\n"
+    "- Keep it short (3-7 words). Write the title in {language}.\n"
+    "Return ONLY the title text, nothing else. No quotes, no punctuation at the end, no prefixes."
+)
+
 
 def _title_language() -> str:
     """Return configured title language, or empty string to match the user."""
@@ -528,6 +567,159 @@ def apply_instant_title(
         return None
 
 
+def _condense_history(
+    conversation_history: list,
+    head_turns: int = 1,
+    tail_turns: int = 3,
+    per_message: int = 400,
+) -> str:
+    """Render a conversation into a compact transcript for whole-conversation
+    title assessment.
+
+    Keeps the OPENING ``head_turns`` user/assistant exchanges (they anchor the
+    conversation's original intent — the thing a good title names) and the most
+    recent ``tail_turns`` exchanges (they reveal genuine topic drift). Anything
+    in between is elided with a ``[…]`` marker so the request stays cheap on a
+    long conversation instead of dumping the entire transcript into the
+    auxiliary model.
+
+    Only ``user`` and ``assistant`` roles are included; system/tool messages
+    are skipped. Each message is truncated to ``per_message`` chars.
+    """
+    msgs = [
+        m for m in (conversation_history or [])
+        if m.get("role") in ("user", "assistant") and (m.get("content") or "").strip()
+    ]
+    if not msgs:
+        return ""
+
+    # A "turn" here is a single message; head/tail are counted in messages so
+    # the head captures the opening user+assistant pair at head_turns=1 -> 2.
+    head_n = max(0, head_turns) * 2
+    tail_n = max(0, tail_turns) * 2
+
+    if len(msgs) <= head_n + tail_n:
+        selected = list(msgs)
+        elided_at = -1
+    else:
+        head = msgs[:head_n]
+        tail = msgs[len(msgs) - tail_n:] if tail_n else []
+        selected = head + tail
+        elided_at = len(head)
+
+    lines = []
+    for i, m in enumerate(selected):
+        if i == elided_at:
+            lines.append("[… earlier turns omitted …]")
+        role = "User" if m.get("role") == "user" else "Assistant"
+        content = (m.get("content") or "").strip()
+        if len(content) > per_message:
+            content = content[:per_message] + "…"
+        lines.append(f"{role}: {content}")
+    return "\n".join(lines)
+
+
+def _looks_like_title(text: str) -> bool:
+    """Return True if ``text`` is shaped like a real title, not prose.
+
+    The retitle model is asked "should the title change?" and sometimes answers
+    CONVERSATIONALLY ("The title remains accurate. The conversation is still
+    about …") instead of returning a title. Without this guard that sentence is
+    sanitized, truncated at 80 chars, and stored AS the title — then pushed to
+    the Discord thread name. A genuine 3-7 word title never trips these signals;
+    prose reliably does. Multi-signal reject (approved threshold):
+
+    - >10 words         → prose, not a 3-7 word title
+    - >80 chars         → longer than any legitimate short title
+    - mid-sentence '. ' → a period followed by more words is a sentence, not a title
+    - internal newline  → multi-line output is never a title
+    """
+    if not text:
+        return False
+    if len(text) > 80:
+        return False
+    if "\n" in text:
+        return False
+    if len(text.split()) > 10:
+        return False
+    # A sentence break — a lowercase word ending in '.' followed by a space and
+    # more text — means prose ("… accurate. The conversation …"). Requiring a
+    # LOWERCASE letter before the dot avoids false-flagging abbreviations whose
+    # dotted component is uppercase or single-letter ("U.S. Visa Renewal",
+    # "e.g. Docker", "Q3 Review"). A trailing period is stripped elsewhere.
+    if re.search(r"[a-z]\.\s+\S", text):
+        return False
+    return True
+
+
+def regenerate_title(
+    conversation_history: list,
+    current_title: str,
+    timeout: float = 30.0,
+    failure_callback: Optional[FailureCallback] = None,
+    main_runtime: dict = None,
+) -> Optional[str]:
+    """Re-assess an existing session title against the WHOLE conversation.
+
+    Unlike :func:`generate_title` (which only sees the first exchange), this
+    reads a condensed view of the entire conversation plus the current title,
+    and is prompted to KEEP the existing title unless the durable topic has
+    clearly drifted. Returns the (possibly unchanged) title, or ``None`` on
+    failure / empty transcript.
+    """
+    transcript = _condense_history(conversation_history)
+    if not transcript:
+        return None
+
+    language = _title_language()
+    prompt = (
+        _RETITLE_PROMPT_PINNED_LANGUAGE.format(language=language)
+        if language else _RETITLE_PROMPT
+    )
+
+    user_block = (
+        f"CURRENT TITLE: {current_title or '(none)'}\n\n"
+        f"CONVERSATION:\n{transcript}"
+    )
+    messages = [
+        {"role": "system", "content": prompt},
+        {"role": "user", "content": user_block},
+    ]
+
+    try:
+        response = call_llm(
+            task="title_generation",
+            messages=messages,
+            max_tokens=500,
+            temperature=0.3,
+            timeout=timeout,
+            main_runtime=main_runtime,
+        )
+        title = (response.choices[0].message.content or "").strip()
+        title = title.strip('"\'')
+        if title.lower().startswith("title:"):
+            title = title[6:].strip()
+        # Reject conversational / prose output instead of truncating it into a
+        # title. When the model answers "should this change?" in prose rather
+        # than returning a title, treat it as "no usable new title" → None,
+        # which the caller reads as "keep the current title, don't rename".
+        # (Do NOT fall back to title[:77]+"..." here — that is exactly how a
+        # 100-char sentence became a Discord thread name.)
+        if not _looks_like_title(title):
+            logger.debug("Retitle: rejected non-title output: %r", title[:120])
+            return None
+        return title if title else None
+    except Exception as e:
+        logger.warning("Title regeneration failed: %s", e)
+        logger.debug("Title regeneration traceback", exc_info=True)
+        if failure_callback is not None:
+            try:
+                failure_callback("title regeneration", e)
+            except Exception:
+                logger.debug("Title regeneration failure_callback raised", exc_info=True)
+        return None
+
+
 def auto_title_session(
     session_db,
     session_id: str,
@@ -700,6 +892,78 @@ def _session_is_untitled(session_db, session_id: str) -> bool:
         return False
 
 
+def maybe_retitle_session(
+    session_db,
+    session_id: str,
+    user_message: str,
+    assistant_response: str,
+    conversation_history: list,
+    failure_callback: Optional[FailureCallback] = None,
+    main_runtime: dict = None,
+    title_callback: Optional[TitleCallback] = None,
+    every_n_turns: int = 6,
+) -> None:
+    """Periodically re-evaluate a session's title to keep it relevant as the
+    conversation evolves. Fires every ``every_n_turns`` user turns AFTER the
+    initial auto-title (so first-turn handling stays exclusively with
+    :func:`maybe_auto_title`).
+
+    Cheap path:
+    - Only runs every Nth turn.
+    - Only acts once conversation_history has at least 3 user messages.
+    - Assesses the WHOLE conversation (condensed) against the current title via
+      :func:`regenerate_title`, which is biased to keep the existing title
+      unless the durable topic has clearly drifted. Only saves + fires the
+      callback (which drives the thread rename) when the title actually changes.
+    """
+    if not session_db or not session_id or not user_message or not assistant_response:
+        return
+    user_msg_count = sum(1 for m in (conversation_history or []) if m.get("role") == "user")
+    # First-turn is handled by maybe_auto_title; only act on 3rd+ user turns.
+    if user_msg_count < 3:
+        return
+    if every_n_turns <= 0 or (user_msg_count % every_n_turns) != 0:
+        return
+
+    # Snapshot history now — the background thread must assess the conversation
+    # as it stands at this turn, not whatever it has mutated into later.
+    history_snapshot = list(conversation_history or [])
+
+    def _runner():
+        try:
+            existing = session_db.get_session_title(session_id) or ""
+        except Exception:
+            return
+        new_title = regenerate_title(
+            history_snapshot, existing,
+            failure_callback=failure_callback, main_runtime=main_runtime,
+        )
+        if not new_title:
+            return
+        new_title = new_title.strip()
+        # regenerate_title is prompted to return the current title verbatim when
+        # nothing has drifted; treat an unchanged title as a no-op so we don't
+        # churn the DB or spuriously rename the thread. Compare case- and
+        # trailing-punctuation-insensitively so "USCIS RFE Response." doesn't
+        # count as a change from "USCIS RFE Response".
+        def _norm(t: str) -> str:
+            return t.strip().rstrip(".!?,;: ").lower()
+        if not new_title or _norm(new_title) == _norm(existing):
+            return
+        try:
+            session_db.set_session_title(session_id, new_title)
+        except Exception:
+            return
+        if title_callback is not None:
+            try:
+                title_callback(new_title)
+            except Exception:
+                logger.debug("Retitle callback failed", exc_info=True)
+
+    thread = threading.Thread(target=_runner, daemon=True, name="retitle")
+    thread.start()
+
+
 def maybe_auto_title(
     session_db,
     session_id: str,
diff --git a/gateway/run.py b/gateway/run.py
index 5d44f273ea647..85d18dabef318 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -7390,6 +7390,46 @@ def _approval_notify_sync(approval_data: dict) -> None:
         # as `_on_session_title` before the run starts (see
         # _attach_session_title_callback), because the titler now fires from
         # inside the turn prologue rather than from here.
+        #
+        # LOCAL CARRY (periodic retitle, PR #29983 residual delta): after the
+        # turn completes we re-evaluate the session title against the WHOLE
+        # (condensed) conversation and rename the thread only when the durable
+        # topic has drifted. Upstream owns first-turn semantic titling; this is
+        # the periodic half it still lacks. Reuses the same `_on_session_title`
+        # rename lane, adapted to its (title, title_source) signature — a
+        # periodic retitle is a real model-derived title, so it carries the
+        # "llm" source that the rename callbacks gate on. Fire-and-forget on a
+        # daemon thread inside maybe_retitle_session; never affects the turn.
+        try:
+            from agent.title_generator import maybe_retitle_session
+            _retitle_cb = getattr(agent, "_on_session_title", None)
+            _history = (
+                ctx.result_holder[0].get("messages", [])
+                if ctx.result_holder[0] else []
+            )
+            maybe_retitle_session(
+                self._session_db,
+                effective_session_id,
+                getattr(ctx, "user_message", None) or "",
+                final_response,
+                _history,
+                failure_callback=(
+                    getattr(agent, "_title_failure_callback", None)
+                    or getattr(agent, "_emit_auxiliary_failure", None)
+                ),
+                main_runtime={
+                    "model": getattr(agent, "model", None),
+                    "provider": getattr(agent, "provider", None),
+                    "base_url": getattr(agent, "base_url", None),
+                    "api_key": getattr(agent, "api_key", None),
+                    "api_mode": getattr(agent, "api_mode", None),
+                },
+                title_callback=(
+                    (lambda t: _retitle_cb(t, "llm")) if _retitle_cb else None
+                ),
+            )
+        except Exception:
+            logger.debug("Periodic retitle dispatch failed", exc_info=True)
 
         return {
             "final_response": final_response,
diff --git a/tests/agent/test_title_generator.py b/tests/agent/test_title_generator.py
index e4ae76b8aed12..fc9c0811c99de 100644
--- a/tests/agent/test_title_generator.py
+++ b/tests/agent/test_title_generator.py
@@ -6,8 +6,11 @@
 
 from agent.title_generator import (
     generate_title,
+    regenerate_title,
+    _condense_history,
     auto_title_session,
     maybe_auto_title,
+    maybe_retitle_session,
     _title_language,
 )
 from hermes_state import SessionDB
@@ -597,3 +600,307 @@ def test_instant_title_skips_marker_uses_real_message(self):
         assert apply_instant_title(db, "sess-1", "南京市秦淮区 小时级天气预报") == (
             "南京市秦淮区 小时级天气预报"
         )
+
+
+class TestCondenseHistory:
+    """Tests for _condense_history() — the whole-conversation renderer."""
+
+    def test_empty_history_returns_empty(self):
+        assert _condense_history([]) == ""
+        assert _condense_history(None) == ""
+
+    def test_skips_system_and_tool_roles(self):
+        history = [
+            {"role": "system", "content": "you are an agent"},
+            {"role": "user", "content": "hello"},
+            {"role": "tool", "content": "tool output"},
+            {"role": "assistant", "content": "hi there"},
+        ]
+        out = _condense_history(history)
+        assert "you are an agent" not in out
+        assert "tool output" not in out
+        assert "User: hello" in out
+        assert "Assistant: hi there" in out
+
+    def test_short_history_not_elided(self):
+        history = [
+            {"role": "user", "content": "q1"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "q2"},
+            {"role": "assistant", "content": "a2"},
+        ]
+        out = _condense_history(history)
+        assert "omitted" not in out
+        assert out.count("User:") == 2
+
+    def test_long_history_keeps_head_and_tail_with_elision(self):
+        # 10 exchanges = 20 messages; head_turns=1 (2 msgs) + tail_turns=3 (6 msgs)
+        history = []
+        for i in range(10):
+            history.append({"role": "user", "content": f"question {i}"})
+            history.append({"role": "assistant", "content": f"answer {i}"})
+        out = _condense_history(history)
+        # Opening turn preserved (anchors intent)
+        assert "question 0" in out
+        assert "answer 0" in out
+        # Latest turns preserved (detect drift)
+        assert "question 9" in out
+        assert "answer 9" in out
+        # A middle turn is gone
+        assert "question 5" not in out
+        # Elision marker present
+        assert "omitted" in out
+
+    def test_truncates_long_messages(self):
+        history = [
+            {"role": "user", "content": "x" * 1000},
+            {"role": "assistant", "content": "y" * 1000},
+        ]
+        out = _condense_history(history)
+        # each message truncated to per_message (400) + ellipsis, not full 1000
+        assert "x" * 401 not in out
+        assert "…" in out
+
+
+class TestRegenerateTitle:
+    """Tests for regenerate_title() — whole-conversation, sticky re-assessment."""
+
+    def _resp(self, text):
+        r = MagicMock()
+        r.choices = [MagicMock()]
+        r.choices[0].message.content = text
+        return r
+
+    def test_returns_none_on_empty_history(self):
+        # No LLM call should happen when there's no transcript.
+        with patch("agent.title_generator.call_llm") as llm:
+            assert regenerate_title([], "Some Title") is None
+            llm.assert_not_called()
+
+    def test_keeps_current_title_when_unchanged(self):
+        history = [
+            {"role": "user", "content": "help me draft the USCIS RFE response"},
+            {"role": "assistant", "content": "Here's the outline..."},
+            {"role": "user", "content": "now write the PDF"},
+            {"role": "assistant", "content": "Generating the PDF..."},
+        ]
+        # Model, seeing the whole conversation, returns the existing title verbatim.
+        with patch("agent.title_generator.call_llm", return_value=self._resp("USCIS RFE Response")):
+            out = regenerate_title(history, "USCIS RFE Response")
+            assert out == "USCIS RFE Response"
+
+    def test_whole_conversation_passed_to_model_not_just_last_exchange(self):
+        """The USCIS-RFE bug: a localized 'write the PDF' detour must not be the
+        only thing the model sees. The opening intent must reach the prompt."""
+        history = [
+            {"role": "user", "content": "help me draft the USCIS RFE response gist"},
+            {"role": "assistant", "content": "Here's the outline of the RFE response..."},
+            {"role": "user", "content": "looks good, keep going"},
+            {"role": "assistant", "content": "Continuing the RFE draft..."},
+            {"role": "user", "content": "now produce the PDF of it"},
+            {"role": "assistant", "content": "Rendering the PDF now..."},
+        ]
+        captured = {}
+
+        def _cap(**kwargs):
+            captured.update(kwargs)
+            return self._resp("USCIS RFE Response")
+
+        with patch("agent.title_generator.call_llm", side_effect=_cap):
+            regenerate_title(history, "USCIS RFE Response")
+
+        user_block = captured["messages"][1]["content"]
+        system_block = captured["messages"][0]["content"]
+        # Current title is handed to the model
+        assert "USCIS RFE Response" in user_block
+        # Opening intent (the real gist) is present, not just the PDF detour
+        assert "RFE response gist" in user_block
+        # The PDF detour is present too (tail), but as context, not the sole input
+        assert "PDF" in user_block
+        # Prompt instructs whole-conversation, keep-biased assessment
+        assert "WHOLE" in system_block
+        assert "UNCHANGED" in system_block
+
+    def test_returns_new_title_on_genuine_drift(self):
+        history = [
+            {"role": "user", "content": "help me draft the USCIS RFE response"},
+            {"role": "assistant", "content": "Here's the outline..."},
+            {"role": "user", "content": "actually forget that, let's debug my docker setup"},
+            {"role": "assistant", "content": "Let's look at your Dockerfile..."},
+            {"role": "user", "content": "the container won't start"},
+            {"role": "assistant", "content": "Check the entrypoint..."},
+        ]
+        with patch("agent.title_generator.call_llm", return_value=self._resp("Debugging Docker Setup")):
+            out = regenerate_title(history, "USCIS RFE Response")
+            assert out == "Debugging Docker Setup"
+
+    def test_pinned_language_prompt(self):
+        history = [
+            {"role": "user", "content": "hola"},
+            {"role": "assistant", "content": "hola, como estas"},
+        ]
+        captured = {}
+
+        def _cap(**kwargs):
+            captured.update(kwargs)
+            return self._resp("Saludo")
+
+        with (
+            patch("agent.title_generator.call_llm", side_effect=_cap),
+            patch("agent.title_generator._title_language", return_value="Spanish"),
+        ):
+            regenerate_title(history, "Greeting")
+
+        system_block = captured["messages"][0]["content"]
+        assert "Write the title in Spanish" in system_block
+
+    def test_returns_none_on_exception(self):
+        history = [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}]
+        with patch("agent.title_generator.call_llm", side_effect=RuntimeError("no provider")):
+            assert regenerate_title(history, "Title") is None
+
+    def test_invokes_failure_callback_on_exception(self):
+        history = [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}]
+        captured = []
+        exc = RuntimeError("boom")
+        with patch("agent.title_generator.call_llm", side_effect=exc):
+            regenerate_title(history, "Title", failure_callback=lambda t, e: captured.append((t, e)))
+        assert captured == [("title regeneration", exc)]
+
+    def test_rejects_conversational_prose_instead_of_truncating(self):
+        """Regression: the model answered the "should this change?" question in
+        PROSE ("The title remains accurate. The conversation is still about …")
+        instead of returning a title. The old code sanitized + truncated it at
+        80 chars and stored the sentence AS the title, which then became the
+        Discord thread name. Prose must be rejected → None → keep current title.
+        """
+        history = [
+            {"role": "user", "content": "run hermes update"},
+            {"role": "assistant", "content": "Starting the triage..."},
+            {"role": "user", "content": "restart and verify"},
+            {"role": "assistant", "content": "Gateway restarted cleanly."},
+        ]
+        prose = (
+            "The title remains accurate. The conversation is still about "
+            "triaging and executing the hermes update"
+        )
+        with patch("agent.title_generator.call_llm", return_value=self._resp(prose)):
+            out = regenerate_title(history, "Hermes Update Triage")
+        assert out is None
+
+    def test_rejects_overlong_prose_not_truncate(self):
+        """A >80-char blob must be rejected (None), never truncated into a title."""
+        history = [
+            {"role": "user", "content": "q"},
+            {"role": "assistant", "content": "a"},
+        ]
+        with patch("agent.title_generator.call_llm", return_value=self._resp("A" * 100)):
+            assert regenerate_title(history, "Existing Title") is None
+
+    def test_accepts_abbreviation_titles_with_internal_dots(self):
+        """The prose guard must NOT false-reject legit titles whose dotted
+        component is an abbreviation ("U.S. Visa Renewal") — only lowercase-
+        word sentence breaks count as prose."""
+        history = [
+            {"role": "user", "content": "help with my visa"},
+            {"role": "assistant", "content": "Sure..."},
+            {"role": "user", "content": "timeline?"},
+            {"role": "assistant", "content": "Here..."},
+        ]
+        with patch("agent.title_generator.call_llm", return_value=self._resp("U.S. Visa Renewal Timeline")):
+            assert regenerate_title(history, "Visa Help") == "U.S. Visa Renewal Timeline"
+
+
+class TestLooksLikeTitle:
+    """Unit tests for the _looks_like_title prose-rejection shape guard."""
+
+    @pytest.mark.parametrize("text", [
+        "USCIS RFE Response",
+        "Debugging Python Import Errors",
+        "Setting Up Docker Environment",
+        "Kubernetes Pod Debugging",
+        "U.S. Visa Renewal Timeline",   # uppercase/single-letter abbreviation dots
+        "Q3 Financial Review",
+    ])
+    def test_accepts_real_titles(self, text):
+        from agent.title_generator import _looks_like_title
+        assert _looks_like_title(text) is True
+
+    @pytest.mark.parametrize("text", [
+        "",
+        "The title remains accurate. The conversation is still about triaging",  # sentence break
+        "A" * 100,                                                                # >80 chars
+        "one two three four five six seven eight nine ten eleven",                # >10 words
+        "Fixing the bug. Then shipping it",                                       # mid-sentence period
+        "Line one\nLine two",                                                     # internal newline
+    ])
+    def test_rejects_prose_and_garbage(self, text):
+        from agent.title_generator import _looks_like_title
+        assert _looks_like_title(text) is False
+
+
+class TestMaybeRetitleSession:
+    """Tests for maybe_retitle_session() — the periodic re-title gate."""
+
+    def _history(self, n_user):
+        h = []
+        for i in range(n_user):
+            h.append({"role": "user", "content": f"q{i}"})
+            h.append({"role": "assistant", "content": f"a{i}"})
+        return h
+
+    def test_skips_before_third_user_turn(self):
+        db = MagicMock()
+        with patch("agent.title_generator.regenerate_title") as regen:
+            maybe_retitle_session(db, "s1", "q", "a", self._history(2), every_n_turns=6)
+            import time
+            time.sleep(0.1)
+            regen.assert_not_called()
+
+    def test_skips_off_cadence(self):
+        db = MagicMock()
+        # 4 user turns, every_n_turns=6 -> 4 % 6 != 0 -> skip
+        with patch("agent.title_generator.regenerate_title") as regen:
+            maybe_retitle_session(db, "s1", "q", "a", self._history(4), every_n_turns=6)
+            import time
+            time.sleep(0.1)
+            regen.assert_not_called()
+
+    def test_fires_on_cadence_and_uses_regenerate_title(self):
+        db = MagicMock()
+        db.get_session_title.return_value = "Old Title"
+        history = self._history(6)  # 6 % 6 == 0 -> fire
+        with patch("agent.title_generator.regenerate_title", return_value="New Title") as regen:
+            maybe_retitle_session(db, "s1", "q", "a", history, every_n_turns=6)
+            import time
+            time.sleep(0.3)
+            regen.assert_called_once()
+            # regenerate_title must receive the full history + current title,
+            # NOT just the last user/assistant message.
+            args, kwargs = regen.call_args
+            assert args[0] == history
+            assert args[1] == "Old Title"
+        db.set_session_title.assert_called_once_with("s1", "New Title")
+
+    def test_no_db_write_when_title_unchanged(self):
+        db = MagicMock()
+        db.get_session_title.return_value = "Same Title"
+        history = self._history(6)
+        with patch("agent.title_generator.regenerate_title", return_value="Same Title"):
+            maybe_retitle_session(db, "s1", "q", "a", history, every_n_turns=6)
+            import time
+            time.sleep(0.3)
+        db.set_session_title.assert_not_called()
+
+    def test_callback_fires_on_change(self):
+        db = MagicMock()
+        db.get_session_title.return_value = "Old"
+        history = self._history(6)
+        seen = []
+        with patch("agent.title_generator.regenerate_title", return_value="Brand New"):
+            maybe_retitle_session(
+                db, "s1", "q", "a", history, every_n_turns=6, title_callback=seen.append
+            )
+            import time
+            time.sleep(0.3)
+        assert seen == ["Brand New"]