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

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

30try:

31 from textual.app import App, ComposeResult

32 from textual.binding import Binding

33 from textual.containers import Horizontal, ScrollableContainer, Vertical

34 from textual.message import Message

35 from textual.widgets import (

36 Footer,

37 Header,

38 Input,

39 Label,

40 ListItem,

41 ListView,

42 RichLog,

43 Static,

44 Tree,

45 )

47 TEXTUAL_AVAILABLE = True

48except ImportError:

49 TEXTUAL_AVAILABLE = False

52# ── Models ──

55@dataclass

56class TUIConfig:

57 """TUI persistent configuration."""

59 theme: str = "dark"

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

61 font_size: int = 14

62 max_history: int = 500

63 auto_scroll: bool = True

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

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

67 p = Path(path).expanduser()

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

69 p.write_text(

70 json.dumps(

71 {

72 "theme": self.theme,

73 "work_dir": self.work_dir,

74 "font_size": self.font_size,

75 "max_history": self.max_history,

76 "auto_scroll": self.auto_scroll,

77 "store_url": self.store_url,

78 },

79 indent=2,

80 )

81 )

83 @classmethod

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

85 p = Path(path).expanduser()

86 if p.exists():

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

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

89 return cls()

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

94if not TEXTUAL_AVAILABLE:

96 class FileTree:

97 pass

99 class ChatArea:

100 pass

101

102 class TerminalPanel:

103 pass

104

105 class StatusBar:

106 pass

107

108 class MarketPanel:

109 pass

110

111 class _StubApp:

112 def run(self):

113 raise RuntimeError("textual not installed: pip install textual")

114

115

116if TEXTUAL_AVAILABLE:

117

118 class FileTree(Vertical):

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

120

121 def compose(self) -> ComposeResult:

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

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

124

125 def on_mount(self) -> None:

126 self._populate_tree()

127

128 def _populate_tree(self) -> None:

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

130 tree.clear()

131 root = tree.root

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

133 try:

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

135 for item in items[:50]:

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

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

138 if item.is_dir():

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

140 except PermissionError:

141 pass

142

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

144 if depth > 1:

145 return

146 try:

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

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

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

150 if child.is_dir():

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

152 except PermissionError:

153 pass

154

155 class ChatArea(Vertical):

156 """Center panel: chat interface."""

157

158 def compose(self) -> ComposeResult:

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

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

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

162

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

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

165 if role == "user":

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

167 elif role == "error":

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

169 else:

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

171

172 class TerminalPanel(Vertical):

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

174

175 def compose(self) -> ComposeResult:

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

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

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

179

180 def on_mount(self) -> None:

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

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

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

184

185 class StatusBar(Horizontal):

186 """Bottom status bar with metrics."""

187

188 def compose(self) -> ComposeResult:

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

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

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

192

193 class MarketPanel(ScrollableContainer):

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

195

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

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

198 sources. External sources open in the system browser.

199

200 Requires the skill store server: agentos skill-store

201 """

202

203 class SkillInstalled(Message):

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

205

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

207 self.skill_name = skill_name

208 self.source = source

209 super().__init__()

210

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

212 super().__init__(**kwargs)

213 self._store_url = store_url

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

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

216 self._active_source: str = "openclaw"

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

218

219 def compose(self) -> ComposeResult:

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

221 with Horizontal():

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

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

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

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

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

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

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

229

230 def on_mount(self) -> None:

231 self._init_sources()

232 self._load_skills()

233

234 def _init_sources(self) -> None:

235 try:

236 import json as _json

237 import urllib.request

238

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

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

241 except Exception:

242 self._sources = [

243 {

244 "id": "openclaw",

245 "name": "OpenClaw Skill Store",

246 "skill_count": "14+",

247 "installable": True,

248 "web_url": "https://github.com/nicepkg/openclaw-skill-store",

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

250 },

251 {

252 "id": "clawhub",

253 "name": "ClawHub",

254 "skill_count": "5,700+",

255 "installable": False,

256 "web_url": "https://github.com/clawhub-community/skills",

257 "description": "ClawHub 社区技能聚合",

258 },

259 {

260 "id": "skillsmp",

261 "name": "SkillsMP",

262 "skill_count": "164万+",

263 "installable": False,

264 "web_url": "https://skills.mp/",

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

266 },

267 {

268 "id": "lobehub",

269 "name": "LobeHub Skills",

270 "skill_count": "28万+",

271 "installable": False,

272 "web_url": "https://lobehub.com/skills",

273 "description": "LobeHub 生态精品平台",

274 },

275 {

276 "id": "skillhub",

277 "name": "SkillHub Club",

278 "skill_count": "1.6万+",

279 "installable": False,

280 "web_url": "https://skillhub.club/",

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

282 },

283 {

284 "id": "skills_sh",

285 "name": "skills.sh",

286 "skill_count": "67万+",

287 "installable": False,

288 "web_url": "https://skills.sh/",

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

290 },

291 {

292 "id": "awesome",

293 "name": "awesome-agent-skills",

294 "skill_count": "380+",

295 "installable": False,

296 "web_url": "https://github.com/nicepkg/awesome-agent-skills",

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

298 },

299 ]

300

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

302 lst.clear()

303 for src in self._sources:

304 icon = (

305 "[bold green]↓[/bold green]"

306 if src.get("installable")

307 else "[bold blue]↗[/bold blue]"

308 )

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

310 lst.append(

311 ListItem(

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

313 name=src["id"],

314 )

315 )

316

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

318 if event.item.name:

319 raw = event.item.name

320 self._active_source = raw.value if hasattr(raw, "value") else str(raw)

321 self._load_skills()

322

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

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

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

326

327 def _load_skills(self) -> None:

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

329 log.clear()

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

331 if not src:

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

333 return

334

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

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

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

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

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

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

341 log.write(

342 "[dim]External marketplace — open the URL above in your browser to browse and install.[/dim]"

343 )

344 return

345

346 try:

347 import json as _json

348 import urllib.request

349

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

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

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

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

354 except Exception:

355 self._skills = self._fallback_skills()

356

357 log.write(

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

359 )

360 for skill in self._skills:

361 name = skill["name"]

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

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

364 installed = (

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

366 )

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

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

369 log.write("")

370 log.write(

371 "[dim]Use 'agentos skill-store' to start the web UI for one-click install.[/dim]"

372 )

373

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

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

376 try:

377 import os

378

379 import yaml

380

381 index_path = os.path.join(

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

383 "agentos",

384 "marketplace",

385 "skills",

386 "_index.yaml",

387 )

388 # Resolve relative to package

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

390 for pkg_dir in (os.path.dirname(os.path.dirname(os.path.dirname(__file__))),):

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

392 if os.path.exists(candidate):

393 index_path = candidate

394 break

395

396 # Try loading from agentos package directly

397 try:

398 import agentos.marketplace.skills as _sk

399

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

401 except Exception:

402 pass

403

404 with open(index_path) as f:

405 data = yaml.safe_load(f)

406

407 skills = []

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

409 # Try to read individual skill.yaml for metadata

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

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

412 try:

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

414 if os.path.exists(skill_yaml):

415 with open(skill_yaml) as sf:

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

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

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

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

420 except Exception:

421 pass

422 skills.append(meta)

423 return skills or self._hardcoded_skills()

424 except Exception:

425 return self._hardcoded_skills()

426

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

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

429 return [

430 {

431 "name": "skill-creator",

432 "description": "Create new skills from templates",

433 "tags": ["meta"],

434 },

435 {

436 "name": "pdf-tools",

437 "description": "PDF manipulation, merge, split, extract",

438 "tags": ["document"],

439 },

440 {

441 "name": "xlsx-tools",

442 "description": "Excel/Spreadsheet processing",

443 "tags": ["document"],

444 },

445 {

446 "name": "docx-tools",

447 "description": "Word document processing",

448 "tags": ["document"],

449 },

450 {

451 "name": "pptx-tools",

452 "description": "PowerPoint generation",

453 "tags": ["document"],

454 },

455 {

456 "name": "image-tools",

457 "description": "Image processing, resize, convert",

458 "tags": ["media"],

459 },

460 {

461 "name": "web-search",

462 "description": "Advanced web search with multiple engines",

463 "tags": ["search"],

464 },

465 {

466 "name": "browser-automation",

467 "description": "Browser automation with Playwright",

468 "tags": ["browser"],

469 },

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

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

472 {

473 "name": "file-organizer",

474 "description": "File organization and cleanup",

475 "tags": ["files"],

476 },

477 {

478 "name": "data-analysis",

479 "description": "Data analysis and visualization",

480 "tags": ["data"],

481 },

482 {

483 "name": "api-tester",

484 "description": "API testing and docs generation",

485 "tags": ["api"],

486 },

487 {

488 "name": "markdown-tools",

489 "description": "Markdown editing and conversion",

490 "tags": ["document"],

491 },

492 ]

493

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

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

496 if not query:

497 self._load_skills()

498 return

499 log.clear()

500 matched = [

501 s

502 for s in self._skills

503 if query.lower() in s["name"].lower()

504 or query.lower() in s.get("description", "").lower()

505 ]

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

507 for skill in matched:

508 name = skill["name"]

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

510 installed = (

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

512 )

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

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

515

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

517 self._store_url = url

518

519 @property

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

521 return self._installed

522

523 # ── Main Application ──

524

525 class AgentOSTUI(App):

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

527

528 CSS = """

529 Screen {

530 layout: grid;

531 grid-size: 3 3;

532 grid-gutter: 1 2;

533 background: $surface;

534 }

535

536 #panel-file {

537 row-span: 2;

538 border: solid $primary;

539 background: $panel;

540 }

541

542 #panel-chat {

543 row-span: 2;

544 border: solid $primary;

545 background: $panel;

546 }

547

548 #panel-terminal {

549 row-span: 2;

550 border: solid $primary;

551 background: $panel;

552 }

553

554 #panel-market {

555 row-span: 2;

556 border: solid $success;

557 background: $panel;

558 }

559

560 #panel-status {

561 column-span: 3;

562 height: 1;

563 background: $primary-darken-2;

564 color: $text;

565 }

566

567 #panel-title {

568 background: $primary-darken-1;

569 color: $text;

570 text-style: bold;

571 padding: 0 1;

572 height: 1;

573 }

574

575 #file-tree {

576 height: 1fr;

577 overflow-y: auto;

578 }

579

580 #chat-log {

581 height: 1fr;

582 overflow-y: auto;

583 }

584

585 #terminal-log {

586 height: 1fr;

587 overflow-y: auto;

588 }

589

590 #chat-input, #terminal-input {

591 dock: bottom;

592 height: 3;

593 }

594

595 .market-sidebar {

596 width: 32;

597 border: solid $primary-darken-2;

598 }

599

600 .market-section-title {

601 background: $primary-darken-1;

602 color: $text;

603 text-style: bold;

604 height: 1;

605 }

606

607 #market-source-list {

608 height: 1fr;

609 }

610

611 #market-search {

612 dock: top;

613 height: 3;

614 }

615

616 #market-skill-log {

617 height: 1fr;

618 overflow-y: auto;

619 }

620 """

621

622 BINDINGS = [

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

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

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

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

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

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

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

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

631 ]

632

633 _config: TUIConfig = TUIConfig()

634 _message_handler: callable | None = None

635 _market_visible: bool = False

636

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

638 super().__init__()

639 if config:

640 self._config = config

641 self._market_visible = start_market

642

643 def compose(self) -> ComposeResult:

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

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

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

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

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

649 yield Footer()

650

651 def on_mount(self) -> None:

652 self.title = "NexusAgentOS TUI"

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

654

655 # Mount market panel

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

657 market.display = self._market_visible

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

659

660 # Update status

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

662 status_sessions.update(" Sessions: 0 ")

663

664 # ── Actions ──

665

666 def action_quit(self) -> None:

667 self.exit()

668

669 def action_save(self) -> None:

670 self._config.save()

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

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

673

674 def action_refresh(self) -> None:

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

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

677

678 def action_focus_terminal(self) -> None:

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

680

681 def action_focus_chat(self) -> None:

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

683

684 def action_focus_files(self) -> None:

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

686

687 def action_toggle_market(self) -> None:

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

689 self._market_visible = not self._market_visible

690 market.display = self._market_visible

691 if self._market_visible:

692 market._init_sources()

693 market._load_skills()

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

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

696 else:

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

698

699 # ── Input Handlers ──

700

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

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

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

704 event.input.clear()

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

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

707 event.input.clear()

708

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

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

711 chat.add_message("user", message)

712 if self._message_handler:

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

714 else:

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

716

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

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

719 try:

720 result = await self._message_handler(message)

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

722 except Exception as e:

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

724

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

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

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

728 try:

729 import subprocess

730

731 result = subprocess.run(

732 command,

733 shell=True,

734 capture_output=True,

735 text=True,

736 timeout=30,

737 cwd=self._config.work_dir,

738 )

739 if result.stdout:

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

741 if result.stderr:

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

743 except Exception as e:

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

745

746 # ── Public API ──

747

748 def set_message_handler(self, handler: callable):

749 self._message_handler = handler

750

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

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

753 chat.add_message(role, text)

754

755

756# ── Entry Point ──

757

758

759def launch_tui(

760 message_handler=None,

761 work_dir: str = "",

762 theme: str = "dark",

763 start_market: bool = False,

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

765) -> None:

766 """Launch the TUI application.

767

768 Args:

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

770 work_dir: Working directory for file tree.

771 theme: 'dark' or 'light'.

772 start_market: Open market panel on launch.

773 store_url: URL of the skill store server.

774 """

775 if not TEXTUAL_AVAILABLE:

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

777 return

778

779 config = TUIConfig.load()

780 if work_dir:

781 config.work_dir = work_dir

782 if theme:

783 config.theme = theme

784 if store_url:

785 config.store_url = store_url

786

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

788 if message_handler:

789 app.set_message_handler(message_handler)

790

791 app.run()

792

793

794if __name__ == "__main__":

795 launch_tui()