diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 1a682b7972..d0e96922fa 100644 --- a/tests/test_tui_gateway_server.py +++ b/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 diff --git a/tui_gateway/server.py b/tui_gateway/server.py index ebfb9c88b3..4e1e99eb05 100644 --- a/tui_gateway/server.py +++ b/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: