feat: honcho-tui — Textual TUI with three-pane layout, braille spinners, dialectic query

This commit is contained in:
Erosika 2026-04-15 17:28:17 -04:00
parent 28c10bfed1
commit ab0ff2f052
11 changed files with 1199 additions and 1 deletions

View File

@ -21,6 +21,7 @@ dependencies = [
"honcho-ai>=2.0.0",
"rich>=13.0.0",
"httpx>=0.27.0",
"textual>=0.70.0",
]
[project.urls]
@ -29,13 +30,14 @@ Repository = "https://github.com/plastic-labs/honcho"
[project.scripts]
honcho = "honcho_cli.main:app"
honcho-tui = "honcho_tui.app:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/honcho_cli"]
packages = ["src/honcho_cli", "src/honcho_tui"]
[tool.pytest.ini_options]
testpaths = ["tests"]

View File

@ -0,0 +1,3 @@
"""Honcho TUI — a Textual interface for Honcho."""
__version__ = "0.1.0"

View File

@ -0,0 +1,315 @@
"""Honcho TUI — main Textual application."""
from __future__ import annotations
import argparse
import os
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical
from textual.reactive import var
from honcho_tui import __version__
from honcho_tui.client import build_client, collect_page
from honcho_tui.theme import TCSS
from honcho_tui.widgets import PeerPanel, SessionsPanel, StatusBar, TranscriptPanel
from honcho_tui.widgets.sessions_panel import SessionSelected
from honcho_tui.widgets.transcript import QuerySubmitted
class HonchoTUI(App[None]):
"""Honcho TUI — memory that reasons."""
TITLE = "HONCHO"
CSS = TCSS
BINDINGS = [
Binding("q", "quit", "quit", priority=True),
Binding("ctrl+c", "quit", "quit", priority=True),
Binding("r", "refresh_all", "refresh"),
Binding("ctrl+d", "focus_query", "dialectic"),
Binding("escape", "blur_query", "back"),
Binding("tab", "focus_sessions", "sessions"),
]
# ── Reactive state ────────────────────────────────────────────────────────
workspace_id: var[str] = var("")
peer_id: var[str] = var("")
active_session: var[str] = var("")
uptime: var[int] = var(0)
def compose(self) -> ComposeResult:
with Horizontal(id="layout"):
yield SessionsPanel(id="sessions-panel")
with Vertical(id="center"):
yield TranscriptPanel(id="transcript")
yield PeerPanel(id="peer-panel")
yield StatusBar(id="status-bar")
# ── Lifecycle ─────────────────────────────────────────────────────────────
def on_mount(self) -> None:
try:
_, config = build_client()
self.workspace_id = config.workspace_id or ""
self.peer_id = config.peer_id or ""
except Exception:
self.workspace_id = ""
self.peer_id = ""
# Sync status bar
bar = self.query_one("#status-bar", StatusBar)
bar.workspace_id = self.workspace_id
bar.peer_id = self.peer_id
self.set_interval(1.0, self._tick_uptime)
transcript = self.query_one("#transcript", TranscriptPanel)
if not self.workspace_id:
transcript.show_no_workspace()
else:
self._load_sessions()
def _tick_uptime(self) -> None:
self.uptime += 1
bar = self.query_one("#status-bar", StatusBar)
bar.uptime = self.uptime
# ── Workers ───────────────────────────────────────────────────────────────
def _load_sessions(self) -> None:
panel = self.query_one("#sessions-panel", SessionsPanel)
panel.set_loading(True)
self.run_worker(self._fetch_sessions, thread=True)
def _fetch_sessions(self) -> None:
try:
client, config = build_client(
workspace_id=self.workspace_id or None,
peer_id=self.peer_id or None,
)
raw = collect_page(client.sessions())
sessions = [
{
"id": s.id,
"is_active": getattr(s, "is_active", True),
}
for s in raw
]
self.call_from_thread(
self.query_one("#sessions-panel", SessionsPanel).load_sessions,
sessions,
)
except Exception as e:
self.call_from_thread(
self.query_one("#sessions-panel", SessionsPanel).show_error,
str(e),
)
def _load_session(self, session_id: str) -> None:
transcript = self.query_one("#transcript", TranscriptPanel)
peer_panel = self.query_one("#peer-panel", PeerPanel)
transcript.show_loading(f"loading session {session_id[:20]}")
peer_panel.set_loading(True)
self.run_worker(
lambda: self._fetch_session(session_id),
thread=True,
)
def _fetch_session(self, session_id: str) -> None:
try:
client, config = build_client(
workspace_id=self.workspace_id or None,
peer_id=self.peer_id or None,
)
sess = client.session(session_id)
# Messages (first page, last 50)
try:
msgs_page = sess.messages()
raw_msgs = list(msgs_page.items)[-50:]
messages = [
{
"id": m.id,
"peer_id": m.peer_id,
"content": m.content,
"created_at": m.created_at,
}
for m in raw_msgs
]
except Exception:
messages = []
# Figure out peer to show
peer_id = self.peer_id
if not peer_id:
try:
session_peers = collect_page(sess.peers())
if session_peers:
peer_id = session_peers[0].id
except Exception:
peer_id = ""
# Peer card + conclusions
card = None
conclusions: list[dict] = []
queue = None
if peer_id:
p = client.peer(peer_id)
try:
card = p.get_card()
except Exception:
card = None
try:
conc_page = p.conclusions.list(size=20)
conclusions = [
{"id": c.id, "content": c.content, "created_at": c.created_at}
for c in conc_page.items
]
except Exception:
conclusions = []
try:
q = client.queue_status()
queue = {
"pending": getattr(q, "pending", 0),
"running": getattr(q, "running", 0) or getattr(q, "processing", 0),
"completed": getattr(q, "completed", 0) or getattr(q, "done", 0),
}
except Exception:
queue = None
# Update UI on main thread
self.call_from_thread(self._on_session_loaded, session_id, messages, peer_id, card, conclusions, queue)
except Exception as e:
self.call_from_thread(self._on_session_error, str(e))
def _on_session_loaded(
self,
session_id: str,
messages: list[dict],
peer_id: str,
card: str | None,
conclusions: list[dict],
queue: dict | None,
) -> None:
self.active_session = session_id
# Update status bar
bar = self.query_one("#status-bar", StatusBar)
bar.session_id = session_id
if peer_id:
bar.peer_id = peer_id
if queue:
bar.pending = queue.get("pending", 0)
bar.running = queue.get("running", 0)
self.query_one("#transcript", TranscriptPanel).load_messages(session_id, messages)
if peer_id:
self.query_one("#peer-panel", PeerPanel).load_peer(peer_id, card, conclusions, queue)
else:
self.query_one("#peer-panel", PeerPanel).set_loading(False)
def _on_session_error(self, msg: str) -> None:
self.query_one("#transcript", TranscriptPanel).show_error(msg)
self.query_one("#peer-panel", PeerPanel).show_error(msg)
def _run_query(self, query: str) -> None:
transcript = self.query_one("#transcript", TranscriptPanel)
transcript.show_loading(f"querying dialectic…")
self.run_worker(
lambda: self._fetch_query(query),
thread=True,
)
def _fetch_query(self, query: str) -> None:
try:
client, config = build_client(
workspace_id=self.workspace_id or None,
peer_id=self.peer_id or None,
)
peer_id = self.peer_id
if not peer_id:
self.call_from_thread(
self.query_one("#transcript", TranscriptPanel).show_error,
"no peer configured — set HONCHO_PEER_ID",
)
return
p = client.peer(peer_id)
response = p.chat(
query,
session=self.active_session or None,
)
self.call_from_thread(
self.query_one("#transcript", TranscriptPanel).append_response,
query,
str(response) if response else "(no response)",
True,
)
except Exception as e:
self.call_from_thread(
self.query_one("#transcript", TranscriptPanel).show_error,
str(e),
)
# ── Message handlers ──────────────────────────────────────────────────────
def on_session_selected(self, event: SessionSelected) -> None:
self._load_session(event.session_id)
def on_query_submitted(self, event: QuerySubmitted) -> None:
if not event.query.strip():
return
self._run_query(event.query)
# ── Actions ───────────────────────────────────────────────────────────────
def action_refresh_all(self) -> None:
if self.workspace_id:
self._load_sessions()
if self.active_session:
self._load_session(self.active_session)
def action_focus_query(self) -> None:
self.query_one("#transcript", TranscriptPanel).focus_input()
def action_blur_query(self) -> None:
self.query_one("#sessions-panel").focus()
def action_focus_sessions(self) -> None:
self.query_one("#sessions-panel").focus()
# ── Entry point ───────────────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(
prog="honcho-tui",
description=f"Honcho TUI v{__version__} — memory that reasons",
)
parser.add_argument("--workspace", "-w", metavar="ID", help="Workspace ID (overrides HONCHO_WORKSPACE_ID)")
parser.add_argument("--peer", "-p", metavar="ID", help="Peer ID (overrides HONCHO_PEER_ID)")
parser.add_argument("--version", "-V", action="version", version=f"honcho-tui {__version__}")
args = parser.parse_args()
if args.workspace:
os.environ["HONCHO_WORKSPACE_ID"] = args.workspace
if args.peer:
os.environ["HONCHO_PEER_ID"] = args.peer
app = HonchoTUI()
app.run()
if __name__ == "__main__":
main()

View File

@ -0,0 +1,36 @@
"""Thin Honcho client builder for the TUI, reusing CLI config."""
from __future__ import annotations
from honcho import Honcho
from honcho_cli.config import CLIConfig, get_client_kwargs
def build_client(workspace_id: str | None = None, peer_id: str | None = None) -> tuple[Honcho, CLIConfig]:
"""Build a Honcho client from the shared CLI config.
Applies optional runtime overrides without modifying the config file.
"""
config = CLIConfig.load()
if workspace_id:
config.workspace_id = workspace_id
if peer_id:
config.peer_id = peer_id
return Honcho(**get_client_kwargs(config)), config
def collect_page(page) -> list:
"""Collect all items from a SyncPage, walking pagination."""
try:
items: list = list(page._raw_items)
while page.has_next_page():
page = page.get_next_page()
if page is None:
break
items.extend(page._raw_items)
except AttributeError:
try:
items = list(page.items)
except AttributeError:
items = list(page)
return items

View File

@ -0,0 +1,249 @@
"""Color palette, spinner frames, and TCSS for Honcho TUI."""
from __future__ import annotations
# ── Color tokens ──────────────────────────────────────────────────────────────
BG = "#0b0e14"
BG_ELEVATED = "#141820"
BG_PANEL = "#0f1318"
BG_INPUT = "#0c1017"
FG = "#c9d1d9"
FG_MUTED = "#5c6370"
FG_DIM = "#8b949e"
BORDER = "#2a2d35"
ACCENT = "#FFBF00" # amber — primary
ACCENT_DIM = "#7a5c00" # dim amber border
BLUE = "#4169e1" # royal blue
CYAN = "#56d4dd"
PURPLE = "#bc8cff"
GREEN = "#8FBC8F" # good
WARN = "#FFD700" # warn
CRITICAL = "#FF6B6B" # critical
ORANGE = "#e6a855"
# ── Spinner frame sets ─────────────────────────────────────────────────────────
FRAMES_HELIX: list[str] = ["", "", "", "", "", "", "", ""]
FRAMES_ORBIT: list[str] = ["", "", "", ""]
FRAMES_DNA: list[str] = ["", "", "", "", "", "", "", "", "", ""]
FRAMES_PULSE: list[str] = ["·", "·", "", "", "", "", "", "·"]
# ── Symbols ────────────────────────────────────────────────────────────────────
DOT = ""
DOT_EMPTY = ""
DOT_HALF = ""
CURSOR = ""
TREE_LAST = ""
TREE_MID = ""
SEP = ""
CHEVRON_OPEN = ""
CHEVRON_CLOSED = ""
# ── TCSS ──────────────────────────────────────────────────────────────────────
TCSS = """
Screen {
background: #0b0e14;
color: #c9d1d9;
layers: base overlay;
}
/* MAIN LAYOUT */
#layout {
height: 1fr;
layout: horizontal;
}
/* LEFT: SESSIONS */
#sessions-panel {
width: 28;
background: #0f1318;
border-right: solid #2a2d35;
layout: vertical;
}
.panel-title {
height: 1;
background: #141820;
color: #FFBF00;
text-style: bold;
content-align: left middle;
padding: 0 1;
}
ListView {
background: #0f1318;
height: 1fr;
}
ListView:focus {
border: none;
}
ListItem {
background: #0f1318;
height: 2;
padding: 0 1;
color: #8b949e;
}
ListItem:hover {
background: #141820;
color: #c9d1d9;
}
ListItem.--highlight {
background: #1a1f2e;
color: #c9d1d9;
}
/* CENTER */
#center {
width: 1fr;
layout: vertical;
}
RichLog {
height: 1fr;
background: #0b0e14;
padding: 0 1;
scrollbar-color: #2a2d35;
scrollbar-background: #0b0e14;
scrollbar-corner-color: #0b0e14;
border: none;
}
/* QUERY BAR */
#query-bar {
height: 3;
background: #0c1017;
border-top: solid #2a2d35;
layout: horizontal;
align: left middle;
padding: 0 1;
}
#query-prefix {
width: 2;
color: #FFBF00;
text-style: bold;
content-align: left middle;
}
#query-input {
width: 1fr;
background: #0c1017;
color: #c9d1d9;
border: none;
padding: 0 0;
}
#query-input:focus {
border: none;
}
/* RIGHT: PEER PANEL */
#peer-panel {
width: 34;
background: #0f1318;
border-left: solid #2a2d35;
layout: vertical;
}
#peer-scroll {
height: 1fr;
background: #0f1318;
padding: 0 1;
scrollbar-color: #2a2d35;
scrollbar-background: #0f1318;
}
.section-label {
color: #FFBF00;
text-style: bold;
height: 1;
margin-top: 1;
}
.section-sep {
color: #2a2d35;
height: 1;
}
.card-body {
color: #8b949e;
}
.conclusion-row {
color: #8b949e;
height: auto;
}
.queue-row {
height: 1;
}
/* STATUS BAR */
StatusBar {
dock: bottom;
height: 1;
background: #141820;
color: #5c6370;
layout: horizontal;
padding: 0 1;
}
/* SPINNER */
.spinner {
color: #FFBF00;
width: 1;
}
/* COLLAPSIBLE */
Collapsible {
background: #0f1318;
border: none;
padding: 0;
margin: 0;
}
CollapsibleTitle {
color: #8b949e;
background: #0f1318;
padding: 0;
}
CollapsibleTitle:hover {
color: #c9d1d9;
background: #141820;
}
/* STARTUP OVERLAY */
#startup-overlay {
layer: overlay;
width: 60;
height: 12;
background: #141820;
border: solid #2a2d35;
align: center middle;
padding: 1 2;
layout: vertical;
}
#startup-title {
color: #FFBF00;
text-style: bold;
text-align: center;
}
#startup-msg {
color: #8b949e;
text-align: center;
margin-top: 1;
}
#workspace-input {
margin-top: 1;
background: #0c1017;
color: #c9d1d9;
border: solid #2a2d35;
}
"""

View File

@ -0,0 +1,8 @@
"""Honcho TUI widget library."""
from honcho_tui.widgets.sessions_panel import SessionsPanel
from honcho_tui.widgets.transcript import TranscriptPanel
from honcho_tui.widgets.peer_panel import PeerPanel
from honcho_tui.widgets.status_bar import StatusBar
__all__ = ["SessionsPanel", "TranscriptPanel", "PeerPanel", "StatusBar"]

View File

@ -0,0 +1,143 @@
"""Right panel: peer card, conclusions, queue status."""
from __future__ import annotations
from textual.app import ComposeResult
from textual.widget import Widget
from textual.widgets import Static
from textual.containers import VerticalScroll
from honcho_tui.theme import (
ACCENT,
ACCENT_DIM,
CRITICAL,
DOT,
DOT_EMPTY,
FG,
FG_DIM,
FG_MUTED,
FRAMES_ORBIT,
GREEN,
TREE_LAST,
TREE_MID,
WARN,
)
class PeerPanel(Widget):
"""Right-side panel: peer card + conclusions + queue status."""
DEFAULT_CSS = """
PeerPanel {
layout: vertical;
width: 100%;
height: 100%;
}
"""
_frame_idx: int = 0
_loading: bool = False
def compose(self) -> ComposeResult:
yield Static(f" [{ACCENT}]PEER[/{ACCENT}]", classes="panel-title")
yield VerticalScroll(id="peer-scroll")
def on_mount(self) -> None:
self._spinner_timer = self.set_interval(0.12, self._tick)
self._show_empty()
def _tick(self) -> None:
if self._loading:
self._frame_idx = (self._frame_idx + 1) % len(FRAMES_ORBIT)
header = self.query_one(".panel-title", Static)
frame = FRAMES_ORBIT[self._frame_idx]
header.update(f" [{ACCENT}]PEER[/{ACCENT}] [{ACCENT_DIM}]{frame}[/{ACCENT_DIM}]")
def set_loading(self, loading: bool) -> None:
self._loading = loading
if not loading:
header = self.query_one(".panel-title", Static)
header.update(f" [{ACCENT}]PEER[/{ACCENT}]")
def _show_empty(self) -> None:
scroll = self.query_one("#peer-scroll", VerticalScroll)
scroll.remove_children()
scroll.mount(Static(f"\n [{FG_DIM}]select a session[/{FG_DIM}]"))
def load_peer(
self,
peer_id: str,
card: str | None,
conclusions: list[dict],
queue: dict | None,
) -> None:
"""Populate panel with peer data (called from main thread)."""
self.set_loading(False)
# Update header with peer ID
short_pid = peer_id[:18] + "" if len(peer_id) > 18 else peer_id
header = self.query_one(".panel-title", Static)
header.update(f" [{ACCENT}]PEER[/{ACCENT}] [{FG_DIM}]{short_pid}[/{FG_DIM}]")
scroll = self.query_one("#peer-scroll", VerticalScroll)
scroll.remove_children()
# ── Card ─────────────────────────────────────────────────────────────
scroll.mount(Static(f"[{ACCENT}]CARD[/{ACCENT}]", classes="section-label"))
if card and card.strip():
# Wrap long lines, show first 400 chars
preview = card.strip()[:400]
if len(card.strip()) > 400:
preview += f"\n[{FG_MUTED}]… {len(card.strip()) - 400} more chars[/{FG_MUTED}]"
scroll.mount(Static(preview, classes="card-body"))
else:
scroll.mount(Static(f"[{FG_DIM}]no card yet[/{FG_DIM}]", classes="card-body"))
# ── Conclusions ───────────────────────────────────────────────────────
count = len(conclusions)
scroll.mount(Static(f"\n[{ACCENT}]CONCLUSIONS[/{ACCENT}] [{FG_MUTED}]({count})[/{FG_MUTED}]", classes="section-label"))
if not conclusions:
scroll.mount(Static(f"[{FG_DIM}]no conclusions yet[/{FG_DIM}]", classes="card-body"))
else:
for i, c in enumerate(conclusions[:12]):
content = c.get("content", "")
# Truncate to fit panel width
preview = content[:55] + "" if len(content) > 55 else content
prefix = TREE_LAST if i == min(len(conclusions), 12) - 1 else TREE_MID
scroll.mount(
Static(f"[{FG_DIM}]{prefix}[/{FG_DIM}][{FG}]{preview}[/{FG}]", classes="conclusion-row")
)
if count > 12:
scroll.mount(Static(f"[{FG_MUTED}] … {count - 12} more[/{FG_MUTED}]"))
# ── Queue ─────────────────────────────────────────────────────────────
scroll.mount(Static(f"\n[{ACCENT}]QUEUE[/{ACCENT}]", classes="section-label"))
if queue is None:
scroll.mount(Static(f"[{FG_DIM}]unavailable[/{FG_DIM}]"))
else:
pending = queue.get("pending", 0)
running = queue.get("running", 0)
completed = queue.get("completed", 0)
p_dot = f"[{WARN}]{DOT}[/{WARN}]" if pending > 0 else f"[{FG_MUTED}]{DOT_EMPTY}[/{FG_MUTED}]"
r_dot = f"[{GREEN}]{DOT}[/{GREEN}]" if running > 0 else f"[{FG_MUTED}]{DOT_EMPTY}[/{FG_MUTED}]"
d_dot = f"[{FG_DIM}]{DOT}[/{FG_DIM}]"
scroll.mount(Static(
f"{p_dot} [{FG_DIM}]{pending} pending[/{FG_DIM}] "
f"{r_dot} [{FG_DIM}]{running} running[/{FG_DIM}]",
classes="queue-row",
))
scroll.mount(Static(
f"{d_dot} [{FG_MUTED}]{completed} done[/{FG_MUTED}]",
classes="queue-row",
))
def show_error(self, msg: str) -> None:
self.set_loading(False)
scroll = self.query_one("#peer-scroll", VerticalScroll)
scroll.remove_children()
scroll.mount(Static(f"\n[red]{msg}[/red]"))

View File

@ -0,0 +1,90 @@
"""Left panel: session list with keyboard navigation."""
from __future__ import annotations
from textual.app import ComposeResult
from textual.message import Message
from textual.widgets import ListItem, ListView, Static
from textual.widget import Widget
from honcho_tui.theme import ACCENT, ACCENT_DIM, FG, FG_DIM, FRAMES_DNA, SEP
class SessionSelected(Message):
"""Posted when the user selects a session."""
def __init__(self, session_id: str) -> None:
self.session_id = session_id
super().__init__()
class _SessionItem(ListItem):
"""Single row in the session list."""
def __init__(self, session_id: str, is_active: bool = True, **kwargs) -> None:
super().__init__(**kwargs)
self.session_id = session_id
self.is_active = is_active
def compose(self) -> ComposeResult:
dot = f"[{ACCENT}]●[/{ACCENT}]" if self.is_active else f"[{FG_DIM}]○[/{FG_DIM}]"
short_id = self.session_id[:18] + "" if len(self.session_id) > 18 else self.session_id
yield Static(f" {dot} [{FG}]{short_id}[/{FG}]")
class SessionsPanel(Widget):
"""Left-side panel showing all sessions in the workspace."""
DEFAULT_CSS = """
SessionsPanel {
layout: vertical;
width: 100%;
height: 100%;
}
"""
_frame_idx: int = 0
_loading: bool = False
def compose(self) -> ComposeResult:
yield Static(f" [{ACCENT}]SESSIONS[/{ACCENT}]", classes="panel-title")
yield ListView(id="sessions-list")
def on_mount(self) -> None:
self._spinner_timer = self.set_interval(0.1, self._tick_spinner)
def _tick_spinner(self) -> None:
if self._loading:
self._frame_idx = (self._frame_idx + 1) % len(FRAMES_DNA)
def set_loading(self, loading: bool) -> None:
self._loading = loading
header = self.query_one(".panel-title", Static)
if loading:
frame = FRAMES_DNA[self._frame_idx]
header.update(f" [{ACCENT}]SESSIONS[/{ACCENT}] [{ACCENT_DIM}]{frame}[/{ACCENT_DIM}]")
else:
header.update(f" [{ACCENT}]SESSIONS[/{ACCENT}]")
def load_sessions(self, sessions: list[dict]) -> None:
"""Populate the list with session data (called from main thread)."""
self.set_loading(False)
lv = self.query_one("#sessions-list", ListView)
lv.clear()
if not sessions:
lv.append(ListItem(Static(f" [{FG_DIM}]no sessions[/{FG_DIM}]")))
return
for s in sessions:
lv.append(_SessionItem(s["id"], is_active=s.get("is_active", True)))
def show_error(self, msg: str) -> None:
self.set_loading(False)
lv = self.query_one("#sessions-list", ListView)
lv.clear()
short = msg[:22] + "" if len(msg) > 22 else msg
lv.append(ListItem(Static(f" [red]{short}[/red]")))
def on_list_view_selected(self, event: ListView.Selected) -> None:
item = event.item
if isinstance(item, _SessionItem):
self.post_message(SessionSelected(item.session_id))

View File

@ -0,0 +1,94 @@
"""Bottom status bar: workspace, peer, queue summary, uptime."""
from __future__ import annotations
from textual.widget import Widget
from textual.app import ComposeResult
from textual.widgets import Static
from textual.reactive import reactive
from honcho_tui.theme import ACCENT, FG_DIM, FG_MUTED, GREEN, WARN, DOT
class StatusBar(Widget):
"""Fixed bottom bar showing workspace, peer, queue, and uptime."""
DEFAULT_CSS = """
StatusBar {
dock: bottom;
height: 1;
layout: horizontal;
background: #141820;
color: #5c6370;
padding: 0 1;
}
"""
workspace_id: reactive[str] = reactive("")
peer_id: reactive[str] = reactive("")
session_id: reactive[str] = reactive("")
pending: reactive[int] = reactive(0)
running: reactive[int] = reactive(0)
uptime: reactive[int] = reactive(0)
def compose(self) -> ComposeResult:
yield Static("", id="status-content")
def on_mount(self) -> None:
self._render_status()
def watch_workspace_id(self, value: str) -> None:
self._render_status()
def watch_peer_id(self, value: str) -> None:
self._render_status()
def watch_session_id(self, value: str) -> None:
self._render_status()
def watch_pending(self, value: int) -> None:
self._render_status()
def watch_running(self, value: int) -> None:
self._render_status()
def watch_uptime(self, value: int) -> None:
self._render_status()
def _render_status(self) -> None:
try:
content = self.query_one("#status-content", Static)
except Exception:
return
parts: list[str] = []
if self.workspace_id:
ws = self.workspace_id[:16] + "" if len(self.workspace_id) > 16 else self.workspace_id
parts.append(f"[{ACCENT}]ws[/{ACCENT}] {ws}")
else:
parts.append(f"[{FG_MUTED}]no workspace[/{FG_MUTED}]")
if self.peer_id:
p = self.peer_id[:14] + "" if len(self.peer_id) > 14 else self.peer_id
parts.append(f"[{FG_DIM}]peer[/{FG_DIM}] {p}")
if self.session_id:
s = self.session_id[:12] + "" if len(self.session_id) > 12 else self.session_id
parts.append(f"[{FG_DIM}]sess[/{FG_DIM}] {s}")
# Queue
if self.running > 0:
parts.append(f"[{GREEN}]{DOT}[/{GREEN}] [{FG_DIM}]{self.running} running[/{FG_DIM}]")
if self.pending > 0:
parts.append(f"[{WARN}]{DOT}[/{WARN}] [{FG_DIM}]{self.pending} pending[/{FG_DIM}]")
# Uptime (right-aligned via spacer)
h = self.uptime // 3600
m = (self.uptime % 3600) // 60
s = self.uptime % 60
uptime_str = f"{h:02d}:{m:02d}:{s:02d}"
sep = f" [{FG_MUTED}]·[/{FG_MUTED}] "
left = sep.join(parts)
content.update(f"{left} [{FG_MUTED}]{uptime_str}[/{FG_MUTED}]")

View File

@ -0,0 +1,201 @@
"""Center panel: message transcript + dialectic query bar."""
from __future__ import annotations
import datetime
from rich.text import Text
from textual.app import ComposeResult
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Input, RichLog, Static
from honcho_tui.theme import (
ACCENT,
ACCENT_DIM,
BLUE,
CURSOR,
FG,
FG_DIM,
FG_MUTED,
FRAMES_HELIX,
GREEN,
ORANGE,
)
class QuerySubmitted(Message):
"""Posted when the user submits a dialectic query."""
def __init__(self, query: str) -> None:
self.query = query
super().__init__()
class TranscriptPanel(Widget):
"""Center pane: message log above, query input below."""
DEFAULT_CSS = """
TranscriptPanel {
layout: vertical;
width: 100%;
height: 100%;
}
"""
_frame_idx: int = 0
_streaming: bool = False
_cursor_on: bool = True
def compose(self) -> ComposeResult:
yield RichLog(highlight=False, markup=True, wrap=True, id="log")
with Widget(id="query-bar"):
yield Static(f"[{ACCENT}]>[/{ACCENT}]", id="query-prefix")
yield Input(
placeholder="query dialectic…",
id="query-input",
)
def on_mount(self) -> None:
self._spinner_timer = self.set_interval(0.08, self._tick)
self._cursor_timer = self.set_interval(0.42, self._blink_cursor)
self.show_welcome()
# ── Timers ────────────────────────────────────────────────────────────────
def _tick(self) -> None:
if self._streaming:
self._frame_idx = (self._frame_idx + 1) % len(FRAMES_HELIX)
def _blink_cursor(self) -> None:
self._cursor_on = not self._cursor_on
if self._streaming:
self._update_stream_cursor()
def _update_stream_cursor(self) -> None:
log = self.query_one("#log", RichLog)
cursor_char = CURSOR if self._cursor_on else " "
# Cursor line is updated via the streaming append path
_ = cursor_char # referenced during actual stream writes
# ── Public API ────────────────────────────────────────────────────────────
def show_welcome(self) -> None:
log = self.query_one("#log", RichLog)
log.clear()
t = Text()
t.append(" HONCHO\n", style=f"bold {ACCENT}")
t.append(" memory that reasons\n\n", style=FG_MUTED)
t.append(f" [{FG_MUTED}]select a session or query the dialectic[/{FG_MUTED}]\n")
log.write(t)
def show_no_workspace(self) -> None:
log = self.query_one("#log", RichLog)
log.clear()
t = Text()
t.append("\n no workspace configured\n\n", style=f"bold {ORANGE}")
t.append(" set HONCHO_WORKSPACE_ID or pass --workspace\n", style=FG_DIM)
t.append(" run 'honcho init' to configure\n", style=FG_DIM)
log.write(t)
def show_loading(self, label: str = "loading…") -> None:
self._streaming = True
log = self.query_one("#log", RichLog)
log.clear()
frame = FRAMES_HELIX[self._frame_idx]
t = Text()
t.append(f"\n {frame} ", style=ACCENT)
t.append(label, style=FG_DIM)
log.write(t)
def clear_loading(self) -> None:
self._streaming = False
def load_messages(self, session_id: str, messages: list[dict]) -> None:
"""Render a session's message history."""
self._streaming = False
log = self.query_one("#log", RichLog)
log.clear()
# Header
short_sid = session_id[:28] + "" if len(session_id) > 28 else session_id
header = Text()
header.append(f" session ", style=FG_MUTED)
header.append(short_sid, style=f"bold {FG}")
header.append(f"\n {'' * 50}\n", style=FG_MUTED)
log.write(header)
if not messages:
log.write(Text(f" [{FG_DIM}]no messages[/{FG_DIM}]\n"))
return
for msg in messages:
self._write_message(log, msg)
def _write_message(self, log: RichLog, msg: dict) -> None:
peer_id = msg.get("peer_id", "unknown")
content = msg.get("content", "")
created_at = msg.get("created_at")
# Timestamp
ts = ""
if created_at:
try:
dt = datetime.datetime.fromisoformat(str(created_at).replace("Z", "+00:00"))
ts = dt.strftime("%H:%M")
except Exception:
ts = str(created_at)[:5]
# Header line
header = Text()
header.append(f" {ts} ", style=FG_MUTED)
header.append(peer_id, style=f"bold {BLUE}")
header.append("\n")
log.write(header)
# Content
body = Text()
for line in content.splitlines():
body.append(f" {line}\n", style=FG)
if not content.strip():
body.append(" (empty)\n", style=FG_MUTED)
log.write(body)
log.write(Text("\n"))
def append_response(self, label: str, content: str, is_query: bool = False) -> None:
"""Append a query + response pair to the log."""
log = self.query_one("#log", RichLog)
self._streaming = False
if is_query:
q_text = Text()
q_text.append(f" > ", style=f"bold {ACCENT}")
q_text.append(f"{label}\n\n", style=f"bold {FG}")
log.write(q_text)
resp = Text()
resp.append(" dialectic\n", style=f"bold {GREEN}")
for line in content.splitlines():
resp.append(f" {line}\n", style=FG)
resp.append("\n")
log.write(resp)
def show_error(self, msg: str) -> None:
self._streaming = False
log = self.query_one("#log", RichLog)
t = Text()
t.append(f"\n error ", style="bold red")
t.append(f"{msg}\n", style=FG_DIM)
log.write(t)
def focus_input(self) -> None:
self.query_one("#query-input", Input).focus()
# ── Events ────────────────────────────────────────────────────────────────
def on_input_submitted(self, event: Input.Submitted) -> None:
query = event.value.strip()
if not query:
return
event.input.clear()
self.post_message(QuerySubmitted(query))

57
uv.lock
View File

@ -1323,6 +1323,7 @@ dependencies = [
{ name = "honcho-ai" },
{ name = "httpx" },
{ name = "rich" },
{ name = "textual" },
{ name = "typer" },
]
@ -1339,6 +1340,7 @@ requires-dist = [
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" },
{ name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.14.0" },
{ name = "rich", specifier = ">=13.0.0" },
{ name = "textual", specifier = ">=0.70.0" },
{ name = "typer", specifier = ">=0.15.0" },
]
provides-extras = ["dev"]
@ -1660,6 +1662,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/be/0a/b84e3e68a690ccfe6d64953c572772c685fcb0915b7f2ee3a87c22e388ab/langfuse-4.2.0-py3-none-any.whl", hash = "sha256:bfd760bf10fd0228f297f6369436620f76d16b589de46393d65706b27e4e4082", size = 475449, upload-time = "2026-04-10T11:55:23.624Z" },
]
[[package]]
name = "linkify-it-py"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "uc-micro-py" },
]
sdist = { url = "https://files.pythonhosted.org/packages/2e/c9/06ea13676ef354f0af6169587ae292d3e2406e212876a413bf9eece4eb23/linkify_it_py-2.1.0.tar.gz", hash = "sha256:43360231720999c10e9328dc3691160e27a718e280673d444c38d7d3aaa3b98b", size = 29158, upload-time = "2026-03-01T07:48:47.683Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b4/de/88b3be5c31b22333b3ca2f6ff1de4e863d8fe45aaea7485f591970ec1d3e/linkify_it_py-2.1.0-py3-none-any.whl", hash = "sha256:0d252c1594ecba2ecedc444053db5d3a9b7ec1b0dd929c8f1d74dce89f86c05e", size = 19878, upload-time = "2026-03-01T07:48:46.098Z" },
]
[[package]]
name = "mako"
version = "1.3.10"
@ -1684,6 +1698,11 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/94/54/e7d793b573f298e1c9013b8c4dade17d481164aa517d1d7148619c2cedbf/markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147", size = 87321, upload-time = "2025-08-11T12:57:51.923Z" },
]
[package.optional-dependencies]
linkify = [
{ name = "linkify-it-py" },
]
[[package]]
name = "markupsafe"
version = "3.0.3"
@ -1758,6 +1777,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/70/bc/6f1c2f612465f5fa89b95bead1f44dcb607670fd42891d8fdcd5d039f4f4/markupsafe-3.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:32001d6a8fc98c8cb5c947787c5d08b0a50663d139f1305bac5885d98d9b40fa", size = 14146, upload-time = "2025-09-27T18:37:28.327Z" },
]
[[package]]
name = "mdit-py-plugins"
version = "0.5.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b2/fd/a756d36c0bfba5f6e39a1cdbdbfdd448dc02692467d83816dff4592a1ebc/mdit_py_plugins-0.5.0.tar.gz", hash = "sha256:f4918cb50119f50446560513a8e311d574ff6aaed72606ddae6d35716fe809c6", size = 44655, upload-time = "2025-08-11T07:25:49.083Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fb/86/dd6e5db36df29e76c7a7699123569a4a18c1623ce68d826ed96c62643cae/mdit_py_plugins-0.5.0-py3-none-any.whl", hash = "sha256:07a08422fc1936a5d26d146759e9155ea466e842f5ab2f7d2266dd084c8dab1f", size = 57205, upload-time = "2025-08-11T07:25:47.597Z" },
]
[[package]]
name = "mdurl"
version = "0.1.2"
@ -3712,6 +3743,23 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/d7/c1/eb8f9debc45d3b7918a32ab756658a0904732f75e555402972246b0b8e71/tenacity-9.1.4-py3-none-any.whl", hash = "sha256:6095a360c919085f28c6527de529e76a06ad89b23659fa881ae0649b867a9d55", size = 28926, upload-time = "2026-02-07T10:45:32.24Z" },
]
[[package]]
name = "textual"
version = "8.2.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py", extra = ["linkify"] },
{ name = "mdit-py-plugins" },
{ name = "platformdirs" },
{ name = "pygments" },
{ name = "rich" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/cf/2f/d44f0f12b3ddb1f0b88f7775652e99c6b5a43fd733badf4ce064bdbfef4a/textual-8.2.3.tar.gz", hash = "sha256:beea7b86b03b03558a2224f0cc35252e60ef8b0c4353b117b2f40972902d976a", size = 1848738, upload-time = "2026-04-05T09:12:45.338Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0e/28/a81d6ce9f4804818bd1231a9a6e4d56ea84ebbe8385c49591444f0234fa2/textual-8.2.3-py3-none-any.whl", hash = "sha256:5008ac581bebf1f6fa0520404261844a231e5715fdbddd10ca73916a3af48ca2", size = 724231, upload-time = "2026-04-05T09:12:48.747Z" },
]
[[package]]
name = "threadpoolctl"
version = "3.6.0"
@ -3906,6 +3954,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b0/70/d460bd685a170790ec89317e9bd33047988e4bce507b831f5db771e142de/tzdata-2026.1-py2.py3-none-any.whl", hash = "sha256:4b1d2be7ac37ceafd7327b961aa3a54e467efbdb563a23655fbfe0d39cfc42a9", size = 348952, upload-time = "2026-04-03T11:25:20.313Z" },
]
[[package]]
name = "uc-micro-py"
version = "2.0.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/78/67/9a363818028526e2d4579334460df777115bdec1bb77c08f9db88f6389f2/uc_micro_py-2.0.0.tar.gz", hash = "sha256:c53691e495c8db60e16ffc4861a35469b0ba0821fe409a8a7a0a71864d33a811", size = 6611, upload-time = "2026-03-01T06:31:27.526Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/61/73/d21edf5b204d1467e06500080a50f79d49ef2b997c79123a536d4a17d97c/uc_micro_py-2.0.0-py3-none-any.whl", hash = "sha256:3603a3859af53e5a39bc7677713c78ea6589ff188d70f4fee165db88e22b242c", size = 6383, upload-time = "2026-03-01T06:31:26.257Z" },
]
[[package]]
name = "urllib3"
version = "2.6.3"