Merge pull request #81985 from NousResearch/bb/session-titles

Name sessions instantly from the opening message, and make the name stick

acp_adapter/server.py

@@ -1904,6 +1904,17 @@ class HermesACPAgent(acp.Agent):
             # never leaks one session's id into the next session's tools.
             previous_session_id = os.environ.get("HERMES_SESSION_ID")
             os.environ["HERMES_SESSION_ID"] = session_id
+            # Auto-titling fires inside the turn prologue now; give the agent
+            # this session's notifier so a new title reaches the client as a
+            # session-info update instead of waiting for the next one.
+            def _notify_title_update(_title: str) -> None:
+                if conn:
+                    loop.call_soon_threadsafe(
+                        asyncio.create_task,
+                        self._send_session_info_update(session_id),
+                    )
+
+            agent._on_session_title = _notify_title_update
             try:
                 result = agent.run_conversation(
                     user_message=user_content,
@@ -2001,43 +2012,6 @@ class HermesACPAgent(acp.Agent):
         suppress_interrupt_response = interrupted and final_response.startswith(
             INTERRUPT_WAITING_FOR_MODEL_PREFIX
         )
-        if final_response and not suppress_interrupt_response:
-            try:
-                from agent.title_generator import maybe_auto_title
-
-                def _notify_title_update(_title: str) -> None:
-                    if conn:
-                        loop.call_soon_threadsafe(
-                            asyncio.create_task,
-                            self._send_session_info_update(session_id),
-                        )
-
-                # Snapshot the runtime identity; the validator lets the
-                # background titler skip its LLM call if the session's model
-                # changed before it fires (#19027).
-                _title_model = getattr(state.agent, "model", None)
-                _title_provider = getattr(state.agent, "provider", None)
-                maybe_auto_title(
-                    self.session_manager._get_db(),
-                    session_id,
-                    user_text,
-                    final_response,
-                    state.history,
-                    main_runtime={
-                        "model": getattr(state.agent, "model", None),
-                        "provider": getattr(state.agent, "provider", None),
-                        "base_url": getattr(state.agent, "base_url", None),
-                        "api_key": getattr(state.agent, "api_key", None),
-                        "api_mode": getattr(state.agent, "api_mode", None),
-                    },
-                    runtime_validator=lambda: (
-                        getattr(state.agent, "model", None) == _title_model
-                        and getattr(state.agent, "provider", None) == _title_provider
-                    ),
-                    title_callback=_notify_title_update,
-                )
-            except Exception:
-                logger.debug("Failed to auto-title ACP session %s", session_id, exc_info=True)
         if (
             final_response
             and conn

agent/auxiliary_client.py

@@ -696,23 +696,131 @@ def _compression_threshold_for_model(
         return _CODEX_SPARK_COMPACTION_THRESHOLD
     return None
 
-# Default auxiliary models for direct API-key providers (cheap/fast for side tasks)
-def _get_aux_model_for_provider(provider_id: str) -> str:
-    """Return the cheap auxiliary model for a provider.
+# Model-family priority for the auxiliary "fast tier", fastest first.
+#
+# Matched as substrings against the provider's LIVE /v1/models catalog rather
+# than pinned as exact ids, because exact ids rot: a hardcoded
+# "google/gemini-3-flash" kept 404ing here once Nous dropped it upstream, and
+# every aux call paid a wasted round-trip before the retry net caught it.
+# Families outlive their version numbers, so a new mini/flash/haiku release is
+# picked up with no source edit.
+#
+# Rolling "-latest" aliases come first where a provider publishes them (Nous
+# serves ~openai/gpt-mini-latest, ~google/gemini-flash-latest, …): they are the
+# only ids that are structurally rot-proof.
+#
+# Order is measured, not guessed — p50 on a real titling prompt against the
+# Nous catalog: gpt-mini-latest 1.40s, claude-haiku-latest 1.55s,
+# gemini-flash-latest 2.13s, step-3.7-flash 7.84s, grok-4.1-fast 8.05s. So the
+# first family a provider actually serves is also the fastest it can offer.
+_FAST_MODEL_FAMILIES: tuple = (
+    "gpt-mini-latest",
+    "gpt-nano-latest",
+    "claude-haiku-latest",
+    "gemini-flash-latest",
+    "gpt-5.4-nano",
+    "gpt-5.4-mini",
+    "gpt-5-mini",
+    "haiku-4.5",
+    "gemini-3.6-flash",
+    "flash-lite",
+    "-nano",
+    "-mini",
+    "-flash",
+    "haiku",
+)
 
-    Reads from ProviderProfile.default_aux_model first, falling back to the
-    legacy hardcoded dict for providers that predate the profiles system.
+# Substrings that disqualify an otherwise-matching id. Reasoning variants
+# ("o3-mini", "gpt-5.4-mini-thinking") think before answering, which is the
+# opposite of what a titler wants; ":batch" is an async queue, not a live
+# endpoint; embedding models ("all-minilm") match "-mini" but aren't chat
+# models at all; ":free" tiers are heavily rate-limited and measured slowest.
+_FAST_MODEL_EXCLUDE: tuple = (
+    "thinking", "reason", "-r1", "minilm", ":batch", ":free",
+    "o1-", "o3-", "o4-", "codex", "audio", "-vl", "embed",
+)
+
+
+def _fast_model_from_catalog(provider_id: str) -> str:
+    """Pick the fastest small model the provider ACTUALLY serves right now.
+
+    Reads the provider's live (cached) ``/v1/models`` catalog and returns the
+    first ``_FAST_MODEL_FAMILIES`` match. Returns "" when the catalog is
+    unavailable or holds no small model, so the caller falls through to the
+    provider's curated default. Never raises and never blocks on a cold
+    network path — the underlying fetch is memory+disk cached with a
+    last-known-good fallback.
     """
     try:
+        from hermes_cli.models import fetch_models_with_pricing
         from providers import get_provider_profile
-        _p = get_provider_profile(provider_id)
-        if _p and _p.default_aux_model:
-            return _p.default_aux_model
+
+        profile = get_provider_profile(provider_id)
+        base_url = str(getattr(profile, "base_url", "") or "").rstrip("/")
+        if not base_url:
+            return ""
+        # fetch_models_with_pricing appends its own /v1/models.
+        if base_url.endswith("/v1"):
+            base_url = base_url[:-3]
+        catalog = fetch_models_with_pricing(base_url=base_url, timeout=3.0) or {}
+    except Exception:
+        logger.debug("Fast-model catalog lookup failed for %s", provider_id, exc_info=True)
+        return ""
+
+    ids = sorted(str(m) for m in catalog)
+    for family in _FAST_MODEL_FAMILIES:
+        for model_id in ids:
+            lowered = model_id.lower()
+            if family in lowered and not any(x in lowered for x in _FAST_MODEL_EXCLUDE):
+                return model_id
+    return ""
+
+
+# Default auxiliary models for direct API-key providers (cheap/fast for side tasks)
+def _get_aux_model_for_provider(provider_id: str, *, prefer_fast: bool = False) -> str:
+    """Return the cheap auxiliary model for a provider.
+
+    Resolution ladder, fastest-and-most-live first:
+
+    1. ``prefer_fast`` only — a family match against the provider's LIVE
+       ``/v1/models`` catalog, preferring rolling ``-latest`` aliases. This is
+       both rot-proof and latency-ordered.
+    2. ``prefer_fast`` only — the provider's own recommendation hook
+       (``ProviderProfile.resolve_aux_model``). Live, but tuned for *quality*
+       on long-context side tasks (Nous returns its compaction pick), so it
+       ranks below the catalog match for latency-critical work.
+    3. ``ProviderProfile.default_aux_model`` — curated, hardcoded, may rot.
+    4. The legacy hardcoded dict, for providers predating the profiles system.
+
+    ``prefer_fast`` is opt-in so this only changes latency-critical tasks
+    (titling). Every other auxiliary caller keeps the existing static
+    behaviour and its cache keys.
+    """
+    profile = None
+    try:
+        from providers import get_provider_profile
+        profile = get_provider_profile(provider_id)
     except Exception:
         pass
+
+    if prefer_fast:
+        catalog_pick = _fast_model_from_catalog(provider_id)
+        if catalog_pick:
+            return catalog_pick
+        if profile is not None:
+            try:
+                live = profile.resolve_aux_model()
+                if live:
+                    return live
+            except Exception:
+                logger.debug("resolve_aux_model failed for %s", provider_id, exc_info=True)
+
+    if profile is not None and profile.default_aux_model:
+        return profile.default_aux_model
     return _API_KEY_PROVIDER_AUX_MODELS_FALLBACK.get(provider_id, "")
 
 
+
 # Fallback for providers not yet migrated to ProviderProfile.default_aux_model,
 # plus providers we intentionally keep pinned here (e.g. Anthropic predates
 # profiles). New providers should set default_aux_model on their profile instead.
@@ -740,6 +848,13 @@ _API_KEY_PROVIDER_AUX_MODELS_FALLBACK: Dict[str, str] = {
 # can still use this dict directly. Kept in sync with _FALLBACK above.
 _API_KEY_PROVIDER_AUX_MODELS: Dict[str, str] = _API_KEY_PROVIDER_AUX_MODELS_FALLBACK
 
+# Auxiliary tasks that prefer the provider's fast/cheap model over the user's
+# main chat model when running in "auto" mode. Restricted to tasks where
+# latency is user-visible and the output is short enough that a small model
+# matches a frontier one. Every other task keeps "auto = my chat model".
+_FAST_MODEL_TASKS: frozenset = frozenset({"title_generation"})
+
+
 # Vision-specific model overrides for direct providers.
 # When the user's main provider has a dedicated vision/multimodal model that
 # differs from their main chat model, map it here.  The vision auto-detect
@@ -5420,7 +5535,6 @@ def _resolve_auto_route(
     runtime_api_key = runtime.get("api_key", "")
     runtime_api_mode = str(runtime.get("api_mode") or "")
 
-
     # ── Warn once if OPENAI_BASE_URL is set but config.yaml uses a named
     #    provider (not 'custom').  This catches the common "env poisoning"
     #    scenario where a user switches providers via `hermes model` but the
@@ -5450,6 +5564,25 @@ def _resolve_auto_route(
     main_provider = str(runtime_provider or _read_main_provider() or "")
     main_model = str(runtime_model or _read_main_model() or "")
 
+    # Latency-critical tasks prefer the provider's registered fast model over
+    # the main chat model. Titling is the only such task: it names a visible
+    # sidebar row, produces ~8 tokens, and running it on a frontier reasoning
+    # model costs seconds per new session. Every comparable tool routes titling
+    # to a small tier (Claude Code → Haiku, OpenCode → small_model, Zed →
+    # default_fast_model, OpenClaw → utilityModel). An explicit
+    # auxiliary.<task>.model in config.yaml still wins — this only redirects
+    # the "auto" default, and only when the provider registered a cheap model.
+    # Every other aux task keeps the "auto means my chat model" contract
+    # documented above: this does NOT change compression, vision, or search.
+    if task in _FAST_MODEL_TASKS and main_provider and main_provider not in {"auto", ""}:
+        fast_model = _get_aux_model_for_provider(main_provider, prefer_fast=True)
+        if fast_model and fast_model != main_model:
+            logger.debug(
+                "Auxiliary task %s: preferring fast model %s over main model %s",
+                task, fast_model, main_model,
+            )
+            main_model = fast_model
+
     # MoA virtual provider: the "model" is a preset name (e.g. "opus-gpt") and
     # there is no real "moa" HTTP endpoint, so resolving an aux client against
     # provider="moa"/model=<preset> sends the preset name as the model id and
@@ -8776,6 +8909,7 @@ def _call_llm_impl(
             api_key=resolved_api_key,
             api_mode=resolved_api_mode,
             main_runtime=main_runtime,
+            task=task,
         )
         effective_provider = _effective_provider_for_client(
             client, resolved_provider,
@@ -9554,6 +9688,7 @@ async def _async_call_llm_impl(
             api_key=resolved_api_key,
             api_mode=resolved_api_mode,
             main_runtime=main_runtime,
+            task=task,
         )
         effective_provider = _effective_provider_for_client(
             client, resolved_provider,

agent/conversation_compression.py

@@ -3331,13 +3331,52 @@ def compress_context(
                         migrate_heartbeat_to_session(old_session_id, agent.session_id)
                     except Exception as _hb_err:
                         logger.debug("Could not migrate heartbeat on compression: %s", _hb_err)
-                    # Auto-number the title for the continuation session
+                    # Carry the title across the compression boundary unchanged.
+                    #
+                    # This used to renumber ("Fix X" → "Fix X #2") on every
+                    # rotation, which is why a long conversation ended up as
+                    # "Smallville Map Architecture Plan #10" — ten forks of ONE
+                    # session, each looking like a separate piece of work in the
+                    # sidebar. Compression is an internal implementation detail;
+                    # the user's conversation did not change topic, so its name
+                    # must not change either. Uniqueness still holds because
+                    # _set_session_title transfers the title off a hidden
+                    # compression ancestor rather than raising on the conflict.
                     if old_title:
+                        # Read provenance BEFORE the write: transferring the
+                        # title off a hidden compression ancestor clears the
+                        # ancestor's row, so reading afterwards always returns
+                        # None and the child would be stamped "user" — freezing
+                        # an auto-title that should still be upgradeable.
+                        _src = None
                         try:
-                            new_title = agent._session_db.get_next_title_in_lineage(old_title)
-                            agent._session_db.set_session_title(agent.session_id, new_title)
+                            _src = agent._session_db.get_session_title_source(
+                                old_session_id
+                            )
+                        except Exception as _src_err:
+                            logger.debug(
+                                "Could not read title provenance: %s", _src_err
+                            )
+                        try:
+                            agent._session_db.set_session_title(
+                                agent.session_id, old_title
+                            )
                         except (ValueError, Exception) as e:
                             logger.debug("Could not propagate title on compression: %s", e)
+                        else:
+                            # set_session_title() records "user"; restore the
+                            # original authority so an inherited auto-title
+                            # stays upgradeable and a manual one stays pinned.
+                            if _src is not None:
+                                try:
+                                    agent._session_db.set_session_title_source(
+                                        agent.session_id, _src
+                                    )
+                                except Exception as _src_err:
+                                    logger.debug(
+                                        "Could not propagate title provenance: %s",
+                                        _src_err,
+                                    )
 
                 # In-place mode still updates/replaces the current row here.
                 # Rotation already published prompt + compacted handoff atomically.

agent/title_generator.py

@@ -1,10 +1,25 @@
-"""Auto-generate short session titles from the first user/assistant exchange.
+"""Auto-generate short session titles from the user's opening message.
 
-Runs asynchronously after the first response is delivered so it never
-adds latency to the user-facing reply.
+Two stages, both off the critical path:
+
+1. **Instant** — a deterministic title derived from the first user message,
+   written before the model is even called. Costs nothing, cannot fail, and
+   means a session is named the moment it starts instead of after the first
+   turn finishes (which measured p50 151s / p90 1212s on real sessions).
+2. **Upgrade** — one small-model call that replaces the derived title with a
+   proper one. Runs on a cheap/fast tier, with thinking disabled and the
+   response constrained to a JSON object, so there is no reasoning preamble to
+   strip and nothing to parse out of prose.
+
+Provenance (``derived`` < ``llm`` < ``user``) is enforced by the storage layer,
+so stage 2 can only ever replace stage 1, and neither can replace a name the
+user typed. That ordering is the industry-standard one — Codex CLI encodes the
+same ``custom > ai > fallback`` precedence in its session importer.
 """
 
+import json
 import logging
+import re
 import threading
 from typing import Callable, Optional
 
@@ -25,18 +40,82 @@ TitleCallback = Callable[[str], None]
 # the request would reload a model the runtime already evicted (#19027).
 RuntimeValidator = Callable[[], bool]
 
-_TITLE_PROMPT = (
-    "Generate a short, descriptive title (3-7 words) for a conversation that starts with the "
-    "following exchange. The title should capture the main topic or intent. "
-    "Write the title in the same language the user is writing in. "
-    "Return ONLY the title text, nothing else. No quotes, no punctuation at the end, no prefixes."
+# Cap on the text handed to the model. Claude Code and OpenClaw independently
+# converged on the same 1000-char budget; a title needs the opening intent, not
+# a pasted stack trace.
+MAX_TITLE_INPUT_CHARS = 1000
+
+# Cap on the instant derived title. Deliberately shorter than the model's
+# budget: a raw sentence fragment reads worse the longer it runs. Cline and
+# Codex CLI independently landed on the same ~50-char slice.
+MAX_DERIVED_TITLE_CHARS = 48
+
+_TITLE_PROMPT_TEMPLATE = (
+    "You name chat sessions. Given the user's opening message, write a title "
+    "that lets them find this conversation again in a list.\n\n"
+    "Rules:\n"
+    "- 3 to 7 words, sentence case (capitalize only the first word and proper nouns).\n"
+    "- Name what the user wants DONE, not that they asked a question.\n"
+    "- Keep technical terms, filenames, numbers, and error codes exact.\n"
+    "- Drop filler words: the, this, my, a, an.\n"
+    "- No trailing punctuation, no quotes, no tool names, no 'Title:' prefix.\n"
+    "- Never answer the message. Name it.\n"
+    "- Always produce something, even for a bare greeting.\n"
+    "__LANGUAGE_RULE__\n"
+    'Good: {"title": "Fix login button on mobile"}\n'
+    'Good: {"title": "Postgres connection pool exhaustion"}\n'
+    'Good: {"title": "Friendly greeting"}\n'
+    'Too vague: {"title": "Code changes"}\n'
+    'Too long: {"title": "Investigate and fix the issue where the login button '
+    'does not respond on mobile devices"}\n\n'
+    'Reply with JSON only: {"title": "..."}'
 )
 
-_TITLE_PROMPT_PINNED_LANGUAGE = (
-    "Generate a short, descriptive title (3-7 words) for a conversation that starts with the "
-    "following exchange. The title should capture the main topic or intent. "
-    "Write the title in {language}. "
-    "Return ONLY the title text, nothing else. No quotes, no punctuation at the end, no prefixes."
+_LANGUAGE_RULE_MATCH_USER = "- Write the title in the same language as the user's message."
+_LANGUAGE_RULE_PINNED = "- Write the title in {language}."
+
+# JSON schema constraining the response to a single title field. Removes the
+# whole class of "model answered the prompt instead of titling it" failures
+# that produced titles like "<title>...</title>" and "User: Yep, that's the
+# catch —" in real session history.
+_TITLE_RESPONSE_FORMAT = {
+    "type": "json_schema",
+    "json_schema": {
+        "name": "session_title",
+        "strict": True,
+        "schema": {
+            "type": "object",
+            "properties": {"title": {"type": "string"}},
+            "required": ["title"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+# Control-tag wrappers that surround machine-authored content inside what is
+# nominally a "user" message. Titling from these is what produces a session
+# named after a slash command or an injected reminder rather than the user's
+# actual request. Ported from Codex CLI's RECOGNIZED_CONTROL_WRAPPERS, which
+# strips them (and keeps titling) rather than refusing outright.
+_CONTROL_WRAPPERS = (
+    ("<command-message>", "</command-message>"),
+    ("<command-name>", "</command-name>"),
+    ("<command-args>", "</command-args>"),
+    ("<local-command-caveat>", "</local-command-caveat>"),
+    ("<local-command-stderr>", "</local-command-stderr>"),
+    ("<local-command-stdout>", "</local-command-stdout>"),
+    ("<task-notification>", "</task-notification>"),
+    ("<system-reminder>", "</system-reminder>"),
+    ("<ide_opened_file>", "</ide_opened_file>"),
+    ("<ide_selection>", "</ide_selection>"),
+)
+
+# Hermes' own machine-authored openers. A compaction handoff or a resumed
+# session must not be titled after the scaffolding that carried it.
+_MACHINE_PREFIXES = (
+    "[CONTEXT COMPACTION",
+    "[Runtime note:",
+    "[SYSTEM]",
 )
 
 
@@ -71,40 +150,177 @@ def _auto_title_enabled() -> bool:
         return True
 
 
+def strip_control_wrappers(text: str) -> str:
+    """Remove leading machine-authored control wrappers, including nested ones.
+
+    Loops so ``<command-message><command-name>/work</command-name></command-message>``
+    reduces to the prose the user actually typed. Unlike a refusal check, this
+    still yields usable text, so a slash-command turn gets a real title instead
+    of staying untitled.
+    """
+    if not text:
+        return ""
+    current = text.strip()
+    # Bounded: each pass must remove at least one wrapper or we stop.
+    for _ in range(len(_CONTROL_WRAPPERS) * 2):
+        stripped = current
+        for open_tag, close_tag in _CONTROL_WRAPPERS:
+            if not stripped.lower().startswith(open_tag):
+                continue
+            end = stripped.lower().find(close_tag)
+            if end == -1:
+                # Unterminated wrapper: drop the opening tag and keep the body.
+                stripped = stripped[len(open_tag):].strip()
+            else:
+                inner = stripped[len(open_tag):end].strip()
+                rest = stripped[end + len(close_tag):].strip()
+                # Prefer the trailing prose when there is any; otherwise the
+                # wrapper's own body is the only content we have.
+                stripped = (rest or inner).strip()
+            break
+        if stripped == current:
+            break
+        current = stripped
+    return current
+
+
 def _summarize_user_message(user_message: str) -> str:
-    """Collapse a slash-skill-expanded turn back to what the user typed.
+    """Reduce a user turn to the text worth titling.
 
     A ``/skill`` invocation expands into a message that embeds the whole skill
     body, so feeding it to the titler verbatim titles the session after the
     *skill's* prose — "Kick off a task in a fresh isolated git worktree" — not
     after the user's request. Reuse the canonical scaffolding parser so the
-    model sees ``/work — fix the title leak`` instead.
+    model sees ``/work — fix the title leak`` instead, then strip any control
+    wrappers left around it.
     """
     if not user_message:
         return ""
+    described = None
     try:
         from agent.skill_commands import describe_skill_invocation
 
         described = describe_skill_invocation(user_message)
     except Exception:
         logger.debug("Skill-scaffolding summary failed; titling raw", exc_info=True)
-        return user_message
-    return described if described is not None else user_message
+    text = described if described is not None else user_message
+    return strip_control_wrappers(text)
+
+
+def is_titleable_user_message(user_message: str) -> bool:
+    """Return whether *user_message* carries real user intent to title from.
+
+    False for machine-authored openers (compaction handoffs, runtime notes) and
+    for turns that reduce to nothing once control scaffolding is stripped.
+    """
+    if not isinstance(user_message, str) or not user_message.strip():
+        return False
+    for prefix in _MACHINE_PREFIXES:
+        if user_message.lstrip().startswith(prefix):
+            return False
+    return bool(_summarize_user_message(user_message).strip())
+
+
+def derive_title(user_message: str) -> Optional[str]:
+    """Build an instant title from the user's message. No model, never fails.
+
+    This is what the user sees within milliseconds of sending their first
+    message. It is intentionally dumb — first meaningful line, trimmed to a
+    word boundary — because its job is to beat the model to the screen, not to
+    beat it on quality. The model's title replaces it moments later.
+    """
+    text = _summarize_user_message(user_message)
+    if not text:
+        return None
+    # First non-empty line: a pasted log or a multi-paragraph brief still gets
+    # named after its opening intent.
+    line = next((ln.strip() for ln in text.splitlines() if ln.strip()), "")
+    if not line:
+        return None
+    line = " ".join(line.split())
+    if len(line) > MAX_DERIVED_TITLE_CHARS:
+        cut = line[:MAX_DERIVED_TITLE_CHARS]
+        # Prefer a word boundary so the title doesn't end mid-token.
+        space = cut.rfind(" ")
+        if space > MAX_DERIVED_TITLE_CHARS // 2:
+            cut = cut[:space]
+        line = cut.rstrip(" ,.;:—-") + "…"
+    return line or None
+
+
+def _extract_title_text(content: str) -> str:
+    """Pull the title out of a model response.
+
+    The JSON schema makes the object shape the expected case, but not every
+    provider honors ``response_format``; fall back through a loose JSON scan
+    and finally to first-line prose so a non-compliant provider still titles.
+    """
+    if not content:
+        return ""
+    raw = content.strip()
+    # Fenced JSON from providers that wrap structured output in markdown.
+    fenced = re.match(r"^```(?:json)?\s*(.*?)\s*```$", raw, re.DOTALL)
+    if fenced:
+        raw = fenced.group(1).strip()
+    try:
+        parsed = json.loads(raw)
+        if isinstance(parsed, dict) and isinstance(parsed.get("title"), str):
+            return parsed["title"].strip()
+    except (ValueError, TypeError):
+        pass
+    # Loose scan: a compliant object embedded in surrounding chatter.
+    match = re.search(r'"title\"\s*:\s*"((?:[^"\\]|\\.)*)"', raw)
+    if match:
+        try:
+            return json.loads(f'"{match.group(1)}"').strip()
+        except ValueError:
+            return match.group(1).strip()
+    # Prose fallback. Reuse the canonical scrubber so reasoning-model output
+    # (<think>…) can't leak into a title, then keep the first real line.
+    try:
+        from agent.agent_runtime_helpers import strip_think_blocks
+
+        raw = strip_think_blocks(None, raw).strip()
+    except Exception:
+        logger.debug("strip_think_blocks unavailable for title output", exc_info=True)
+    raw = next((ln.strip() for ln in raw.splitlines() if ln.strip()), "")
+    if raw.lower().startswith("title:"):
+        raw = raw[6:].strip()
+    return raw.strip("\"'").strip()
+
+
+def _clean_title(text: str) -> Optional[str]:
+    """Normalize a model-produced title, or None when nothing usable remains."""
+    title = " ".join((text or "").split())
+    title = title.strip("\"'").strip()
+    if title.lower().startswith("title:"):
+        title = title[6:].strip()
+    # Trailing sentence punctuation reads wrong in a sidebar list.
+    title = title.rstrip(".!,;:")
+    if not title:
+        return None
+    if len(title) > 80:
+        title = title[:77].rstrip() + "..."
+    return title
 
 
 def generate_title(
     user_message: str,
-    assistant_response: str,
     timeout: Optional[float] = None,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
     runtime_validator: Optional[RuntimeValidator] = None,
 ) -> Optional[str]:
-    """Generate a session title from the first exchange.
+    """Generate a session title from the user's opening message.
 
-    Uses the main runtime's model when available, falling back to the
-    auxiliary LLM client (cheapest/fastest available model).
-    Returns the title string or None on failure.
+    Runs on the ``title_generation`` auxiliary task, which resolves to a
+    small/fast model tier. Thinking is disabled and the response is constrained
+    to ``{"title": "..."}`` so there is no preamble or reasoning to strip.
+
+    Titles come from the user's message alone — every surveyed implementation
+    that titles well (Claude Code, OpenCode, Cursor, OpenClaw) does the same.
+    Waiting for the assistant is what made this slow, and it bought nothing:
+    the user's opening message already states the intent worth naming.
 
     ``failure_callback`` is invoked with ``(task, exception)`` when the
     auxiliary call raises — the caller typically wires this to
@@ -130,49 +346,39 @@ def generate_title(
             # Fail open: a broken validator must not disable titling.
             logger.debug("Title runtime validator raised; proceeding", exc_info=True)
 
-    # Truncate long messages to keep the request small
-    user_snippet = _summarize_user_message(user_message)[:500]
-    assistant_snippet = assistant_response[:500] if assistant_response else ""
+    user_snippet = _summarize_user_message(user_message)[:MAX_TITLE_INPUT_CHARS]
+    if not user_snippet.strip():
+        return None
 
     language = _title_language()
-    prompt = _TITLE_PROMPT_PINNED_LANGUAGE.format(language=language) if language else _TITLE_PROMPT
+    language_rule = (
+        _LANGUAGE_RULE_PINNED.format(language=language)
+        if language
+        else _LANGUAGE_RULE_MATCH_USER
+    )
+    # Placeholder substitution, not str.format: the prompt embeds literal JSON
+    # braces as few-shot examples, which format() would try to interpolate.
+    prompt = _TITLE_PROMPT_TEMPLATE.replace("__LANGUAGE_RULE__", language_rule)
 
     messages = [
         {"role": "system", "content": prompt},
-        {"role": "user", "content": f"User: {user_snippet}\n\nAssistant: {assistant_snippet}"},
+        {"role": "user", "content": user_snippet},
     ]
 
     try:
         response = call_llm(
             task="title_generation",
             messages=messages,
-            max_tokens=500,
+            # A title is a handful of tokens. The old 500-token ceiling let a
+            # chatty model burn seconds generating prose we then threw away.
+            max_tokens=64,
             temperature=0.3,
             timeout=timeout,
             main_runtime=main_runtime,
+            extra_body={"response_format": _TITLE_RESPONSE_FORMAT},
         )
         content = response.choices[0].message.content or ""
-        # Strip thinking/reasoning blocks that think-enabled models
-        # (MiniMax M2.7, DeepSeek, etc.) emit even for simple prompts like
-        # title generation. Without this the raw <think>...</think> XML
-        # leaks into session titles. Reuses the canonical scrubber so all
-        # tag variants (unterminated blocks, orphan closes, mixed case)
-        # are handled, not just a single literal <think> pair.
-        from agent.agent_runtime_helpers import strip_think_blocks
-        title = strip_think_blocks(None, content).strip()
-        # Clean up: remove quotes, trailing punctuation, prefixes like "Title: "
-        title = title.strip('"\'')
-        if title.lower().startswith("title:"):
-            title = title[6:].strip()
-        # A title is one line. A model that ignores "return ONLY the title" and
-        # answers the prompt instead (a shell transcript, a bulleted plan) would
-        # otherwise be stored verbatim and truncated mid-command. Keep the first
-        # non-empty line — the closest thing to a title in that response.
-        title = next((line.strip() for line in title.splitlines() if line.strip()), "")
-        # Enforce reasonable length
-        if len(title) > 80:
-            title = title[:77] + "..."
-        return title if title else None
+        return _clean_title(_extract_title_text(content))
     except Exception as e:
         # Log at WARNING so this shows up in agent.log without debug mode.
         # Full detail at debug level for operators who need the stack.
@@ -186,40 +392,38 @@ def generate_title(
         return None
 
 
-def _persist_session_title(session_db, session_id, title):
-    """Persist a generated title, recovering from duplicate-title collisions.
+def _persist_session_title(session_db, session_id, title, *, source):
+    """Persist a title at *source* authority, recovering from name collisions.
 
-    The write goes through ``set_auto_title_if_empty`` (predicate + write in
-    one transaction) so a manual ``/title`` set while LLM generation was in
-    flight is never overwritten — a plain ``set_session_title`` fallback keeps
-    older stores working. ``set_session_title`` raises ValueError when the
-    title would collide with another session (the unique-title index). Rather
-    than swallow it and leave the session untitled (#50537), append a #N
-    suffix via get_next_title_in_lineage() when the store supports lineage
-    dedup; otherwise re-raise so the caller can decide.
+    The write goes through ``set_auto_title`` (precedence check + write in one
+    transaction) so a manual ``/title`` set while generation was in flight is
+    never overwritten. ``ValueError`` means the name is taken by an unrelated
+    session (the unique-title index); rather than leave the session untitled
+    (#50537), append a ``#N`` suffix via ``get_next_title_in_lineage``.
 
-    Returns the title actually persisted, or None when a concurrent manual
-    title won the race (nothing was written).
+    Returns the title actually persisted, or None when a higher-authority
+    title already held the row (nothing was written).
     """
-    atomic_fn = getattr(session_db, "set_auto_title_if_empty", None)
+    auto_fn = getattr(session_db, "set_auto_title", None)
 
-    def _set(t):
-        if atomic_fn is not None:
-            if not atomic_fn(session_id, t):
-                # Predicate failed: a title appeared while generation was in
-                # flight (manual /title wins), or the session vanished.
+    def _set(candidate):
+        if auto_fn is not None:
+            if not auto_fn(session_id, candidate, source=source):
                 logger.debug(
-                    "Skipping auto-generated session title because a title "
-                    "was set while generation was in flight"
+                    "Skipping %s title: a higher-authority title already holds "
+                    "session %s",
+                    source, session_id,
                 )
                 return None
-            return t
-        ok = session_db.set_session_title(session_id, t)
+            return candidate
+        # Older store without provenance support.
+        legacy_fn = getattr(session_db, "set_auto_title_if_empty", None)
+        if legacy_fn is not None:
+            return candidate if legacy_fn(session_id, candidate) else None
+        ok = session_db.set_session_title(session_id, candidate)
         if ok is False:
-            raise RuntimeError(
-                f"session {session_id} not found when storing title"
-            )
-        return t
+            raise RuntimeError(f"session {session_id} not found when storing title")
+        return candidate
 
     try:
         return _set(title)
@@ -233,22 +437,54 @@ def _persist_session_title(session_db, session_id, title):
         return _set(deduped)
 
 
+def apply_instant_title(
+    session_db,
+    session_id: str,
+    user_message: str,
+    title_callback: Optional[TitleCallback] = None,
+) -> Optional[str]:
+    """Write the derived title synchronously. Cheap enough to run inline.
+
+    Returns the title written, or None when nothing was written (no usable
+    text, or the session already carries a title of at least ``derived``
+    authority). Never raises: a titling failure must not affect the turn.
+    """
+    if not session_db or not session_id:
+        return None
+    try:
+        if not is_titleable_user_message(user_message):
+            return None
+        title = derive_title(user_message)
+        if not title:
+            return None
+        persisted = _persist_session_title(
+            session_db, session_id, title, source="derived"
+        )
+        if persisted and title_callback is not None:
+            try:
+                title_callback(persisted)
+            except Exception:
+                logger.debug("Instant-title callback failed", exc_info=True)
+        return persisted
+    except Exception:
+        logger.debug("Instant title failed", exc_info=True)
+        return None
+
+
 def auto_title_session(
     session_db,
     session_id: str,
     user_message: str,
-    assistant_response: str,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
     title_callback: Optional[TitleCallback] = None,
     runtime_validator: Optional[RuntimeValidator] = None,
 ) -> None:
-    """Generate and set a session title if one doesn't already exist.
+    """Generate and store the model title for a session.
 
-    Called in a background thread after the first exchange completes.
-    Silently skips if:
+    Called on a background thread. Silently skips if:
     - session_db is None
-    - session already has a title (user-set or previously auto-generated)
+    - the session already carries an ``llm`` or ``user`` title
     - title generation fails
     - runtime_validator returns False (model was switched)
 
@@ -266,7 +502,6 @@ def auto_title_session(
             session_db,
             session_id,
             user_message,
-            assistant_response,
             failure_callback=failure_callback,
             main_runtime=main_runtime,
             title_callback=title_callback,
@@ -292,7 +527,6 @@ def _auto_title_session(
     session_db,
     session_id: str,
     user_message: str,
-    assistant_response: str,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
     title_callback: Optional[TitleCallback] = None,
@@ -302,10 +536,15 @@ def _auto_title_session(
     if not session_db or not session_id:
         return
 
-    # Check if title already exists (user may have set one via /title before first response)
+    # Skip when a title of at least LLM authority is already stored. A derived
+    # title is expected here — upgrading it is the whole point of this call.
     try:
-        existing = session_db.get_session_title(session_id)
-        if existing:
+        source_fn = getattr(session_db, "get_session_title_source", None)
+        if source_fn is not None:
+            existing_source = source_fn(session_id)
+            if existing_source is not None and existing_source != "derived":
+                return
+        elif session_db.get_session_title(session_id):
             return
     except Exception:
         return
@@ -314,8 +553,7 @@ def _auto_title_session(
     # conversation context was reset, so publish it here from the session id
     # we already hold — the title-generation LLM call then carries the same
     # ``conversation=`` Portal tag as the turn it titles. Root-of-lineage for
-    # consistency with the agent loop (a no-op on first exchange, where
-    # titling happens, but correct if this ever runs on a continuation).
+    # consistency with the agent loop.
     from agent.aux_accounting import set_accounting_context
     from agent.portal_tags import set_conversation_context
 
@@ -331,7 +569,6 @@ def _auto_title_session(
 
     title = generate_title(
         user_message,
-        assistant_response,
         failure_callback=failure_callback,
         main_runtime=main_runtime,
         runtime_validator=runtime_validator,
@@ -340,7 +577,9 @@ def _auto_title_session(
         return
 
     try:
-        persisted = _persist_session_title(session_db, session_id, title)
+        persisted = _persist_session_title(
+            session_db, session_id, title, source="llm"
+        )
         if persisted is None:
             return
         logger.debug("Auto-generated session title: %s", persisted)
@@ -357,39 +596,51 @@ def maybe_auto_title(
     session_db,
     session_id: str,
     user_message: str,
-    assistant_response: str,
-    conversation_history: list,
+    conversation_history: Optional[list] = None,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
     title_callback: Optional[TitleCallback] = None,
     runtime_validator: Optional[RuntimeValidator] = None,
 ) -> None:
-    """Fire-and-forget title generation after the first exchange.
+    """Title a session from its opening message: instant, then upgraded.
 
-    Only generates a title when:
-    - This appears to be the first user→assistant exchange
-    - No title is already set
+    Call this at the START of a turn, before the model is invoked. The derived
+    title is written inline (sub-millisecond) and the model upgrade is forked
+    onto a daemon thread, so nothing here is on the critical path.
+
+    Only acts on the session's opening exchange, and only when the message
+    carries real user intent (machine-authored compaction handoffs are skipped).
     """
-    if not session_db or not session_id or not user_message or not assistant_response:
+    if not session_db or not session_id or not user_message:
         return
 
-    # Count user messages in history to detect first exchange.
-    # conversation_history includes the exchange that just happened,
-    # so for a first exchange we expect exactly 1 user message
-    # (or 2 counting system). Be generous: generate on first 2 exchanges.
-    user_msg_count = sum(1 for m in (conversation_history or []) if m.get("role") == "user")
-    if user_msg_count > 2:
+    # Count user messages to detect the opening turn. ``conversation_history``
+    # is the state BEFORE this turn's message is appended when called from the
+    # turn prologue, and after it when called post-response, so accept both.
+    # Entries are dicts; anything else means a caller passed the wrong
+    # positional and titling must degrade quietly rather than raise.
+    user_msg_count = sum(
+        1
+        for m in (conversation_history or [])
+        if isinstance(m, dict) and m.get("role") == "user"
+    )
+    if user_msg_count > 1:
         return
 
-    # Config read comes after the cheap first-exchange guard so the file
-    # isn't touched on every subsequent turn of a long session.
+    if not is_titleable_user_message(user_message):
+        return
+
+    # Config read comes after the cheap guards so the file isn't touched on
+    # every subsequent turn of a long session.
     if not _auto_title_enabled():
         logger.debug("Auto-title skipped: auxiliary.title_generation.enabled=false")
         return
 
+    apply_instant_title(session_db, session_id, user_message, title_callback)
+
     thread = threading.Thread(
         target=auto_title_session,
-        args=(session_db, session_id, user_message, assistant_response),
+        args=(session_db, session_id, user_message),
         kwargs={
             "failure_callback": failure_callback,
             "main_runtime": main_runtime,

agent/turn_context.py

@@ -170,6 +170,75 @@ def append_notes_to_multimodal_content(content: Any, notes: str) -> bool:
         return False
 
 
+def _maybe_title_session_at_turn_start(agent: Any, messages: List[Any]) -> None:
+    """Kick off auto-titling for this session's first user message.
+
+    Called from the turn prologue, so every surface (CLI, gateway, TUI/desktop,
+    ACP) gets identical behavior without each one re-implementing the call.
+    Fully defensive: titling is cosmetic and must never break a turn.
+    """
+    session_db = getattr(agent, "_session_db", None)
+    session_id = getattr(agent, "session_id", None)
+    if not session_db or not session_id:
+        return
+
+    try:
+        from agent.message_content import flatten_message_text
+        from agent.title_generator import maybe_auto_title
+
+        # The turn's own user message, as text. Multimodal turns flatten to
+        # their text parts; an image-only turn yields "" and is skipped, since
+        # there is nothing to title from.
+        user_text = ""
+        for msg in reversed(messages or []):
+            if isinstance(msg, dict) and msg.get("role") == "user":
+                user_text = flatten_message_text(msg.get("content")).strip()
+                break
+        if not user_text:
+            return
+
+        # The session row is created lazily on the first persist, which happens
+        # later in the turn. Force it now, or the title write matches zero rows
+        # and the session stays untitled for the whole turn anyway.
+        if not getattr(agent, "_session_db_created", False):
+            ensure = getattr(agent, "_ensure_db_session", None)
+            if callable(ensure):
+                ensure()
+            if not getattr(agent, "_session_db_created", False):
+                return
+
+        # Snapshot the runtime identity; the validator lets the background
+        # titler skip its LLM call if the user switches models before it fires
+        # (a stale request would reload an unloaded Ollama model, #19027).
+        _model = getattr(agent, "model", None)
+        _provider = getattr(agent, "provider", None)
+
+        maybe_auto_title(
+            session_db,
+            session_id,
+            user_text,
+            conversation_history=messages,
+            failure_callback=(
+                getattr(agent, "_title_failure_callback", None)
+                or getattr(agent, "_emit_auxiliary_failure", None)
+            ),
+            main_runtime={
+                "model": _model,
+                "provider": _provider,
+                "base_url": getattr(agent, "base_url", None),
+                "api_key": getattr(agent, "api_key", None),
+                "api_mode": getattr(agent, "api_mode", None),
+            },
+            title_callback=getattr(agent, "_on_session_title", None),
+            runtime_validator=lambda: (
+                getattr(agent, "model", None) == _model
+                and getattr(agent, "provider", None) == _provider
+            ),
+        )
+    except Exception:
+        logger.debug("Turn-start auto-title dispatch failed", exc_info=True)
+
+
 def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> int:
     """Locate this turn's user message after compaction rebuilt ``messages``.
 
@@ -1265,6 +1334,16 @@ def build_turn_context(
         if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"):
             agent._pending_cli_user_message = None
 
+    # Title the session from this user message, now — the row exists and the
+    # turn has not called the model yet. Titling is derived from the user's
+    # ask alone, so it runs concurrently with the turn instead of waiting for
+    # a final response; on a long tool-heavy first turn that is the difference
+    # between a title in ~1s and a title minutes later (or never, when the
+    # turn failed before producing one). Fire-and-forget on a daemon thread,
+    # a no-op once the session has a title, and shared by every surface
+    # because every surface enters the turn through this prologue.
+    _maybe_title_session_at_turn_start(agent, messages)
+
     return TurnContext(
         user_message=user_message,
         original_user_message=original_user_message,

cli.py

@@ -14324,44 +14324,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
             # Get the final response
             response = result.get("final_response", "") if result else ""
 
-            # Auto-generate session title after first exchange (non-blocking)
-            if response and result and not result.get("failed") and not result.get("partial"):
-                try:
-                    from agent.title_generator import maybe_auto_title
-                    # Route title-generation failures through the agent's
-                    # user-visible warning channel so a depleted auxiliary
-                    # provider doesn't silently leave sessions untitled
-                    # (issue #15775).
-                    _title_failure_cb = getattr(
-                        self.agent, "_emit_auxiliary_failure", None
-                    ) if self.agent else None
-                    # Snapshot the runtime identity; the validator lets the
-                    # background titler skip its LLM call if the user switches
-                    # models before it fires (a stale request would reload an
-                    # unloaded Ollama model, #19027).
-                    _title_model = self.model
-                    _title_provider = self.provider
-                    maybe_auto_title(
-                        self._session_db,
-                        self.session_id,
-                        message,
-                        response,
-                        self.conversation_history,
-                        failure_callback=_title_failure_cb,
-                        main_runtime={
-                            "model": self.model,
-                            "provider": self.provider,
-                            "base_url": self.base_url,
-                            "api_key": self.api_key,
-                            "api_mode": self.api_mode,
-                        },
-                        runtime_validator=lambda: (
-                            getattr(self, "model", None) == _title_model
-                            and getattr(self, "provider", None) == _title_provider
-                        ),
-                    )
-                except Exception:
-                    pass
+            # Session titling now runs at TURN START (agent/turn_context.py)
+            # from the user's message alone, so it is already done — or in
+            # flight — by the time we get here, instead of waiting on a final
+            # response that a failed or interrupted turn never produces.
 
             # Handle failed or partial results (e.g., non-retryable errors, rate limits,
             # truncated output, invalid tool calls). Both "failed" and "partial" with

gateway/run.py

@@ -4418,6 +4418,56 @@ class TurnRunner:
         except Exception as _e:
             logger.debug("event_callback hook error: %s", _e)
 
+    def _attach_session_title_callback(self, agent, ctx) -> None:
+        """Wire the platform thread-rename lane onto the agent as `_on_session_title`.
+
+        The session titler runs inside the turn prologue now (it derives the
+        title from the user's first message, so it no longer needs the
+        response), which means the callback has to be attached before the run
+        rather than registered after it. The lane predicates and their
+        rationale are unchanged from the old post-response registration.
+        """
+        try:
+            # Gateway auto-title failures must NOT be surfaced as user-visible
+            # messages (#23246) — they are not actionable to the end user.
+            # Overriding the failure sink here keeps CLI mode on the agent's
+            # _emit_auxiliary_failure path while the gateway logs at debug.
+            def _title_failure_cb(task: str, exc: BaseException) -> None:
+                logger.debug(
+                    "Gateway auto-title failure suppressed (not user-visible): %s: %s",
+                    task, exc,
+                )
+
+            agent._title_failure_callback = _title_failure_cb
+
+            session_id = getattr(agent, "session_id", None)
+            source = ctx.source
+
+            if self._runner._is_telegram_topic_lane(source):
+                agent._on_session_title = lambda title: (
+                    self._runner._schedule_telegram_topic_title_rename(
+                        source, session_id, title,
+                    )
+                )
+            elif self._runner._is_discord_auto_thread_lane(source) or (
+                self._runner._is_relay_discord_channel_lane(source)
+            ):
+                # Relay note: the second predicate is shape-only (relay
+                # Discord channel event). Whether the connector actually
+                # auto-threaded our reply is only knowable AFTER delivery
+                # (send-result feedback), so the callback must be registered
+                # eagerly and the rename lane performs the cache lookup at
+                # fire time (staging repro 2026-07-31: gating registration on
+                # the cache read meant it never registered and no
+                # thread_rename op was ever sent).
+                agent._on_session_title = lambda title: (
+                    self._runner._schedule_discord_semantic_thread_rename(
+                        source, session_id, title,
+                    )
+                )
+        except Exception:
+            logger.debug("Failed to attach session title callback", exc_info=True)
+
     def _status_callback_sync(self, event_type: str, message: str) -> None:
         ctx = self._ctx
         if not ctx._status_adapter or not ctx._run_still_current():
@@ -5136,6 +5186,10 @@ class TurnRunner:
         agent.thinking_progress = ctx._thinking_enabled
         # Store agent reference for interrupt support
         ctx.agent_holder[0] = agent
+        # Wire the platform thread-rename lane onto the agent, because the
+        # session titler now fires from the turn prologue rather than after
+        # the response. Titles are pushed here the moment they land.
+        self._attach_session_title_callback(agent, ctx)
         # Publish turn ownership for explicit /stop, /new, disconnect, and
         # shutdown interrupts. Older session processes are outside this
         # baseline and remain alive.
@@ -5707,74 +5761,13 @@ class TurnRunner:
                     unique_tags.insert(0, "[[audio_as_voice]]")
                 final_response = final_response + "\n" + "\n".join(unique_tags)
 
-        # Auto-generate session title after first exchange (non-blocking)
-        if final_response and self._runner._session_db:
-            try:
-                from agent.title_generator import maybe_auto_title
-                all_msgs = ctx.result_holder[0].get("messages", []) if ctx.result_holder[0] else []
-                # In Gateway mode, auto-title failures must NOT be
-                # surfaced as user-visible messages (fixes #23246).
-                # Log them at debug level only — they are not actionable
-                # to the end user. CLI mode keeps the existing behaviour
-                # via the agent's _emit_auxiliary_failure path.
-                def _title_failure_cb(task: str, exc: BaseException) -> None:
-                    logger.debug(
-                        "Gateway auto-title failure suppressed (not user-visible): %s: %s",
-                        task, exc,
-                    )
-                # Snapshot the runtime identity; the validator lets the
-                # background titler skip its LLM call if the session's
-                # model changed before it fires (a stale request would
-                # reload an unloaded Ollama model, #19027).
-                _title_model = getattr(agent, "model", None) if agent else None
-                _title_provider = getattr(agent, "provider", None) if agent else None
-                maybe_auto_title_kwargs = {
-                    "failure_callback": _title_failure_cb,
-                    "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),
-                    } if agent else None,
-                    "runtime_validator": (lambda: (
-                        getattr(agent, "model", None) == _title_model
-                        and getattr(agent, "provider", None) == _title_provider
-                    )) if agent else None,
-                }
-                if self._runner._is_telegram_topic_lane(ctx.source):
-                    maybe_auto_title_kwargs["title_callback"] = lambda title: self._runner._schedule_telegram_topic_title_rename(
-                        ctx.source,
-                        effective_session_id,
-                        title,
-                    )
-                elif self._runner._is_discord_auto_thread_lane(ctx.source) or (
-                    self._runner._is_relay_discord_channel_lane(ctx.source)
-                ):
-                    # Relay note: the second predicate is shape-only (relay
-                    # Discord channel event). Whether the connector actually
-                    # auto-threaded our reply is only knowable AFTER delivery
-                    # (send-result feedback), which on the non-streaming lane
-                    # happens after this registration runs — so the callback
-                    # must be registered eagerly and the rename lane performs
-                    # the cache lookup at fire time (staging repro 2026-07-31:
-                    # gating registration on the cache read meant it never
-                    # registered and no thread_rename op was ever sent).
-                    maybe_auto_title_kwargs["title_callback"] = lambda title: self._runner._schedule_discord_semantic_thread_rename(
-                        ctx.source,
-                        effective_session_id,
-                        title,
-                    )
-                maybe_auto_title(
-                    getattr(self._runner._session_db, "_db", self._runner._session_db),
-                    effective_session_id,
-                    ctx.message,
-                    final_response,
-                    all_msgs,
-                    **maybe_auto_title_kwargs,
-                )
-            except Exception:
-                pass
+        # Auto-titling runs at TURN START (agent/turn_context.py) from the
+        # user's message alone, so it no longer waits on final_response — a
+        # failed or interrupted turn still gets a titled session. The
+        # platform-specific thread-rename callbacks are attached to the agent
+        # 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.
 
         return {
             "final_response": final_response,

hermes_cli/sessions_cmd.py

@@ -950,8 +950,7 @@ def cmd_sessions(args, sessions_parser=None):
         for row in candidates:
             session_id = row["id"]
             typed = describe_skill_invocation(row["content"]) or ""
-            first_reply = db.get_first_assistant_text(session_id) or ""
-            new_title = generate_title(typed, first_reply)
+            new_title = generate_title(typed)
             if not new_title or new_title == row["title"]:
                 continue
             if not _is_titlelike(new_title):

hermes_state.py

@@ -5897,6 +5897,33 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
     # Maximum length for session titles
     MAX_TITLE_LENGTH = 100
 
+    # Title provenance, lowest to highest authority. An auto-titling write may
+    # only replace a title of strictly lower authority, so the instant
+    # ``derived`` title upgrades to the model's ``llm`` title exactly once and
+    # nothing the agent generates can ever clobber a name the user typed.
+    TITLE_SOURCE_DERIVED = "derived"
+    TITLE_SOURCE_LLM = "llm"
+    TITLE_SOURCE_USER = "user"
+    _TITLE_SOURCE_RANK = {
+        TITLE_SOURCE_DERIVED: 0,
+        TITLE_SOURCE_LLM: 1,
+        TITLE_SOURCE_USER: 2,
+    }
+
+    @classmethod
+    def _title_rank(cls, source: Optional[str]) -> int:
+        """Rank a stored title_source. NULL means a pre-provenance row.
+
+        Rows written before this column existed carry NULL. They were almost
+        always set by the old auto-titler, but a manual ``/title`` from that
+        era is indistinguishable — so treat NULL as ``user`` and refuse to
+        overwrite it. Auto-titling only ever fills genuinely empty titles on
+        legacy rows, which is the conservative direction.
+        """
+        if source is None:
+            return cls._TITLE_SOURCE_RANK[cls.TITLE_SOURCE_USER]
+        return cls._TITLE_SOURCE_RANK.get(str(source), 0)
+
     @staticmethod
     def sanitize_title(title: Optional[str]) -> Optional[str]:
         """Validate and sanitize a session title.
@@ -5987,17 +6014,35 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         session_id: str,
         title: str,
         *,
-        only_if_empty: bool,
+        source: str,
     ) -> bool:
+        """Write a title, enforcing provenance precedence.
+
+        ``source`` is one of ``TITLE_SOURCE_{DERIVED,LLM,USER}``. A ``user``
+        write always lands — an explicit rename is authoritative. An automatic
+        write (``derived``/``llm``) lands only when the row is untitled or the
+        stored title has strictly lower authority, so the instant ``derived``
+        title upgrades to ``llm`` exactly once and neither can ever overwrite a
+        name the user typed. Re-running the titler on an already-``llm`` row is
+        a no-op, which is what stops a session renaming itself.
+
+        The read and the write are one compare-and-swap inside a single
+        transaction, so a manual ``/title`` racing an in-flight generation
+        cannot be clobbered by the late arrival.
+        """
         title = self.sanitize_title(title)
+        is_user = source == self.TITLE_SOURCE_USER
+        new_rank = self._title_rank(source) if not is_user else None
 
         def _do(conn):
-            if only_if_empty:
-                current = conn.execute(
-                    "SELECT title FROM sessions WHERE id = ?",
-                    (session_id,),
-                ).fetchone()
-                if current is None or current["title"] is not None:
+            current = conn.execute(
+                "SELECT title, title_source FROM sessions WHERE id = ?",
+                (session_id,),
+            ).fetchone()
+            if current is None:
+                return 0
+            if not is_user and current["title"] is not None:
+                if self._title_rank(current["title_source"]) >= new_rank:
                     return 0
 
             if title:
@@ -6031,10 +6076,19 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         raise ValueError(
                             f"Title '{title}' is already in use by session {conflict_id}"
                         )
-            predicate = " AND title IS NULL" if only_if_empty else ""
+            # Compare-and-swap on the exact values we just read (``IS`` is
+            # NULL-safe in SQLite), so a concurrent write between the SELECT
+            # and here loses instead of being silently overwritten.
             cursor = conn.execute(
-                f"UPDATE sessions SET title = ? WHERE id = ?{predicate}",
-                (title, session_id),
+                "UPDATE sessions SET title = ?, title_source = ? "
+                "WHERE id = ? AND title IS ? AND title_source IS ?",
+                (
+                    title,
+                    source if title else None,
+                    session_id,
+                    current["title"],
+                    current["title_source"],
+                ),
             )
             return cursor.rowcount
 
@@ -6042,23 +6096,40 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         return rowcount > 0
 
     def set_session_title(self, session_id: str, title: str) -> bool:
-        """Set or update a session's title.
+        """Set or update a session's title on the user's behalf.
 
         Returns True if session was found and title was set.
         Raises ValueError if title is already in use by another session,
         or if the title fails validation (too long, invalid characters).
         Empty/whitespace-only strings are normalized to None (clearing the title).
+
+        This records ``user`` provenance, so auto-titling will never replace
+        the result. Automatic callers must use :meth:`set_auto_title`.
         """
-        return self._set_session_title(session_id, title, only_if_empty=False)
+        return self._set_session_title(
+            session_id, title, source=self.TITLE_SOURCE_USER
+        )
+
+    def set_auto_title(self, session_id: str, title: str, *, source: str) -> bool:
+        """Set an automatically generated title, honoring provenance precedence.
+
+        Returns True when the title was written, False when a higher-authority
+        title already holds the row (nothing is modified in that case).
+        """
+        if source not in (self.TITLE_SOURCE_DERIVED, self.TITLE_SOURCE_LLM):
+            raise ValueError(f"invalid automatic title source: {source!r}")
+        return self._set_session_title(session_id, title, source=source)
 
     def set_auto_title_if_empty(self, session_id: str, title: str) -> bool:
-        """Set an auto-generated title only when the current title is NULL.
+        """Back-compat shim: set an LLM title only if nothing better exists.
 
-        The predicate and write run in one transaction so a concurrent manual
-        rename cannot be overwritten. Validation and uniqueness behavior match
-        :meth:`set_session_title`.
+        Retained because older callers (and third-party plugins) reference it
+        by name. New code should call :meth:`set_auto_title` with an explicit
+        source.
         """
-        return self._set_session_title(session_id, title, only_if_empty=True)
+        return self.set_auto_title(
+            session_id, title, source=self.TITLE_SOURCE_LLM
+        )
 
     def get_session_title(self, session_id: str) -> Optional[str]:
         """Get the title for a session, or None."""
@@ -6069,6 +6140,38 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             row = cursor.fetchone()
         return row["title"] if row else None
 
+    def get_session_title_source(self, session_id: str) -> Optional[str]:
+        """Get the provenance of a session's title, or None when untitled."""
+        with self._lock:
+            cursor = self._conn.execute(
+                "SELECT title, title_source FROM sessions WHERE id = ?",
+                (session_id,),
+            )
+            row = cursor.fetchone()
+        if not row or row["title"] is None:
+            return None
+        return row["title_source"]
+
+    def set_session_title_source(self, session_id: str, source: str) -> bool:
+        """Overwrite a title's provenance without touching the title text.
+
+        Used when a title is carried across a session boundary (compression
+        rotation) and the copy must keep the original's authority rather than
+        the authority of whichever setter performed the copy.
+        """
+        if source not in self._TITLE_SOURCE_RANK:
+            raise ValueError(f"invalid title source: {source!r}")
+
+        def _do(conn):
+            cursor = conn.execute(
+                "UPDATE sessions SET title_source = ? "
+                "WHERE id = ? AND title IS NOT NULL",
+                (source, session_id),
+            )
+            return cursor.rowcount
+
+        return self._execute_write(_do) > 0
+
     def set_session_archived(self, session_id: str, archived: bool) -> bool:
         """Archive or unarchive a session.
 

hermes_state_common.py

@@ -242,6 +242,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     cost_source TEXT,
     pricing_version TEXT,
     title TEXT,
+    title_source TEXT,
     last_activity_at REAL,
     last_activity_description TEXT,
     last_activity_provenance TEXT,

plugins/model-providers/nous/__init__.py

@@ -11,6 +11,22 @@ from providers.base import ProviderProfile
 class NousProfile(ProviderProfile):
     """Nous Portal — product tags, reasoning with Nous-specific omission."""
 
+    def resolve_aux_model(self, *, vision: bool = False) -> str:
+        """Ask the Portal which cheap model it currently recommends.
+
+        ``/api/nous/recommended-models`` is the authoritative, tier-aware
+        source (free vs paid), so the auxiliary fast tier tracks the live
+        catalog instead of a hardcoded id that 404s the day Nous retires it.
+        The underlying fetch is memory- and disk-cached with a last-known-good
+        fallback, so this is cheap to call and safe offline.
+        """
+        try:
+            from hermes_cli.models import get_nous_recommended_aux_model
+
+            return get_nous_recommended_aux_model(vision=vision) or ""
+        except Exception:
+            return ""
+
     def build_extra_body(
         self, *, session_id: str | None = None, **context
     ) -> dict[str, Any]:

providers/base.py

@@ -101,6 +101,22 @@ class ProviderProfile:
 
     # ── Hooks (override in subclass for complex providers) ───
 
+    def resolve_aux_model(self, *, vision: bool = False) -> str:
+        """Return a LIVE cheap-model id for auxiliary tasks, or "".
+
+        ``default_aux_model`` is a hardcoded id in source, so it rots: when the
+        provider retires that model every auxiliary call spends a round-trip
+        404ing before the retry net catches it. Providers that publish a
+        machine-readable recommendation should override this and query it, so
+        the cheap tier tracks the upstream catalog instead of a constant a human
+        has to remember to bump.
+
+        Contract: cheap to call (implementations must cache — this runs on
+        client-resolution paths), never raises, and returns "" when it has no
+        answer so the caller falls through to ``default_aux_model``.
+        """
+        return ""
+
     def get_hostname(self) -> str:
         """Return the provider's base hostname for URL-based detection.
 

tests/agent/test_auxiliary_client.py

@@ -4488,3 +4488,61 @@ class TestAutoRoutedProviderProfileHooks:
         assert relay.call_args_list[1].args[1]["extra_headers"] == {
             "Authorization": "Bearer token-2",
         }
+
+
+class TestFastModelTier:
+    """The titling fast tier: rot-proof resolution, scoped to titling only."""
+
+    def test_catalog_match_prefers_rolling_alias_over_pinned_id(self):
+        """A "-latest" alias wins: it is the only id that cannot go stale."""
+        from agent import auxiliary_client as ac
+
+        catalog = {
+            "z-ai/glm-5.2": {},
+            "openai/gpt-5.4-mini": {},
+            "~openai/gpt-mini-latest": {},
+            "stepfun/step-3.7-flash:free": {},
+        }
+        with patch("hermes_cli.models.fetch_models_with_pricing", return_value=catalog):
+            assert ac._fast_model_from_catalog("nous") == "~openai/gpt-mini-latest"
+
+    def test_catalog_match_skips_reasoning_batch_and_embedding_lookalikes(self):
+        """Substring matching must not pick a thinker, a queue, or an encoder."""
+        from agent import auxiliary_client as ac
+
+        catalog = {
+            "openai/o3-mini": {},
+            "openai/gpt-5.4-mini:batch": {},
+            "sentence-transformers/all-minilm-l6-v2": {},
+            "google/gemini-3.6-flash": {},
+        }
+        with patch("hermes_cli.models.fetch_models_with_pricing", return_value=catalog):
+            assert ac._fast_model_from_catalog("nous") == "google/gemini-3.6-flash"
+
+    def test_falls_back_to_curated_default_when_catalog_unavailable(self):
+        """An offline catalog degrades to the provider's pinned default."""
+        from agent import auxiliary_client as ac
+
+        with patch.object(ac, "_fast_model_from_catalog", return_value=""):
+            assert (
+                ac._get_aux_model_for_provider("anthropic", prefer_fast=True)
+                == ac._get_aux_model_for_provider("anthropic")
+            )
+
+    def test_fast_tier_is_opt_in(self):
+        """Without prefer_fast the resolver must not touch the live catalog."""
+        from agent import auxiliary_client as ac
+
+        with patch.object(ac, "_fast_model_from_catalog") as spy:
+            ac._get_aux_model_for_provider("nous")
+        spy.assert_not_called()
+
+    def test_only_titling_is_in_the_fast_tier(self):
+        """Compression/vision/search keep 'auto means my chat model'."""
+        from agent.auxiliary_client import _FAST_MODEL_TASKS
+
+        assert "title_generation" in _FAST_MODEL_TASKS
+        overlap = {"compression", "vision", "web_extract"}.intersection(
+            _FAST_MODEL_TASKS
+        )
+        assert not overlap

tests/agent/test_title_generator.py

@@ -41,7 +41,7 @@ class TestGenerateTitle:
             return resp
 
         with patch("agent.title_generator.call_llm", side_effect=mock_call_llm):
-            assert generate_title("question", "answer") == "Configured Timeout"
+            assert generate_title("question") == "Configured Timeout"
 
         assert captured_kwargs["task"] == "title_generation"
         assert captured_kwargs["timeout"] is None
@@ -59,7 +59,7 @@ class TestGenerateTitle:
         )
 
         with patch("agent.title_generator.call_llm", return_value=mock_response):
-            title = generate_title("help me fix this import", "Sure...")
+            title = generate_title("help me fix this import")
             assert title == "Debugging Python Import Errors"
             assert "<think>" not in title
             assert "summarize" not in title
@@ -74,7 +74,7 @@ class TestGenerateTitle:
         )
 
         with patch("agent.title_generator.call_llm", return_value=mock_response):
-            title = generate_title("hello", "hi there")
+            title = generate_title("hello")
             # Everything from the unterminated open tag onward is stripped,
             # leaving nothing → None.
             assert title is None
@@ -86,7 +86,7 @@ class TestGenerateTitle:
         mock_response.choices[0].message.content = "A" * 100
 
         with patch("agent.title_generator.call_llm", return_value=mock_response):
-            title = generate_title("question", "answer")
+            title = generate_title("question")
             assert len(title) == 80
             assert title.endswith("...")
 
@@ -143,7 +143,6 @@ class TestAutoTitleSession:
                 db,
                 "sess-1",
                 "hi",
-                "hello",
                 title_callback=seen.append,
             )
 
@@ -152,20 +151,39 @@ class TestAutoTitleSession:
 
     def test_invokes_title_callback_after_setting_title(self):
         db = MagicMock()
-        db.get_session_title.return_value = None
-        db.set_auto_title_if_empty.return_value = True
+        db.get_session_title_source.return_value = None
+        db.set_auto_title.return_value = True
         seen = []
         with patch("agent.title_generator.generate_title", return_value="Readable Session"):
             auto_title_session(
                 db,
                 "sess-1",
                 "hello",
-                "hi there",
                 title_callback=seen.append,
             )
-        db.set_auto_title_if_empty.assert_called_once_with("sess-1", "Readable Session")
+        db.set_auto_title.assert_called_once_with(
+            "sess-1", "Readable Session", source="llm"
+        )
         assert seen == ["Readable Session"]
 
+    def test_upgrades_a_derived_title_but_not_an_llm_one(self, tmp_path):
+        """The instant title is provisional; a model title is final.
+
+        This is the "session renames itself" guard: re-running the titler on a
+        session that already has an LLM title must be a no-op.
+        """
+        db = SessionDB(tmp_path / "state.db")
+        db.create_session(session_id="sess-1", source="cli")
+        db.set_auto_title("sess-1", "fix the flaky auth test", source="derived")
+
+        with patch("agent.title_generator.generate_title", return_value="Fix flaky auth test"):
+            auto_title_session(db, "sess-1", "fix the flaky auth test")
+        assert db.get_session_title("sess-1") == "Fix flaky auth test"
+
+        with patch("agent.title_generator.generate_title", return_value="Totally Different"):
+            auto_title_session(db, "sess-1", "fix the flaky auth test")
+        assert db.get_session_title("sess-1") == "Fix flaky auth test"
+
 
 
     def test_body_exception_routed_to_failure_callback(self):
@@ -179,7 +197,6 @@ class TestAutoTitleSession:
                 db,
                 "sess-1",
                 "hi",
-                "hello",
                 failure_callback=lambda task, exc: seen.append((task, exc)),
             )
         assert seen == [("title generation", boom)]
@@ -190,7 +207,7 @@ class TestMaybeAutoTitle:
     """Tests for maybe_auto_title() — the fire-and-forget entry point."""
 
     def test_skips_if_not_first_exchange(self):
-        """Should not fire for conversations with more than 2 user messages."""
+        """Should not fire once the conversation is past its opening turn."""
         db = MagicMock()
         history = [
             {"role": "user", "content": "first"},
@@ -202,26 +219,25 @@ class TestMaybeAutoTitle:
         ]
 
         with patch("agent.title_generator.auto_title_session") as mock_auto:
-            maybe_auto_title(db, "sess-1", "third", "response 3", history)
+            maybe_auto_title(db, "sess-1", "third", history)
             # Wait briefly for any thread to start
             import time
             time.sleep(0.1)
             mock_auto.assert_not_called()
 
     def test_fires_on_first_exchange(self):
-        """Should fire a background thread for the first exchange."""
+        """Should fire a background thread for the opening message."""
         db = MagicMock()
         db.get_session_title.return_value = None
         history = [
             {"role": "user", "content": "hello"},
-            {"role": "assistant", "content": "hi there"},
         ]
 
         with patch("agent.title_generator.auto_title_session") as mock_auto:
             import threading
             called = threading.Event()
             mock_auto.side_effect = lambda *a, **k: called.set()
-            maybe_auto_title(db, "sess-1", "hello", "hi there", history)
+            maybe_auto_title(db, "sess-1", "hello", history)
             # Event-based wait: sleep-sync flaked when the daemon thread
             # wasn't scheduled within the fixed nap on a loaded runner.
             assert called.wait(timeout=10), "auto_title thread never ran"
@@ -229,13 +245,37 @@ class TestMaybeAutoTitle:
                 db,
                 "sess-1",
                 "hello",
-                "hi there",
                 failure_callback=None,
                 main_runtime=None,
                 title_callback=None,
                 runtime_validator=None,
             )
 
+    def test_writes_instant_title_before_the_model_runs(self, tmp_path):
+        """The derived title lands synchronously — no LLM, no waiting."""
+        db = SessionDB(tmp_path / "state.db")
+        db.create_session(session_id="sess-1", source="cli")
+        with patch("agent.title_generator.auto_title_session"):
+            maybe_auto_title(
+                db, "sess-1", "fix the flaky auth test in login", []
+            )
+        assert db.get_session_title("sess-1") == "fix the flaky auth test in login"
+        assert db.get_session_title_source("sess-1") == "derived"
+
+    def test_skips_machine_authored_opening_messages(self, tmp_path):
+        """A compaction handoff is not a user request and must not title."""
+        db = SessionDB(tmp_path / "state.db")
+        db.create_session(session_id="sess-1", source="cli")
+        with patch("agent.title_generator.auto_title_session") as mock_auto:
+            maybe_auto_title(
+                db,
+                "sess-1",
+                "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted",
+                [],
+            )
+        assert db.get_session_title("sess-1") is None
+        mock_auto.assert_not_called()
+
 
 
 
@@ -246,18 +286,18 @@ class TestAutoTitleDuplicateHandling:
 
     def test_dedupes_duplicate_title_via_lineage(self):
         db = MagicMock()
-        db.get_session_title.return_value = None
+        db.get_session_title_source.return_value = None
         # Atomic write path: collision raises ValueError, retry persists.
-        db.set_auto_title_if_empty.side_effect = [ValueError("in use"), True]
+        db.set_auto_title.side_effect = [ValueError("in use"), True]
         db.get_next_title_in_lineage.return_value = "Debugging Import Error #2"
         with patch(
             "agent.title_generator.generate_title",
             return_value="Debugging Import Error",
         ):
             seen = []
-            auto_title_session(db, "sess-1", "hi", "hello", title_callback=seen.append)
+            auto_title_session(db, "sess-1", "hi", title_callback=seen.append)
         db.get_next_title_in_lineage.assert_called_once_with("Debugging Import Error")
-        assert db.set_auto_title_if_empty.call_args_list[-1][0] == (
+        assert db.set_auto_title.call_args_list[-1][0] == (
             "sess-1",
             "Debugging Import Error #2",
         )
@@ -267,12 +307,14 @@ class TestAutoTitleDuplicateHandling:
 
 
     def test_manual_title_race_skips_without_callback(self):
-        # Atomic predicate fails (manual /title landed while generation was in
+        # Precedence check fails (manual /title landed while generation was in
         # flight) -> nothing persisted, no callback fired.
         from agent.title_generator import _persist_session_title
         db = MagicMock()
-        db.set_auto_title_if_empty.return_value = False
-        assert _persist_session_title(db, "sess-1", "Some Title") is None
+        db.set_auto_title.return_value = False
+        assert (
+            _persist_session_title(db, "sess-1", "Some Title", source="llm") is None
+        )
         db.set_session_title.assert_not_called()
 
 
@@ -314,7 +356,7 @@ class TestRuntimeValidator:
             import threading
             called = threading.Event()
             mock_auto.side_effect = lambda *a, **k: called.set()
-            maybe_auto_title(db, "sess-1", "hello", "hi there", history, runtime_validator=_v)
+            maybe_auto_title(db, "sess-1", "hello", history, runtime_validator=_v)
             assert called.wait(timeout=10), "auto_title thread never ran"
             kwargs = mock_auto.call_args.kwargs
             assert kwargs["runtime_validator"] is _v

tests/run_agent/test_in_place_compaction.py

@@ -185,13 +185,16 @@ class TestRotationFallbackWhenFlagOff:
 
             # Identity rotated to a fresh id.
             assert agent.session_id != sid
-            # Old session ended via compression; continuation forked + renamed.
+            # Old session ended via compression; continuation forked and
+            # carries the SAME name. Compression is an internal detail — the
+            # conversation didn't change topic, so it must not be renumbered
+            # into "my-research #2" and shown as a separate piece of work.
             assert db.get_session(sid)["end_reason"] == "compression"
             child = db._conn.execute(
                 "SELECT id, title FROM sessions WHERE parent_session_id = ?", (sid,)
             ).fetchall()
             assert len(child) == 1
-            assert child[0]["title"] == "my-research #2"
+            assert child[0]["title"] == "my-research"
             # The compacted child is persisted atomically at the rotation
             # boundary, so a headless process killed before finalization can
             # still resume it without duplicating the two handoff messages.

tests/test_tui_gateway_server.py

@@ -12193,8 +12193,13 @@ class _ImmediateThread:
         self._target()
 
 
-def test_prompt_submit_auto_titles_session_on_complete(monkeypatch):
-    """maybe_auto_title is called after a successful (complete) prompt."""
+def test_prompt_submit_wires_live_title_rename_callback(monkeypatch):
+    """The gateway hands the agent a hook so a new title repaints the sidebar.
+
+    Titling itself moved into the shared turn prologue (agent/turn_context.py),
+    so the gateway's only remaining job is delivering the rename event. Asserted
+    by calling the hook the gateway installed and checking what it emits.
+    """
 
     class _Agent:
         model = "gpt-5.6-sol"
@@ -12212,93 +12217,32 @@ def test_prompt_submit_auto_titles_session_on_complete(monkeypatch):
                 ],
             }
 
-    server._sessions["sid"] = _session(agent=_Agent())
+    agent = _Agent()
+    server._sessions["sid"] = _session(agent=agent)
+    emitted = []
     monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
-    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
+    monkeypatch.setattr(
+        server, "_emit", lambda kind, sid, payload=None, **kw: emitted.append((kind, payload))
+    )
     monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
     monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
     monkeypatch.setattr(server, "_get_db", lambda: None)
 
-    with patch("agent.title_generator.maybe_auto_title") as mock_title:
-        server.handle_request(
-            {
-                "id": "1",
-                "method": "prompt.submit",
-                "params": {"session_id": "sid", "text": "Tell me about Rome"},
-            }
-        )
+    server.handle_request(
+        {
+            "id": "1",
+            "method": "prompt.submit",
+            "params": {"session_id": "sid", "text": "Tell me about Rome"},
+        }
+    )
 
-    mock_title.assert_called_once()
-    args = mock_title.call_args.args
-    assert args[1] == "session-key"
-    assert args[2] == "Tell me about Rome"
-    assert args[3] == "Rome was founded in 753 BC."
-    assert mock_title.call_args.kwargs["main_runtime"] == {
-        "model": "gpt-5.6-sol",
-        "provider": "openai-codex",
-        "base_url": "https://chatgpt.example.test/backend-api/codex",
-        "api_key": _Agent.api_key,
-        "api_mode": "codex_responses",
-    }
-
-
-def test_prompt_submit_skips_auto_title_when_interrupted(monkeypatch):
-    """maybe_auto_title must NOT be called when the agent was interrupted."""
-
-    class _Agent:
-        def run_conversation(self, prompt, conversation_history=None, stream_callback=None, **_kwargs):
-            return {
-                "final_response": "partial answer",
-                "interrupted": True,
-                "messages": [],
-            }
-
-    server._sessions["sid"] = _session(agent=_Agent())
-    monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
-    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
-    monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
-    monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
-    monkeypatch.setattr(server, "_get_db", lambda: None)
-
-    with patch("agent.title_generator.maybe_auto_title") as mock_title:
-        server.handle_request(
-            {
-                "id": "1",
-                "method": "prompt.submit",
-                "params": {"session_id": "sid", "text": "Tell me about Rome"},
-            }
-        )
-
-    mock_title.assert_not_called()
-
-
-def test_prompt_submit_skips_auto_title_when_response_empty(monkeypatch):
-    """maybe_auto_title must NOT be called when the agent returns an empty reply."""
-
-    class _Agent:
-        def run_conversation(self, prompt, conversation_history=None, stream_callback=None, **_kwargs):
-            return {
-                "final_response": "",
-                "messages": [],
-            }
-
-    server._sessions["sid"] = _session(agent=_Agent())
-    monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
-    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
-    monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
-    monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
-    monkeypatch.setattr(server, "_get_db", lambda: None)
-
-    with patch("agent.title_generator.maybe_auto_title") as mock_title:
-        server.handle_request(
-            {
-                "id": "1",
-                "method": "prompt.submit",
-                "params": {"session_id": "sid", "text": "Tell me about Rome"},
-            }
-        )
-
-    mock_title.assert_not_called()
+    hook = getattr(agent, "_on_session_title", None)
+    assert callable(hook), "gateway did not install a live title-rename hook"
+    hook("Founding of Rome")
+    assert (
+        "session.title",
+        {"session_id": "session-key", "title": "Founding of Rome"},
+    ) in emitted
 
 
 def test_prompt_submit_surfaces_backend_error_as_visible_text(monkeypatch):

tui_gateway/server.py

@@ -9849,6 +9849,14 @@ def _run_prompt_submit(
             if display_kind and "persist_user_display_kind" in _run_params:
                 run_kwargs["persist_user_display_kind"] = display_kind
                 run_kwargs["persist_user_display_metadata"] = display_metadata
+            # Auto-titling now fires inside the turn prologue (shared by every
+            # surface). Hand the agent this session's live-rename hook so the
+            # sidebar repaints the moment a title lands, rather than waiting
+            # for the next list refresh.
+            _title_key = session.get("session_key") or sid
+            agent._on_session_title = lambda t, _k=_title_key: _emit(
+                "session.title", sid, {"session_id": _k, "title": t}
+            )
             result = agent.run_conversation(run_message, **run_kwargs)
             if display_kind and isinstance(text, str):
                 db = getattr(agent, "_session_db", None)
@@ -10132,53 +10140,6 @@ def _run_prompt_submit(
                     # Transient DB failure — keep pending_title for retry.
                     pass
 
-            if (
-                status == "complete"
-                and isinstance(raw, str)
-                and raw.strip()
-                and isinstance(text, str)
-                and text.strip()
-            ):
-                try:
-                    from agent.title_generator import maybe_auto_title
-
-                    _title_key = session.get("session_key") or sid
-                    # Snapshot the runtime identity; the validator lets the
-                    # background titler skip its LLM call if the session's
-                    # model changed before it fires (#19027).
-                    _title_model = getattr(agent, "model", None)
-                    _title_provider = getattr(agent, "provider", None)
-                    maybe_auto_title(
-                        _get_db(),
-                        _title_key,
-                        text,
-                        raw,
-                        session.get("history", []),
-                        # Keep auxiliary auto-detection aligned with the active
-                        # Desktop/Webapp session. Without this, providers that
-                        # rely on runtime auth (for example OpenAI Codex OAuth)
-                        # are skipped and the new session remains untitled.
-                        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),
-                        },
-                        runtime_validator=lambda: (
-                            getattr(agent, "model", None) == _title_model
-                            and getattr(agent, "provider", None) == _title_provider
-                        ),
-                        # Push the generated title live so the sidebar renames
-                        # without waiting for the next list refresh (the titler
-                        # runs async, after this turn's refresh already fired).
-                        title_callback=lambda t, _k=_title_key: _emit(
-                            "session.title", sid, {"session_id": _k, "title": t}
-                        ),
-                    )
-                except Exception:
-                    pass
-
             # Voice TTS fallback: when the streaming pipeline couldn't start
             # (no provider / missing deps probed at turn start), speak the
             # final text whole (cli.py:_voice_speak_response parity). The