#!/usr/bin/env python3
"""
Chat list component: left drawer with sessions and session management.
Holds the session list, chat switching / renaming / pinning / deletion,
the delete-user-session flow, and the drawer rail toggle. All identity
and active-chat reads go through the shared :class:`PageContext`, so a
user-identity reset (delete all data) takes effect immediately for
every handler without a closure rebind.
File: klea_utils/ui/web/nicegui/components/chat_list.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import logging
import uuid
from collections.abc import Mapping
from datetime import datetime
from typing import Any
import coolname
from nicegui import background_tasks, ui
from klea_utils.ui.client import (
create_chat_on_server,
delete_chat_on_server,
rename_chat_on_server,
)
from klea_utils.ui.state import (
chat_indicator,
chat_preview_text,
chats,
ensure_chat,
get_chats_sorted,
)
from klea_utils.ui.web.nicegui.components import stream
from klea_utils.ui.web.nicegui.components.context import PageContext
from klea_utils.ui.web.nicegui.components.storage import safe_set_user
logger = logging.getLogger(__name__)
def _chat_preview(chat: Mapping[str, Any]) -> str:
"""Return a single-line preview for a session row.
Prefers the server's first-user-query ``preview`` (hydrated from
``GET /chat/{user_id}``); falls back to the first local user message so a
chat created or used in the current page session has a tooltip before the
next hydration.
"""
preview = chat.get("preview")
if not preview:
preview = next(
(
str(message.get("text", ""))
for message in chat.get("messages", [])
if message.get("role") == "user"
),
"",
)
return chat_preview_text(str(preview or ""), max_chars=200)
[docs]
def attach_chat_list(ctx: PageContext) -> None:
"""Build the left drawer (session rail) and register its handlers.
Registers ``refresh_chat_list`` and ``switch_chat`` on the context.
:param ctx: The shared page context; ``left_drawer`` and
``toggle_icon`` are filled in here.
"""
def _switch_chat(chat_id: str) -> None:
"""Switch the active chat without a page reload."""
safe_set_user("chat_id", chat_id)
ctx.chat_id = chat_id
logger.debug("chat_id=%s user_id=%s", chat_id, ctx.user_id)
ctx.render_chat_area()
ctx.refresh_chat_list()
ctx.refresh_status_pane()
# Re-render the status region too: a chat switched to while a HITL
# ask is pending must show the ask (the composer is hidden).
ctx.refresh_turn_status()
ctx.refresh_stream_button()
ctx.reset_center_tab()
ctx.refresh_inspector()
if ctx.fetch_model_info is not None:
background_tasks.create(ctx.fetch_model_info())
def _delete_chat(chat_id: str) -> None:
"""Remove a chat session from the store and server.
If the currently active chat session is deleted, the next
available chat session becomes active (or a new one is created).
"""
logger.debug(
"deleting chat_id=%s user_id=%s (current=%s)",
chat_id,
ctx.user_id,
ctx.chat_id,
)
# Stop an in-flight run for this chat before deleting it, so the graph
# run is cancelled both locally and server-side.
if ctx.chat_is_streaming(chat_id):
stream.stop_stream(ctx, chat_id)
background_tasks.create(
delete_chat_on_server(ctx.server_url, ctx.user_id, chat_id)
)
chats.pop(f"{ctx.user_id}:{chat_id}", None)
if ctx.chat_id == chat_id:
remaining = get_chats_sorted(ctx.user_id)
if remaining:
_switch_chat(remaining[0][0])
else:
ctx.chat_id = ""
safe_set_user("chat_id", "")
ctx.render_chat_area()
ctx.refresh_chat_list()
ctx.refresh_status_pane()
ctx.refresh_send_state()
ctx.reset_center_tab()
ctx.refresh_inspector()
else:
ctx.refresh_chat_list()
def _confirm_delete_chat(chat_id: str) -> None:
"""Show a confirmation dialog before deleting a single chat.
Shared by the sidebar's per-chat delete and the client
``/sessions delete`` command (ADR-0047), so both paths confirm.
"""
chat = ensure_chat(ctx.user_id, chat_id)
name = chat["name"]
logger.debug(
"opening delete-chat dialog for chat_id=%s user_id=%s",
chat_id,
ctx.user_id,
)
dialog = ui.dialog()
def _do_delete() -> None:
_delete_chat(chat_id)
dialog.close()
with dialog, ui.card():
ui.label("Delete session?").classes("text-lg font-bold")
ui.label(
f"This will permanently delete the session {name!r} and its "
"messages. This cannot be undone."
).classes("text-sm")
with ui.row().classes("w-full justify-end"):
ui.button("Cancel", on_click=dialog.close)
ui.button("Delete", on_click=_do_delete).props(
"unelevated color=negative"
)
dialog.open()
def _toggle_pin(chat_id: str) -> None:
"""Flip the pinned flag for a chat and refresh the list."""
current_chat = ensure_chat(ctx.user_id, chat_id)
current_chat["pinned"] = not current_chat["pinned"]
logger.debug(
"toggled pin for chat=%s pinned=%s", chat_id, current_chat["pinned"]
)
ctx.refresh_chat_list()
def _apply_rename(chat_id: str, name: str) -> None:
"""Rename *chat_id* locally and persist it on the server.
Shared by the chat-list rename dialog and the client ``/rename``
command (ADR-0047). The server PATCH runs in the background so callers
stay synchronous.
"""
chat = ensure_chat(ctx.user_id, chat_id)
chat["name"] = name
logger.debug("renaming chat=%s to %r", chat_id, name)
background_tasks.create(
rename_chat_on_server(ctx.server_url, ctx.user_id, chat_id, name)
)
ctx.refresh_chat_list()
ctx.refresh_status_pane()
def _rename_chat(chat_id: str) -> None:
"""Open a dialog to rename a chat and persist on server."""
chat = ensure_chat(ctx.user_id, chat_id)
logger.debug("opening rename dialog for chat=%s", chat_id)
dialog = ui.dialog()
async def _save():
_apply_rename(chat_id, inp.value)
dialog.close()
with dialog, ui.card():
ui.label("Rename chat").classes("text-lg font-bold")
inp = ui.input(value=chat["name"]).on("keydown.enter", _save)
with ui.row().classes("w-full justify-end"):
ui.button("Cancel", on_click=dialog.close)
ui.button("Save", on_click=_save).props("unelevated color=primary")
dialog.open()
def _new_chat() -> None:
"""Create a new chat and switch to it."""
chat_id = coolname.generate_slug(2)
logger.debug("creating chat_id=%s", chat_id)
new_chat = ensure_chat(ctx.user_id, chat_id)
for hook in ctx.chat_created_hooks:
hook(new_chat)
_switch_chat(chat_id)
background_tasks.create(
create_chat_on_server(ctx.server_url, ctx.user_id, chat_id)
)
@ui.refreshable
def _render_chat_list():
"""Render the sorted list of chat sessions in the left drawer.
Each entry shows the chat name (bold for the active chat),
a tooltip with the creation timestamp, and a three-dot context
menu for rename / pin / delete. Refresh this to pick up newly
created chat sessions without a page reload.
"""
# Underscore suffix avoids shadowing the enclosing chat_id params
for chat_id_, sdata in get_chats_sorted(ctx.user_id):
is_current = chat_id_ == ctx.chat_id
with (
ui.item(
on_click=lambda s=chat_id_: _switch_chat(s),
)
.props("dense")
.classes("w-full")
.on("dblclick", lambda s=chat_id_: _rename_chat(s))
):
streaming = ctx.chat_is_streaming(chat_id_)
awaiting = sdata["turn_status"]["kind"] == "awaiting_input"
indicator = chat_indicator(
streaming=streaming,
awaiting=awaiting,
pinned=sdata["pinned"],
active=is_current,
)
# The avatar cell is the only section Quasar keeps when the
# drawer is in mini mode, so the run/awaiting/active state is
# shown there (the label below is hidden when collapsed).
with ui.item_section().props("avatar"):
if indicator.kind == "spinner":
ui.spinner().classes(indicator.classes)
else:
ui.icon(indicator.icon).classes(indicator.classes)
with ui.item_section():
label_cls = "text-xs font-bold" if is_current else "text-xs"
ui.label(sdata["name"]).classes(label_cls)
tooltip = "Created: " + datetime.fromtimestamp(
sdata["created"]
).astimezone().strftime("%a %d %b %Y at %X")
if streaming:
tooltip += " - running"
elif awaiting:
tooltip += " - awaiting your input"
preview = _chat_preview(sdata)
with ui.tooltip():
if preview:
ui.label(preview).classes("text-xs")
ui.label(tooltip).classes("text-xs")
# Three-dot context menu (right-aligned).
with (
ui.item_section().props("side"),
ui.button(icon="more_vert")
.props("flat dense round")
.classes("icon-btn")
.on("click.stop", lambda: None),
ui.menu(),
):
with ui.menu_item(on_click=lambda s=chat_id_: _rename_chat(s)):
with ui.item_section().props("avatar"):
ui.icon("edit")
with ui.item_section():
ui.label("Rename")
with ui.menu_item(on_click=lambda s=chat_id_: _toggle_pin(s)):
with ui.item_section().props("avatar"):
ui.icon("push_pin")
with ui.item_section():
ui.label("Unpin" if sdata["pinned"] else "Pin")
with ui.menu_item(
on_click=lambda s=chat_id_: _confirm_delete_chat(s)
):
with ui.item_section().props("avatar"):
ui.icon("delete")
with ui.item_section():
ui.label("Delete")
def _delete_all_data():
"""Show a confirmation dialog before deleting the user session."""
logger.debug("opening delete-user-session dialog for user_id=%s", ctx.user_id)
dialog = ui.dialog()
with dialog, ui.card():
ui.label("Delete user session?").classes("text-lg font-bold")
ui.label(
"This will permanently delete all your chats, messages, and "
"checkpoints from the server. This cannot be undone."
).classes("text-sm")
with ui.row().classes("w-full justify-end"):
ui.button("Cancel", on_click=dialog.close)
ui.button("Delete", on_click=lambda: _confirm_delete_all(dialog)).props(
"unelevated color=negative"
)
dialog.open()
async def _confirm_delete_all(dialog: ui.dialog):
"""DELETE all server data, reset in-memory state, and generate a new user ID."""
import httpx
old_id = ctx.user_id
logger.debug("confirming delete of user session for user_id=%s", old_id)
try:
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.delete(f"{ctx.server_url}/chat/{old_id}")
if resp.status_code == 200:
logger.debug("delete_user_session succeeded for user_id=%s", old_id)
else:
logger.warning(
"delete_user_session failed: HTTP %s for user_id=%s",
resp.status_code,
old_id,
)
ui.notification(
f"Failed to delete session (HTTP {resp.status_code}). "
"No data was cleared.",
type="negative",
timeout=10000,
close_button=True,
)
dialog.close()
return
except Exception as e: # noqa: BLE001
logger.warning("Failed to delete user session: %s", e)
ui.notification(
f"Failed to delete session: {e}",
type="negative",
timeout=10000,
close_button=True,
)
dialog.close()
return
logger.debug("clearing in-memory chats for user_id=%s", old_id)
chat_key_prefix = f"{old_id}:"
for key in list(chats.keys()):
if key.startswith(chat_key_prefix):
chats.pop(key, None)
ctx.chat_id = ""
safe_set_user("chat_id", "")
new_id = str(uuid.uuid4())
safe_set_user("user_id", new_id)
# Rebind the shared context so every handler (new chat, send,
# status pane, ...) uses the fresh identity for the rest of this
# page session.
ctx.user_id = new_id
logger.debug(
"reset local state: new user_id=%s (was %s)",
ctx.user_id,
old_id,
)
# The new identity has no model overrides or credentials --- they
# were purged with the old user's data --- so drop the cached
# config and re-resolve it. Without this the status pane and the
# chat-area "choose a model" prompt keep showing the old user's
# setup.
ctx.session_model_info = {}
ctx.credentials = []
if ctx.fetch_session_model_info is not None:
await ctx.fetch_session_model_info()
ctx.render_chat_area()
ctx.refresh_chat_list()
ctx.refresh_status_pane()
ctx.refresh_send_state()
dialog.close()
def _toggle_left_drawer():
"""Switch the left drawer between mini (rail) and full width.
When *mini_state* is ``True`` the drawer shows a 64 px narrow
rail with only item icons; otherwise it expands to ``w-80``
(320 px) showing icons and labels. The toggle-button icon
changes direction to hint at the available action.
"""
ctx.mini_state = not ctx.mini_state
logger.debug("toggling left drawer: mini_state=%s", ctx.mini_state)
if ctx.mini_state:
ctx.left_drawer.props("mini")
ctx.toggle_icon.name = "keyboard_double_arrow_right"
else:
ctx.left_drawer.props(remove="mini")
ctx.toggle_icon.name = "keyboard_double_arrow_left"
ctx.switch_chat = _switch_chat
ctx.new_chat = _new_chat
ctx.rename_chat = _apply_rename
ctx.delete_chat = _confirm_delete_chat
ctx.refresh_chat_list = _render_chat_list.refresh
# ---- Left drawer (rail mode by default) ----
with (
ui.left_drawer(value=True)
.props("mini")
.props("width=320")
.classes("overflow-x-hidden p-2") as left_drawer
):
ctx.left_drawer = left_drawer
# Items use QItem + QItemSection(avatar) so that Quasar
# automatically hides the label when the drawer is in mini mode.
with (
ui.item(on_click=_new_chat)
.props("dense")
.classes("w-full")
.mark("new-chat")
):
with ui.item_section().props("avatar"):
ui.icon("add")
ui.tooltip("Start a new conversation")
with ui.item_section():
ui.label("New Chat")
# Session list header (no icon --- plain text signals a section heading)
with ui.item().props("dense").classes("w-full"), ui.item_section():
ui.label("Chats").classes("text-sm font-bold")
_render_chat_list()
ui.space()
with ui.item(on_click=_delete_all_data).props("dense").classes("w-full"):
with ui.item_section().props("avatar"):
ui.icon("delete")
ui.tooltip("Delete all data for this user session")
with ui.item_section():
ui.label("Delete user session").classes("text-xs")
# Toggle button at the bottom of the drawer.
with ui.item(on_click=_toggle_left_drawer).props("dense").classes("w-full"):
with ui.item_section().props("avatar"):
ctx.toggle_icon = ui.icon("keyboard_double_arrow_right")
ui.tooltip("Expand or collapse the sidebar")
with ui.item_section():
ui.label("").classes("text-xs")