diff --git a/src/memory_manager/cli/main.py b/src/memory_manager/cli/main.py index 390c052..9e7ec07 100644 --- a/src/memory_manager/cli/main.py +++ b/src/memory_manager/cli/main.py @@ -1,338 +1 @@ -import asyncio -import os -from datetime import datetime - -import click - -from memory_manager import __version__ -from memory_manager.core.services import MemoryManager -from memory_manager.db.models import MemoryCategory -from memory_manager.db.repository import MemoryRepository - - -def get_db_path() -> str: - return os.getenv("MEMORY_DB_PATH", ".memory/codebase_memory.db") - - -async def get_memory_manager() -> MemoryManager: - repository = MemoryRepository(get_db_path()) - await repository.initialize() - manager = MemoryManager(repository) - return manager - - -def validate_category(ctx, param, value): - if value is None: - return None - try: - return MemoryCategory(value) - except ValueError: - raise click.BadParameter(f"Invalid category. Must be one of: {[c.value for c in MemoryCategory]}") - - -@click.group() -@click.version_option(version=__version__) -def cli(): - """Agentic Codebase Memory Manager - A centralized memory store for AI coding agents.""" - pass - - -@click.command() -@click.option("--title", "-t", required=True, help="Entry title") -@click.option("--content", "-c", required=True, help="Entry content") -@click.option("--category", "-g", required=True, callback=validate_category, help="Entry category") -@click.option("--tags", "-T", multiple=True, help="Entry tags") -@click.option("--agent-id", help="Agent ID (defaults to AGENT_ID env var)") -@click.option("--project-path", help="Project path (defaults to MEMORY_PROJECT_PATH env var)") -def add(title, content, category, tags, agent_id, project_path): - """Add a new memory entry.""" - asyncio.run(_add(title, content, category, list(tags), agent_id, project_path)) - - -async def _add(title, content, category, tags, agent_id, project_path): - manager = await get_memory_manager() - try: - entry = await manager.memory_service.create_entry( - title=title, - content=content, - category=category, - tags=tags, - agent_id=agent_id, - project_path=project_path, - ) - click.echo(f"Created entry {entry['id']}: {entry['title']}") - finally: - await manager.close() - - -@click.command() -@click.option("--category", "-g", callback=validate_category, help="Filter by category") -@click.option("--agent-id", help="Filter by agent ID") -@click.option("--project-path", help="Filter by project path") -@click.option("--limit", "-n", default=100, help="Number of entries to show") -@click.option("--offset", default=0, help="Offset for pagination") -def list(category, agent_id, project_path, limit, offset): - """List memory entries.""" - asyncio.run(_list(category, agent_id, project_path, limit, offset)) - - -async def _list(category, agent_id, project_path, limit, offset): - manager = await get_memory_manager() - try: - entries = await manager.memory_service.list_entries( - category=category, - agent_id=agent_id, - project_path=project_path, - limit=limit, - offset=offset, - ) - if not entries: - click.echo("No entries found.") - return - - for entry in entries: - created = entry["created_at"] - if created: - created = datetime.fromisoformat(created).strftime("%Y-%m-%d %H:%M") - click.echo(f"[{entry['id']}] {entry['category']} | {entry['title']} | {created} | {entry['agent_id']}") - click.echo(f" {entry['content'][:100]}...") - if entry["tags"]: - click.echo(f" Tags: {', '.join(entry['tags'])}") - click.echo() - finally: - await manager.close() - - -@click.command() -@click.argument("query") -@click.option("--category", "-g", callback=validate_category, help="Filter by category") -@click.option("--agent-id", help="Filter by agent ID") -@click.option("--project-path", help="Filter by project path") -@click.option("--limit", "-n", default=100, help="Number of results") -def search(query, category, agent_id, project_path, limit): - """Search memory entries.""" - asyncio.run(_search(query, category, agent_id, project_path, limit)) - - -async def _search(query, category, agent_id, project_path, limit): - manager = await get_memory_manager() - try: - results = await manager.search_service.search( - query=query, - category=category, - agent_id=agent_id, - project_path=project_path, - limit=limit, - ) - if not results: - click.echo("No results found.") - return - - click.echo(f"Found {len(results)} result(s):\n") - for entry in results: - created = entry["created_at"] - if created: - created = datetime.fromisoformat(created).strftime("%Y-%m-%d %H:%M") - click.echo(f"[{entry['id']}] {entry['category']} | {entry['title']} | {created}") - click.echo(f" {entry['content'][:200]}...") - if entry["tags"]: - click.echo(f" Tags: {', '.join(entry['tags'])}") - click.echo() - finally: - await manager.close() - - -@click.command() -@click.argument("entry_id", type=int) -def get(entry_id): - """Get a specific memory entry by ID.""" - asyncio.run(_get(entry_id)) - - -async def _get(entry_id): - manager = await get_memory_manager() - try: - entry = await manager.memory_service.get_entry(entry_id) - if not entry: - click.echo(f"Entry {entry_id} not found.", err=True) - return - - click.echo(f"ID: {entry['id']}") - click.echo(f"Title: {entry['title']}") - click.echo(f"Category: {entry['category']}") - click.echo(f"Agent: {entry['agent_id']}") - click.echo(f"Project: {entry['project_path']}") - click.echo(f"Tags: {', '.join(entry['tags']) if entry['tags'] else '(none)'}") - click.echo(f"Created: {entry['created_at']}") - click.echo(f"Updated: {entry['updated_at']}") - click.echo(f"\nContent:\n{entry['content']}") - finally: - await manager.close() - - -@click.command() -@click.argument("entry_id", type=int) -@click.option("--title", "-t", help="New title") -@click.option("--content", "-c", help="New content") -@click.option("--category", "-g", callback=validate_category, help="New category") -@click.option("--tags", "-T", multiple=True, help="New tags") -def update(entry_id, title, content, category, tags): - """Update a memory entry.""" - asyncio.run(_update(entry_id, title, content, category, list(tags) if tags else None)) - - -async def _update(entry_id, title, content, category, tags): - manager = await get_memory_manager() - try: - result = await manager.memory_service.update_entry( - entry_id=entry_id, - title=title, - content=content, - category=category, - tags=tags, - ) - if not result: - click.echo(f"Entry {entry_id} not found.", err=True) - return - click.echo(f"Updated entry {entry_id}: {result['title']}") - finally: - await manager.close() - - -@click.command() -@click.argument("entry_id", type=int) -def delete(entry_id): - """Delete a memory entry.""" - asyncio.run(_delete(entry_id)) - - -async def _delete(entry_id): - manager = await get_memory_manager() - try: - deleted = await manager.memory_service.delete_entry(entry_id) - if not deleted: - click.echo(f"Entry {entry_id} not found.", err=True) - return - click.echo(f"Deleted entry {entry_id}.") - finally: - await manager.close() - - -@click.command() -@click.option("--message", "-m", required=True, help="Commit message") -@click.option("--agent-id", help="Agent ID") -@click.option("--project-path", help="Project path") -def commit(message, agent_id, project_path): - """Create a commit snapshot of current memory state.""" - asyncio.run(_commit(message, agent_id, project_path)) - - -async def _commit(message, agent_id, project_path): - manager = await get_memory_manager() - try: - result = await manager.commit_service.create_commit( - message=message, - agent_id=agent_id, - project_path=project_path, - ) - click.echo(f"Created commit {result['hash']}: {result['message']}") - finally: - await manager.close() - - -@click.command() -@click.option("--agent-id", help="Filter by agent ID") -@click.option("--project-path", help="Filter by project path") -@click.option("--limit", "-n", default=100, help="Number of commits to show") -@click.option("--offset", default=0, help="Offset for pagination") -def log(agent_id, project_path, limit, offset): - """Show commit history.""" - asyncio.run(_log(agent_id, project_path, limit, offset)) - - -async def _log(agent_id, project_path, limit, offset): - manager = await get_memory_manager() - try: - commits = await manager.commit_service.list_commits( - agent_id=agent_id, - project_path=project_path, - limit=limit, - offset=offset, - ) - if not commits: - click.echo("No commits found.") - return - - for commit in commits: - created = commit["created_at"] - if created: - created = datetime.fromisoformat(created).strftime("%Y-%m-%d %H:%M") - click.echo(f"commit {commit['hash']}") - click.echo(f"Author: {commit['agent_id']}") - click.echo(f"Date: {created}") - click.echo(f"\n {commit['message']}\n") - finally: - await manager.close() - - -@click.command() -@click.argument("hash1") -@click.argument("hash2") -def diff(hash1, hash2): - """Show diff between two commits.""" - asyncio.run(_diff(hash1, hash2)) - - -async def _diff(hash1, hash2): - manager = await get_memory_manager() - try: - result = await manager.commit_service.diff(hash1, hash2) - if not result: - click.echo("One or both commits not found. Check available commits with 'memory log'.", err=True) - return - - if result["added"]: - click.echo("Added entries:") - for entry in result["added"]: - click.echo(f" + [{entry['id']}] {entry['title']}") - if result["removed"]: - click.echo("\nRemoved entries:") - for entry in result["removed"]: - click.echo(f" - [{entry['id']}] {entry['title']}") - if result["modified"]: - click.echo("\nModified entries:") - for mod in result["modified"]: - click.echo(f" ~ [{mod['after']['id']}] {mod['after']['title']}") - - if not any([result["added"], result["removed"], result["modified"]]): - click.echo("No differences found.") - finally: - await manager.close() - - -@click.command() -@click.option("--host", default=os.getenv("MEMORY_API_HOST", "127.0.0.1"), help="Server host") -@click.option("--port", default=int(os.getenv("MEMORY_API_PORT", "8080")), help="Server port") -def serve(host, port): - """Start the API server.""" - import uvicorn - - from memory_manager.api.app import app as fastapi_app - - click.echo(f"Starting server at http://{host}:{port}") - click.echo(f"API documentation available at http://{host}:{port}/docs") - - uvicorn.run(fastapi_app, host=host, port=port) - - -@click.command() -def tui(): - """Launch the TUI dashboard.""" - from memory_manager.tui.app import TUIApp - - app = TUIApp() - app.run() - - -if __name__ == "__main__": - cli() +cli main content \ No newline at end of file