Coverage for agentos/desktop/tui.py: 11%
310 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +0800
1"""
2Terminal User Interface (TUI) — Textual-based native terminal agent cockpit.
4OpenCode/Cursor-style terminal experience:
5 - Four-panel layout: file tree | chat/result | terminal/editor | market
6 - Full keyboard navigation (vim-style + standard shortcuts)
7 - Real-time streaming output
8 - Session persistence
9 - Dark/light themes
10 - Skill marketplace browser (ctrl+m)
12Requirements: pip install textual
14Usage:
15 agentos tui # Launch TUI
16 agentos tui --safe # Safe mode (read-only)
17 agentos tui --theme dark # Dark theme (default)
18 agentos tui --market # Open market panel on start
19 agentos tui --store-url :18900 # Custom skill store server URL
20"""
22from __future__ import annotations
24import asyncio
25import json
26import os
27import time
28from dataclasses import dataclass, field
29from pathlib import Path
30from typing import Any, Optional
32try:
33 from textual.app import App, ComposeResult
34 from textual.widgets import (
35 Header, Footer, Tree, TextArea, Input, Static, RichLog,
36 ListView, ListItem, Label, Button, TabbedContent, TabPane,
37 )
38 from textual.containers import Horizontal, Vertical, Container, ScrollableContainer
39 from textual.binding import Binding
40 from textual.reactive import reactive
41 from textual.screen import ModalScreen
42 from textual.message import Message
43 TEXTUAL_AVAILABLE = True
44except ImportError:
45 TEXTUAL_AVAILABLE = False
48# ── Models ──
50@dataclass
51class TUIConfig:
52 """TUI persistent configuration."""
53 theme: str = "dark"
54 work_dir: str = field(default_factory=lambda: str(Path.home()))
55 font_size: int = 14
56 max_history: int = 500
57 auto_scroll: bool = True
58 store_url: str = "http://127.0.0.1:18900"
60 def save(self, path: str = "~/.agentos/tui.json"):
61 p = Path(path).expanduser()
62 p.parent.mkdir(parents=True, exist_ok=True)
63 p.write_text(json.dumps({
64 "theme": self.theme, "work_dir": self.work_dir,
65 "font_size": self.font_size, "max_history": self.max_history,
66 "auto_scroll": self.auto_scroll, "store_url": self.store_url,
67 }, indent=2))
69 @classmethod
70 def load(cls, path: str = "~/.agentos/tui.json") -> "TUIConfig":
71 p = Path(path).expanduser()
72 if p.exists():
73 data = json.loads(p.read_text())
74 return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
75 return cls()
78# ── Stubs (when textual not installed) ──
80if not TEXTUAL_AVAILABLE:
81 class FileTree: pass
82 class ChatArea: pass
83 class TerminalPanel: pass
84 class StatusBar: pass
85 class MarketPanel: pass
86 class _StubApp:
87 def run(self): raise RuntimeError("textual not installed: pip install textual")
90if TEXTUAL_AVAILABLE:
92 class FileTree(Vertical):
93 """Left panel: clickable file tree."""
94 def compose(self) -> ComposeResult:
95 yield Static(" File Tree ", id="panel-title")
96 yield Tree("~/", id="file-tree")
98 def on_mount(self) -> None:
99 self._populate_tree()
101 def _populate_tree(self) -> None:
102 tree = self.query_one("#file-tree", Tree)
103 tree.clear()
104 root = tree.root
105 root.set_label(str(Path.home()))
106 try:
107 items = sorted(Path.home().iterdir(), key=lambda p: (p.is_file(), p.name))
108 for item in items[:50]:
109 icon = " " if item.is_dir() else " "
110 node = root.add(f"{icon}{item.name}", data=str(item))
111 if item.is_dir():
112 self._add_dir_children(node, item, depth=0)
113 except PermissionError:
114 pass
116 def _add_dir_children(self, parent, path: Path, depth: int):
117 if depth > 1:
118 return
119 try:
120 for child in sorted(path.iterdir(), key=lambda p: (p.is_file(), p.name))[:15]:
121 icon = " " if child.is_dir() else " "
122 node = parent.add(f"{icon}{child.name}", data=str(child))
123 if child.is_dir():
124 self._add_dir_children(node, child, depth + 1)
125 except PermissionError:
126 pass
129 class ChatArea(Vertical):
130 """Center panel: chat interface."""
131 def compose(self) -> ComposeResult:
132 yield Static(" Chat ", id="panel-title")
133 yield RichLog(id="chat-log", highlight=True, markup=True)
134 yield Input(placeholder="Ask anything... (Enter to send)", id="chat-input")
136 def add_message(self, role: str, text: str):
137 log = self.query_one("#chat-log", RichLog)
138 if role == "user":
139 log.write(f"\n[bold green]>[/bold green] {text}")
140 elif role == "error":
141 log.write(f"\n[bold red]![/bold red] {text}")
142 else:
143 log.write(f"\n[bold blue]•[/bold blue] {text}")
146 class TerminalPanel(Vertical):
147 """Right panel: terminal / shell output."""
148 def compose(self) -> ComposeResult:
149 yield Static(" Terminal ", id="panel-title")
150 yield RichLog(id="terminal-log", highlight=True, markup=True)
151 yield Input(placeholder="$ command...", id="terminal-input")
153 def on_mount(self) -> None:
154 log = self.query_one("#terminal-log", RichLog)
155 log.write("[dim]$ agentos tui started[/dim]")
156 log.write(f"[dim] cwd: {os.getcwd()}[/dim]")
159 class StatusBar(Horizontal):
160 """Bottom status bar with metrics."""
161 def compose(self) -> ComposeResult:
162 yield Static(" Sessions: 0 ", id="status-sessions")
163 yield Static(" Tasks: 0 ", id="status-tasks")
164 yield Static(" Mode: READY ", id="status-mode")
167 class MarketPanel(ScrollableContainer):
168 """Skill marketplace panel with embedded web-like view.
170 Shows skill sources from multiple marketplaces (OpenClaw, ClawHub,
171 SkillsMP, LobeHub, etc.) and supports one-click install for compatible
172 sources. External sources open in the system browser.
174 Requires the skill store server: agentos skill-store
175 """
177 class SkillInstalled(Message):
178 """Posted when a skill is installed via the market."""
179 def __init__(self, skill_name: str, source: str) -> None:
180 self.skill_name = skill_name
181 self.source = source
182 super().__init__()
184 def __init__(self, store_url: str = "http://127.0.0.1:18900", **kwargs):
185 super().__init__(**kwargs)
186 self._store_url = store_url
187 self._sources: list[dict] = []
188 self._installed: set[str] = set()
189 self._active_source: str = "openclaw"
190 self._skills: list[dict] = []
192 def compose(self) -> ComposeResult:
193 yield Static(" Skill Marketplace ", id="panel-title")
194 with Horizontal():
195 with Vertical(classes="market-sidebar", id="market-sidebar-container"):
196 yield Static(" Sources", classes="market-section-title")
197 yield ListView(id="market-source-list")
198 with Vertical(classes="market-content", id="market-content-container"):
199 yield Static(" Skills", classes="market-section-title")
200 yield Input(placeholder="Search skills...", id="market-search")
201 yield RichLog(id="market-skill-log", highlight=True, markup=True)
203 def on_mount(self) -> None:
204 self._init_sources()
205 self._load_skills()
207 def _init_sources(self) -> None:
208 try:
209 import urllib.request, json as _json
210 with urllib.request.urlopen(f"{self._store_url}/api/sources", timeout=5) as resp:
211 self._sources = _json.loads(resp.read())
212 except Exception:
213 self._sources = [
214 {"id": "openclaw", "name": "OpenClaw Skill Store", "skill_count": "14+",
215 "installable": True, "web_url": "https://github.com/nicepkg/openclaw-skill-store",
216 "description": "OpenClaw 官方社区技能商店"},
217 {"id": "clawhub", "name": "ClawHub", "skill_count": "5,700+",
218 "installable": False, "web_url": "https://github.com/clawhub-community/skills",
219 "description": "ClawHub 社区技能聚合"},
220 {"id": "skillsmp", "name": "SkillsMP", "skill_count": "164万+",
221 "installable": False, "web_url": "https://skills.mp/",
222 "description": "技能界的 Google，最大索引平台"},
223 {"id": "lobehub", "name": "LobeHub Skills", "skill_count": "28万+",
224 "installable": False, "web_url": "https://lobehub.com/skills",
225 "description": "LobeHub 生态精品平台"},
226 {"id": "skillhub", "name": "SkillHub Club", "skill_count": "1.6万+",
227 "installable": False, "web_url": "https://skillhub.club/",
228 "description": "AI 评分品质筛选市集"},
229 {"id": "skills_sh", "name": "skills.sh", "skill_count": "67万+",
230 "installable": False, "web_url": "https://skills.sh/",
231 "description": "Vercel Labs 一键安装平台"},
232 {"id": "awesome", "name": "awesome-agent-skills", "skill_count": "380+",
233 "installable": False, "web_url": "https://github.com/nicepkg/awesome-agent-skills",
234 "description": "人工审核精选技能合集"},
235 ]
237 lst = self.query_one("#market-source-list", ListView)
238 lst.clear()
239 for src in self._sources:
240 icon = "[bold green]↓[/bold green]" if src.get("installable") else "[bold blue]↗[/bold blue]"
241 cnt = src.get("skill_count", "?")
242 lst.append(ListItem(
243 Label(f"{icon} {src['name']} [dim]({cnt})[/dim]"),
244 name=src["id"],
245 ))
247 def on_list_view_selected(self, event: ListView.Selected) -> None:
248 if event.item.name:
249 raw = event.item.name
250 self._active_source = raw.value if hasattr(raw, 'value') else str(raw)
251 self._load_skills()
253 def on_input_submitted(self, event: Input.Submitted) -> None:
254 if event.input.id == "market-search":
255 self._filter_skills(event.value.strip())
257 def _load_skills(self) -> None:
258 log = self.query_one("#market-skill-log", RichLog)
259 log.clear()
260 src = next((s for s in self._sources if s["id"] == self._active_source), None)
261 if not src:
262 log.write("[bold red]Source not found[/bold red]")
263 return
265 if not src.get("installable"):
266 log.write(f"[bold blue]{src['name']}[/bold blue]")
267 log.write(f"[dim]{src.get('description', 'External marketplace')}[/dim]\n")
268 log.write(f"[bold]Open in browser:[/bold]")
269 log.write(f" [link={src['web_url']}]{src['web_url']}[/link]")
270 log.write(f" [link={src.get('url', src['web_url'])}]{src.get('url', '')}[/link]\n")
271 log.write("[dim]External marketplace — open the URL above in your browser to browse and install.[/dim]")
272 return
274 try:
275 import urllib.request, json as _json
276 url = f"{self._store_url}/api/skills?source={self._active_source}"
277 with urllib.request.urlopen(url, timeout=5) as resp:
278 data = _json.loads(resp.read())
279 self._skills = data.get("skills", [])
280 except Exception:
281 self._skills = self._fallback_skills()
283 log.write(f"[bold blue]{src['name']}[/bold blue] [dim]({len(self._skills)} skills)[/dim]\n")
284 for skill in self._skills:
285 name = skill["name"]
286 desc = skill.get("description", "")
287 tags = " ".join(f"[dim]#{t}[/dim]" for t in skill.get("tags", []))
288 installed = "[bold green]✓[/bold green]" if name in self._installed else "[dim]○[/dim]"
289 log.write(f" {installed} [bold]{name}[/bold] {tags}")
290 log.write(f" [dim]{desc}[/dim]")
291 log.write("")
292 log.write(f"[dim]Use 'agentos skill-store' to start the web UI for one-click install.[/dim]")
294 def _fallback_skills(self) -> list[dict]:
295 return [
296 {"name": "skill-creator", "description": "Create new skills from templates", "tags": ["meta"]},
297 {"name": "pdf-tools", "description": "PDF manipulation, merge, split, extract", "tags": ["document"]},
298 {"name": "xlsx-tools", "description": "Excel/Spreadsheet processing", "tags": ["document"]},
299 {"name": "docx-tools", "description": "Word document processing", "tags": ["document"]},
300 {"name": "pptx-tools", "description": "PowerPoint generation", "tags": ["document"]},
301 {"name": "image-tools", "description": "Image processing, resize, convert", "tags": ["media"]},
302 {"name": "web-search", "description": "Advanced web search with multiple engines", "tags": ["search"]},
303 {"name": "browser-automation", "description": "Browser automation with Playwright", "tags": ["browser"]},
304 {"name": "code-review", "description": "Automated code review", "tags": ["code"]},
305 {"name": "git-tools", "description": "Git workflow automation", "tags": ["git"]},
306 {"name": "file-organizer", "description": "File organization and cleanup", "tags": ["files"]},
307 {"name": "data-analysis", "description": "Data analysis and visualization", "tags": ["data"]},
308 {"name": "api-tester", "description": "API testing and docs generation", "tags": ["api"]},
309 {"name": "markdown-tools", "description": "Markdown editing and conversion", "tags": ["document"]},
310 ]
312 def _filter_skills(self, query: str) -> None:
313 log = self.query_one("#market-skill-log", RichLog)
314 if not query:
315 self._load_skills()
316 return
317 log.clear()
318 matched = [s for s in self._skills if
319 query.lower() in s["name"].lower() or
320 query.lower() in s.get("description", "").lower()]
321 log.write(f"[dim]Search: '{query}' — {len(matched)} results[/dim]\n")
322 for skill in matched:
323 name = skill["name"]
324 desc = skill.get("description", "")
325 installed = "[bold green]✓[/bold green]" if name in self._installed else "[dim]○[/dim]"
326 log.write(f" {installed} [bold]{name}[/bold]")
327 log.write(f" [dim]{desc}[/dim]")
329 def set_store_url(self, url: str) -> None:
330 self._store_url = url
332 @property
333 def installed_skills(self) -> set[str]:
334 return self._installed
337 # ── Main Application ──
339 class AgentOSTUI(App):
340 """Main TUI application — four-panel agent cockpit."""
342 CSS = """
343 Screen {
344 layout: grid;
345 grid-size: 3 3;
346 grid-gutter: 1 2;
347 background: $surface;
348 }
350 #panel-file {
351 row-span: 2;
352 border: solid $primary;
353 background: $panel;
354 }
356 #panel-chat {
357 row-span: 2;
358 border: solid $primary;
359 background: $panel;
360 }
362 #panel-terminal {
363 row-span: 2;
364 border: solid $primary;
365 background: $panel;
366 }
368 #panel-market {
369 row-span: 2;
370 border: solid $success;
371 background: $panel;
372 }
374 #panel-status {
375 column-span: 3;
376 height: 1;
377 background: $primary-darken-2;
378 color: $text;
379 }
381 #panel-title {
382 background: $primary-darken-1;
383 color: $text;
384 text-style: bold;
385 padding: 0 1;
386 height: 1;
387 }
389 #file-tree {
390 height: 1fr;
391 overflow-y: auto;
392 }
394 #chat-log {
395 height: 1fr;
396 overflow-y: auto;
397 }
399 #terminal-log {
400 height: 1fr;
401 overflow-y: auto;
402 }
404 #chat-input, #terminal-input {
405 dock: bottom;
406 height: 3;
407 }
409 .market-sidebar {
410 width: 32;
411 border: solid $primary-darken-2;
412 }
414 .market-section-title {
415 background: $primary-darken-1;
416 color: $text;
417 text-style: bold;
418 height: 1;
419 }
421 #market-source-list {
422 height: 1fr;
423 }
425 #market-search {
426 dock: top;
427 height: 3;
428 }
430 #market-skill-log {
431 height: 1fr;
432 overflow-y: auto;
433 }
434 """
436 BINDINGS = [
437 Binding("ctrl+q", "quit", "Quit", show=True),
438 Binding("ctrl+s", "save", "Save", show=True),
439 Binding("ctrl+r", "refresh", "Refresh", show=True),
440 Binding("ctrl+t", "focus_terminal", "Terminal", show=True),
441 Binding("ctrl+c", "focus_chat", "Chat", show=True),
442 Binding("ctrl+f", "focus_files", "Files", show=True),
443 Binding("ctrl+m", "toggle_market", "Market", show=True),
444 Binding("f5", "refresh", "Refresh", show=False),
445 ]
447 _config: TUIConfig = TUIConfig()
448 _message_handler: Optional[callable] = None
449 _market_visible: bool = False
451 def __init__(self, config: TUIConfig = None, start_market: bool = False):
452 super().__init__()
453 if config:
454 self._config = config
455 self._market_visible = start_market
457 def compose(self) -> ComposeResult:
458 yield Header("NexusAgentOS", icon="")
459 yield FileTree(id="panel-file")
460 yield ChatArea(id="panel-chat")
461 yield TerminalPanel(id="panel-terminal")
462 yield StatusBar(id="panel-status")
463 yield Footer()
465 def on_mount(self) -> None:
466 self.title = "NexusAgentOS TUI"
467 self.sub_title = f"v1.7.6 — {self._config.work_dir}"
469 # Mount market panel
470 market = MarketPanel(store_url=self._config.store_url, id="panel-market")
471 market.display = self._market_visible
472 self.mount(market, before="#panel-status")
474 # Update status
475 status_sessions = self.query_one("#status-sessions", Static)
476 status_sessions.update(" Sessions: 0 ")
478 # ── Actions ──
480 def action_quit(self) -> None:
481 self.exit()
483 def action_save(self) -> None:
484 self._config.save()
485 chat = self.query_one("#chat-log", RichLog)
486 chat.write("[dim]Config saved.[/dim]")
488 def action_refresh(self) -> None:
489 self.query_one("#file-tree", Tree).root.remove_children()
490 self.query_one("#panel-file", FileTree)._populate_tree()
492 def action_focus_terminal(self) -> None:
493 self.query_one("#terminal-input", Input).focus()
495 def action_focus_chat(self) -> None:
496 self.query_one("#chat-input", Input).focus()
498 def action_focus_files(self) -> None:
499 self.query_one("#file-tree", Tree).focus()
501 def action_toggle_market(self) -> None:
502 market = self.query_one("#panel-market", MarketPanel)
503 self._market_visible = not self._market_visible
504 market.display = self._market_visible
505 if self._market_visible:
506 market._init_sources()
507 market._load_skills()
508 market.query_one("#market-search", Input).focus()
509 self.sub_title = f"v1.7.6 — Market | {self._config.work_dir}"
510 else:
511 self.sub_title = f"v1.7.6 — {self._config.work_dir}"
513 # ── Input Handlers ──
515 def on_input_submitted(self, event: Input.Submitted) -> None:
516 if event.input.id == "chat-input" and event.value.strip():
517 self._handle_chat(event.value.strip())
518 event.input.clear()
519 elif event.input.id == "terminal-input" and event.value.strip():
520 self._handle_terminal(event.value.strip())
521 event.input.clear()
523 def _handle_chat(self, message: str) -> None:
524 chat = self.query_one("#panel-chat", ChatArea)
525 chat.add_message("user", message)
526 if self._message_handler:
527 asyncio.create_task(self._dispatch_to_handler(message))
528 else:
529 chat.add_message("agent", f"Echo: {message}")
531 async def _dispatch_to_handler(self, message: str) -> None:
532 chat = self.query_one("#panel-chat", ChatArea)
533 try:
534 result = await self._message_handler(message)
535 chat.add_message("agent", str(result))
536 except Exception as e:
537 chat.add_message("error", f"Error: {e}")
539 def _handle_terminal(self, command: str) -> None:
540 log = self.query_one("#terminal-log", RichLog)
541 log.write(f"\n$ {command}")
542 try:
543 import subprocess
544 result = subprocess.run(
545 command, shell=True, capture_output=True,
546 text=True, timeout=30, cwd=self._config.work_dir,
547 )
548 if result.stdout:
549 log.write(result.stdout.rstrip())
550 if result.stderr:
551 log.write(f"[bold red]{result.stderr.rstrip()}[/bold red]")
552 except Exception as e:
553 log.write(f"[bold red]{e}[/bold red]")
555 # ── Public API ──
557 def set_message_handler(self, handler: callable):
558 self._message_handler = handler
560 def add_message(self, role: str, text: str):
561 chat = self.query_one("#panel-chat", ChatArea)
562 chat.add_message(role, text)
565# ── Entry Point ──
567def launch_tui(
568 message_handler=None,
569 work_dir: str = "",
570 theme: str = "dark",
571 start_market: bool = False,
572 store_url: str = "http://127.0.0.1:18900",
573) -> None:
574 """Launch the TUI application.
576 Args:
577 message_handler: async callable(msg: str) -> str for chat responses.
578 work_dir: Working directory for file tree.
579 theme: 'dark' or 'light'.
580 start_market: Open market panel on launch.
581 store_url: URL of the skill store server.
582 """
583 if not TEXTUAL_AVAILABLE:
584 print("ERROR: textual not installed. Run: pip install textual")
585 return
587 config = TUIConfig.load()
588 if work_dir:
589 config.work_dir = work_dir
590 if theme:
591 config.theme = theme
592 if store_url:
593 config.store_url = store_url
595 app = AgentOSTUI(config=config, start_market=start_market)
596 if message_handler:
597 app.set_message_handler(message_handler)
599 app.run()
602if __name__ == "__main__":
603 launch_tui()