#!/usr/bin/env python3
"""
Chat area component: message list, scroll, and empty-state.
Builds the scroll area that holds the message bubbles and the stream
progress container, and registers the render/scroll callbacks on the
page context.
File: klea_utils/ui/web/nicegui/components/chat_area.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import json
import logging
from typing import Any
from nicegui import ui
from klea_utils.llm import missing_required_roles
from klea_utils.ui.linkify import linkify_md
from klea_utils.ui.state import TurnStatus, chats, missing_credentials
from klea_utils.ui.web.nicegui.components.chat_bubble import ChatBubble
from klea_utils.ui.web.nicegui.components.choice import choice_buttons
from klea_utils.ui.web.nicegui.components.context import PageContext
from klea_utils.ui.web.nicegui.components.state_edit import render_state_edit
logger = logging.getLogger(__name__)
[docs]
def render_turn_status(ctx: PageContext, retry_cb=None) -> None:
"""Render the turn status region for the active chat from its state.
The region is a single slot under the transcript, owned by the chat's
``status`` dict (``state.ChatData``), so it is restored on chat switch
and replaced wholesale on the next turn. ``kind`` selects the content:
* ``progress`` - a spinner plus the live ``heading``.
* ``error`` - the message and, when ``resumable``, a Retry action.
* ``stopped`` - a plain "Stopped" line.
* ``idle``/missing - nothing.
:param ctx: The shared page context (uses ``turn_status_container``).
:param retry_cb: Callable invoked by the Retry button; the stream
component passes it when rendering a live error.
"""
status = _current_turn_status(ctx)
kind = status.get("kind", "idle")
container = ctx.turn_status_container
container.clear()
ctx.interrupt_container.clear()
if kind == "idle":
return
if kind == "awaiting_input":
# The ask replaces the composer: render it in the pinned container
# between the transcript and the input row (the input area hides its
# row while awaiting, mirroring the TUI).
with ctx.interrupt_container:
_render_interrupt_form(ctx)
return
with container:
if kind == "progress":
with ui.row().classes("w-full items-center gap-2 p-2"):
ui.spinner(type="dots").classes("w-4 h-4")
label = ui.label(status.get("heading", "")).classes(
"text-xs text-grey-5 italic"
)
ctx.turn_status_label = label
elif kind == "error":
with ui.row().classes("w-full items-center gap-2 p-2"):
ui.icon("error").classes("text-negative")
message = f"Error: {status.get('message', '')}"
if "No model configured" in message:
message += (
" Use the settings (gear) icon to choose a model for "
"this chat, then retry."
)
ui.label(message).classes("text-xs text-negative flex-grow")
if status.get("resumable") and retry_cb is not None:
ui.button("Retry", on_click=retry_cb).props(
"flat dense color=primary"
)
elif kind == "stopped":
ui.label("Stopped").classes("text-xs text-grey-5 italic p-2")
def _current_turn_status(ctx: PageContext) -> TurnStatus:
"""Return the active chat's status dict, or ``{"kind": "idle"}``."""
if not ctx.chat_id:
return {"kind": "idle"}
chat = chats.get(f"{ctx.user_id}:{ctx.chat_id}")
if not chat:
return {"kind": "idle"}
return chat.get("turn_status") or {"kind": "idle"}
def _interrupt_ask(ctx: PageContext) -> dict[str, Any]:
"""Return the active chat's pending HITL ask, or ``{}``."""
if not ctx.chat_id:
return {}
chat = chats.get(f"{ctx.user_id}:{ctx.chat_id}")
return (chat or {}).get("interrupt") or {}
def _render_ask_sections(ask: dict[str, Any]) -> None:
"""Render optional HITL sections (review plan/progress, input plan history).
The sections are rendered monospace so tool names and paths stay literal;
full results remain in the inspection pane.
"""
sections = [
section for section in (ask.get("sections") or []) if isinstance(section, dict)
]
for index, section in enumerate(sections):
title = str(section.get("title", ""))
if len(sections) > 1:
label = f"Option {index + 1}"
title = f"{label}: {title}" if title else label
if title:
ui.label(title).classes("text-xs font-bold")
if body := section.get("body"):
ui.code(body, language=None).classes("text-xs w-full nicegui-code-noformat")
def _render_interrupt_form(ctx: PageContext) -> None:
"""Render the HITL ask form in the turn status region (ADR-0046).
A ``review`` ask gets Approve / Request changes / Cancel; an ``input``
ask gets one text field per question plus Answer / Cancel; a
``permission`` ask (ADR-0007) gets one allow-once / allow-session / deny
choice per requested path; a ``state_edit`` ask renders the generic
session-state editor. Submitting calls
:attr:`PageContext.submit_interrupt` (registered by the input area).
"""
ask = _interrupt_ask(ctx)
kind = ask.get("kind", "input")
interrupt_id = ask.get("interrupt_id")
if kind == "permission":
_render_permission_form(ctx, ask, interrupt_id)
return
if kind == "state_edit":
render_state_edit(ctx, ask, interrupt_id)
return
heading = "Plan review" if kind == "review" else "User input"
with ui.column().classes("w-full gap-2 p-3 border border-primary rounded"):
ui.label(heading).classes("text-sm font-bold")
if kind == "review":
# ``headline`` + ``sections`` adapt to the review's position
# (ADR-0051): a plan approval, or a checkpoint showing the work
# done since the last review plus what approval unblocks. The
# sections are rendered monospace so tool names/paths stay literal
# (full results are in the inspection pane).
if headline := ask.get("headline"):
ui.label(headline).classes("text-sm font-bold")
if question := ask.get("question"):
ui.label(question).classes("text-sm")
_render_ask_sections(ask)
feedback = (
ui.textarea(placeholder="Feedback (required to request changes)")
.props("outlined autogrow")
.classes("w-full")
)
with ui.row().classes("gap-2"):
ui.button(
"Approve",
on_click=lambda: ctx.submit_interrupt(
{"decision": "approve"}, interrupt_id, False
),
).props("unelevated color=primary")
ui.button(
"Request changes",
on_click=lambda: ctx.submit_interrupt(
{"decision": "revise", "feedback": feedback.value or ""},
interrupt_id,
False,
),
).props("outline")
ui.button(
"Cancel",
on_click=lambda: ctx.submit_interrupt(None, interrupt_id, True),
).props("flat color=negative")
return
# input kind: plan history (when present), then one text field per
# question.
_render_ask_sections(ask)
questions = ask.get("questions")
if not isinstance(questions, list) or not questions:
fallback = ask.get("question") or "More information is needed to continue."
questions = [{"question": fallback}]
fields = []
for question in questions:
ui.label(str(question.get("question", ""))).classes("text-sm")
fields.append(ui.input().props("outlined dense").classes("w-full"))
def _submit() -> None:
ctx.submit_interrupt(
{"answers": [field.value or "" for field in fields]},
interrupt_id,
False,
)
with ui.row().classes("gap-2"):
ui.button("Answer", on_click=_submit).props("unelevated color=primary")
ui.button(
"Cancel",
on_click=lambda: ctx.submit_interrupt(None, interrupt_id, True),
).props("flat color=negative")
def _permission_key(request: dict[str, Any]) -> str:
"""Return the approval key for a serialized path request (ADR-0007).
The resolved file for a ``sensitive`` request, the resolved directory for
an ``outside`` one.
"""
if request.get("kind") == "sensitive":
return str(request.get("path", ""))
return str(request.get("directory", ""))
def _permission_path_display(request: dict[str, Any]) -> str:
"""Return the bounded target path for a permission request."""
return str(
request.get("path_display")
or request.get("path")
or request.get("directory", "")
)
def _permission_directory_display(request: dict[str, Any]) -> str:
"""Return the bounded approval directory for a permission request."""
return str(request.get("directory_display") or request.get("directory", ""))
def _permission_scope(request: dict[str, Any]) -> str:
"""Return the session-approval scope when it differs from the target path."""
path = str(request.get("path", ""))
directory = str(request.get("directory", ""))
if request.get("kind") != "sensitive" and path and directory and path != directory:
return f"Session approval covers: {_permission_directory_display(request)}"
return ""
def _permission_hint(request: dict[str, Any]) -> str:
"""Return the tool-call context for a request (tool, argument, value).
Gives the user the tool call behind the ask so they can judge it. Includes
the confidence tier: a ``guess`` is a heuristic path-like value (e.g. a
regex fragment), which is also the only tier offered the "Not a path"
choice. The argument value is a bounded preview; the full target path is
shown in the row title.
"""
tool = str(request.get("tool", ""))
argument = str(request.get("argument", ""))
confidence = str(request.get("confidence", ""))
value = str(request.get("argument_preview") or request.get("raw", ""))
raw = str(request.get("raw_preview") or request.get("raw", ""))
source = str(request.get("source", ""))
detail = f"{tool}: {argument} = {value} ({confidence})" if tool or argument else ""
if raw and (source.startswith("value:") or raw not in value):
detail = (
f"{detail}\nDetected token: {raw}" if detail else f"Detected token: {raw}"
)
if source:
detail = f"{detail}\nSource: {source}" if detail else f"Source: {source}"
if request.get("kind") == "sensitive":
note = "Commonly holds credentials."
return f"{detail} - {note}" if detail else note
return detail
def _permission_coverage(request: dict[str, Any]) -> str:
"""Return the paths covered by this request's session approval, if plural."""
paths = request.get("covered_paths_display") or request.get("covered_paths") or []
if not isinstance(paths, list) or len(paths) < 2:
return ""
lines = "\n".join(f"- {path}" for path in paths)
return f"Covers:\n{lines}"
def _permission_options(request: dict[str, Any]) -> dict[str, str]:
"""Return the per-request decision options.
The "Not a path" dismissal is offered only for a ``guess``-tier request,
where the gate may have mistaken a non-path value for a path.
"""
options = {"once": "Once", "session": "Session"}
if request.get("confidence") == "guess":
options["not_a_path"] = "Not a path"
options["deny"] = "Deny"
return options
def _render_permission_form(
ctx: PageContext, ask: dict[str, Any], interrupt_id: Any
) -> None:
"""Render a path-permission ask: one decision per requested path.
Each request gets an allow-once / allow-session / deny choice; selecting
one re-renders the rows so it shows its active state. Submit sends the
per-key decisions to :meth:`PageContext.submit_interrupt` (ADR-0007 update
2026-10-08). In-progress choices live on *ask* (the chat's stored
interrupt), so they survive a status-region re-render and are cleared with
the ask on resume.
"""
requests = ask.get("requests")
if not isinstance(requests, list) or not requests:
requests = [{"kind": "outside", "path": ask.get("question", "")}]
logger.debug(f"Rendering permission form\n{len(requests) = }\n{interrupt_id = }")
decisions: dict[str, str] = ask.setdefault("choices", {})
for request in requests:
if isinstance(request, dict):
decisions.setdefault(_permission_key(request), "deny")
with ui.column().classes("w-full gap-2 p-3 border border-primary rounded"):
count = len(requests)
noun = "path" if count == 1 else "paths"
ui.label(f"Permission required ({count} {noun})").classes("text-sm font-bold")
if question := ask.get("question"):
ui.label(str(question)).classes("text-xs text-grey-6")
# Re-render on select so the chosen option shows its active state
# (``choice_buttons`` computes it at build time).
@ui.refreshable
def _rows() -> None:
for index, request in enumerate(requests, start=1):
if not isinstance(request, dict):
continue
path = _permission_path_display(request)
if request.get("kind") == "sensitive":
title = f"Sensitive file: {path}"
else:
title = f"Outside project path: {path}"
title = f"{index}. {title}"
scope = _permission_scope(request)
hint = _permission_hint(request)
key = _permission_key(request)
def _set(value: str, k: str = key) -> None:
decisions[k] = value
_rows.refresh()
choice_buttons(
title,
_permission_options(request),
decisions.get(key, "deny"),
_set,
colors={"deny": "red-5", "not_a_path": "grey-7"},
label_width="20ch",
group_width="16rem",
)
if scope:
ui.label(scope).classes("text-xs text-grey-6")
if hint:
ui.label(hint).classes("text-xs text-grey-6").style(
"white-space: pre-line"
)
coverage = _permission_coverage(request)
if coverage:
ui.label(coverage).classes("text-xs text-grey-6").style(
"white-space: pre-line"
)
_rows()
def _submit() -> None:
logger.debug(f"Submitting permission decisions\n{decisions = }")
ctx.submit_interrupt(
{
"decisions": [
{"key": key, "decision": value}
for key, value in decisions.items()
]
},
interrupt_id,
False,
)
with ui.row().classes("gap-2"):
ui.button("Submit", on_click=_submit).props("unelevated color=primary")
ui.button(
"Cancel",
on_click=lambda: ctx.submit_interrupt(None, interrupt_id, True),
).props("flat color=negative")
def _render_messages(ctx: PageContext) -> None:
"""Rebuild the scroll-area content (welcome or messages) for the page.
Uses explicit clear+rebuild instead of ``@ui.refreshable``
to avoid issues with the welcome-to-empty-chat transition.
"""
current = ctx.chat_id
logger.debug(
"current=%s msgs=%d",
current,
len(chats.get(f"{ctx.user_id}:{current}", {}).get("messages", []))
if current
else 0,
)
ctx.chat_area.clear()
# The turn status region is rendered from the active chat's turn status,
# so switching chats restores that chat's own status and no stale
# marker bleeds across. A live run re-renders this from state too, so
# there is a single source of truth for the region.
render_turn_status(ctx)
with ctx.chat_area:
if not current:
with (
ui.column()
.classes("w-full h-full items-center justify-center gap-4")
.style("flex: 1; display: flex;")
):
ui.label("Start a conversation").classes("text-xl text-grey-5")
ui.label("Type your message below to begin").classes(
"text-sm text-grey-5"
)
# First-run setup call-to-action: point the user straight
# at the model dialog when anything required is missing.
missing_models = missing_required_roles(ctx.session_model_info)
missing_keys = missing_credentials(ctx.session_model_info)
if missing_models or missing_keys:
with ui.card().classes("items-center gap-1 mt-2 p-4"):
ui.label("Models are not configured").classes(
"text-sm font-bold"
)
if missing_models:
ui.label(
"Set the required models to start: "
+ ", ".join(
role.capitalize() for role in missing_models
)
).classes("text-xs text-grey-6")
elif missing_keys:
providers = sorted(
{
(
ctx.session_model_info[role].get("credential")
or {}
).get("provider", "")
for role in missing_keys
}
- {""}
)
ui.label(
"Add an API key for: " + ", ".join(providers)
).classes("text-xs text-grey-6")
ui.button(
"Choose models", on_click=ctx.model_config_dialog
).props("unelevated color=primary")
else:
current_chat = chats.get(f"{ctx.user_id}:{current}")
msgs = current_chat["messages"] if current_chat else []
expanded = current_chat["expanded"] if current_chat else set()
for idx, msg in enumerate(msgs):
collapsed = idx not in expanded
text = msg.get("text", "")
role = msg.get("role", "agent")
ChatBubble(
# A user message is rendered verbatim (no markdown), so it
# is not linkified into markdown links either.
text=text if role == "user" else linkify_md(text),
stamp=msg.get("stamp", ""),
role=role,
header=msg.get("header", ""),
mime=msg.get("mime", ""),
data=msg.get("data", ""),
meta=msg.get("meta", {}),
is_error=msg.get("is_error", False),
collapsed=collapsed,
idx=idx,
on_copy=lambda t=text: ui.run_javascript(
f"navigator.clipboard.writeText({json.dumps(t)})"
),
on_expand=lambda i=idx: (
(expanded.discard(i) if i in expanded else expanded.add(i))
or ctx.render_chat_area()
),
)
ctx.scroll_chat_bottom()
def _scroll_to_bottom(ctx: PageContext) -> None:
"""Scroll chat area to the bottom.
Uses Quasar's ``setScrollPosition`` (via NiceGUI's
``scroll_to(pixels=99999)``) instead of raw JavaScript because
NiceGUI batches UI updates and JS into the same WebSocket packet
--- by the time a ``setTimeout`` or ``requestAnimationFrame``
callback fires the new DOM may not be laid out yet, so
``scrollTop = scrollHeight`` or ``scrollIntoView`` land at the
wrong position.
``setScrollPosition`` is Quasar's own scroll API on
``QScrollArea``; it coordinates with its internal layout cycle
so the scroll lands correctly after the content updates are
painted. The large pixel value is safe --- Quasar clamps it to
the actual scrollable extent.
"""
logger.debug("attempting scroll for chat=%s", ctx.chat_id)
ctx.scroll_area.scroll_to(pixels=99999)
[docs]
def attach_chat_area(ctx: PageContext) -> None:
"""Build the chat scroll area and register its render callbacks.
Must be called with the chat tab panel as the ambient NiceGUI
context (the surrounding layout decides where the tab panels go).
:param ctx: The shared page context; ``chat_area``,
``scroll_area`` and ``turn_status_container`` are filled in here.
"""
with ui.scroll_area().classes("w-full grow chat-scroll-area") as scroll_area:
ctx.scroll_area = scroll_area
ctx.chat_area = ui.column().classes("w-full")
ctx.turn_status_container = ui.column().classes("w-full")
# Pinned between the transcript and the composer: while a HITL ask is
# active it is rendered here and the input row is hidden, so the ask
# replaces the composer (mirroring the TUI; ADR-0046).
ctx.interrupt_container = ui.column().classes("w-full")
ctx.render_chat_area = lambda: _render_messages(ctx)
ctx.refresh_turn_status = lambda: render_turn_status(
ctx, retry_cb=ctx.retry_cbs.get(f"{ctx.user_id}:{ctx.chat_id}")
)
ctx.scroll_chat_bottom = lambda: _scroll_to_bottom(ctx)
_render_messages(ctx)