diff --git a/src/memory_manager/tui/app.py b/src/memory_manager/tui/app.py index 0e1f6eb..2252c44 100644 --- a/src/memory_manager/tui/app.py +++ b/src/memory_manager/tui/app.py @@ -1,103 +1,360 @@ -"""Textual TUI application.""" -import os -import asyncio -from typing import Optional +from datetime import datetime +from typing import Any +from textual import on from textual.app import App, ComposeResult -from textual.widgets import Header, Footer, DataTable, Static, Tabs, Tab -from textual.containers import Container, VerticalScroll +from textual.binding import Binding +from textual.containers import Container, Horizontal, ScrollableContainer, Vertical from textual.screen import Screen +from textual.widgets import ( + Button, + Footer, + Header, + Input, + Label, + ListItem, + ListView, + Static, +) -from memory_manager.core.services import MemoryService, CommitService +from memory_manager.core.services import MemoryManager +from memory_manager.db.models import MemoryCategory from memory_manager.db.repository import MemoryRepository - db_path = os.getenv("MEMORY_DB_PATH", ".memory/codebase_memory.db") -os.makedirs(os.path.dirname(db_path), exist_ok=True) -repository = MemoryRepository(db_path) -memory_service = MemoryService(repository) -commit_service = CommitService(repository) + +async def get_memory_manager() -> MemoryManager: + repository = MemoryRepository(db_path) + await repository.initialize() + manager = MemoryManager(repository) + return manager class DashboardScreen(Screen): + CSS = """ + Screen { + background: $surface; + } + .stats-container { + height: auto; + padding: 1 2; + background: $panel; + border: solid $border; + } + .stat-label { + color: $text-muted; + } + .stat-value { + color: $text; + bold: true; + } + .entry-list { + height: 1fr; + } + """ + + def __init__(self, manager: MemoryManager): + super().__init__() + self.manager = manager + def compose(self) -> ComposeResult: yield Header() - yield Static("Memory Manager Dashboard", id="title") - yield DataTable() + yield Container( + Vertical( + Label("Memory Manager Dashboard", classes="header-title"), + ScrollableContainer(classes="stats-container", id="stats-container"), + ListView(id="recent-entries", classes="entry-list"), + id="dashboard-content", + ) + ) yield Footer() - async def on_mount(self): - stats = await repository.get_stats() - table = self.query_one(DataTable) - table.add_columns("Metric", "Value") - table.add_row("Total Entries", str(stats["total_entries"])) - table.add_row("Total Commits", str(stats["total_commits"])) - for category, count in stats["category_counts"].items(): - table.add_row(category.title(), str(count)) + async def on_mount(self) -> None: + await self.load_stats() + + async def load_stats(self) -> None: + stats_container = self.query_one("#stats-container", ScrollableContainer) + + entries = await self.manager.memory_service.list_entries(limit=10000) + commits = await self.manager.commit_service.list_commits(limit=10000) + + entries_by_category: dict[str, int] = {} + for entry in entries: + cat = entry["category"] + entries_by_category[cat] = entries_by_category.get(cat, 0) + 1 + + stats_text = f""" +Total Entries: {len(entries)} | Total Commits: {len(commits)} + +Entries by Category: +""" + for cat, count in entries_by_category.items(): + stats_text += f" {cat}: {count}\n" + + stats_container.remove_children() + stats_container.mount(Static(stats_text)) + + list_view = self.query_one("#recent-entries", ListView) + list_view.clear() + for entry in entries[:10]: + created = entry["created_at"] + if created: + created = datetime.fromisoformat(created).strftime("%m/%d %H:%M") + list_item = ListItem( + Label(f"[{entry['category']}] {entry['title'][:40]} - {created}"), + ) + await list_view.mount(list_item) class MemoryListScreen(Screen): + CSS = """ + Screen { + background: $surface; + } + .filter-bar { + height: 3; + padding: 1; + background: $panel; + } + .entry-detail { + height: 1fr; + padding: 1 2; + } + """ + + def __init__(self, manager: MemoryManager, category: MemoryCategory | None = None): + super().__init__() + self.manager = manager + self.current_category = category + self.entries: list[dict[str, Any]] = [] + def compose(self) -> ComposeResult: yield Header() - yield Static("Memory Entries", id="title") - yield DataTable() + yield Container( + Horizontal( + Button("All", id="filter-all"), + Button("Decision", id="filter-decision"), + Button("Feature", id="filter-feature"), + Button("Refactoring", id="filter-refactoring"), + Button("Architecture", id="filter-architecture"), + Button("Bug", id="filter-bug"), + Button("Note", id="filter-note"), + classes="filter-bar", + ), + Horizontal( + ListView(id="entries-list", classes="column"), + ScrollableContainer(id="entry-detail", classes="column entry-detail"), + classes="main-content", + ), + ) yield Footer() - async def on_mount(self): - entries = await memory_service.list_entries(limit=100) - table = self.query_one(DataTable) - table.add_columns("ID", "Category", "Title", "Agent") - for entry in entries: - table.add_row( - str(entry.id), - entry.category.value, - entry.title[:40], - entry.agent_id, + async def on_mount(self) -> None: + await self.load_entries() + + async def load_entries(self, category: MemoryCategory | None = None) -> None: + self.entries = await self.manager.memory_service.list_entries( + category=category, + limit=1000, + ) + + list_view = self.query_one("#entries-list", ListView) + list_view.clear() + + for entry in self.entries: + created = entry["created_at"] + if created: + created = datetime.fromisoformat(created).strftime("%m/%d %H:%M") + list_item = ListItem( + Label(f"[{entry['category']}] {entry['title']}"), + Label(f"{entry['agent_id']} | {created}"), ) + await list_view.mount(list_item) + + @on(ListView.Selected) + async def on_entry_selected(self, event: ListView.Selected) -> None: + index = event.list_view.index + if index is not None and 0 <= index < len(self.entries): + entry = self.entries[index] + detail_container = self.query_one("#entry-detail", ScrollableContainer) + detail_container.remove_children() + + content = f""" +Title: {entry['title']} +Category: {entry['category']} +Agent: {entry['agent_id']} +Project: {entry['project_path']} +Tags: {', '.join(entry['tags']) if entry['tags'] else '(none)'} +Created: {entry['created_at']} +Updated: {entry['updated_at']} + +Content: +{entry['content']} +""" + await detail_container.mount(Static(content)) + + @on(Button.Pressed) + async def on_filter_pressed(self, event: Button.Pressed) -> None: + button_id = event.button.id + category = None + + if button_id == "filter-decision": + category = MemoryCategory.DECISION + elif button_id == "filter-feature": + category = MemoryCategory.FEATURE + elif button_id == "filter-refactoring": + category = MemoryCategory.REFACTORING + elif button_id == "filter-architecture": + category = MemoryCategory.ARCHITECTURE + elif button_id == "filter-bug": + category = MemoryCategory.BUG + elif button_id == "filter-note": + category = MemoryCategory.NOTE + + await self.load_entries(category) class CommitHistoryScreen(Screen): + CSS = """ + Screen { + background: $surface; + } + .commit-list { + height: 1fr; + } + """ + + def __init__(self, manager: MemoryManager): + super().__init__() + self.manager = manager + self.commits: list[dict[str, Any]] = [] + def compose(self) -> ComposeResult: yield Header() - yield Static("Commit History", id="title") - yield DataTable() + yield Container( + ScrollableContainer( + ListView(id="commits-list", classes="commit-list"), + id="commits-container", + ), + ) yield Footer() - async def on_mount(self): - commits = await commit_service.get_commits(limit=50) - table = self.query_one(DataTable) - table.add_columns("Hash", "Message", "Agent", "Date") - for commit in commits: - table.add_row( - commit.hash[:8], - commit.message[:40], - commit.agent_id, - str(commit.created_at)[:10], + async def on_mount(self) -> None: + await self.load_commits() + + async def load_commits(self) -> None: + self.commits = await self.manager.commit_service.list_commits(limit=100) + + list_view = self.query_one("#commits-list", ListView) + list_view.clear() + + for commit in self.commits: + created = commit["created_at"] + if created: + created = datetime.fromisoformat(created).strftime("%Y-%m-%d %H:%M:%S") + + content = f"commit {commit['hash'][:8]}\n{commit['agent_id']} | {created}\n\n {commit['message']}" + + list_item = ListItem( + Static(content, markup=False), ) + await list_view.mount(list_item) -class MemoryTUIApp(App): - def __init__(self): +class SearchScreen(Screen): + CSS = """ + Screen { + background: $surface; + } + .search-input { + height: 3; + padding: 1; + } + .results-list { + height: 1fr; + } + """ + + def __init__(self, manager: MemoryManager): super().__init__() - self.title = "Memory Manager TUI" - self.SCREENS = { - "dashboard": DashboardScreen(), - "list": MemoryListScreen(), - "history": CommitHistoryScreen(), - } - - def on_mount(self): - self.push_screen("dashboard") + self.manager = manager + self.results: list[dict[str, Any]] = [] def compose(self) -> ComposeResult: - yield Tabs( - Tab("Dashboard", id="dashboard"), - Tab("Entries", id="list"), - Tab("History", id="history"), + yield Header() + yield Container( + Horizontal( + Input(placeholder="Search query...", id="search-input"), + Button("Search", id="search-button"), + classes="search-input", + ), + ScrollableContainer( + ListView(id="results-list", classes="results-list"), + id="results-container", + ), ) + yield Footer() + + @on(Button.Pressed, "#search-button") + @on(Input.Submitted, "#search-input") + async def on_search(self) -> None: + input_widget = self.query_one("#search-input", Input) + query = input_widget.value + + if not query: + return + + self.results = await self.manager.search_service.search(query=query, limit=100) + + list_view = self.query_one("#results-list", ListView) + list_view.clear() + + for entry in self.results: + created = entry["created_at"] + if created: + created = datetime.fromisoformat(created).strftime("%m/%d %H:%M") + + list_item = ListItem( + Label(f"[{entry['category']}] {entry['title'][:40]}"), + Label(f"{entry['content'][:80]}... | {created}"), + ) + await list_view.mount(list_item) -def run_tui(): - app = MemoryTUIApp() - app.run() +class TUIApp(App): + BINDINGS = [ + Binding("d", "switch_dashboard", "Dashboard"), + Binding("l", "switch_memory_list", "Memory List"), + Binding("c", "switch_commits", "Commits"), + Binding("s", "switch_search", "Search"), + Binding("q", "quit", "Quit"), + ] + + def __init__(self): + super().__init__() + self.manager = None + + async def on_mount(self) -> None: + self.manager = await get_memory_manager() + await self.push_screen(DashboardScreen(self.manager)) + + async def on_unmount(self) -> None: + if self.manager: + await self.manager.close() + + async def switch_dashboard(self) -> None: + if self.manager: + await self.push_screen(DashboardScreen(self.manager)) + + async def switch_memory_list(self) -> None: + if self.manager: + await self.push_screen(MemoryListScreen(self.manager)) + + async def switch_commits(self) -> None: + if self.manager: + await self.push_screen(CommitHistoryScreen(self.manager)) + + async def switch_search(self) -> None: + if self.manager: + await self.push_screen(SearchScreen(self.manager))