fix(tui): restore ValueError-drop on pending-title flush

tests/test_tui_gateway_server.py

@@ -754,56 +754,29 @@ def set_session_title(self, _key, _title):
 
 
 def test_session_create_drops_pending_title_on_valueerror(monkeypatch):
-    unblock_agent = threading.Event()
+    """Verify that the pending-title flush drops the queue on ``ValueError``.
 
-    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 = ""
+    Pre-#18370 this was exercised in the ``session.create`` path, but the
+    DB row is now created lazily and the flush moved to a helper invoked
+    from the prompt-submit handler. Drive that helper directly: that's
+    where the contract introduced for #14334 (queued title + duplicate
+    error → drop the queue) actually lives.
+    """
 
     class _FakeDB:
-        def create_session(self, _key, source="tui", model=None):
-            return None
-
         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)
 
-    assert session["pending_title"] is None
-    server._sessions.pop(sid, None)
+    sid = "sid-pending-valueerror"
+    session = _session(pending_title="duplicate title")
+    server._sessions[sid] = session
+    try:
+        server._apply_pending_title(sid, session)
+        assert session["pending_title"] is None
+    finally:
+        server._sessions.pop(sid, None)
 
 
 def test_config_set_yolo_toggles_session_scope():

tui_gateway/server.py

@@ -604,6 +604,42 @@ def _build() -> None:
     threading.Thread(target=_build, daemon=True).start()
 
 
+def _apply_pending_title(sid: str, session: dict) -> None:
+    """Persist a queued ``pending_title`` once the session DB row exists.
+
+    Session creation is lazy (#18370) — the DB row only materialises on the
+    first ``run_conversation()`` call, so any title queued before then has
+    to be applied post-first-message. This helper centralises that flush
+    so it can be invoked from the prompt-submit handler and exercised by
+    tests without standing up a full prompt round-trip.
+
+    Semantics, in order of try blocks below:
+
+    * If the queued title applies cleanly, clear ``pending_title``.
+    * If ``set_session_title`` raises ``ValueError`` (duplicate / invalid
+      by the time the row exists), **drop** the queued title and log the
+      reason so ``/title`` doesn't keep surfacing a stuck pending value.
+      This preserves the contract introduced for #14334.
+    * Any other exception is best-effort swallowed; auto-title can take
+      over from the next message.
+    """
+    pending = session.get("pending_title")
+    if not pending:
+        return
+    db = _get_db()
+    if 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 ve:
+        session["pending_title"] = None
+        logger.info("Dropping pending title for session %s: %s", sid, ve)
+    except Exception:
+        pass  # Best effort — auto-title will handle it.
+
+
 def _sess_nowait(params, rid):
     s = _sessions.get(params.get("session_id") or "")
     return (s, None) if s else (None, _err(rid, 4001, "session not found"))
@@ -2982,15 +3018,8 @@ def _stream(delta):
             _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_title(sid, session)
 
             if (
                 status == "complete"