fix(tui-gateway): harden session title RPC edge cases

Handle session.title read failures without crashing, distinguish no-op title writes from missing session rows, and use a distinct empty-title error code with regression coverage.

tests/test_tui_gateway_server.py

@@ -263,6 +263,9 @@ def test_session_title_queues_when_db_row_not_ready(monkeypatch):
         def get_session_title(self, _key):
             return None
 
+        def get_session(self, _key):
+            return None
+
         def set_session_title(self, _key, _title):
             return False
 
@@ -297,6 +300,9 @@ def test_session_title_clears_pending_after_persist(monkeypatch):
         def get_session_title(self, _key):
             return self.title
 
+        def get_session(self, _key):
+            return {"id": _key, "title": self.title}
+
         def set_session_title(self, _key, title):
             self.title = title
             return True
@@ -320,6 +326,76 @@ def test_session_title_clears_pending_after_persist(monkeypatch):
         server._sessions.pop("sid", None)
 
 
+def test_session_title_does_not_queue_noop_when_row_exists(monkeypatch):
+    class _FakeDB:
+        def __init__(self):
+            self.title = "same title"
+
+        def get_session_title(self, _key):
+            return self.title
+
+        def get_session(self, _key):
+            return {"id": _key, "title": self.title}
+
+        def set_session_title(self, _key, _title):
+            # Simulate sqlite UPDATE rowcount==0 for no-op update.
+            return False
+
+    server._sessions["sid"] = _session(pending_title="stale")
+    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
+    try:
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.title",
+                "params": {"session_id": "sid", "title": "same title"},
+            }
+        )
+
+        assert resp["result"]["pending"] is False
+        assert resp["result"]["title"] == "same title"
+        assert server._sessions["sid"]["pending_title"] is None
+    finally:
+        server._sessions.pop("sid", None)
+
+
+def test_session_title_get_falls_back_to_pending_when_db_read_throws(monkeypatch):
+    class _FakeDB:
+        def get_session_title(self, _key):
+            raise RuntimeError("db temporarily locked")
+
+    server._sessions["sid"] = _session(pending_title="queued title")
+    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
+    try:
+        resp = server.handle_request(
+            {"id": "1", "method": "session.title", "params": {"session_id": "sid"}}
+        )
+        assert resp["result"]["title"] == "queued title"
+    finally:
+        server._sessions.pop("sid", None)
+
+
+def test_session_title_rejects_empty_title_with_specific_error_code(monkeypatch):
+    class _FakeDB:
+        def get_session_title(self, _key):
+            return ""
+
+    server._sessions["sid"] = _session()
+    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
+    try:
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.title",
+                "params": {"session_id": "sid", "title": "   "},
+            }
+        )
+        assert "error" in resp
+        assert resp["error"]["code"] == 4021
+    finally:
+        server._sessions.pop("sid", None)
+
+
 def test_config_set_yolo_toggles_session_scope():
     from tools.approval import clear_session, is_session_yolo_enabled
 

tui_gateway/server.py

@@ -1746,20 +1746,41 @@ def _(rid, params: dict) -> dict:
         return _db_unavailable_error(rid, code=5007)
     key = session["session_key"]
     if "title" not in params:
+        fallback = session.get("pending_title") or ""
+        try:
+            resolved_title = db.get_session_title(key) or fallback
+        except Exception:
+            resolved_title = fallback
         return _ok(
             rid,
             {
-                "title": db.get_session_title(key) or session.get("pending_title") or "",
+                "title": resolved_title,
                 "session_key": key,
             },
         )
     title = (params.get("title", "") or "").strip()
     if not title:
-        return _err(rid, 4007, "title required")
+        return _err(rid, 4021, "title required")
     try:
         if db.set_session_title(key, title):
             session["pending_title"] = None
             return _ok(rid, {"pending": False, "title": title})
+        # rowcount == 0 can mean "same value" as well as "missing row".
+        # Queue only when the session row truly does not exist yet.
+        existing_row = None
+        try:
+            existing_row = db.get_session(key)
+        except Exception:
+            existing_row = None
+        if existing_row:
+            session["pending_title"] = None
+            return _ok(
+                rid,
+                {
+                    "pending": False,
+                    "title": (existing_row.get("title") or title),
+                },
+            )
         session["pending_title"] = title
         return _ok(rid, {"pending": True, "title": title})
     except Exception as e: