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

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()

100

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

115

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

127

128

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")

135

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}")

144

145

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")

152

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]")

157

158

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")

165

166

167 class MarketPanel(ScrollableContainer):

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

169

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.

173

174 Requires the skill store server: agentos skill-store

175 """

176

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__()

183

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] = []

191

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)

202

203 def on_mount(self) -> None:

204 self._init_sources()

205 self._load_skills()

206

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 ]

236

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 ))

246

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()

252

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

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

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

256

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

264

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

273

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()

282

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]")

293

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

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

296 try:

297 import yaml, os

298 index_path = os.path.join(

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

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

301 )

302 # Resolve relative to package

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

304 for pkg_dir in (

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

306 ):

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

308 if os.path.exists(candidate):

309 index_path = candidate

310 break

311

312 # Try loading from agentos package directly

313 try:

314 import agentos.marketplace.skills as _sk

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

316 except Exception:

317 pass

318

319 with open(index_path) as f:

320 data = yaml.safe_load(f)

321

322 skills = []

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

324 # Try to read individual skill.yaml for metadata

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

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

327 try:

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

329 if os.path.exists(skill_yaml):

330 with open(skill_yaml) as sf:

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

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

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

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

335 except Exception:

336 pass

337 skills.append(meta)

338 return skills or self._hardcoded_skills()

339 except Exception:

340 return self._hardcoded_skills()

341

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

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

344 return [

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

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

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

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

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

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

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

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

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

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

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

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

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

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

359 ]

360

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

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

363 if not query:

364 self._load_skills()

365 return

366 log.clear()

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

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

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

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

371 for skill in matched:

372 name = skill["name"]

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

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

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

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

377

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

379 self._store_url = url

380

381 @property

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

383 return self._installed

384

385

386 # ── Main Application ──

387

388 class AgentOSTUI(App):

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

390

391 CSS = """

392 Screen {

393 layout: grid;

394 grid-size: 3 3;

395 grid-gutter: 1 2;

396 background: $surface;

397 }

398

399 #panel-file {

400 row-span: 2;

401 border: solid $primary;

402 background: $panel;

403 }

404

405 #panel-chat {

406 row-span: 2;

407 border: solid $primary;

408 background: $panel;

409 }

410

411 #panel-terminal {

412 row-span: 2;

413 border: solid $primary;

414 background: $panel;

415 }

416

417 #panel-market {

418 row-span: 2;

419 border: solid $success;

420 background: $panel;

421 }

422

423 #panel-status {

424 column-span: 3;

425 height: 1;

426 background: $primary-darken-2;

427 color: $text;

428 }

429

430 #panel-title {

431 background: $primary-darken-1;

432 color: $text;

433 text-style: bold;

434 padding: 0 1;

435 height: 1;

436 }

437

438 #file-tree {

439 height: 1fr;

440 overflow-y: auto;

441 }

442

443 #chat-log {

444 height: 1fr;

445 overflow-y: auto;

446 }

447

448 #terminal-log {

449 height: 1fr;

450 overflow-y: auto;

451 }

452

453 #chat-input, #terminal-input {

454 dock: bottom;

455 height: 3;

456 }

457

458 .market-sidebar {

459 width: 32;

460 border: solid $primary-darken-2;

461 }

462

463 .market-section-title {

464 background: $primary-darken-1;

465 color: $text;

466 text-style: bold;

467 height: 1;

468 }

469

470 #market-source-list {

471 height: 1fr;

472 }

473

474 #market-search {

475 dock: top;

476 height: 3;

477 }

478

479 #market-skill-log {

480 height: 1fr;

481 overflow-y: auto;

482 }

483 """

484

485 BINDINGS = [

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

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

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

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

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

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

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

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

494 ]

495

496 _config: TUIConfig = TUIConfig()

497 _message_handler: Optional[callable] = None

498 _market_visible: bool = False

499

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

501 super().__init__()

502 if config:

503 self._config = config

504 self._market_visible = start_market

505

506 def compose(self) -> ComposeResult:

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

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

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

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

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

512 yield Footer()

513

514 def on_mount(self) -> None:

515 self.title = "NexusAgentOS TUI"

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

517

518 # Mount market panel

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

520 market.display = self._market_visible

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

522

523 # Update status

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

525 status_sessions.update(" Sessions: 0 ")

526

527 # ── Actions ──

528

529 def action_quit(self) -> None:

530 self.exit()

531

532 def action_save(self) -> None:

533 self._config.save()

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

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

536

537 def action_refresh(self) -> None:

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

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

540

541 def action_focus_terminal(self) -> None:

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

543

544 def action_focus_chat(self) -> None:

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

546

547 def action_focus_files(self) -> None:

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

549

550 def action_toggle_market(self) -> None:

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

552 self._market_visible = not self._market_visible

553 market.display = self._market_visible

554 if self._market_visible:

555 market._init_sources()

556 market._load_skills()

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

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

559 else:

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

561

562 # ── Input Handlers ──

563

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

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

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

567 event.input.clear()

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

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

570 event.input.clear()

571

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

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

574 chat.add_message("user", message)

575 if self._message_handler:

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

577 else:

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

579

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

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

582 try:

583 result = await self._message_handler(message)

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

585 except Exception as e:

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

587

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

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

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

591 try:

592 import subprocess

593 result = subprocess.run(

594 command, shell=True, capture_output=True,

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

596 )

597 if result.stdout:

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

599 if result.stderr:

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

601 except Exception as e:

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

603

604 # ── Public API ──

605

606 def set_message_handler(self, handler: callable):

607 self._message_handler = handler

608

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

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

611 chat.add_message(role, text)

612

613

614# ── Entry Point ──

615

616def launch_tui(

617 message_handler=None,

618 work_dir: str = "",

619 theme: str = "dark",

620 start_market: bool = False,

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

622) -> None:

623 """Launch the TUI application.

624

625 Args:

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

627 work_dir: Working directory for file tree.

628 theme: 'dark' or 'light'.

629 start_market: Open market panel on launch.

630 store_url: URL of the skill store server.

631 """

632 if not TEXTUAL_AVAILABLE:

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

634 return

635

636 config = TUIConfig.load()

637 if work_dir:

638 config.work_dir = work_dir

639 if theme:

640 config.theme = theme

641 if store_url:

642 config.store_url = store_url

643

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

645 if message_handler:

646 app.set_message_handler(message_handler)

647

648 app.run()

649

650

651if __name__ == "__main__":

652 launch_tui()