#!/usr/bin/env python3
"""
HITL review/input/permission panels for the Klea TUI (ADR-0046).
When a run pauses at an interrupt, the app mounts an inline panel in the chat
panel rather than a modal, so the transcript stays scrollable while the user
reads the ask and decides. A ``review`` ask offers Approve / Request changes /
Cancel; an ``input`` ask offers one field per question plus Answer / Cancel; a
``permission`` ask offers once/session/not-a-path/deny per requested path
(ADR-0007).
Each panel posts :class:`HitlDecisionPanel.Decided` (carrying a
:class:`HitlDecision`) instead of dismissing; the app removes the panel and
resumes the same run.
File: klea_utils/ui/tui/components/hitl.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail dot com>
"""
from __future__ import annotations
import logging
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, ClassVar
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.message import Message
from textual.widgets import Button, Collapsible, Input, RadioButton, RadioSet, Rule
from klea_utils.ui.tui.components.buttons import action_button
from klea_utils.ui.tui.components.selectable import SelectableStatic
logger = logging.getLogger(__name__)
#: Path-permission decision options and their labels. "Not a path" is offered
#: only for a ``guess``-tier request (see :meth:`PermissionPanel._options`).
_PERMISSION_OPTION_LABELS = {
"once": "Once",
"session": "Session",
"not_a_path": "Not a path",
"deny": "Deny",
}
[docs]
@dataclass(frozen=True)
class HitlDecision:
"""The user's answer to a HITL ask.
:param response: The resume mapping sent to the server, or ``None`` on a
cancel.
:param cancel: Whether the user cancelled the interrupt instead of
answering.
"""
response: dict[str, Any] | None
cancel: bool = False
[docs]
class HitlDecisionPanel(VerticalScroll):
"""Base for the inline HITL panels.
A panel owns its form and posts :class:`Decided` when the user answers; the
app removes it and resumes the run. It is a sibling of the transcript (not
a modal), so the chat stays scrollable while the ask is open. It is
bounded (``max-height``) and scrolls internally so a long plan/digest does
not push the transcript off screen.
"""
BINDINGS: ClassVar[list[Binding]] = [Binding("escape", "cancel", "Cancel")]
DEFAULT_CSS = """
HitlDecisionPanel {
height: auto;
max-height: 60%;
border: round $primary;
padding: 0 2;
background: $surface;
}
HitlDecisionPanel #hitl-title,
HitlDecisionPanel #perm-title {
text-style: bold;
height: 1;
}
HitlDecisionPanel #hitl-headline {
text-style: bold;
height: auto;
}
HitlDecisionPanel .hitl-section {
height: auto;
margin-top: 1;
}
HitlDecisionPanel .hitl-section-title {
text-style: bold;
color: $text-muted;
height: auto;
margin-top: 0;
}
HitlDecisionPanel .hitl-section-body {
height: auto;
padding-left: 2;
}
HitlDecisionPanel Input {
width: 1fr;
margin-top: 1;
}
HitlDecisionPanel RadioSet {
margin-top: 1;
}
HitlDecisionPanel .hitl-hint {
color: $text-muted;
height: auto;
}
HitlDecisionPanel .perm-request {
height: auto;
margin-top: 1;
}
"""
[docs]
class Decided(Message):
"""Posted when the user answers an ask.
:param decision: The user's :class:`HitlDecision`.
"""
def __init__(self, decision: HitlDecision) -> None:
self.decision = decision
super().__init__()
[docs]
def action_cancel(self) -> None:
"""Cancel the ask (Escape or the Cancel button)."""
self.post_message(self.Decided(HitlDecision(None, cancel=True)))
[docs]
class HitlPanel(HitlDecisionPanel):
"""An inline review / input ask.
A ``review`` ask renders the headline and sections (the plan/digest) with an
approve / request-changes / cancel row; an ``input`` ask renders one field
per question with an answer / cancel row.
"""
def __init__(self, ask: dict[str, Any]) -> None:
"""Build the panel for the interrupt *ask*.
:param ask: The interrupt ``data`` (``kind``, ``question``/
``questions``, ``headline``/``sections``).
"""
super().__init__()
self.ask = ask
self._questions: list[dict[str, Any]] = []
[docs]
def compose(self) -> ComposeResult:
"""Compose the review or input form."""
kind = self.ask.get("kind", "input")
yield SelectableStatic(
"Plan review" if kind == "review" else "User input", id="hitl-title"
)
if kind == "review":
yield from self._compose_review()
else:
yield from self._compose_input()
def _compose_sections(self) -> ComposeResult:
"""Compose optional ask sections (review plan/progress, input history).
Each section is rendered inside a :class:`~textual.widgets.Collapsible`
so a multi-section ask (for example several candidate plans) can be
reviewed one entry at a time. When an ask carries more than one
section the titles are numbered (``Option 1``, ``Option 2``, ...) and
only the first section starts expanded, so comparing against the
history opens on the first entry while the rest stay collapsed until
the user expands them.
"""
sections = [
section
for section in (self.ask.get("sections") or [])
if isinstance(section, Mapping)
]
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
with Collapsible(
title=title,
collapsed=index > 0,
id=f"hitl-section-{index}",
classes="hitl-section",
):
yield SelectableStatic(
title,
id=f"hitl-section-{index}-title",
classes="hitl-section-title",
markup=False,
)
yield SelectableStatic(
str(section.get("body", "")),
id=f"hitl-section-{index}-body",
classes="hitl-section-body",
markup=False,
)
def _compose_review(self) -> ComposeResult:
"""Compose the review form (headline, sections, feedback, buttons).
The payload's ``headline`` and ``sections`` adapt to the review's
position (a plan approval or a progress checkpoint; ADR-0051); both
frontends render the same sections.
"""
headline = self.ask.get("headline")
if headline:
yield SelectableStatic(str(headline), id="hitl-headline", markup=False)
question = self.ask.get("question")
if question:
yield SelectableStatic(str(question), id="hitl-question", markup=False)
yield from self._compose_sections()
yield Input(
placeholder="Feedback (required to request changes)", id="hitl-feedback"
)
yield SelectableStatic("", id="hitl-feedback-hint", classes="hitl-hint")
with Horizontal(classes="actions"):
yield action_button(
"Approve", id="hitl-approve", variant="primary", compact=True
)
yield action_button("Request changes", id="hitl-revise", compact=True)
yield action_button(
"Cancel", id="hitl-cancel", variant="default", compact=True
)
def _compose_input(self) -> ComposeResult:
"""Compose the input form (plan history, questions, buttons)."""
yield from self._compose_sections()
questions = self.ask.get("questions")
if not isinstance(questions, list) or not questions:
fallback = self.ask.get("question") or "More information is needed."
questions = [{"question": fallback}]
self._questions = [q for q in questions if isinstance(q, dict)]
for index, question in enumerate(self._questions):
yield SelectableStatic(str(question.get("question", "")))
yield Input(id=f"hitl-answer-{index}")
with Horizontal(classes="actions"):
yield action_button(
"Answer", id="hitl-answer", variant="primary", compact=True
)
yield action_button(
"Cancel", id="hitl-cancel", variant="default", compact=True
)
[docs]
def on_mount(self) -> None:
"""Focus the first input field."""
fields = list(self.query(Input))
if fields:
fields[0].focus()
def _permission_key(request: Mapping[str, Any]) -> str:
"""Return the approval key for a path request (ADR-0007).
The resolved file for a ``sensitive`` request, the resolved directory for an
``outside`` one.
:param request: A permission request dict.
:returns: The key the decision is recorded against.
"""
if request.get("kind") == "sensitive":
return str(request.get("path", ""))
return str(request.get("directory", ""))
def _permission_path_display(request: Mapping[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: Mapping[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: Mapping[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 ""
[docs]
class PermissionPanel(HitlDecisionPanel):
"""An inline ask collecting a once/session/deny decision per requested path.
Each request (a sensitive file or an outside-project path, ADR-0007)
gets a :class:`~textual.widgets.RadioSet` defaulting to ``deny``; Submit
posts the per-key decisions, Cancel (or Escape) cancels the ask.
"""
def __init__(self, ask: dict[str, Any]) -> None:
"""Build the panel for the permission interrupt *ask*.
:param ask: The interrupt ``data`` (``requests``, ``question``).
"""
super().__init__()
self.ask = ask
def _requests(self) -> list[Mapping[str, Any]]:
"""Return the ask's path requests (a fallback when the list is empty)."""
requests = self.ask.get("requests")
if not isinstance(requests, list) or not requests:
requests = [{"kind": "outside", "path": self.ask.get("question", "")}]
return [request for request in requests if isinstance(request, Mapping)]
@staticmethod
def _title(request: Mapping[str, Any]) -> str:
"""Return the row title for a request."""
path = _permission_path_display(request)
if request.get("kind") == "sensitive":
return f"Sensitive file: {path}"
return f"Outside project path: {path}"
@staticmethod
def _hint(request: Mapping[str, Any]) -> str:
"""Return the tool-call context (tool, argument, value, confidence)."""
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
@staticmethod
def _coverage(request: Mapping[str, Any]) -> str:
"""Return the paths covered by this request's session approval."""
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}"
@staticmethod
def _options(request: Mapping[str, Any]) -> tuple[str, ...]:
"""Return the decision options for a request (``RadioSet`` order).
"Not a path" is offered only for a ``guess``-tier request, where the
gate may have mistaken a non-path value for a path.
"""
options = ["once", "session"]
if request.get("confidence") == "guess":
options.append("not_a_path")
options.append("deny")
return tuple(options)
[docs]
def compose(self) -> ComposeResult:
"""Compose one radio group per requested path, plus Submit / Cancel."""
yield SelectableStatic("Permission required", id="perm-title")
question = self.ask.get("question")
if question:
yield SelectableStatic(str(question), id="perm-question")
for index, request in enumerate(self._requests()):
if index:
yield Rule()
with Vertical(classes="perm-request"):
yield SelectableStatic(
f"{index + 1}. {self._title(request)}", markup=False
)
scope = _permission_scope(request)
if scope:
yield SelectableStatic(scope, classes="hitl-hint", markup=False)
hint = self._hint(request)
if hint:
yield SelectableStatic(hint, classes="hitl-hint", markup=False)
coverage = self._coverage(request)
if coverage:
yield SelectableStatic(coverage, classes="hitl-hint", markup=False)
with RadioSet(id=f"perm-{index}"):
for option in self._options(request):
yield RadioButton(
_PERMISSION_OPTION_LABELS[option],
id=f"perm-{index}-{option}",
value=option == "deny",
)
with Horizontal(classes="actions"):
yield action_button("Submit", id="perm-submit", variant="primary")
yield action_button("Cancel", id="perm-cancel", variant="default")
def _submit(self) -> None:
"""Post one decision per requested path."""
decisions: list[dict[str, str]] = []
for index, request in enumerate(self._requests()):
radio_set = self.query_one(f"#perm-{index}", RadioSet)
options = self._options(request)
pressed = radio_set.pressed_index
decision = options[pressed] if 0 <= pressed < len(options) else "deny"
decisions.append({"key": _permission_key(request), "decision": decision})
self.post_message(self.Decided(HitlDecision({"decisions": decisions})))