Coverage for agentos/desktop/tui.py: 10%

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

27from dataclasses import dataclass, field

28from pathlib import Path

29from typing import Optional

31try:

32 from textual.app import App, ComposeResult

33 from textual.widgets import (

34 Header, Footer, Tree, TextArea, Input, Static, RichLog,

35 ListView, ListItem, Label, Button, TabbedContent, TabPane,

36 )

37 from textual.containers import Horizontal, Vertical, Container, ScrollableContainer

38 from textual.binding import Binding

39 from textual.reactive import reactive

40 from textual.screen import ModalScreen

41 from textual.message import Message

42 TEXTUAL_AVAILABLE = True

43except ImportError:

44 TEXTUAL_AVAILABLE = False

47# ── Models ──

49@dataclass

50class TUIConfig:

51 """TUI persistent configuration."""

52 theme: str = "dark"

53 work_dir: str = field(default_factory=lambda: str(Path.home()))

54 font_size: int = 14

55 max_history: int = 500

56 auto_scroll: bool = True

57 store_url: str = "http://127.0.0.1:18900"

59 def save(self, path: str = "~/.agentos/tui.json"):

60 p = Path(path).expanduser()

61 p.parent.mkdir(parents=True, exist_ok=True)

62 p.write_text(json.dumps({

63 "theme": self.theme, "work_dir": self.work_dir,

64 "font_size": self.font_size, "max_history": self.max_history,

65 "auto_scroll": self.auto_scroll, "store_url": self.store_url,

66 }, indent=2))

68 @classmethod

69 def load(cls, path: str = "~/.agentos/tui.json") -> "TUIConfig":

70 p = Path(path).expanduser()

71 if p.exists():

72 data = json.loads(p.read_text())

73 return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})

74 return cls()

77# ── Stubs (when textual not installed) ──

79if not TEXTUAL_AVAILABLE:

80 class FileTree: pass

81 class ChatArea: pass

82 class TerminalPanel: pass

83 class StatusBar: pass

84 class MarketPanel: pass

85 class _StubApp:

86 def run(self): raise RuntimeError("textual not installed: pip install textual")

89if TEXTUAL_AVAILABLE:

91 class FileTree(Vertical):

92 """Left panel: clickable file tree."""

93 def compose(self) -> ComposeResult:

94 yield Static(" File Tree ", id="panel-title")

95 yield Tree("~/", id="file-tree")

97 def on_mount(self) -> None:

98 self._populate_tree()

100 def _populate_tree(self) -> None:

101 tree = self.query_one("#file-tree", Tree)

102 tree.clear()

103 root = tree.root

104 root.set_label(str(Path.home()))

105 try:

106 items = sorted(Path.home().iterdir(), key=lambda p: (p.is_file(), p.name))

107 for item in items[:50]:

108 icon = " " if item.is_dir() else " "

109 node = root.add(f"{icon}{item.name}", data=str(item))

110 if item.is_dir():

111 self._add_dir_children(node, item, depth=0)

112 except PermissionError:

113 pass

114

115 def _add_dir_children(self, parent, path: Path, depth: int):

116 if depth > 1:

117 return

118 try:

119 for child in sorted(path.iterdir(), key=lambda p: (p.is_file(), p.name))[:15]:

120 icon = " " if child.is_dir() else " "

121 node = parent.add(f"{icon}{child.name}", data=str(child))

122 if child.is_dir():

123 self._add_dir_children(node, child, depth + 1)

124 except PermissionError:

125 pass

126

127

128 class ChatArea(Vertical):

129 """Center panel: chat interface."""

130 def compose(self) -> ComposeResult:

131 yield Static(" Chat ", id="panel-title")

132 yield RichLog(id="chat-log", highlight=True, markup=True)

133 yield Input(placeholder="Ask anything... (Enter to send)", id="chat-input")

134

135 def add_message(self, role: str, text: str):

136 log = self.query_one("#chat-log", RichLog)

137 if role == "user":

138 log.write(f"\n[bold green]>[/bold green] {text}")

139 elif role == "error":

140 log.write(f"\n[bold red]![/bold red] {text}")

141 else:

142 log.write(f"\n[bold blue]•[/bold blue] {text}")

143

144

145 class TerminalPanel(Vertical):

146 """Right panel: terminal / shell output."""

147 def compose(self) -> ComposeResult:

148 yield Static(" Terminal ", id="panel-title")

149 yield RichLog(id="terminal-log", highlight=True, markup=True)

150 yield Input(placeholder="$ command...", id="terminal-input")

151

152 def on_mount(self) -> None:

153 log = self.query_one("#terminal-log", RichLog)

154 log.write("[dim]$ agentos tui started[/dim]")

155 log.write(f"[dim] cwd: {os.getcwd()}[/dim]")

156

157

158 class StatusBar(Horizontal):

159 """Bottom status bar with metrics."""

160 def compose(self) -> ComposeResult:

161 yield Static(" Sessions: 0 ", id="status-sessions")

162 yield Static(" Tasks: 0 ", id="status-tasks")

163 yield Static(" Mode: READY ", id="status-mode")

164

165

166 class MarketPanel(ScrollableContainer):

167 """Skill marketplace panel with embedded web-like view.

168

169 Shows skill sources from multiple marketplaces (OpenClaw, ClawHub,

170 SkillsMP, LobeHub, etc.) and supports one-click install for compatible

171 sources. External sources open in the system browser.

172

173 Requires the skill store server: agentos skill-store

174 """

175

176 class SkillInstalled(Message):

177 """Posted when a skill is installed via the market."""

178 def __init__(self, skill_name: str, source: str) -> None:

179 self.skill_name = skill_name

180 self.source = source

181 super().__init__()

182

183 def __init__(self, store_url: str = "http://127.0.0.1:18900", **kwargs):

184 super().__init__(**kwargs)

185 self._store_url = store_url

186 self._sources: list[dict] = []

187 self._installed: set[str] = set()

188 self._active_source: str = "openclaw"

189 self._skills: list[dict] = []

190

191 def compose(self) -> ComposeResult:

192 yield Static(" Skill Marketplace ", id="panel-title")

193 with Horizontal():

194 with Vertical(classes="market-sidebar", id="market-sidebar-container"):

195 yield Static(" Sources", classes="market-section-title")

196 yield ListView(id="market-source-list")

197 with Vertical(classes="market-content", id="market-content-container"):

198 yield Static(" Skills", classes="market-section-title")

199 yield Input(placeholder="Search skills...", id="market-search")

200 yield RichLog(id="market-skill-log", highlight=True, markup=True)

201

202 def on_mount(self) -> None:

203 self._init_sources()

204 self._load_skills()

205

206 def _init_sources(self) -> None:

207 try:

208 import urllib.request, json as _json

209 with urllib.request.urlopen(f"{self._store_url}/api/sources", timeout=5) as resp:

210 self._sources = _json.loads(resp.read())

211 except Exception:

212 self._sources = [

213 {"id": "openclaw", "name": "OpenClaw Skill Store", "skill_count": "14+",

214 "installable": True, "web_url": "https://github.com/nicepkg/openclaw-skill-store",

215 "description": "OpenClaw 官方社区技能商店"},

216 {"id": "clawhub", "name": "ClawHub", "skill_count": "5,700+",

217 "installable": False, "web_url": "https://github.com/clawhub-community/skills",

218 "description": "ClawHub 社区技能聚合"},

219 {"id": "skillsmp", "name": "SkillsMP", "skill_count": "164万+",

220 "installable": False, "web_url": "https://skills.mp/",

221 "description": "技能界的 Google，最大索引平台"},

222 {"id": "lobehub", "name": "LobeHub Skills", "skill_count": "28万+",

223 "installable": False, "web_url": "https://lobehub.com/skills",

224 "description": "LobeHub 生态精品平台"},

225 {"id": "skillhub", "name": "SkillHub Club", "skill_count": "1.6万+",

226 "installable": False, "web_url": "https://skillhub.club/",

227 "description": "AI 评分品质筛选市集"},

228 {"id": "skills_sh", "name": "skills.sh", "skill_count": "67万+",

229 "installable": False, "web_url": "https://skills.sh/",

230 "description": "Vercel Labs 一键安装平台"},

231 {"id": "awesome", "name": "awesome-agent-skills", "skill_count": "380+",

232 "installable": False, "web_url": "https://github.com/nicepkg/awesome-agent-skills",

233 "description": "人工审核精选技能合集"},

234 ]

235

236 lst = self.query_one("#market-source-list", ListView)

237 lst.clear()

238 for src in self._sources:

239 icon = "[bold green]↓[/bold green]" if src.get("installable") else "[bold blue]↗[/bold blue]"

240 cnt = src.get("skill_count", "?")

241 lst.append(ListItem(

242 Label(f"{icon} {src['name']} [dim]({cnt})[/dim]"),

243 name=src["id"],

244 ))

245

246 def on_list_view_selected(self, event: ListView.Selected) -> None:

247 if event.item.name:

248 raw = event.item.name

249 self._active_source = raw.value if hasattr(raw, 'value') else str(raw)

250 self._load_skills()

251

252 def on_input_submitted(self, event: Input.Submitted) -> None:

253 if event.input.id == "market-search":

254 self._filter_skills(event.value.strip())

255

256 def _load_skills(self) -> None:

257 log = self.query_one("#market-skill-log", RichLog)

258 log.clear()

259 src = next((s for s in self._sources if s["id"] == self._active_source), None)

260 if not src:

261 log.write("[bold red]Source not found[/bold red]")

262 return

263

264 if not src.get("installable"):

265 log.write(f"[bold blue]{src['name']}[/bold blue]")

266 log.write(f"[dim]{src.get('description', 'External marketplace')}[/dim]\n")

267 log.write("[bold]Open in browser:[/bold]")

268 log.write(f" [link={src['web_url']}]{src['web_url']}[/link]")

269 log.write(f" [link={src.get('url', src['web_url'])}]{src.get('url', '')}[/link]\n")

270 log.write("[dim]External marketplace — open the URL above in your browser to browse and install.[/dim]")

271 return

272

273 try:

274 import urllib.request, json as _json

275 url = f"{self._store_url}/api/skills?source={self._active_source}"

276 with urllib.request.urlopen(url, timeout=5) as resp:

277 data = _json.loads(resp.read())

278 self._skills = data.get("skills", [])

279 except Exception:

280 self._skills = self._fallback_skills()

281

282 log.write(f"[bold blue]{src['name']}[/bold blue] [dim]({len(self._skills)} skills)[/dim]\n")

283 for skill in self._skills:

284 name = skill["name"]

285 desc = skill.get("description", "")

286 tags = " ".join(f"[dim]#{t}[/dim]" for t in skill.get("tags", []))

287 installed = "[bold green]✓[/bold green]" if name in self._installed else "[dim]○[/dim]"

288 log.write(f" {installed} [bold]{name}[/bold] {tags}")

289 log.write(f" [dim]{desc}[/dim]")

290 log.write("")

291 log.write("[dim]Use 'agentos skill-store' to start the web UI for one-click install.[/dim]")

292

293 def _fallback_skills(self) -> list[dict]:

294 """Load real seed skills from the registry index."""

295 try:

296 import yaml, os

297 index_path = os.path.join(

298 os.path.dirname(os.path.dirname(os.path.dirname(__file__))),

299 "agentos", "marketplace", "skills", "_index.yaml",

300 )

301 # Resolve relative to package

302 pkgs = ["agentos", "marketplace", "skills", "_index.yaml"]

303 for pkg_dir in (

304 os.path.dirname(os.path.dirname(os.path.dirname(__file__))),

305 ):

306 candidate = os.path.join(pkg_dir, *pkgs)

307 if os.path.exists(candidate):

308 index_path = candidate

309 break

310

311 # Try loading from agentos package directly

312 try:

313 import agentos.marketplace.skills as _sk

314 index_path = os.path.join(os.path.dirname(_sk.__file__), "_index.yaml")

315 except Exception:

316 pass

317

318 with open(index_path) as f:

319 data = yaml.safe_load(f)

320

321 skills = []

322 for name in data.get("skills", []):

323 # Try to read individual skill.yaml for metadata

324 skill_dir = os.path.join(os.path.dirname(index_path), name)

325 meta = {"name": name, "description": "", "tags": []}

326 try:

327 skill_yaml = os.path.join(skill_dir, "skill.yaml")

328 if os.path.exists(skill_yaml):

329 with open(skill_yaml) as sf:

330 sm = yaml.safe_load(sf) or {}

331 meta["description"] = sm.get("description", "")

332 meta["tags"] = sm.get("tags", [])

333 meta["version"] = sm.get("version", "")

334 except Exception:

335 pass

336 skills.append(meta)

337 return skills or self._hardcoded_skills()

338 except Exception:

339 return self._hardcoded_skills()

340

341 def _hardcoded_skills(self) -> list[dict]:

342 """Ultimate fallback — 14 generic skills."""

343 return [

344 {"name": "skill-creator", "description": "Create new skills from templates", "tags": ["meta"]},

345 {"name": "pdf-tools", "description": "PDF manipulation, merge, split, extract", "tags": ["document"]},

346 {"name": "xlsx-tools", "description": "Excel/Spreadsheet processing", "tags": ["document"]},

347 {"name": "docx-tools", "description": "Word document processing", "tags": ["document"]},

348 {"name": "pptx-tools", "description": "PowerPoint generation", "tags": ["document"]},

349 {"name": "image-tools", "description": "Image processing, resize, convert", "tags": ["media"]},

350 {"name": "web-search", "description": "Advanced web search with multiple engines", "tags": ["search"]},

351 {"name": "browser-automation", "description": "Browser automation with Playwright", "tags": ["browser"]},

352 {"name": "code-review", "description": "Automated code review", "tags": ["code"]},

353 {"name": "git-tools", "description": "Git workflow automation", "tags": ["git"]},

354 {"name": "file-organizer", "description": "File organization and cleanup", "tags": ["files"]},

355 {"name": "data-analysis", "description": "Data analysis and visualization", "tags": ["data"]},

356 {"name": "api-tester", "description": "API testing and docs generation", "tags": ["api"]},

357 {"name": "markdown-tools", "description": "Markdown editing and conversion", "tags": ["document"]},

358 ]

359

360 def _filter_skills(self, query: str) -> None:

361 log = self.query_one("#market-skill-log", RichLog)

362 if not query:

363 self._load_skills()

364 return

365 log.clear()

366 matched = [s for s in self._skills if

367 query.lower() in s["name"].lower() or

368 query.lower() in s.get("description", "").lower()]

369 log.write(f"[dim]Search: '{query}' — {len(matched)} results[/dim]\n")

370 for skill in matched:

371 name = skill["name"]

372 desc = skill.get("description", "")

373 installed = "[bold green]✓[/bold green]" if name in self._installed else "[dim]○[/dim]"

374 log.write(f" {installed} [bold]{name}[/bold]")

375 log.write(f" [dim]{desc}[/dim]")

376

377 def set_store_url(self, url: str) -> None:

378 self._store_url = url

379

380 @property

381 def installed_skills(self) -> set[str]:

382 return self._installed

383

384

385 # ── Main Application ──

386

387 class AgentOSTUI(App):

388 """Main TUI application — four-panel agent cockpit."""

389

390 CSS = """

391 Screen {

392 layout: grid;

393 grid-size: 3 3;

394 grid-gutter: 1 2;

395 background: $surface;

396 }

397

398 #panel-file {

399 row-span: 2;

400 border: solid $primary;

401 background: $panel;

402 }

403

404 #panel-chat {

405 row-span: 2;

406 border: solid $primary;

407 background: $panel;

408 }

409

410 #panel-terminal {

411 row-span: 2;

412 border: solid $primary;

413 background: $panel;

414 }

415

416 #panel-market {

417 row-span: 2;

418 border: solid $success;

419 background: $panel;

420 }

421

422 #panel-status {

423 column-span: 3;

424 height: 1;

425 background: $primary-darken-2;

426 color: $text;

427 }

428

429 #panel-title {

430 background: $primary-darken-1;

431 color: $text;

432 text-style: bold;

433 padding: 0 1;

434 height: 1;

435 }

436

437 #file-tree {

438 height: 1fr;

439 overflow-y: auto;

440 }

441

442 #chat-log {

443 height: 1fr;

444 overflow-y: auto;

445 }

446

447 #terminal-log {

448 height: 1fr;

449 overflow-y: auto;

450 }

451

452 #chat-input, #terminal-input {

453 dock: bottom;

454 height: 3;

455 }

456

457 .market-sidebar {

458 width: 32;

459 border: solid $primary-darken-2;

460 }

461

462 .market-section-title {

463 background: $primary-darken-1;

464 color: $text;

465 text-style: bold;

466 height: 1;

467 }

468

469 #market-source-list {

470 height: 1fr;

471 }

472

473 #market-search {

474 dock: top;

475 height: 3;

476 }

477

478 #market-skill-log {

479 height: 1fr;

480 overflow-y: auto;

481 }

482 """

483

484 BINDINGS = [

485 Binding("ctrl+q", "quit", "Quit", show=True),

486 Binding("ctrl+s", "save", "Save", show=True),

487 Binding("ctrl+r", "refresh", "Refresh", show=True),

488 Binding("ctrl+t", "focus_terminal", "Terminal", show=True),

489 Binding("ctrl+c", "focus_chat", "Chat", show=True),

490 Binding("ctrl+f", "focus_files", "Files", show=True),

491 Binding("ctrl+m", "toggle_market", "Market", show=True),

492 Binding("f5", "refresh", "Refresh", show=False),

493 ]

494

495 _config: TUIConfig = TUIConfig()

496 _message_handler: Optional[callable] = None

497 _market_visible: bool = False

498

499 def __init__(self, config: TUIConfig = None, start_market: bool = False):

500 super().__init__()

501 if config:

502 self._config = config

503 self._market_visible = start_market

504

505 def compose(self) -> ComposeResult:

506 yield Header("NexusAgentOS", icon="")

507 yield FileTree(id="panel-file")

508 yield ChatArea(id="panel-chat")

509 yield TerminalPanel(id="panel-terminal")

510 yield StatusBar(id="panel-status")

511 yield Footer()

512

513 def on_mount(self) -> None:

514 self.title = "NexusAgentOS TUI"

515 self.sub_title = f"v1.7.6 — {self._config.work_dir}"

516

517 # Mount market panel

518 market = MarketPanel(store_url=self._config.store_url, id="panel-market")

519 market.display = self._market_visible

520 self.mount(market, before="#panel-status")

521

522 # Update status

523 status_sessions = self.query_one("#status-sessions", Static)

524 status_sessions.update(" Sessions: 0 ")

525

526 # ── Actions ──

527

528 def action_quit(self) -> None:

529 self.exit()

530

531 def action_save(self) -> None:

532 self._config.save()

533 chat = self.query_one("#chat-log", RichLog)

534 chat.write("[dim]Config saved.[/dim]")

535

536 def action_refresh(self) -> None:

537 self.query_one("#file-tree", Tree).root.remove_children()

538 self.query_one("#panel-file", FileTree)._populate_tree()

539

540 def action_focus_terminal(self) -> None:

541 self.query_one("#terminal-input", Input).focus()

542

543 def action_focus_chat(self) -> None:

544 self.query_one("#chat-input", Input).focus()

545

546 def action_focus_files(self) -> None:

547 self.query_one("#file-tree", Tree).focus()

548

549 def action_toggle_market(self) -> None:

550 market = self.query_one("#panel-market", MarketPanel)

551 self._market_visible = not self._market_visible

552 market.display = self._market_visible

553 if self._market_visible:

554 market._init_sources()

555 market._load_skills()

556 market.query_one("#market-search", Input).focus()

557 self.sub_title = f"v1.7.6 — Market | {self._config.work_dir}"

558 else:

559 self.sub_title = f"v1.7.6 — {self._config.work_dir}"

560

561 # ── Input Handlers ──

562

563 def on_input_submitted(self, event: Input.Submitted) -> None:

564 if event.input.id == "chat-input" and event.value.strip():

565 self._handle_chat(event.value.strip())

566 event.input.clear()

567 elif event.input.id == "terminal-input" and event.value.strip():

568 self._handle_terminal(event.value.strip())

569 event.input.clear()

570

571 def _handle_chat(self, message: str) -> None:

572 chat = self.query_one("#panel-chat", ChatArea)

573 chat.add_message("user", message)

574 if self._message_handler:

575 asyncio.create_task(self._dispatch_to_handler(message))

576 else:

577 chat.add_message("agent", f"Echo: {message}")

578

579 async def _dispatch_to_handler(self, message: str) -> None:

580 chat = self.query_one("#panel-chat", ChatArea)

581 try:

582 result = await self._message_handler(message)

583 chat.add_message("agent", str(result))

584 except Exception as e:

585 chat.add_message("error", f"Error: {e}")

586

587 def _handle_terminal(self, command: str) -> None:

588 log = self.query_one("#terminal-log", RichLog)

589 log.write(f"\n$ {command}")

590 try:

591 import subprocess

592 result = subprocess.run(

593 command, shell=True, capture_output=True,

594 text=True, timeout=30, cwd=self._config.work_dir,

595 )

596 if result.stdout:

597 log.write(result.stdout.rstrip())

598 if result.stderr:

599 log.write(f"[bold red]{result.stderr.rstrip()}[/bold red]")

600 except Exception as e:

601 log.write(f"[bold red]{e}[/bold red]")

602

603 # ── Public API ──

604

605 def set_message_handler(self, handler: callable):

606 self._message_handler = handler

607

608 def add_message(self, role: str, text: str):

609 chat = self.query_one("#panel-chat", ChatArea)

610 chat.add_message(role, text)

611

612

613# ── Entry Point ──

614

615def launch_tui(

616 message_handler=None,

617 work_dir: str = "",

618 theme: str = "dark",

619 start_market: bool = False,

620 store_url: str = "http://127.0.0.1:18900",

621) -> None:

622 """Launch the TUI application.

623

624 Args:

625 message_handler: async callable(msg: str) -> str for chat responses.

626 work_dir: Working directory for file tree.

627 theme: 'dark' or 'light'.

628 start_market: Open market panel on launch.

629 store_url: URL of the skill store server.

630 """

631 if not TEXTUAL_AVAILABLE:

632 print("ERROR: textual not installed. Run: pip install textual")

633 return

634

635 config = TUIConfig.load()

636 if work_dir:

637 config.work_dir = work_dir

638 if theme:

639 config.theme = theme

640 if store_url:

641 config.store_url = store_url

642

643 app = AgentOSTUI(config=config, start_market=start_market)

644 if message_handler:

645 app.set_message_handler(message_handler)

646

647 app.run()

648

649

650if __name__ == "__main__":

651 launch_tui()