From 492c4c6573b43c9d887d0161ca43809526661834 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Mon, 27 Apr 2026 11:15:37 -0500 Subject: [PATCH] fix(tui-gateway): address follow-up Copilot title threads Tighten pending-title flush during session init and treat row lookup failures during title-set no-op detection as RPC errors instead of silently queueing. --- tests/test_tui_gateway_server.py | 28 ++++++++++++++++++++++++++++ tui_gateway/server.py | 27 ++++++++++++++++++++------- 2 files changed, 48 insertions(+), 7 deletions(-) diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index d0e96922f..ff6a022b5 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -396,6 +396,34 @@ def test_session_title_rejects_empty_title_with_specific_error_code(monkeypatch) server._sessions.pop("sid", None) +def test_session_title_set_errors_when_row_lookup_fails_after_noop(monkeypatch): + class _FakeDB: + def get_session_title(self, _key): + return "" + + def get_session(self, _key): + raise RuntimeError("row lookup failed") + + def set_session_title(self, _key, _title): + return False + + 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": "fresh"}, + } + ) + assert "error" in resp + assert resp["error"]["code"] == 5007 + assert "row lookup failed" in resp["error"]["message"] + 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 4e1e99eb0..f47b21f0f 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1571,10 +1571,27 @@ def _(rid, params: dict) -> dict: pending_title = (session.get("pending_title") or "").strip() if pending_title: try: - if db.set_session_title(key, pending_title): + title_applied = db.set_session_title(key, pending_title) + if title_applied: session["pending_title"] = None + else: + existing_row = db.get_session(key) + existing_title = ((existing_row or {}).get("title") or "").strip() + if existing_title == pending_title: + session["pending_title"] = None + else: + logger.info( + "Pending title still queued for session %s (wanted=%r, current=%r)", + sid, + pending_title, + existing_title, + ) except Exception: - pass + logger.warning( + "Failed to apply pending title for session %s", + sid, + exc_info=True, + ) session["agent"] = agent try: @@ -1767,11 +1784,7 @@ def _(rid, params: dict) -> dict: 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 + existing_row = db.get_session(key) if existing_row: session["pending_title"] = None return _ok(