fix(tui_gateway): drop pending_title on ValueError, retain on transient errors

Production bug + missing test coverage. c5b4c48 ("fix: lazy session
creation — defer DB row until first message (#18370)") moved
pending_title application from the eager _start_agent_build path to a
post-message-complete handler. The original block had:

    except ValueError as e:
        current["pending_title"] = None
        logger.info("Dropping pending title for session %s: %s", sid, e)
    except Exception:
        logger.warning("Failed to apply pending title ...", exc_info=True)

…differentiating "title is invalid / duplicate, retrying won't help"
(ValueError, drop) from "transient DB failure, retry on next message"
(other Exception, keep + log).

The replacement block collapsed both into:

    except Exception:
        pass  # Best effort — auto-title will handle it below

…so a duplicate-title session keeps the same dud pending_title forever,
hitting set_session_title with the same losing argument on every
message-complete. Auto-title can't kick in because pending_title still
shadows it.

Fix: extract a documented _apply_pending_session_title helper that
restores the three-branch semantics (success → clear, ValueError →
drop, other Exception → retain). Call it from the
message-complete handler instead of the inline try/except.

Test rewrite: the previous test_session_create_drops_pending_title_on_valueerror
exercised an obsolete code path (eager apply during session.create) that
no longer existed after c5b4c48. Replace with four focused tests against
the helper:

  - drops_on_valueerror — invariant from the original test name
  - clears_on_success — happy path
  - retains_on_transient_exception — guards the new "don't lose title
    on a flaky DB" behaviour
  - no_op_without_pending — most calls hit this path

Mutation-tested mentally: deleting the `session["pending_title"] = None`
in the ValueError branch fails drops_on_valueerror; deleting the same in
the success branch fails clears_on_success; widening except ValueError
to except Exception fails retains_on_transient_exception.

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

tests/test_tui_gateway_server.py

@@ -753,57 +753,63 @@ def test_session_title_set_errors_when_row_lookup_fails_after_noop(monkeypatch):
         server._sessions.pop("sid", None)
 
 
-def test_session_create_drops_pending_title_on_valueerror(monkeypatch):
-    unblock_agent = threading.Event()
-
-    class _FakeWorker:
-        def __init__(self, key, model):
-            self.key = key
-
-        def close(self):
-            return None
-
-    class _FakeAgent:
-        model = "x"
-        provider = "openrouter"
-        base_url = ""
-        api_key = ""
-
-    class _FakeDB:
-        def create_session(self, _key, source="tui", model=None):
-            return None
+def test_apply_pending_session_title_drops_on_valueerror():
+    """ValueError from set_session_title (e.g. duplicate title) must drop
+    the pending_title so a stuck title doesn't keep retrying forever.
 
+    Originally tested via the eager-apply path in _start_agent_build, which
+    was removed by c5b4c48 (#18370, lazy session creation) and replaced by
+    a post-message-complete apply that only `except Exception: pass`'d —
+    losing the ValueError-specific drop semantics. The helper restores
+    them; this test asserts that.
+    """
+    class _RaisingDB:
         def set_session_title(self, _key, _title):
             raise ValueError("Title already in use")
 
-    def _make_agent(_sid, _key):
-        unblock_agent.wait(timeout=2.0)
-        return _FakeAgent()
-
-    monkeypatch.setattr(server, "_make_agent", _make_agent)
-    monkeypatch.setattr(server, "_SlashWorker", _FakeWorker)
-    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
-    monkeypatch.setattr(server, "_session_info", lambda _a: {"model": "x"})
-    monkeypatch.setattr(server, "_probe_credentials", lambda _a: None)
-    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
-    monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
-
-    import tools.approval as _approval
-
-    monkeypatch.setattr(_approval, "register_gateway_notify", lambda key, cb: None)
-    monkeypatch.setattr(_approval, "load_permanent_allowlist", lambda: None)
-
-    resp = server.handle_request(
-        {"id": "1", "method": "session.create", "params": {"cols": 80}}
-    )
-    sid = resp["result"]["session_id"]
-    session = server._sessions[sid]
-    session["pending_title"] = "duplicate title"
-    unblock_agent.set()
-    session["agent_ready"].wait(timeout=2.0)
+    session = {"session_key": "k1", "pending_title": "duplicate title"}
+    server._apply_pending_session_title(session, "sid-1", _RaisingDB())
+
+    assert session["pending_title"] is None
+
+
+def test_apply_pending_session_title_clears_on_success():
+    class _OkDB:
+        def set_session_title(self, _key, _title):
+            return True
+
+    session = {"session_key": "k2", "pending_title": "Real title"}
+    server._apply_pending_session_title(session, "sid-2", _OkDB())
+
+    assert session["pending_title"] is None
+
+
+def test_apply_pending_session_title_retains_on_transient_exception():
+    """A transient (non-ValueError) DB failure should keep the pending
+    title queued so the next message-complete can retry. Without this
+    behaviour, a single flaky DB call would silently lose the title."""
+    class _FlakyDB:
+        def set_session_title(self, _key, _title):
+            raise RuntimeError("transient db blip")
+
+    session = {"session_key": "k3", "pending_title": "Keep retrying"}
+    server._apply_pending_session_title(session, "sid-3", _FlakyDB())
+
+    assert session["pending_title"] == "Keep retrying"
+
+
+def test_apply_pending_session_title_no_op_without_pending():
+    """Helper must be a no-op when pending_title is None — most calls
+    look like this (every message-complete on a session that already has
+    a title applied)."""
+    class _ShouldNotBeCalledDB:
+        def set_session_title(self, _key, _title):
+            raise AssertionError("DB must not be touched when no pending title")
+
+    session = {"session_key": "k4", "pending_title": None}
+    server._apply_pending_session_title(session, "sid-4", _ShouldNotBeCalledDB())
 
     assert session["pending_title"] is None
-    server._sessions.pop(sid, None)
 
 
 def test_config_set_yolo_toggles_session_scope():

tui_gateway/server.py

@@ -515,6 +515,50 @@ def _wait_agent(session: dict, rid: str, timeout: float = 30.0) -> dict | None:
     return _err(rid, 5032, err) if err else None
 
 
+def _apply_pending_session_title(
+    session: dict, sid: str, db: object | None
+) -> None:
+    """Apply session["pending_title"] to the DB via db.set_session_title.
+
+    Pending titles are queued during session.create (before the DB row
+    exists, since c5b4c48 deferred row creation to first message) and
+    flushed here once a message-complete event lands.
+
+    Outcome by branch:
+      - set_session_title returns truthy: pending_title cleared.
+      - ValueError (title invalid / duplicate): pending_title dropped,
+        because retrying with the same value will fail the same way.
+        Auto-title later picks a fresh title from message content.
+      - other Exception: pending_title retained — likely a transient DB
+        failure worth retrying on the next message-complete.
+
+    No-ops when there is no pending title or no DB.
+
+    Pre-c5b4c48 (#18370) the same semantics lived inline in
+    _start_agent_build. Extracting them here both restores the lost
+    ValueError handling and makes the invariant testable without
+    simulating a full message turn.
+    """
+    pending = session.get("pending_title")
+    if not pending or db is None:
+        return
+    key = session.get("session_key") or sid
+    try:
+        if db.set_session_title(key, pending):
+            session["pending_title"] = None
+    except ValueError as exc:
+        # Title invalid / duplicate — retrying is futile; drop and let
+        # auto-title pick something.
+        session["pending_title"] = None
+        logger.info("Dropping pending title for session %s: %s", sid, exc)
+    except Exception:
+        # Likely transient — keep pending_title so the next
+        # message-complete can retry. Auto-title is the eventual fallback.
+        logger.warning(
+            "Failed to apply pending title for session %s", sid, exc_info=True,
+        )
+
+
 def _start_agent_build(sid: str, session: dict) -> None:
     """Start building the real AIAgent for a TUI session, once.
 
@@ -2982,15 +3026,8 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
             _emit("message.complete", sid, payload)
 
             # Apply pending_title now that the DB row exists.
-            _pending = session.get("pending_title")
-            if _pending and status == "complete":
-                _pdb = _get_db()
-                if _pdb:
-                    try:
-                        if _pdb.set_session_title(session.get("session_key") or sid, _pending):
-                            session["pending_title"] = None
-                    except Exception:
-                        pass  # Best effort — auto-title will handle it below
+            if status == "complete":
+                _apply_pending_session_title(session, sid, _get_db())
 
             if (
                 status == "complete"