feat: Beta work (#17)

feat: Add multiple projects support 
feat: enhanced read_note for when initial result is not found
fix: merge frontmatter when updating note
fix: handle directory removed on sync watch

src/basic_memory/__init__.py

@@ -1,3 +1,8 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
+# Set this at the package level to ensure it's set before any modules import logfire
+import os
+
+os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
+
 __version__ = "0.8.0"

src/basic_memory/alembic/migrations.py

@@ -1,6 +1,5 @@
 """Functions for managing database migrations."""
 
-import asyncio
 from pathlib import Path
 from loguru import logger
 from alembic.config import Config
@@ -10,20 +9,16 @@ from alembic import command
 def get_alembic_config() -> Config:  # pragma: no cover
     """Get alembic config with correct paths."""
     migrations_path = Path(__file__).parent
-    alembic_ini = migrations_path.parent.parent.parent / "alembic.ini"
+    alembic_ini = migrations_path / "alembic.ini"
 
     config = Config(alembic_ini)
     config.set_main_option("script_location", str(migrations_path))
     return config
 
 
-async def reset_database():  # pragma: no cover
+def reset_database():  # pragma: no cover
     """Drop and recreate all tables."""
     logger.info("Resetting database...")
     config = get_alembic_config()
-
-    def _reset(cfg):
-        command.downgrade(cfg, "base")
-        command.upgrade(cfg, "head")
-
-    await asyncio.get_event_loop().run_in_executor(None, _reset, config)
+    command.downgrade(config, "base")
+    command.upgrade(config, "head")

src/basic_memory/api/app.py

@@ -1,5 +1,10 @@
 """FastAPI application for basic-memory knowledge graph API."""
 
+# Suppress logfire warnings
+import os
+
+os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
+
 from contextlib import asynccontextmanager
 
 import logfire
@@ -43,6 +48,12 @@ app.include_router(resource.router)
 @app.exception_handler(Exception)
 async def exception_handler(request, exc):  # pragma: no cover
     logger.exception(
-        f"An unhandled exception occurred for request '{request.url}', exception: {exc}"
+        "API unhandled exception",
+        url=str(request.url),
+        method=request.method,
+        client=request.client.host if request.client else None,
+        path=request.url.path,
+        error_type=type(exc).__name__,
+        error=str(exc),
     )
     return await http_exception_handler(request, HTTPException(status_code=500, detail=str(exc)))

src/basic_memory/api/routers/knowledge_router.py

@@ -33,7 +33,9 @@ async def create_entity(
     search_service: SearchServiceDep,
 ) -> EntityResponse:
     """Create an entity."""
-    logger.info(f"request: create_entity with data={data}")
+    logger.info(
+        "API request", endpoint="create_entity", entity_type=data.entity_type, title=data.title
+    )
 
     entity = await entity_service.create_entity(data)
 
@@ -41,7 +43,13 @@ async def create_entity(
     await search_service.index_entity(entity, background_tasks=background_tasks)
     result = EntityResponse.model_validate(entity)
 
-    logger.info(f"response: create_entity with result={result}")
+    logger.info(
+        "API response",
+        endpoint="create_entity",
+        title=result.title,
+        permalink=result.permalink,
+        status_code=201,
+    )
     return result
 
 
@@ -55,10 +63,23 @@ async def create_or_update_entity(
     search_service: SearchServiceDep,
 ) -> EntityResponse:
     """Create or update an entity. If entity exists, it will be updated, otherwise created."""
-    logger.info(f"request: create_or_update_entity with permalink={permalink}, data={data}")
+    logger.info(
+        "API request",
+        endpoint="create_or_update_entity",
+        permalink=permalink,
+        entity_type=data.entity_type,
+        title=data.title,
+    )
 
     # Validate permalink matches
     if data.permalink != permalink:
+        logger.warning(
+            "API validation error",
+            endpoint="create_or_update_entity",
+            permalink=permalink,
+            data_permalink=data.permalink,
+            error="Permalink mismatch",
+        )
         raise HTTPException(status_code=400, detail="Entity permalink must match URL path")
 
     # Try create_or_update operation
@@ -70,7 +91,12 @@ async def create_or_update_entity(
     result = EntityResponse.model_validate(entity)
 
     logger.info(
-        f"response: create_or_update_entity with result={result}, status_code={response.status_code}"
+        "API response",
+        endpoint="create_or_update_entity",
+        title=result.title,
+        permalink=result.permalink,
+        created=created,
+        status_code=response.status_code,
     )
     return result
 

src/basic_memory/cli/app.py

@@ -1,4 +1,5 @@
 import asyncio
+from typing import Optional
 
 import typer
 
@@ -6,13 +7,63 @@ from basic_memory import db
 from basic_memory.config import config
 
 
-asyncio.run(db.run_migrations(config))
+def version_callback(value: bool) -> None:
+    """Show version and exit."""
+    if value:  # pragma: no cover
+        import basic_memory
+
+        typer.echo(f"Basic Memory version: {basic_memory.__version__}")
+        raise typer.Exit()
+
 
 app = typer.Typer(name="basic-memory")
 
-import_app = typer.Typer()
-app.add_typer(import_app, name="import")
 
+@app.callback()
+def app_callback(
+    project: Optional[str] = typer.Option(
+        None,
+        "--project",
+        "-p",
+        help="Specify which project to use",
+        envvar="BASIC_MEMORY_PROJECT",
+    ),
+    version: Optional[bool] = typer.Option(
+        None,
+        "--version",
+        "-v",
+        help="Show version and exit.",
+        callback=version_callback,
+        is_eager=True,
+    ),
+) -> None:
+    """Basic Memory - Local-first personal knowledge management."""
+    # We use the project option to set the BASIC_MEMORY_PROJECT environment variable
+    # The config module will pick this up when loading
+    if project:  # pragma: no cover
+        import os
+        import importlib
+        from basic_memory import config as config_module
+
+        # Set the environment variable
+        os.environ["BASIC_MEMORY_PROJECT"] = project
+
+        # Reload the config module to pick up the new project
+        importlib.reload(config_module)
+
+        # Update the local reference
+        global config
+        from basic_memory.config import config as new_config
+
+        config = new_config
+
+
+# Run database migrations
+asyncio.run(db.run_migrations(config))
+
+# Register sub-command groups
+import_app = typer.Typer(help="Import data from various sources")
+app.add_typer(import_app, name="import")
 
 claude_app = typer.Typer()
 import_app.add_typer(claude_app, name="claude")

src/basic_memory/cli/commands/__init__.py

@@ -1,5 +1,17 @@
 """CLI commands for basic-memory."""
 
-from . import status, sync, db, import_memory_json, mcp
+from . import status, sync, db, import_memory_json, mcp, import_claude_conversations
+from . import import_claude_projects, import_chatgpt, tool, project
 
-__all__ = ["status", "sync", "db", "import_memory_json", "mcp"]
+__all__ = [
+    "status",
+    "sync",
+    "db",
+    "import_memory_json",
+    "mcp",
+    "import_claude_conversations",
+    "import_claude_projects",
+    "import_chatgpt",
+    "tool",
+    "project",
+]

src/basic_memory/cli/commands/db.py

@@ -1,7 +1,5 @@
 """Database management commands."""
 
-import asyncio
-
 import logfire
 import typer
 from loguru import logger
@@ -10,19 +8,19 @@ from basic_memory.alembic import migrations
 from basic_memory.cli.app import app
 
 
+@logfire.instrument()
 @app.command()
 def reset(
-    reindex: bool = typer.Option(False, "--reindex", help="Rebuild indices from filesystem"),
+    reindex: bool = typer.Option(False, "--reindex", help="Rebuild db index from filesystem"),
 ):  # pragma: no cover
     """Reset database (drop all tables and recreate)."""
-    with logfire.span("reset"):  # pyright: ignore [reportGeneralTypeIssues]
-        if typer.confirm("This will delete all data in your db. Are you sure?"):
-            logger.info("Resetting database...")
-            asyncio.run(migrations.reset_database())
+    if typer.confirm("This will delete all data in your db. Are you sure?"):
+        logger.info("Resetting database...")
+        migrations.reset_database()
 
-            if reindex:
-                # Import and run sync
-                from basic_memory.cli.commands.sync import sync
+        if reindex:
+            # Import and run sync
+            from basic_memory.cli.commands.sync import sync
 
-                logger.info("Rebuilding search index from filesystem...")
-                sync(watch=False)  # pyright: ignore
+            logger.info("Rebuilding search index from filesystem...")
+            sync(watch=False)  # pyright: ignore

src/basic_memory/cli/commands/import_chatgpt.py

@@ -208,6 +208,7 @@ async def get_markdown_processor() -> MarkdownProcessor:
 
 
 @import_app.command(name="chatgpt", help="Import conversations from ChatGPT JSON export.")
+@logfire.instrument(extract_args=False)
 def import_chatgpt(
     conversations_json: Annotated[
         Path, typer.Argument(help="Path to ChatGPT conversations.json file")
@@ -226,38 +227,35 @@ def import_chatgpt(
     After importing, run 'basic-memory sync' to index the new files.
     """
 
-    with logfire.span("import chatgpt"):  # pyright: ignore [reportGeneralTypeIssues]
-        try:
-            if conversations_json:
-                if not conversations_json.exists():
-                    typer.echo(f"Error: File not found: {conversations_json}", err=True)
-                    raise typer.Exit(1)
+    try:
+        if conversations_json:
+            if not conversations_json.exists():
+                typer.echo(f"Error: File not found: {conversations_json}", err=True)
+                raise typer.Exit(1)
 
-                # Get markdown processor
-                markdown_processor = asyncio.run(get_markdown_processor())
+            # Get markdown processor
+            markdown_processor = asyncio.run(get_markdown_processor())
 
-                # Process the file
-                base_path = config.home / folder
-                console.print(
-                    f"\nImporting chats from {conversations_json}...writing to {base_path}"
-                )
-                results = asyncio.run(
-                    process_chatgpt_json(conversations_json, folder, markdown_processor)
+            # Process the file
+            base_path = config.home / folder
+            console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
+            results = asyncio.run(
+                process_chatgpt_json(conversations_json, folder, markdown_processor)
+            )
+
+            # Show results
+            console.print(
+                Panel(
+                    f"[green]Import complete![/green]\n\n"
+                    f"Imported {results['conversations']} conversations\n"
+                    f"Containing {results['messages']} messages",
+                    expand=False,
                 )
+            )
 
-                # Show results
-                console.print(
-                    Panel(
-                        f"[green]Import complete![/green]\n\n"
-                        f"Imported {results['conversations']} conversations\n"
-                        f"Containing {results['messages']} messages",
-                        expand=False,
-                    )
-                )
+        console.print("\nRun 'basic-memory sync' to index the new files.")
 
-            console.print("\nRun 'basic-memory sync' to index the new files.")
-
-        except Exception as e:
-            logger.error("Import failed")
-            typer.echo(f"Error during import: {e}", err=True)
-            raise typer.Exit(1)
+    except Exception as e:
+        logger.error("Import failed")
+        typer.echo(f"Error during import: {e}", err=True)
+        raise typer.Exit(1)

src/basic_memory/cli/commands/import_claude_conversations.py

@@ -161,6 +161,7 @@ async def get_markdown_processor() -> MarkdownProcessor:
 
 
 @claude_app.command(name="conversations", help="Import chat conversations from Claude.ai.")
+@logfire.instrument(extract_args=False)
 def import_claude(
     conversations_json: Annotated[
         Path, typer.Argument(..., help="Path to conversations.json file")
@@ -179,35 +180,34 @@ def import_claude(
     After importing, run 'basic-memory sync' to index the new files.
     """
 
-    with logfire.span("import claude conversations"):  # pyright: ignore [reportGeneralTypeIssues]
-        try:
-            if not conversations_json.exists():
-                typer.echo(f"Error: File not found: {conversations_json}", err=True)
-                raise typer.Exit(1)
-
-            # Get markdown processor
-            markdown_processor = asyncio.run(get_markdown_processor())
-
-            # Process the file
-            base_path = config.home / folder
-            console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
-            results = asyncio.run(
-                process_conversations_json(conversations_json, base_path, markdown_processor)
-            )
-
-            # Show results
-            console.print(
-                Panel(
-                    f"[green]Import complete![/green]\n\n"
-                    f"Imported {results['conversations']} conversations\n"
-                    f"Containing {results['messages']} messages",
-                    expand=False,
-                )
-            )
-
-            console.print("\nRun 'basic-memory sync' to index the new files.")
-
-        except Exception as e:
-            logger.error("Import failed")
-            typer.echo(f"Error during import: {e}", err=True)
+    try:
+        if not conversations_json.exists():
+            typer.echo(f"Error: File not found: {conversations_json}", err=True)
             raise typer.Exit(1)
+
+        # Get markdown processor
+        markdown_processor = asyncio.run(get_markdown_processor())
+
+        # Process the file
+        base_path = config.home / folder
+        console.print(f"\nImporting chats from {conversations_json}...writing to {base_path}")
+        results = asyncio.run(
+            process_conversations_json(conversations_json, base_path, markdown_processor)
+        )
+
+        # Show results
+        console.print(
+            Panel(
+                f"[green]Import complete![/green]\n\n"
+                f"Imported {results['conversations']} conversations\n"
+                f"Containing {results['messages']} messages",
+                expand=False,
+            )
+        )
+
+        console.print("\nRun 'basic-memory sync' to index the new files.")
+
+    except Exception as e:
+        logger.error("Import failed")
+        typer.echo(f"Error during import: {e}", err=True)
+        raise typer.Exit(1)

src/basic_memory/cli/commands/import_claude_projects.py

@@ -144,6 +144,7 @@ async def get_markdown_processor() -> MarkdownProcessor:
 
 
 @claude_app.command(name="projects", help="Import projects from Claude.ai.")
+@logfire.instrument(extract_args=False)
 def import_projects(
     projects_json: Annotated[Path, typer.Argument(..., help="Path to projects.json file")] = Path(
         "projects.json"
@@ -161,36 +162,35 @@ def import_projects(
 
     After importing, run 'basic-memory sync' to index the new files.
     """
-    with logfire.span("import claude projects"):  # pyright: ignore [reportGeneralTypeIssues]
-        try:
-            if projects_json:
-                if not projects_json.exists():
-                    typer.echo(f"Error: File not found: {projects_json}", err=True)
-                    raise typer.Exit(1)
+    try:
+        if projects_json:
+            if not projects_json.exists():
+                typer.echo(f"Error: File not found: {projects_json}", err=True)
+                raise typer.Exit(1)
 
-                # Get markdown processor
-                markdown_processor = asyncio.run(get_markdown_processor())
+            # Get markdown processor
+            markdown_processor = asyncio.run(get_markdown_processor())
 
-                # Process the file
-                base_path = config.home / base_folder if base_folder else config.home
-                console.print(f"\nImporting projects from {projects_json}...writing to {base_path}")
-                results = asyncio.run(
-                    process_projects_json(projects_json, base_path, markdown_processor)
+            # Process the file
+            base_path = config.home / base_folder if base_folder else config.home
+            console.print(f"\nImporting projects from {projects_json}...writing to {base_path}")
+            results = asyncio.run(
+                process_projects_json(projects_json, base_path, markdown_processor)
+            )
+
+            # Show results
+            console.print(
+                Panel(
+                    f"[green]Import complete![/green]\n\n"
+                    f"Imported {results['documents']} project documents\n"
+                    f"Imported {results['prompts']} prompt templates",
+                    expand=False,
                 )
+            )
 
-                # Show results
-                console.print(
-                    Panel(
-                        f"[green]Import complete![/green]\n\n"
-                        f"Imported {results['documents']} project documents\n"
-                        f"Imported {results['prompts']} prompt templates",
-                        expand=False,
-                    )
-                )
+        console.print("\nRun 'basic-memory sync' to index the new files.")
 
-            console.print("\nRun 'basic-memory sync' to index the new files.")
-
-        except Exception as e:
-            logger.error("Import failed")
-            typer.echo(f"Error during import: {e}", err=True)
-            raise typer.Exit(1)
+    except Exception as e:
+        logger.error("Import failed")
+        typer.echo(f"Error during import: {e}", err=True)
+        raise typer.Exit(1)

src/basic_memory/cli/commands/import_memory_json.py

@@ -99,6 +99,7 @@ async def get_markdown_processor() -> MarkdownProcessor:
 
 
 @import_app.command()
+@logfire.instrument(extract_args=False)
 def memory_json(
     json_path: Annotated[Path, typer.Argument(..., help="Path to memory.json file")] = Path(
         "memory.json"
@@ -114,33 +115,32 @@ def memory_json(
     After importing, run 'basic-memory sync' to index the new files.
     """
 
-    with logfire.span("import memory_json"):  # pyright: ignore [reportGeneralTypeIssues]
-        if not json_path.exists():
-            typer.echo(f"Error: File not found: {json_path}", err=True)
-            raise typer.Exit(1)
+    if not json_path.exists():
+        typer.echo(f"Error: File not found: {json_path}", err=True)
+        raise typer.Exit(1)
 
-        try:
-            # Get markdown processor
-            markdown_processor = asyncio.run(get_markdown_processor())
+    try:
+        # Get markdown processor
+        markdown_processor = asyncio.run(get_markdown_processor())
 
-            # Process the file
-            base_path = config.home
-            console.print(f"\nImporting from {json_path}...writing to {base_path}")
-            results = asyncio.run(process_memory_json(json_path, base_path, markdown_processor))
+        # Process the file
+        base_path = config.home
+        console.print(f"\nImporting from {json_path}...writing to {base_path}")
+        results = asyncio.run(process_memory_json(json_path, base_path, markdown_processor))
 
-            # Show results
-            console.print(
-                Panel(
-                    f"[green]Import complete![/green]\n\n"
-                    f"Created {results['entities']} entities\n"
-                    f"Added {results['relations']} relations",
-                    expand=False,
-                )
+        # Show results
+        console.print(
+            Panel(
+                f"[green]Import complete![/green]\n\n"
+                f"Created {results['entities']} entities\n"
+                f"Added {results['relations']} relations",
+                expand=False,
             )
+        )
 
-            console.print("\nRun 'basic-memory sync' to index the new files.")
+        console.print("\nRun 'basic-memory sync' to index the new files.")
 
-        except Exception as e:
-            logger.error("Import failed")
-            typer.echo(f"Error during import: {e}", err=True)
-            raise typer.Exit(1)
+    except Exception as e:
+        logger.error("Import failed")
+        typer.echo(f"Error during import: {e}", err=True)
+        raise typer.Exit(1)

src/basic_memory/cli/commands/mcp.py

@@ -1,6 +1,8 @@
 """MCP server command."""
 
 from loguru import logger
+
+import basic_memory
 from basic_memory.cli.app import app
 from basic_memory.config import config
 
@@ -15,6 +17,10 @@ import basic_memory.mcp.tools  # noqa: F401  # pragma: no cover
 def mcp():  # pragma: no cover
     """Run the MCP server for Claude Desktop integration."""
     home_dir = config.home
+    project_name = config.project
+
     logger.info(f"Starting Basic Memory MCP server {basic_memory.__version__}")
-    logger.info(f"Home directory: {home_dir}")
+    logger.info(f"Project: {project_name}")
+    logger.info(f"Project directory: {home_dir}")
+
     mcp_server.run()

src/basic_memory/cli/commands/project.py

@@ -0,0 +1,119 @@
+"""Command module for basic-memory project management."""
+
+import os
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from basic_memory.cli.app import app
+from basic_memory.config import ConfigManager, config
+
+console = Console()
+
+# Create a project subcommand
+project_app = typer.Typer(help="Manage multiple Basic Memory projects")
+app.add_typer(project_app, name="project")
+
+
+def format_path(path: str) -> str:
+    """Format a path for display, using ~ for home directory."""
+    home = str(Path.home())
+    if path.startswith(home):
+        return path.replace(home, "~", 1)
+    return path
+
+
+@project_app.command("list")
+def list_projects() -> None:
+    """List all configured projects."""
+    config_manager = ConfigManager()
+    projects = config_manager.projects
+
+    table = Table(title="Basic Memory Projects")
+    table.add_column("Name", style="cyan")
+    table.add_column("Path", style="green")
+    table.add_column("Default", style="yellow")
+    table.add_column("Active", style="magenta")
+
+    default_project = config_manager.default_project
+    active_project = config.project
+
+    for name, path in projects.items():
+        is_default = "✓" if name == default_project else ""
+        is_active = "✓" if name == active_project else ""
+        table.add_row(name, format_path(path), is_default, is_active)
+
+    console.print(table)
+
+
+@project_app.command("add")
+def add_project(
+    name: str = typer.Argument(..., help="Name of the project"),
+    path: str = typer.Argument(..., help="Path to the project directory"),
+) -> None:
+    """Add a new project."""
+    config_manager = ConfigManager()
+
+    try:
+        # Resolve to absolute path
+        resolved_path = os.path.abspath(os.path.expanduser(path))
+        config_manager.add_project(name, resolved_path)
+        console.print(f"[green]Project '{name}' added at {format_path(resolved_path)}[/green]")
+
+        # Display usage hint
+        console.print("\nTo use this project:")
+        console.print(f"  basic-memory --project={name} <command>")
+        console.print("  # or")
+        console.print(f"  basic-memory project default {name}")
+    except ValueError as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("remove")
+def remove_project(
+    name: str = typer.Argument(..., help="Name of the project to remove"),
+) -> None:
+    """Remove a project from configuration."""
+    config_manager = ConfigManager()
+
+    try:
+        config_manager.remove_project(name)
+        console.print(f"[green]Project '{name}' removed from configuration[/green]")
+        console.print("[yellow]Note: The project files have not been deleted from disk.[/yellow]")
+    except ValueError as e:  # pragma: no cover
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("default")
+def set_default_project(
+    name: str = typer.Argument(..., help="Name of the project to set as default"),
+) -> None:
+    """Set the default project."""
+    config_manager = ConfigManager()
+
+    try:
+        config_manager.set_default_project(name)
+        console.print(f"[green]Project '{name}' set as default[/green]")
+    except ValueError as e:  # pragma: no cover
+        console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+@project_app.command("current")
+def show_current_project() -> None:
+    """Show the current project."""
+    config_manager = ConfigManager()
+    current = os.environ.get("BASIC_MEMORY_PROJECT", config_manager.default_project)
+
+    try:
+        path = config_manager.get_project_path(current)
+        console.print(f"Current project: [cyan]{current}[/cyan]")
+        console.print(f"Path: [green]{format_path(str(path))}[/green]")
+        console.print(f"Database: [blue]{format_path(str(config.database_path))}[/blue]")
+    except ValueError:  # pragma: no cover
+        console.print(f"[yellow]Warning: Project '{current}' not found in configuration[/yellow]")
+        console.print(f"Using default project: [cyan]{config_manager.default_project}[/cyan]")

src/basic_memory/cli/commands/status.py

@@ -130,15 +130,15 @@ async def run_status(sync_service: SyncService, verbose: bool = False):
 
 
 @app.command()
+@logfire.instrument(extract_args=False)
 def status(
     verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed file information"),
 ):
     """Show sync status between files and database."""
-    with logfire.span("status"):  # pyright: ignore [reportGeneralTypeIssues]
-        try:
-            sync_service = asyncio.run(get_sync_service())
-            asyncio.run(run_status(sync_service, verbose))  # pragma: no cover
-        except Exception as e:
-            logger.exception(f"Error checking status: {e}")
-            typer.echo(f"Error checking status: {e}", err=True)
-            raise typer.Exit(code=1)  # pragma: no cover
+    try:
+        sync_service = asyncio.run(get_sync_service())
+        asyncio.run(run_status(sync_service, verbose))  # pragma: no cover
+    except Exception as e:
+        logger.exception(f"Error checking status: {e}")
+        typer.echo(f"Error checking status: {e}", err=True)
+        raise typer.Exit(code=1)  # pragma: no cover

src/basic_memory/cli/commands/sync.py

@@ -93,8 +93,10 @@ def group_issues_by_directory(issues: List[ValidationIssue]) -> Dict[str, List[V
 def display_sync_summary(knowledge: SyncReport):
     """Display a one-line summary of sync changes."""
     total_changes = knowledge.total
+    project_name = config.project
+
     if total_changes == 0:
-        console.print("[green]Everything up to date[/green]")
+        console.print(f"[green]Project '{project_name}': Everything up to date[/green]")
         return
 
     # Format as: "Synced X files (A new, B modified, C moved, D deleted)"
@@ -113,16 +115,18 @@ def display_sync_summary(knowledge: SyncReport):
     if del_count:
         changes.append(f"[red]{del_count} deleted[/red]")
 
-    console.print(f"Synced {total_changes} files ({', '.join(changes)})")
+    console.print(f"Project '{project_name}': Synced {total_changes} files ({', '.join(changes)})")
 
 
 def display_detailed_sync_results(knowledge: SyncReport):
     """Display detailed sync results with trees."""
+    project_name = config.project
+
     if knowledge.total == 0:
-        console.print("\n[green]Everything up to date[/green]")
+        console.print(f"\n[green]Project '{project_name}': Everything up to date[/green]")
         return
 
-    console.print("\n[bold]Sync Results[/bold]")
+    console.print(f"\n[bold]Sync Results for Project '{project_name}'[/bold]")
 
     if knowledge.total > 0:
         knowledge_tree = Tree("[bold]Knowledge Files[/bold]")
@@ -150,23 +154,52 @@ def display_detailed_sync_results(knowledge: SyncReport):
 
 async def run_sync(verbose: bool = False, watch: bool = False, console_status: bool = False):
     """Run sync operation."""
+    import time
+
+    start_time = time.time()
+
+    logger.info(
+        "Sync command started",
+        project=config.project,
+        watch_mode=watch,
+        verbose=verbose,
+        directory=str(config.home),
+    )
 
     sync_service = await get_sync_service()
 
     # Start watching if requested
     if watch:
+        logger.info("Starting watch service after initial sync")
         watch_service = WatchService(
             sync_service=sync_service,
             file_service=sync_service.entity_service.file_service,
             config=config,
         )
-        # full sync
-        await sync_service.sync(config.home)
+
+        # full sync - no progress bars in watch mode
+        await sync_service.sync(config.home, show_progress=False)
+
         # watch changes
         await watch_service.run()  # pragma: no cover
     else:
-        # one time sync
-        knowledge_changes = await sync_service.sync(config.home)
+        # one time sync - use progress bars for better UX
+        logger.info("Running one-time sync")
+        knowledge_changes = await sync_service.sync(config.home, show_progress=True)
+
+        # Log results
+        duration_ms = int((time.time() - start_time) * 1000)
+        logger.info(
+            "Sync command completed",
+            project=config.project,
+            total_changes=knowledge_changes.total,
+            new_files=len(knowledge_changes.new),
+            modified_files=len(knowledge_changes.modified),
+            deleted_files=len(knowledge_changes.deleted),
+            moved_files=len(knowledge_changes.moves),
+            duration_ms=duration_ms,
+        )
+
         # Display results
         if verbose:
             display_detailed_sync_results(knowledge_changes)
@@ -191,12 +224,24 @@ def sync(
 ) -> None:
     """Sync knowledge files with the database."""
     try:
+        # Show which project we're syncing
+        if not watch:  # Don't show in watch mode as it would break the UI
+            typer.echo(f"Syncing project: {config.project}")
+            typer.echo(f"Project path: {config.home}")
+
         # Run sync
         asyncio.run(run_sync(verbose=verbose, watch=watch))
 
     except Exception as e:  # pragma: no cover
         if not isinstance(e, typer.Exit):
-            logger.exception("Sync failed", e)
+            logger.exception(
+                "Sync command failed",
+                project=config.project,
+                error=str(e),
+                error_type=type(e).__name__,
+                watch_mode=watch,
+                directory=str(config.home),
+            )
             typer.echo(f"Error during sync: {e}", err=True)
             raise typer.Exit(1)
         raise

src/basic_memory/cli/commands/tool.py

@@ -1,6 +1,7 @@
 """CLI tool commands for Basic Memory."""
 
 import asyncio
+import sys
 from typing import Optional, List, Annotated
 
 import typer
@@ -19,24 +20,78 @@ from basic_memory.mcp.prompts.continue_conversation import (
     continue_conversation as mcp_continue_conversation,
 )
 
+from basic_memory.mcp.prompts.recent_activity import (
+    recent_activity_prompt as recent_activity_prompt,
+)
+
 from basic_memory.schemas.base import TimeFrame
 from basic_memory.schemas.memory import MemoryUrl
 from basic_memory.schemas.search import SearchQuery, SearchItemType
 
 tool_app = typer.Typer()
-app.add_typer(tool_app, name="tools", help="cli versions mcp tools")
+app.add_typer(tool_app, name="tool", help="Direct access to MCP tools via CLI")
 
 
 @tool_app.command()
 def write_note(
     title: Annotated[str, typer.Option(help="The title of the note")],
-    content: Annotated[str, typer.Option(help="The content of the note")],
     folder: Annotated[str, typer.Option(help="The folder to create the note in")],
+    content: Annotated[
+        Optional[str],
+        typer.Option(
+            help="The content of the note. If not provided, content will be read from stdin. This allows piping content from other commands, e.g.: cat file.md | basic-memory tools write-note"
+        ),
+    ] = None,
     tags: Annotated[
         Optional[List[str]], typer.Option(help="A list of tags to apply to the note")
     ] = None,
 ):
+    """Create or update a markdown note. Content can be provided as an argument or read from stdin.
+
+    Content can be provided in two ways:
+    1. Using the --content parameter
+    2. Piping content through stdin (if --content is not provided)
+
+    Examples:
+
+    # Using content parameter
+    basic-memory tools write-note --title "My Note" --folder "notes" --content "Note content"
+
+    # Using stdin pipe
+    echo "# My Note Content" | basic-memory tools write-note --title "My Note" --folder "notes"
+
+    # Using heredoc
+    cat << EOF | basic-memory tools write-note --title "My Note" --folder "notes"
+    # My Document
+
+    This is my document content.
+
+    - Point 1
+    - Point 2
+    EOF
+
+    # Reading from a file
+    cat document.md | basic-memory tools write-note --title "Document" --folder "docs"
+    """
     try:
+        # If content is not provided, read from stdin
+        if content is None:
+            # Check if we're getting data from a pipe or redirect
+            if not sys.stdin.isatty():
+                content = sys.stdin.read()
+            else:  # pragma: no cover
+                # If stdin is a terminal (no pipe/redirect), inform the user
+                typer.echo(
+                    "No content provided. Please provide content via --content or by piping to stdin.",
+                    err=True,
+                )
+                raise typer.Exit(1)
+
+        # Also check for empty content
+        if content is not None and not content.strip():
+            typer.echo("Empty content provided. Please provide non-empty content.", err=True)
+            raise typer.Exit(1)
+
         note = asyncio.run(mcp_write_note(title, content, folder, tags))
         rprint(note)
     except Exception as e:  # pragma: no cover
@@ -166,7 +221,7 @@ def continue_conversation(
         Optional[str], typer.Option(help="How far back to look for activity")
     ] = None,
 ):
-    """Continue a previous conversation or work session."""
+    """Prompt to continue a previous conversation or work session."""
     try:
         # Prompt functions return formatted strings directly
         session = asyncio.run(mcp_continue_conversation(topic=topic, timeframe=timeframe))
@@ -177,3 +232,22 @@ def continue_conversation(
             typer.echo(f"Error continuing conversation: {e}", err=True)
             raise typer.Exit(1)
         raise
+
+
+# @tool_app.command(name="show-recent-activity")
+# def show_recent_activity(
+#     timeframe: Annotated[
+#         str, typer.Option(help="How far back to look for activity")
+#     ] = "7d",
+# ):
+#     """Prompt to show recent activity."""
+#     try:
+#         # Prompt functions return formatted strings directly
+#         session = asyncio.run(recent_activity_prompt(timeframe=timeframe))
+#         rprint(session)
+#     except Exception as e:  # pragma: no cover
+#         if not isinstance(e, typer.Exit):
+#             logger.exception("Error continuing conversation", e)
+#             typer.echo(f"Error continuing conversation: {e}", err=True)
+#             raise typer.Exit(1)
+#         raise

src/basic_memory/cli/main.py

@@ -1,6 +1,7 @@
 """Main CLI entry point for basic-memory."""  # pragma: no cover
 
 from basic_memory.cli.app import app  # pragma: no cover
+import typer
 
 # Register commands
 from basic_memory.cli.commands import (  # noqa: F401  # pragma: no cover
@@ -12,8 +13,46 @@ from basic_memory.cli.commands import (  # noqa: F401  # pragma: no cover
     import_claude_conversations,
     import_claude_projects,
     import_chatgpt,
-    tools,
+    tool,
+    project,
 )
 
+
+# Version command
+@app.callback(invoke_without_command=True)
+def main(
+    ctx: typer.Context,
+    project: str = typer.Option(  # noqa
+        "main",
+        "--project",
+        "-p",
+        help="Specify which project to use",
+        envvar="BASIC_MEMORY_PROJECT",
+    ),
+    version: bool = typer.Option(
+        False,
+        "--version",
+        "-V",
+        help="Show version information and exit.",
+        is_eager=True,
+    ),
+):
+    """Basic Memory - Local-first personal knowledge management system."""
+    if version:  # pragma: no cover
+        from basic_memory import __version__
+        from basic_memory.config import config
+
+        typer.echo(f"Basic Memory v{__version__}")
+        typer.echo(f"Current project: {config.project}")
+        typer.echo(f"Project path: {config.home}")
+        raise typer.Exit()
+
+    # Handle project selection via environment variable
+    if project:
+        import os
+
+        os.environ["BASIC_MEMORY_PROJECT"] = project
+
+
 if __name__ == "__main__":  # pragma: no cover
     app()

src/basic_memory/config.py

@@ -1,7 +1,9 @@
 """Configuration management for basic-memory."""
 
+import json
+import os
 from pathlib import Path
-from typing import Literal
+from typing import Any, Dict, Literal, Optional
 
 from loguru import logger
 from pydantic import Field, field_validator
@@ -12,6 +14,7 @@ from basic_memory.utils import setup_logging
 
 DATABASE_NAME = "memory.db"
 DATA_DIR_NAME = ".basic-memory"
+CONFIG_FILE_NAME = "config.json"
 
 Environment = Literal["test", "dev", "user"]
 
@@ -62,15 +65,160 @@ class ProjectConfig(BaseSettings):
         return v
 
 
-# Load project config
-config = ProjectConfig()
+class BasicMemoryConfig(BaseSettings):
+    """Pydantic model for Basic Memory global configuration."""
+
+    projects: Dict[str, str] = Field(
+        default_factory=lambda: {"main": str(Path.home() / "basic-memory")},
+        description="Mapping of project names to their filesystem paths",
+    )
+    default_project: str = Field(
+        default="main",
+        description="Name of the default project to use",
+    )
+
+    model_config = SettingsConfigDict(
+        env_prefix="BASIC_MEMORY_",
+        extra="ignore",
+    )
+
+    def model_post_init(self, __context: Any) -> None:
+        """Ensure configuration is valid after initialization."""
+        # Ensure main project exists
+        if "main" not in self.projects:
+            self.projects["main"] = str(Path.home() / "basic-memory")
+
+        # Ensure default project is valid
+        if self.default_project not in self.projects:
+            self.default_project = "main"
+
+
+class ConfigManager:
+    """Manages Basic Memory configuration."""
+
+    def __init__(self) -> None:
+        """Initialize the configuration manager."""
+        self.config_dir = Path.home() / DATA_DIR_NAME
+        self.config_file = self.config_dir / CONFIG_FILE_NAME
+
+        # Ensure config directory exists
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+
+        # Load or create configuration
+        self.config = self.load_config()
+
+    def load_config(self) -> BasicMemoryConfig:
+        """Load configuration from file or create default."""
+        if self.config_file.exists():
+            try:
+                data = json.loads(self.config_file.read_text())
+                return BasicMemoryConfig(**data)
+            except Exception as e:
+                logger.error(f"Failed to load config: {e}")
+                config = BasicMemoryConfig()
+                self.save_config(config)
+                return config
+        else:
+            config = BasicMemoryConfig()
+            self.save_config(config)
+            return config
+
+    def save_config(self, config: BasicMemoryConfig) -> None:
+        """Save configuration to file."""
+        try:
+            self.config_file.write_text(json.dumps(config.model_dump(), indent=2))
+        except Exception as e:  # pragma: no cover
+            logger.error(f"Failed to save config: {e}")
+
+    @property
+    def projects(self) -> Dict[str, str]:
+        """Get all configured projects."""
+        return self.config.projects.copy()
+
+    @property
+    def default_project(self) -> str:
+        """Get the default project name."""
+        return self.config.default_project
+
+    def get_project_path(self, project_name: Optional[str] = None) -> Path:
+        """Get the path for a specific project or the default project."""
+        name = project_name or self.config.default_project
+
+        # Check if specified in environment variable
+        if not project_name and "BASIC_MEMORY_PROJECT" in os.environ:
+            name = os.environ["BASIC_MEMORY_PROJECT"]
+
+        if name not in self.config.projects:
+            raise ValueError(f"Project '{name}' not found in configuration")
+
+        return Path(self.config.projects[name])
+
+    def add_project(self, name: str, path: str) -> None:
+        """Add a new project to the configuration."""
+        if name in self.config.projects:
+            raise ValueError(f"Project '{name}' already exists")
+
+        # Ensure the path exists
+        project_path = Path(path)
+        project_path.mkdir(parents=True, exist_ok=True)
+
+        self.config.projects[name] = str(project_path)
+        self.save_config(self.config)
+
+    def remove_project(self, name: str) -> None:
+        """Remove a project from the configuration."""
+        if name not in self.config.projects:
+            raise ValueError(f"Project '{name}' not found")
+
+        if name == self.config.default_project:
+            raise ValueError(f"Cannot remove the default project '{name}'")
+
+        del self.config.projects[name]
+        self.save_config(self.config)
+
+    def set_default_project(self, name: str) -> None:
+        """Set the default project."""
+        if name not in self.config.projects:  # pragma: no cover
+            raise ValueError(f"Project '{name}' not found")
+
+        self.config.default_project = name
+        self.save_config(self.config)
+
+
+def get_project_config(project_name: Optional[str] = None) -> ProjectConfig:
+    """Get a project configuration for the specified project."""
+    config_manager = ConfigManager()
+
+    # Get project name from environment variable or use provided name or default
+    actual_project_name = os.environ.get(
+        "BASIC_MEMORY_PROJECT", project_name or config_manager.default_project
+    )
+
+    try:
+        project_path = config_manager.get_project_path(actual_project_name)
+        return ProjectConfig(home=project_path, project=actual_project_name)
+    except ValueError:  # pragma: no cover
+        logger.warning(f"Project '{actual_project_name}' not found, using default")
+        project_path = config_manager.get_project_path(config_manager.default_project)
+        return ProjectConfig(home=project_path, project=config_manager.default_project)
+
+
+# Create config manager
+config_manager = ConfigManager()
+
+# Load project config for current context
+config = get_project_config()
+
+# setup logging to a single log file in user home directory
+user_home = Path.home()
+log_dir = user_home / DATA_DIR_NAME
+log_dir.mkdir(parents=True, exist_ok=True)
 
-# setup logging
 setup_logging(
     env=config.env,
-    home_dir=config.home,
+    home_dir=user_home,  # Use user home for logs
     log_level=config.log_level,
-    log_file=".basic-memory/basic-memory.log",
+    log_file=f"{DATA_DIR_NAME}/basic-memory.log",
     console=False,
 )
-logger.info(f"Starting Basic Memory {basic_memory.__version__}")
+logger.info(f"Starting Basic Memory {basic_memory.__version__} (Project: {config.project})")

src/basic_memory/db.py

@@ -86,8 +86,16 @@ async def get_or_create_db(
         _engine = create_async_engine(db_url, connect_args={"check_same_thread": False})
         _session_maker = async_sessionmaker(_engine, expire_on_commit=False)
 
-    assert _engine is not None  # for type checker
-    assert _session_maker is not None  # for type checker
+    # These checks should never fail since we just created the engine and session maker
+    # if they were None, but we'll check anyway for the type checker
+    if _engine is None:
+        logger.error("Failed to create database engine", db_path=str(db_path))
+        raise RuntimeError("Database engine initialization failed")
+
+    if _session_maker is None:
+        logger.error("Failed to create session maker", db_path=str(db_path))
+        raise RuntimeError("Session maker initialization failed")
+
     return _engine, _session_maker
 
 
@@ -121,8 +129,15 @@ async def engine_session_factory(
     try:
         _session_maker = async_sessionmaker(_engine, expire_on_commit=False)
 
-        assert _engine is not None  # for type checker
-        assert _session_maker is not None  # for type checker
+        # Verify that engine and session maker are initialized
+        if _engine is None:  # pragma: no cover
+            logger.error("Database engine is None in engine_session_factory")
+            raise RuntimeError("Database engine initialization failed")
+
+        if _session_maker is None:  # pragma: no cover
+            logger.error("Session maker is None in engine_session_factory")
+            raise RuntimeError("Session maker initialization failed")
+
         yield _engine, _session_maker
     finally:
         if _engine:

src/basic_memory/file_utils.py

@@ -2,11 +2,13 @@
 
 import hashlib
 from pathlib import Path
-from typing import Dict, Any, Union
+from typing import Any, Dict, Union
 
 import yaml
 from loguru import logger
 
+from basic_memory.utils import FilePath
+
 
 class FileError(Exception):
     """Base exception for file operations."""
@@ -48,42 +50,47 @@ async def compute_checksum(content: Union[str, bytes]) -> str:
         raise FileError(f"Failed to compute checksum: {e}")
 
 
-async def ensure_directory(path: Path) -> None:
+async def ensure_directory(path: FilePath) -> None:
     """
     Ensure directory exists, creating if necessary.
 
     Args:
-        path: Directory path to ensure
+        path: Directory path to ensure (Path or string)
 
     Raises:
         FileWriteError: If directory creation fails
     """
     try:
-        path.mkdir(parents=True, exist_ok=True)
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        path_obj.mkdir(parents=True, exist_ok=True)
     except Exception as e:  # pragma: no cover
-        logger.error(f"Failed to create directory: {path}: {e}")
+        logger.error("Failed to create directory", path=str(path), error=str(e))
         raise FileWriteError(f"Failed to create directory {path}: {e}")
 
 
-async def write_file_atomic(path: Path, content: str) -> None:
+async def write_file_atomic(path: FilePath, content: str) -> None:
     """
     Write file with atomic operation using temporary file.
 
     Args:
-        path: Target file path
+        path: Target file path (Path or string)
         content: Content to write
 
     Raises:
         FileWriteError: If write operation fails
     """
-    temp_path = path.with_suffix(".tmp")
+    # Convert string to Path if needed
+    path_obj = Path(path) if isinstance(path, str) else path
+    temp_path = path_obj.with_suffix(".tmp")
+
     try:
         temp_path.write_text(content)
-        temp_path.replace(path)
-        logger.debug(f"wrote file: {path}")
+        temp_path.replace(path_obj)
+        logger.debug("Wrote file atomically", path=str(path_obj), content_length=len(content))
     except Exception as e:  # pragma: no cover
         temp_path.unlink(missing_ok=True)
-        logger.error(f"Failed to write file: {path}: {e}")
+        logger.error("Failed to write file", path=str(path_obj), error=str(e))
         raise FileWriteError(f"Failed to write file {path}: {e}")
 
 
@@ -173,7 +180,7 @@ def remove_frontmatter(content: str) -> str:
     return parts[2].strip()
 
 
-async def update_frontmatter(path: Path, updates: Dict[str, Any]) -> str:
+async def update_frontmatter(path: FilePath, updates: Dict[str, Any]) -> str:
     """Update frontmatter fields in a file while preserving all content.
 
     Only modifies the frontmatter section, leaving all content untouched.
@@ -181,7 +188,7 @@ async def update_frontmatter(path: Path, updates: Dict[str, Any]) -> str:
     Returns checksum of updated file.
 
     Args:
-        path: Path to markdown file
+        path: Path to markdown file (Path or string)
         updates: Dict of frontmatter fields to update
 
     Returns:
@@ -192,8 +199,11 @@ async def update_frontmatter(path: Path, updates: Dict[str, Any]) -> str:
         ParseError: If frontmatter parsing fails
     """
     try:
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+
         # Read current content
-        content = path.read_text()
+        content = path_obj.read_text()
 
         # Parse current frontmatter
         current_fm = {}
@@ -208,9 +218,15 @@ async def update_frontmatter(path: Path, updates: Dict[str, Any]) -> str:
         yaml_fm = yaml.dump(new_fm, sort_keys=False)
         final_content = f"---\n{yaml_fm}---\n\n{content.strip()}"
 
-        await write_file_atomic(path, final_content)
+        logger.debug("Updating frontmatter", path=str(path_obj), update_keys=list(updates.keys()))
+
+        await write_file_atomic(path_obj, final_content)
         return await compute_checksum(final_content)
 
     except Exception as e:  # pragma: no cover
-        logger.error(f"Failed to update frontmatter in {path}: {e}")
+        logger.error(
+            "Failed to update frontmatter",
+            path=str(path) if isinstance(path, (str, Path)) else "<unknown>",
+            error=str(e),
+        )
         raise FileError(f"Failed to update frontmatter: {e}")

src/basic_memory/markdown/utils.py

@@ -5,6 +5,7 @@ from typing import Optional, Any
 
 from frontmatter import Post
 
+from basic_memory.file_utils import has_frontmatter, remove_frontmatter
 from basic_memory.markdown import EntityMarkdown
 from basic_memory.models import Entity, Observation as ObservationModel
 from basic_memory.utils import generate_permalink
@@ -78,6 +79,10 @@ async def schema_to_markdown(schema: Any) -> Post:
     content = schema.content or ""
     frontmatter_metadata = dict(schema.entity_metadata or {})
 
+    # if the content contains frontmatter, remove it and merge
+    if has_frontmatter(content):
+        content = remove_frontmatter(content)
+
     # Remove special fields for ordered frontmatter
     for field in ["type", "title", "permalink"]:
         frontmatter_metadata.pop(field, None)

src/basic_memory/mcp/prompts/__init__.py

@@ -10,12 +10,10 @@ from basic_memory.mcp.prompts import continue_conversation
 from basic_memory.mcp.prompts import recent_activity
 from basic_memory.mcp.prompts import search
 from basic_memory.mcp.prompts import ai_assistant_guide
-from basic_memory.mcp.prompts import json_canvas_spec
 
 __all__ = [
     "ai_assistant_guide",
     "continue_conversation",
-    "json_canvas_spec",
     "recent_activity",
     "search",
 ]

src/basic_memory/mcp/prompts/ai_assistant_guide.py

@@ -11,6 +11,7 @@ from basic_memory.mcp.server import mcp
     name="ai assistant guide",
     description="Give an AI assistant guidance on how to use Basic Memory tools effectively",
 )
+@logfire.instrument(extract_args=False)
 def ai_assistant_guide() -> str:
     """Return a concise guide on Basic Memory tools and how to use them.
 
@@ -20,9 +21,8 @@ def ai_assistant_guide() -> str:
     Returns:
         A focused guide on Basic Memory usage.
     """
-    with logfire.span("Getting Basic Memory guide"):  # pyright: ignore
-        logger.info("Loading AI assistant guide resource")
-        guide_doc = Path(__file__).parent.parent.parent.parent.parent / "data/ai_assistant_guide.md"
-        content = guide_doc.read_text()
-        logger.info(f"Loaded AI assistant guide ({len(content)} chars)")
-        return content
+    logger.info("Loading AI assistant guide resource")
+    guide_doc = Path(__file__).parent.parent.parent.parent.parent / "data/ai_assistant_guide.md"
+    content = guide_doc.read_text()
+    logger.info(f"Loaded AI assistant guide ({len(content)} chars)")
+    return content

src/basic_memory/mcp/prompts/continue_conversation.py

@@ -5,12 +5,13 @@ providing context from previous interactions to maintain continuity.
 """
 
 from textwrap import dedent
-from typing import Optional, List, Annotated
+from typing import Optional, Annotated
 
 from loguru import logger
 import logfire
 from pydantic import Field
 
+from basic_memory.mcp.prompts.utils import format_prompt_context, PromptContext, PromptContextItem
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.build_context import build_context
 from basic_memory.mcp.tools.recent_activity import recent_activity
@@ -21,9 +22,10 @@ from basic_memory.schemas.search import SearchQuery, SearchItemType
 
 
 @mcp.prompt(
-    name="continue conversation",
+    name="Continue Conversation",
     description="Continue a previous conversation",
 )
+@logfire.instrument(extract_args=False)
 async def continue_conversation(
     topic: Annotated[Optional[str], Field(description="Topic or keyword to search for")] = None,
     timeframe: Annotated[
@@ -43,140 +45,69 @@ async def continue_conversation(
     Returns:
         Context from previous sessions on this topic
     """
-    with logfire.span("Continuing session", topic=topic, timeframe=timeframe):  # pyright: ignore
-        logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}")
+    logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}")
 
-        # If topic provided, search for it
-        if topic:
-            search_results = await search(
-                SearchQuery(text=topic, after_date=timeframe, types=[SearchItemType.ENTITY])
-            )
+    # If topic provided, search for it
+    if topic:
+        search_results = await search(
+            SearchQuery(text=topic, after_date=timeframe, types=[SearchItemType.ENTITY])
+        )
 
-            # Build context from results
-            contexts = []
-            for result in search_results.results:
-                if hasattr(result, "permalink") and result.permalink:
-                    context = await build_context(f"memory://{result.permalink}")
-                    contexts.append(context)
+        # Build context from results
+        contexts = []
+        for result in search_results.results:
+            if hasattr(result, "permalink") and result.permalink:
+                context: GraphContext = await build_context(f"memory://{result.permalink}")
+                if context.primary_results:
+                    contexts.append(
+                        PromptContextItem(
+                            primary_results=context.primary_results[:1],  # pyright: ignore
+                            related_results=context.related_results[:3],  # pyright: ignore
+                        )
+                    )
 
-            # get context for the top 3 results
-            return format_continuation_context(topic, contexts[:3], timeframe)
+        # get context for the top 3 results
+        prompt_context = format_prompt_context(
+            PromptContext(topic=topic, timeframe=timeframe, results=contexts)  # pyright: ignore
+        )
 
+    else:
         # If no topic, get recent activity
-        recent = await recent_activity(timeframe=timeframe)
-        return format_continuation_context("Recent Activity", [recent], timeframe)
+        timeframe = timeframe or "7d"
+        recent: GraphContext = await recent_activity(
+            timeframe=timeframe, type=[SearchItemType.ENTITY]
+        )
+        prompt_context = format_prompt_context(
+            PromptContext(
+                topic=f"Recent Activity from ({timeframe})",
+                timeframe=timeframe,
+                results=[
+                    PromptContextItem(
+                        primary_results=recent.primary_results[:5],  # pyright: ignore
+                        related_results=recent.related_results[:2],  # pyright: ignore
+                    )
+                ],
+            )
+        )
 
-
-def format_continuation_context(
-    topic: str, contexts: List[GraphContext], timeframe: TimeFrame | None
-) -> str:
-    """Format continuation context into a helpful summary.
-
-    Args:
-        topic: The topic or focus of continuation
-        contexts: List of context graphs
-        timeframe: How far back to look for activity
-
-    Returns:
-        Formatted continuation summary
-    """
-    if not contexts or all(not context.primary_results for context in contexts):
-        return dedent(f"""
-            # Continuing conversation on: {topic}
-            
-            This is a memory retrieval session. 
-            Please use the available basic-memory tools to gather relevant context before responding. 
-            Start by executing one of the suggested commands below to retrieve content.
-            
-            I couldn't find any recent work specifically on this topic.
-            
-            ## Suggestions
-            - Try a different search term
-            - Check recent activity with `recent_activity(timeframe="1w")`
-            - Start a new topic with `write_note(...)`
-            """)
-
-    # Start building our summary with header
-    summary = dedent(f"""
-        # Continuing conversation on: {topic}
-        
-        This is a memory retrieval session. 
-        Please use the available basic-memory tools to gather relevant context before responding. 
-        Start by executing one of the suggested commands below to retrieve content.
-        
-        Here's what I found about the previous conversation:
-        """)
-
-    # Track what we've added to avoid duplicates
-    added_permalinks = set()
-    sections = []
-
-    # Process each context
-    for context in contexts:
-        # Add primary results
-        for primary in context.primary_results:
-            if hasattr(primary, "permalink") and primary.permalink not in added_permalinks:
-                added_permalinks.add(primary.permalink)
-
-                section = dedent(f"""
-                    ## {primary.title}
-                    - **Type**: {primary.type}
-                    """)
-
-                # Add creation date if available
-                if hasattr(primary, "created_at"):
-                    section += f"- **Created**: {primary.created_at.strftime('%Y-%m-%d %H:%M')}\n"
-
-                # Add content snippet
-                if hasattr(primary, "content") and primary.content:  # pyright: ignore
-                    content = primary.content or ""  # pyright: ignore
-                    if content:
-                        section += f"- **Content Snippet**: {content}\n"
-
-                section += dedent(f"""
-                    
-                    You can read this document with: `read_note("{primary.permalink}")`
-                    """)
-
-                # Add related documents if available
-                related_by_type = {}
-                if context.related_results:
-                    for related in context.related_results:
-                        if hasattr(related, "relation_type") and related.relation_type:  # pyright: ignore
-                            if related.relation_type not in related_by_type:  # pyright: ignore
-                                related_by_type[related.relation_type] = []  # pyright: ignore
-                            related_by_type[related.relation_type].append(related)  # pyright: ignore
-
-                if related_by_type:
-                    section += dedent("""
-                        ### Related Documents
-                        """)
-                    for rel_type, relations in related_by_type.items():
-                        display_type = rel_type.replace("_", " ").title()
-                        section += f"- **{display_type}**:\n"
-                        for rel in relations[:3]:  # Limit to avoid overwhelming
-                            if hasattr(rel, "to_entity") and rel.to_entity:
-                                section += f"  - `{rel.to_entity}`\n"
-
-                sections.append(section)
-
-    # Add all sections
-    summary += "\n".join(sections)
-
-    # Add next steps
+    # Add next steps with strong encouragement to write
     next_steps = dedent(f"""
         ## Next Steps
-        
+
         You can:
         - Explore more with: `search({{"text": "{topic}"}})`
-        - See what's changed: `recent_activity(timeframe="{timeframe}")`
+        - See what's changed: `recent_activity(timeframe="{timeframe or "7d"}")`
+        - **Record new learnings or decisions from this conversation:** `write_note(title="[Create a meaningful title]", content="[Content with observations and relations]")`
+        
+        ## Knowledge Capture Recommendation
+        
+        As you continue this conversation, **actively look for opportunities to:**
+        1. Record key information, decisions, or insights that emerge
+        2. Link new knowledge to existing topics 
+        3. Suggest capturing important context when appropriate
+        4. Create forward references to topics that might be created later
+        
+        Remember that capturing knowledge during conversations is one of the most valuable aspects of Basic Memory.
         """)
 
-    # Add specific exploration based on what we found
-    if added_permalinks:
-        first_permalink = next(iter(added_permalinks))
-        next_steps += dedent(f"""
-            - Continue the conversation: `build_context("memory://{first_permalink}")`
-            """)
-
-    return summary + next_steps
+    return prompt_context + next_steps

src/basic_memory/mcp/prompts/json_canvas_spec.py

@@ -1,27 +0,0 @@
-from pathlib import Path
-
-import logfire
-from loguru import logger
-
-from basic_memory.mcp.server import mcp
-
-
-@mcp.resource(
-    uri="memory://json_canvas_spec",
-    name="json canvas spec",
-    description="JSON Canvas specification for visualizing knowledge graphs in Obsidian",
-)
-def json_canvas_spec() -> str:
-    """Return the JSON Canvas specification for Obsidian visualizations.
-
-    Returns:
-        The JSON Canvas specification document.
-    """
-    with logfire.span("Getting JSON Canvas spec"):  # pyright: ignore
-        logger.info("Loading JSON Canvas spec resource")
-        canvas_spec = (
-            Path(__file__).parent.parent.parent.parent.parent / "data/json_canvas_spec_1_0.md"
-        )
-        content = canvas_spec.read_text()
-        logger.info(f"Loaded JSON Canvas spec ({len(content)} chars)")
-        return content

src/basic_memory/mcp/prompts/recent_activity.py

@@ -3,27 +3,29 @@
 These prompts help users see what has changed in their knowledge base recently.
 """
 
-from typing import Annotated, Optional
+from typing import Annotated
 
 from loguru import logger
 import logfire
 from pydantic import Field
 
-from basic_memory.mcp.prompts.utils import format_context_summary
+from basic_memory.mcp.prompts.utils import format_prompt_context, PromptContext, PromptContextItem
 from basic_memory.mcp.server import mcp
-from basic_memory.mcp.tools.recent_activity import recent_activity as recent_activity_tool
+from basic_memory.mcp.tools.recent_activity import recent_activity
 from basic_memory.schemas.base import TimeFrame
+from basic_memory.schemas.search import SearchItemType
 
 
 @mcp.prompt(
-    name="recent activity",
+    name="Share Recent Activity",
     description="Get recent activity from across the knowledge base",
 )
+@logfire.instrument(extract_args=False)
 async def recent_activity_prompt(
     timeframe: Annotated[
-        Optional[TimeFrame],
+        TimeFrame,
         Field(description="How far back to look for activity (e.g. '1d', '1 week')"),
-    ] = None,
+    ] = "7d",
 ) -> str:
     """Get recent activity from across the knowledge base.
 
@@ -36,11 +38,53 @@ async def recent_activity_prompt(
     Returns:
         Formatted summary of recent activity
     """
-    with logfire.span("Getting recent activity", timeframe=timeframe):  # pyright: ignore
-        logger.info(f"Getting recent activity, timeframe: {timeframe}")
+    logger.info(f"Getting recent activity, timeframe: {timeframe}")
 
-        results = await recent_activity_tool(timeframe=timeframe)
+    recent = await recent_activity(timeframe=timeframe, type=[SearchItemType.ENTITY])
 
-        time_display = f" ({timeframe})" if timeframe else ""
-        header = f"# Recent Activity{time_display}"
-        return format_context_summary(header, results)
+    prompt_context = format_prompt_context(
+        PromptContext(
+            topic=f"Recent Activity from ({timeframe})",
+            timeframe=timeframe,
+            results=[
+                PromptContextItem(
+                    primary_results=recent.primary_results[:5],
+                    related_results=recent.related_results[:2],
+                )
+            ],
+        )
+    )
+
+    # Add suggestions for summarizing recent activity
+    capture_suggestions = f"""
+    ## Opportunity to Capture Activity Summary
+    
+    Consider creating a summary note of recent activity:
+    
+    ```python
+    await write_note(
+        title="Activity Summary {timeframe}",
+        content='''
+        # Activity Summary for {timeframe}
+        
+        ## Overview
+        [Summary of key changes and developments over this period]
+        
+        ## Key Updates
+        [List main updates and their significance]
+        
+        ## Observations
+        - [trend] [Observation about patterns in recent activity]
+        - [insight] [Connection between different activities]
+        
+        ## Relations
+        - summarizes [[{recent.primary_results[0].title if recent.primary_results else "Recent Topic"}]]
+        - relates_to [[Project Overview]]
+        '''
+    )
+    ```
+    
+    Summarizing periodic activity helps create high-level insights and connections between topics.
+    """
+
+    return prompt_context + capture_suggestions

src/basic_memory/mcp/prompts/search.py

@@ -17,9 +17,10 @@ from basic_memory.schemas.base import TimeFrame
 
 
 @mcp.prompt(
-    name="search",
+    name="Search Knowledge Base",
     description="Search across all content in basic-memory",
 )
+@logfire.instrument(extract_args=False)
 async def search_prompt(
     query: str,
     timeframe: Annotated[
@@ -39,11 +40,10 @@ async def search_prompt(
     Returns:
         Formatted search results with context
     """
-    with logfire.span("Searching knowledge base", query=query, timeframe=timeframe):  # pyright: ignore
-        logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
+    logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
 
-        search_results = await search_tool(SearchQuery(text=query, after_date=timeframe))
-        return format_search_results(query, search_results, timeframe)
+    search_results = await search_tool(SearchQuery(text=query, after_date=timeframe))
+    return format_search_results(query, search_results, timeframe)
 
 
 def format_search_results(
@@ -65,11 +65,33 @@ def format_search_results(
             
             I couldn't find any results for this query.
             
-            ## Suggestions
+            ## Opportunity to Capture Knowledge!
+            
+            This is an excellent opportunity to create new knowledge on this topic. Consider:
+            
+            ```python
+            await write_note(
+                title="{query.capitalize()}",
+                content=f'''
+                # {query.capitalize()}
+                
+                ## Overview
+                [Summary of what we've discussed about {query}]
+                
+                ## Observations
+                - [category] [First observation about {query}]
+                - [category] [Second observation about {query}]
+                
+                ## Relations
+                - relates_to [[Other Relevant Topic]]
+                '''
+            )
+            ```
+            
+            ## Other Suggestions
             - Try a different search term
             - Broaden your search criteria
             - Check recent activity with `recent_activity(timeframe="1w")`
-            - Create new content with `write_note(...)`
             """)
 
     # Start building our summary with header
@@ -88,32 +110,38 @@ def format_search_results(
     for i, result in enumerate(results.results[:5]):  # Limit to top 5 results
         summary += dedent(f"""
             ## {i + 1}. {result.title}
-            - **Type**: {result.type}
+            - **Type**: {result.type.value}
             """)
 
         # Add creation date if available in metadata
-        if hasattr(result, "metadata") and result.metadata and "created_at" in result.metadata:
+        if result.metadata and "created_at" in result.metadata:
             created_at = result.metadata["created_at"]
             if hasattr(created_at, "strftime"):
-                summary += f"- **Created**: {created_at.strftime('%Y-%m-%d %H:%M')}\n"
+                summary += (
+                    f"- **Created**: {created_at.strftime('%Y-%m-%d %H:%M')}\n"  # pragma: no cover
+                )
             elif isinstance(created_at, str):
                 summary += f"- **Created**: {created_at}\n"
 
         # Add score and excerpt
         summary += f"- **Relevance Score**: {result.score:.2f}\n"
+
         # Add excerpt if available in metadata
-        if hasattr(result, "metadata") and result.metadata and "excerpt" in result.metadata:
-            summary += f"- **Excerpt**: {result.metadata['excerpt']}\n"
+        if result.content:
+            summary += f"- **Excerpt**:\n{result.content}\n"
 
         # Add permalink for retrieving content
-        if hasattr(result, "permalink") and result.permalink:
+        if result.permalink:
             summary += dedent(f"""
-                
                 You can view this content with: `read_note("{result.permalink}")`
                 Or explore its context with: `build_context("memory://{result.permalink}")`
                 """)
+        else:
+            summary += dedent(f"""
+                You can view this file with: `read_file("{result.file_path}")`
+                """)  # pragma: no cover
 
-    # Add next steps
+    # Add next steps with strong write encouragement
     summary += dedent(f"""
         ## Next Steps
         
@@ -122,6 +150,35 @@ def format_search_results(
         - Exclude terms: `search("{query} NOT exclude_term")`
         - View more results: `search("{query}", after_date=None)`
         - Check recent activity: `recent_activity()`
+        
+        ## Synthesize and Capture Knowledge
+        
+        Consider creating a new note that synthesizes what you've learned:
+        
+        ```python
+        await write_note(
+            title="Synthesis of {query.capitalize()} Information",
+            content='''
+            # Synthesis of {query.capitalize()} Information
+            
+            ## Overview
+            [Synthesis of the search results and your conversation]
+            
+            ## Key Insights
+            [Summary of main points learned from these results]
+            
+            ## Observations
+            - [insight] [Important observation from search results]
+            - [connection] [How this connects to other topics]
+            
+            ## Relations
+            - relates_to [[{results.results[0].title if results.results else "Related Topic"}]]
+            - extends [[Another Relevant Topic]]
+            '''
+        )
+        ```
+        
+        Remember that capturing synthesized knowledge is one of the most valuable features of Basic Memory.
         """)
 
     return summary

src/basic_memory/mcp/prompts/utils.py

@@ -4,95 +4,152 @@ These utilities help format data from various tools into consistent,
 user-friendly markdown summaries.
 """
 
-from basic_memory.schemas.memory import GraphContext
+from dataclasses import dataclass
+from textwrap import dedent
+from typing import List
+
+from basic_memory.schemas.base import TimeFrame
+from basic_memory.schemas.memory import (
+    normalize_memory_url,
+    EntitySummary,
+    RelationSummary,
+    ObservationSummary,
+)
 
 
-def format_context_summary(header: str, context: GraphContext) -> str:
-    """Format GraphContext as a helpful markdown summary.
+@dataclass
+class PromptContextItem:
+    primary_results: List[EntitySummary]
+    related_results: List[EntitySummary | RelationSummary | ObservationSummary]
 
-    This creates a user-friendly markdown response that explains the context
-    and provides guidance on how to explore further.
 
-    Args:
-        header: The title to use for the summary
-        context: The GraphContext object to format
+@dataclass
+class PromptContext:
+    timeframe: TimeFrame
+    topic: str
+    results: List[PromptContextItem]
 
+
+def format_prompt_context(context: PromptContext) -> str:
+    """Format continuation context into a helpful summary.
     Returns:
-        Formatted markdown string with the context summary
+        Formatted continuation summary
     """
-    summary = []
+    if not context.results:
+        return dedent(f"""
+            # Continuing conversation on: {context.topic}
 
-    # Extract URI for reference
-    uri = context.metadata.uri or "a/permalink-value"
-
-    # Add header
-    summary.append(f"{header}")
-    summary.append("")
-
-    # Primary document section
-    if context.primary_results:
-        summary.append(f"## Primary Documents ({len(context.primary_results)})")
-
-        for primary in context.primary_results:
-            summary.append(f"### {primary.title}")
-            summary.append(f"- **Type**: {primary.type}")
-            summary.append(f"- **Path**: {primary.file_path}")
-            summary.append(f"- **Created**: {primary.created_at.strftime('%Y-%m-%d %H:%M')}")
-            summary.append("")
-            summary.append(
-                f'To view this document\'s content: `read_note("{primary.permalink}")` or `read_note("{primary.title}")` '
+            This is a memory retrieval session. 
+            The supplied query did not return any information specifically on this topic.
+            
+            ## Opportunity to Capture New Knowledge!
+            
+            This is an excellent chance to start documenting this topic:
+            
+            ```python
+            await write_note(
+                title="{context.topic}",
+                content=f'''
+                # {context.topic}
+                
+                ## Overview
+                [Summary of what we know about {context.topic}]
+                
+                ## Key Points
+                [Main aspects or components of {context.topic}]
+                
+                ## Observations
+                - [category] [First important observation about {context.topic}]
+                - [category] [Second observation about {context.topic}]
+                
+                ## Relations
+                - relates_to [[Related Topic]]
+                - part_of [[Broader Context]]
+                '''
             )
-            summary.append("")
-    else:
-        summary.append("\nNo primary documents found.")
+            ```
+            
+            ## Other Options
+            
+            Please use the available basic-memory tools to gather relevant context before responding.
+            You can also:
+            - Try a different search term
+            - Check recent activity with `recent_activity(timeframe="1w")`
+            """)
 
-    # Related documents section
-    if context.related_results:
-        summary.append(f"## Related Documents ({len(context.related_results)})")
+    # Start building our summary with header - add knowledge capture emphasis
+    summary = dedent(f"""
+        # Continuing conversation on: {context.topic}
 
-        # Group by relation type for better organization
-        relation_types = {}
-        for rel in context.related_results:
-            if hasattr(rel, "relation_type"):
-                rel_type = rel.relation_type  # pyright: ignore
-                if rel_type not in relation_types:
-                    relation_types[rel_type] = []
-                relation_types[rel_type].append(rel)
+        This is a memory retrieval session. 
+        
+        Please use the available basic-memory tools to gather relevant context before responding. 
+        Start by executing one of the suggested commands below to retrieve content.
 
-        # Display relations grouped by type
-        for rel_type, relations in relation_types.items():
-            summary.append(f"### {rel_type.replace('_', ' ').title()} ({len(relations)})")
+        Here's what I found from previous conversations:
+        
+        > **Knowledge Capture Recommendation:** As you continue this conversation, actively look for opportunities to record new information, decisions, or insights that emerge. Use `write_note()` to document important context.
+        """)
 
-            for rel in relations:
-                if hasattr(rel, "to_id") and rel.to_id:
-                    summary.append(f"- **{rel.to_id}**")
-                    summary.append(f'  - View document: `read_note("{rel.to_id}")` ')
-                    summary.append(
-                        f'  - Explore connections: `build_context("memory://{rel.to_id}")` '
+    # Track what we've added to avoid duplicates
+    added_permalinks = set()
+    sections = []
+
+    # Process each context
+    for context in context.results:  # pyright: ignore
+        for primary in context.primary_results:  # pyright: ignore
+            if primary.permalink not in added_permalinks:
+                primary_permalink = primary.permalink
+
+                added_permalinks.add(primary_permalink)
+
+                memory_url = normalize_memory_url(primary_permalink)
+                section = dedent(f"""
+                    --- {memory_url}
+                
+                    ## {primary.title}
+                    - **Type**: {primary.type}
+                    """)
+
+                # Add creation date
+                section += f"- **Created**: {primary.created_at.strftime('%Y-%m-%d %H:%M')}\n"
+
+                # Add content snippet
+                if hasattr(primary, "content") and primary.content:  # pyright: ignore
+                    content = primary.content or ""  # pyright: ignore
+                    if content:
+                        section += f"\n**Excerpt**:\n{content}\n"
+
+                section += dedent(f"""
+    
+                    You can read this document with: `read_note("{primary_permalink}")`
+                    """)
+                sections.append(section)
+
+        if context.related_results:  # pyright: ignore
+            section += dedent(  # pyright: ignore
+                """   
+                ## Related Context
+                """
+            )
+
+            for related in context.related_results:  # pyright: ignore
+                section_content = dedent(f"""
+                    - type: **{related.type}**
+                    - title: {related.title}
+                    """)
+                if related.permalink:
+                    section_content += (
+                        f'You can view this document with: `read_note("{related.permalink}")`'
                     )
                 else:
-                    summary.append(f"- **Unresolved relation**: {rel.permalink}")
-            summary.append("")
+                    section_content += (
+                        f'You can view this file with: `read_file("{related.file_path}")`'
+                    )
 
-    # Next steps section
-    summary.append("## Next Steps")
-    summary.append("Here are some ways to explore further:")
+                section += section_content
+                sections.append(section)
 
-    search_term = uri.split("/")[-1]
-    summary.append(f'- **Search related topics**: `search({{"text": "{search_term}"}})`')
-
-    summary.append('- **Check recent changes**: `recent_activity(timeframe="3 days")`')
-    summary.append(f'- **Explore all relations**: `build_context("memory://{uri}/*")`')
-
-    # Tips section
-    summary.append("")
-    summary.append("## Tips")
-    summary.append(
-        f'- For more specific context, increase depth: `build_context("memory://{uri}", depth=2)`'
-    )
-    summary.append(
-        "- You can follow specific relation types using patterns like: `memory://document/relation-type/*`"
-    )
-    summary.append("- Look for connected documents by checking relations between them")
-
-    return "\n".join(summary)
+    # Add all sections
+    summary += "\n".join(sections)
+    return summary

src/basic_memory/mcp/tools/build_context.py

@@ -17,6 +17,7 @@ from basic_memory.schemas.memory import (
 from basic_memory.schemas.base import TimeFrame
 
 
+@logfire.instrument(extract_args=False)
 @mcp.tool(
     description="""Build context from a memory:// URI to continue conversations naturally.
     
@@ -70,18 +71,17 @@ async def build_context(
         # Research the history of a feature
         build_context("memory://features/knowledge-graph", timeframe="3 months ago")
     """
-    with logfire.span("Building context", url=url, depth=depth, timeframe=timeframe):  # pyright: ignore [reportGeneralTypeIssues]
-        logger.info(f"Building context from {url}")
-        url = normalize_memory_url(url)
-        response = await call_get(
-            client,
-            f"/memory/{memory_url_path(url)}",
-            params={
-                "depth": depth,
-                "timeframe": timeframe,
-                "page": page,
-                "page_size": page_size,
-                "max_related": max_related,
-            },
-        )
-        return GraphContext.model_validate(response.json())
+    logger.info(f"Building context from {url}")
+    url = normalize_memory_url(url)
+    response = await call_get(
+        client,
+        f"/memory/{memory_url_path(url)}",
+        params={
+            "depth": depth,
+            "timeframe": timeframe,
+            "page": page,
+            "page_size": page_size,
+            "max_related": max_related,
+        },
+    )
+    return GraphContext.model_validate(response.json())

src/basic_memory/mcp/tools/canvas.py

@@ -17,6 +17,7 @@ from basic_memory.mcp.tools.utils import call_put
 @mcp.tool(
     description="Create an Obsidian canvas file to visualize concepts and connections.",
 )
+@logfire.instrument(extract_args=False)
 async def canvas(
     nodes: List[Dict[str, Any]],
     edges: List[Dict[str, Any]],
@@ -73,27 +74,26 @@ async def canvas(
     }
     ```
     """
-    with logfire.span("Creating canvas", folder=folder, title=title):  # type: ignore
-        # Ensure path has .canvas extension
-        file_title = title if title.endswith(".canvas") else f"{title}.canvas"
-        file_path = f"{folder}/{file_title}"
+    # Ensure path has .canvas extension
+    file_title = title if title.endswith(".canvas") else f"{title}.canvas"
+    file_path = f"{folder}/{file_title}"
 
-        # Create canvas data structure
-        canvas_data = {"nodes": nodes, "edges": edges}
+    # Create canvas data structure
+    canvas_data = {"nodes": nodes, "edges": edges}
 
-        # Convert to JSON
-        canvas_json = json.dumps(canvas_data, indent=2)
+    # Convert to JSON
+    canvas_json = json.dumps(canvas_data, indent=2)
 
-        # Write the file using the resource API
-        logger.info(f"Creating canvas file: {file_path}")
-        response = await call_put(client, f"/resource/{file_path}", json=canvas_json)
+    # Write the file using the resource API
+    logger.info(f"Creating canvas file: {file_path}")
+    response = await call_put(client, f"/resource/{file_path}", json=canvas_json)
 
-        # Parse response
-        result = response.json()
-        logger.debug(result)
+    # Parse response
+    result = response.json()
+    logger.debug(result)
 
-        # Build summary
-        action = "Created" if response.status_code == 201 else "Updated"
-        summary = [f"# {action}: {file_path}", "\nThe canvas is ready to open in Obsidian."]
+    # Build summary
+    action = "Created" if response.status_code == 201 else "Updated"
+    summary = [f"# {action}: {file_path}", "\nThe canvas is ready to open in Obsidian."]
 
-        return "\n".join(summary)
+    return "\n".join(summary)

src/basic_memory/mcp/tools/delete_note.py

@@ -9,6 +9,7 @@ from basic_memory.schemas import DeleteEntitiesResponse
 
 
 @mcp.tool(description="Delete a note by title or permalink")
+@logfire.instrument(extract_args=False)
 async def delete_note(identifier: str) -> bool:
     """Delete a note from the knowledge base.
 
@@ -25,7 +26,6 @@ async def delete_note(identifier: str) -> bool:
         # Delete by permalink
         delete_note("notes/project-planning")
     """
-    with logfire.span("Deleting note", identifier=identifier):  # pyright: ignore [reportGeneralTypeIssues]
-        response = await call_delete(client, f"/knowledge/entities/{identifier}")
-        result = DeleteEntitiesResponse.model_validate(response.json())
-        return result.deleted
+    response = await call_delete(client, f"/knowledge/entities/{identifier}")
+    result = DeleteEntitiesResponse.model_validate(response.json())
+    return result.deleted

src/basic_memory/mcp/tools/read_note.py

@@ -1,36 +1,37 @@
 """Read note tool for Basic Memory MCP server."""
 
+from textwrap import dedent
+
 import logfire
 from loguru import logger
 
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.search import search
 from basic_memory.mcp.tools.utils import call_get
 from basic_memory.schemas.memory import memory_url_path
+from basic_memory.schemas.search import SearchQuery
 
 
 @mcp.tool(
     description="Read a markdown note by title or permalink.",
 )
+@logfire.instrument(extract_args=False)
 async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
     """Read a markdown note from the knowledge base.
 
-    This tool finds and retrieves a note by its title or permalink, returning
-    the raw markdown content including observations, relations, and metadata.
-    Unlike read_file, this tool is aware of the knowledge graph structure and
-    will attempt to resolve entity references if the file path doesn't exist.
+    This tool finds and retrieves a note by its title, permalink, or content search,
+    returning the raw markdown content including observations, relations, and metadata.
+    It will try multiple lookup strategies to find the most relevant note.
 
     Args:
         identifier: The title or permalink of the note to read
-                   Can be a full memory:// URL, a permalink, or a title
+                   Can be a full memory:// URL, a permalink, a title, or search text
         page: Page number for paginated results (default: 1)
         page_size: Number of items per page (default: 10)
 
     Returns:
-        The full markdown content of the note, either from file content
-        or constructed from entity data if direct file access fails.
-        For entities without markdown content, returns a message indicating
-        the entity was found but has no content.
+        The full markdown content of the note if found, or helpful guidance if not found.
 
     Examples:
         # Read by permalink
@@ -45,16 +46,147 @@ async def read_note(identifier: str, page: int = 1, page_size: int = 10) -> str:
         # Read with pagination
         read_note("Project Updates", page=2, page_size=5)
     """
-    with logfire.span("Reading note", identifier=identifier):  # pyright: ignore [reportGeneralTypeIssues]
-        # Get the file via REST API
-        entity_path = memory_url_path(identifier)
-        path = f"/resource/{entity_path}"
-        logger.info(f"Reading note from URL: {path}")
+    # Get the file via REST API - first try direct permalink lookup
+    entity_path = memory_url_path(identifier)
+    path = f"/resource/{entity_path}"
+    logger.info(f"Attempting to read note from URL: {path}")
 
+    try:
+        # Try direct lookup first
         response = await call_get(client, path, params={"page": page, "page_size": page_size})
 
-        # Just return the content as a string
+        # If successful, return the content
         if response.status_code == 200:
+            logger.info("Returning read_note result from resource: {path}", path=entity_path)
             return response.text
-        else:
-            return f"Error: Could not find entity at {identifier}"
+    except Exception as e:  # pragma: no cover
+        logger.info(f"Direct lookup failed for '{path}': {e}")
+        # Continue to fallback methods
+
+    # Fallback 1: Try title search via API
+    logger.info(f"Search title for: {identifier}")
+    title_results = await search(SearchQuery(title=identifier))
+
+    if title_results and title_results.results:
+        result = title_results.results[0]  # Get the first/best match
+        if result.permalink:
+            try:
+                # Try to fetch the content using the found permalink
+                path = f"/resource/{result.permalink}"
+                response = await call_get(
+                    client, path, params={"page": page, "page_size": page_size}
+                )
+
+                if response.status_code == 200:
+                    logger.info(f"Found note by title search: {result.permalink}")
+                    return response.text
+            except Exception as e:  # pragma: no cover
+                logger.info(
+                    f"Failed to fetch content for found title match {result.permalink}: {e}"
+                )
+    else:
+        logger.info(f"No results in title search for: {identifier}")
+
+    # Fallback 2: Text search as a last resort
+    logger.info(f"Title search failed, trying text search for: {identifier}")
+    text_results = await search(SearchQuery(text=identifier))
+
+    # We didn't find a direct match, construct a helpful error message
+    if not text_results or not text_results.results:
+        # No results at all
+        return format_not_found_message(identifier)
+    else:
+        # We found some related results
+        return format_related_results(identifier, text_results.results[:5])
+
+
+def format_not_found_message(identifier: str) -> str:
+    """Format a helpful message when no note was found."""
+    return dedent(f"""
+        # Note Not Found: "{identifier}"
+        
+        I couldn't find any notes matching "{identifier}". Here are some suggestions:
+        
+        ## Check Identifier Type
+        - If you provided a title, try using the exact permalink instead
+        - If you provided a permalink, check for typos or try a broader search
+        
+        ## Search Instead
+        Try searching for related content:
+        ```
+        search(query="{identifier}")
+        ```
+        
+        ## Recent Activity
+        Check recently modified notes:
+        ```
+        recent_activity(timeframe="7d")
+        ```
+        
+        ## Create New Note
+        This might be a good opportunity to create a new note on this topic:
+        ```
+        write_note(
+            title="{identifier.capitalize()}",
+            content='''
+            # {identifier.capitalize()}
+            
+            ## Overview
+            [Your content here]
+            
+            ## Observations
+            - [category] [Observation about {identifier}]
+            
+            ## Relations
+            - relates_to [[Related Topic]]
+            ''',
+            folder="notes"
+        )
+        ```
+    """)
+
+
+def format_related_results(identifier: str, results) -> str:
+    """Format a helpful message with related results when an exact match wasn't found."""
+    message = dedent(f"""
+        # Note Not Found: "{identifier}"
+        
+        I couldn't find an exact match for "{identifier}", but I found some related notes:
+        
+        """)
+
+    for i, result in enumerate(results):
+        message += dedent(f"""
+            ## {i + 1}. {result.title}
+            - **Type**: {result.type.value}
+            - **Permalink**: {result.permalink}
+            
+            You can read this note with:
+            ```
+            read_note("{result.permalink}")
+            ```
+            
+            """)
+
+    message += dedent("""
+        ## Try More Specific Lookup
+        For exact matches, try using the full permalink from one of the results above.
+        
+        ## Search For More Results
+        To see more related content:
+        ```
+        search(query="{identifier}")
+        ```
+        
+        ## Create New Note
+        If none of these match what you're looking for, consider creating a new note:
+        ```
+        write_note(
+            title="[Your title]",
+            content="[Your content]",
+            folder="notes"
+        )
+        ```
+    """)
+
+    return message

src/basic_memory/mcp/tools/recent_activity.py

@@ -25,6 +25,7 @@ from basic_memory.schemas.search import SearchItemType
     Or standard formats like "7d"
     """,
 )
+@logfire.instrument(extract_args=False)
 async def recent_activity(
     type: Optional[List[SearchItemType]] = None,
     depth: Optional[int] = 1,
@@ -74,29 +75,28 @@ async def recent_activity(
         - For focused queries, consider using build_context with a specific URI
         - Max timeframe is 1 year in the past
     """
-    with logfire.span("Getting recent activity", type=type, depth=depth, timeframe=timeframe):  # pyright: ignore [reportGeneralTypeIssues]
-        logger.info(
-            f"Getting recent activity from {type}, depth={depth}, timeframe={timeframe}, page={page}, page_size={page_size}, max_related={max_related}"
-        )
-        params = {
-            "page": page,
-            "page_size": page_size,
-            "max_related": max_related,
-        }
-        if depth:
-            params["depth"] = depth
-        if timeframe:
-            params["timeframe"] = timeframe  # pyright: ignore
+    logger.info(
+        f"Getting recent activity from type={type}, depth={depth}, timeframe={timeframe}, page={page}, page_size={page_size}, max_related={max_related}"
+    )
+    params = {
+        "page": page,
+        "page_size": page_size,
+        "max_related": max_related,
+    }
+    if depth:
+        params["depth"] = depth
+    if timeframe:
+        params["timeframe"] = timeframe  # pyright: ignore
 
-        # send enum values if we have an enum, else send string value
-        if type:
-            params["type"] = [  # pyright: ignore
-                type.value if isinstance(type, SearchItemType) else type for type in type
-            ]
+    # send enum values if we have an enum, else send string value
+    if type:
+        params["type"] = [  # pyright: ignore
+            type.value if isinstance(type, SearchItemType) else type for type in type
+        ]
 
-        response = await call_get(
-            client,
-            "/memory/recent",
-            params=params,
-        )
-        return GraphContext.model_validate(response.json())
+    response = await call_get(
+        client,
+        "/memory/recent",
+        params=params,
+    )
+    return GraphContext.model_validate(response.json())

src/basic_memory/mcp/tools/search.py

@@ -9,6 +9,7 @@ from basic_memory.schemas.search import SearchQuery, SearchResponse
 from basic_memory.mcp.async_client import client
 
 
+@logfire.instrument(extract_args=False)
 @mcp.tool(
     description="Search across all content in basic-memory, including documents and entities",
 )
@@ -65,12 +66,11 @@ async def search(query: SearchQuery, page: int = 1, page_size: int = 10) -> Sear
             permalink_match="docs/meeting-*"
         ))
     """
-    with logfire.span("Searching for {query}", query=query):  # pyright: ignore [reportGeneralTypeIssues]
-        logger.info(f"Searching for {query}")
-        response = await call_post(
-            client,
-            "/search/",
-            json=query.model_dump(),
-            params={"page": page, "page_size": page_size},
-        )
-        return SearchResponse.model_validate(response.json())
+    logger.info(f"Searching for {query}")
+    response = await call_post(
+        client,
+        "/search/",
+        json=query.model_dump(),
+        params={"page": page, "page_size": page_size},
+    )
+    return SearchResponse.model_validate(response.json())

src/basic_memory/mcp/tools/write_note.py

@@ -15,6 +15,7 @@ from basic_memory.mcp.tools.utils import call_put
 @mcp.tool(
     description="Create or update a markdown note. Returns a markdown formatted summary of the semantic content.",
 )
+@logfire.instrument(extract_args=False)
 async def write_note(
     title: str,
     content: str,
@@ -57,53 +58,69 @@ async def write_note(
         - Relation counts (resolved/unresolved)
         - Tags if present
     """
-    with logfire.span("Writing note", title=title, folder=folder):  # pyright: ignore [reportGeneralTypeIssues]
-        logger.info(f"Writing note folder:'{folder}' title: '{title}'")
+    logger.info("MCP tool call", tool="write_note", folder=folder, title=title, tags=tags)
 
-        # Create the entity request
-        metadata = {"tags": [f"#{tag}" for tag in tags]} if tags else None
-        entity = Entity(
-            title=title,
-            folder=folder,
-            entity_type="note",
-            content_type="text/markdown",
-            content=content,
-            entity_metadata=metadata,
-        )
+    # Create the entity request
+    metadata = {"tags": [f"#{tag}" for tag in tags]} if tags else None
+    entity = Entity(
+        title=title,
+        folder=folder,
+        entity_type="note",
+        content_type="text/markdown",
+        content=content,
+        entity_metadata=metadata,
+    )
 
-        # Create or update via knowledge API
-        logger.info(f"Creating {entity.permalink}")
-        url = f"/knowledge/entities/{entity.permalink}"
-        response = await call_put(client, url, json=entity.model_dump())
-        result = EntityResponse.model_validate(response.json())
+    # Create or update via knowledge API
+    logger.debug("Creating entity via API", permalink=entity.permalink)
+    url = f"/knowledge/entities/{entity.permalink}"
+    response = await call_put(client, url, json=entity.model_dump())
+    result = EntityResponse.model_validate(response.json())
 
-        # Format semantic summary based on status code
-        action = "Created" if response.status_code == 201 else "Updated"
-        summary = [
-            f"# {action} {result.file_path} ({result.checksum[:8] if result.checksum else 'unknown'})",
-            f"permalink: {result.permalink}",
-        ]
+    # Format semantic summary based on status code
+    action = "Created" if response.status_code == 201 else "Updated"
+    summary = [
+        f"# {action} {result.file_path} ({result.checksum[:8] if result.checksum else 'unknown'})",
+        f"permalink: {result.permalink}",
+    ]
 
-        if result.observations:
-            categories = {}
-            for obs in result.observations:
-                categories[obs.category] = categories.get(obs.category, 0) + 1
+    # Count observations by category
+    categories = {}
+    if result.observations:
+        for obs in result.observations:
+            categories[obs.category] = categories.get(obs.category, 0) + 1
 
-            summary.append("\n## Observations")
-            for category, count in sorted(categories.items()):
-                summary.append(f"- {category}: {count}")
+        summary.append("\n## Observations")
+        for category, count in sorted(categories.items()):
+            summary.append(f"- {category}: {count}")
 
-        if result.relations:
-            unresolved = sum(1 for r in result.relations if not r.to_id)
-            resolved = len(result.relations) - unresolved
+    # Count resolved/unresolved relations
+    unresolved = 0
+    resolved = 0
+    if result.relations:
+        unresolved = sum(1 for r in result.relations if not r.to_id)
+        resolved = len(result.relations) - unresolved
 
-            summary.append("\n## Relations")
-            summary.append(f"- Resolved: {resolved}")
-            if unresolved:
-                summary.append(f"- Unresolved: {unresolved}")
-                summary.append("\nUnresolved relations will be retried on next sync.")
+        summary.append("\n## Relations")
+        summary.append(f"- Resolved: {resolved}")
+        if unresolved:
+            summary.append(f"- Unresolved: {unresolved}")
+            summary.append("\nUnresolved relations will be retried on next sync.")
 
-        if tags:
-            summary.append(f"\n## Tags\n- {', '.join(tags)}")
+    if tags:
+        summary.append(f"\n## Tags\n- {', '.join(tags)}")
 
-        return "\n".join(summary)
+    # Log the response with structured data
+    logger.info(
+        "MCP tool response",
+        tool="write_note",
+        action=action,
+        permalink=result.permalink,
+        observations_count=len(result.observations),
+        relations_count=len(result.relations),
+        resolved_relations=resolved,
+        unresolved_relations=unresolved,
+        status_code=response.status_code,
+    )
+
+    return "\n".join(summary)

src/basic_memory/repository/repository.py

@@ -70,7 +70,15 @@ class Repository[T: Base]:
 
             # Query within same session
             found = await self.select_by_id(session, model.id)  # pyright: ignore [reportAttributeAccessIssue]
-            assert found is not None, "can't find model after session.add"
+            if found is None:  # pragma: no cover
+                logger.error(
+                    "Failed to retrieve model after add",
+                    model_type=self.Model.__name__,
+                    model_id=model.id,  # pyright: ignore
+                )
+                raise ValueError(
+                    f"Can't find {self.Model.__name__} with ID {model.id} after session.add"  # pyright: ignore
+                )
             return found
 
     async def add_all(self, models: List[T]) -> Sequence[T]:
@@ -152,7 +160,15 @@ class Repository[T: Base]:
             await session.flush()
 
             return_instance = await self.select_by_id(session, model.id)  # pyright: ignore [reportAttributeAccessIssue]
-            assert return_instance is not None, "can't find model after session.add"
+            if return_instance is None:  # pragma: no cover
+                logger.error(
+                    "Failed to retrieve model after create",
+                    model_type=self.Model.__name__,
+                    model_id=model.id,  # pyright: ignore
+                )
+                raise ValueError(
+                    f"Can't find {self.Model.__name__} with ID {model.id} after session.add"  # pyright: ignore
+                )
             return return_instance
 
     async def create_all(self, data_list: List[dict]) -> Sequence[T]:

src/basic_memory/repository/search_repository.py

@@ -206,7 +206,7 @@ class SearchRepository:
             OFFSET :offset
         """
 
-        logger.debug(f"Search {sql} params: {params}")
+        logger.trace(f"Search {sql} params: {params}")
         async with db.scoped_session(self.session_maker) as session:
             result = await session.execute(text(sql), params)
             rows = result.fetchall()

src/basic_memory/schemas/search.py

@@ -59,6 +59,7 @@ class SearchQuery(BaseModel):
         return (
             self.permalink is None
             and self.permalink_match is None
+            and self.title is None
             and self.text is None
             and self.after_date is None
             and self.types is None

src/basic_memory/services/entity_service.py

@@ -144,7 +144,7 @@ class EntityService(BaseService[EntityModel]):
         post = await schema_to_markdown(schema)
 
         # write file
-        final_content = frontmatter.dumps(post)
+        final_content = frontmatter.dumps(post, sort_keys=False)
         checksum = await self.file_service.write_file(file_path, final_content)
 
         # parse entity from file
@@ -171,7 +171,13 @@ class EntityService(BaseService[EntityModel]):
                 entity = await self.get_by_permalink(permalink_or_id)
             else:
                 entities = await self.get_entities_by_id([permalink_or_id])
-                assert len(entities) == 1, f"Expected 1 entity, got {len(entities)}"
+                if len(entities) != 1:  # pragma: no cover
+                    logger.error(
+                        "Entity lookup error", entity_id=permalink_or_id, found_count=len(entities)
+                    )
+                    raise ValueError(
+                        f"Expected 1 entity with ID {permalink_or_id}, got {len(entities)}"
+                    )
                 entity = entities[0]
 
             # Delete file first

src/basic_memory/services/file_service.py

@@ -3,7 +3,7 @@
 import mimetypes
 from os import stat_result
 from pathlib import Path
-from typing import Tuple, Union, Dict, Any
+from typing import Any, Dict, Tuple, Union
 
 from loguru import logger
 
@@ -13,6 +13,7 @@ from basic_memory.markdown.markdown_processor import MarkdownProcessor
 from basic_memory.models import Entity as EntityModel
 from basic_memory.schemas import Entity as EntitySchema
 from basic_memory.services.exceptions import FileOperationError
+from basic_memory.utils import FilePath
 
 
 class FileService:
@@ -60,7 +61,7 @@ class FileService:
         Returns:
             Raw content string without metadata sections
         """
-        logger.debug(f"Reading entity with permalink: {entity.permalink}")
+        logger.debug("Reading entity content", entity_id=entity.id, permalink=entity.permalink)
 
         file_path = self.get_entity_path(entity)
         markdown = await self.markdown_processor.read_file(file_path)
@@ -78,13 +79,13 @@ class FileService:
         path = self.get_entity_path(entity)
         await self.delete_file(path)
 
-    async def exists(self, path: Union[Path, str]) -> bool:
+    async def exists(self, path: FilePath) -> bool:
         """Check if file exists at the provided path.
 
         If path is relative, it is assumed to be relative to base_path.
 
         Args:
-            path: Path to check (Path object or string)
+            path: Path to check (Path or string)
 
         Returns:
             True if file exists, False otherwise
@@ -93,23 +94,25 @@ class FileService:
             FileOperationError: If check fails
         """
         try:
-            path = Path(path)
-            if path.is_absolute():
-                return path.exists()
+            # Convert string to Path if needed
+            path_obj = Path(path) if isinstance(path, str) else path
+
+            if path_obj.is_absolute():
+                return path_obj.exists()
             else:
-                return (self.base_path / path).exists()
+                return (self.base_path / path_obj).exists()
         except Exception as e:
-            logger.error(f"Failed to check file existence {path}: {e}")
+            logger.error("Failed to check file existence", path=str(path), error=str(e))
             raise FileOperationError(f"Failed to check file existence: {e}")
 
-    async def write_file(self, path: Union[Path, str], content: str) -> str:
+    async def write_file(self, path: FilePath, content: str) -> str:
         """Write content to file and return checksum.
 
         Handles both absolute and relative paths. Relative paths are resolved
         against base_path.
 
         Args:
-            path: Where to write (Path object or string)
+            path: Where to write (Path or string)
             content: Content to write
 
         Returns:
@@ -118,34 +121,43 @@ class FileService:
         Raises:
             FileOperationError: If write fails
         """
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
 
         try:
             # Ensure parent directory exists
             await file_utils.ensure_directory(full_path.parent)
 
             # Write content atomically
+            logger.info(
+                "Writing file",
+                operation="write_file",
+                path=str(full_path),
+                content_length=len(content),
+                is_markdown=full_path.suffix.lower() == ".md",
+            )
+
             await file_utils.write_file_atomic(full_path, content)
 
             # Compute and return checksum
             checksum = await file_utils.compute_checksum(content)
-            logger.debug(f"wrote file: {full_path}, checksum: {checksum}")
+            logger.debug("File write completed", path=str(full_path), checksum=checksum)
             return checksum
 
         except Exception as e:
-            logger.error(f"Failed to write file {full_path}: {e}")
+            logger.exception("File write error", path=str(full_path), error=str(e))
             raise FileOperationError(f"Failed to write file: {e}")
 
     # TODO remove read_file
-    async def read_file(self, path: Union[Path, str]) -> Tuple[str, str]:
+    async def read_file(self, path: FilePath) -> Tuple[str, str]:
         """Read file and compute checksum.
 
         Handles both absolute and relative paths. Relative paths are resolved
         against base_path.
 
         Args:
-            path: Path to read (Path object or string)
+            path: Path to read (Path or string)
 
         Returns:
             Tuple of (content, checksum)
@@ -153,45 +165,74 @@ class FileService:
         Raises:
             FileOperationError: If read fails
         """
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
 
         try:
+            logger.debug("Reading file", operation="read_file", path=str(full_path))
+
             content = full_path.read_text()
             checksum = await file_utils.compute_checksum(content)
-            logger.debug(f"read file: {full_path}, checksum: {checksum}")
+
+            logger.debug(
+                "File read completed",
+                path=str(full_path),
+                checksum=checksum,
+                content_length=len(content),
+            )
             return content, checksum
 
         except Exception as e:
-            logger.error(f"Failed to read file {full_path}: {e}")
+            logger.exception("File read error", path=str(full_path), error=str(e))
             raise FileOperationError(f"Failed to read file: {e}")
 
-    async def delete_file(self, path: Union[Path, str]) -> None:
+    async def delete_file(self, path: FilePath) -> None:
         """Delete file if it exists.
 
         Handles both absolute and relative paths. Relative paths are resolved
         against base_path.
 
         Args:
-            path: Path to delete (Path object or string)
+            path: Path to delete (Path or string)
         """
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         full_path.unlink(missing_ok=True)
 
-    async def update_frontmatter(self, path: Union[Path, str], updates: Dict[str, Any]) -> str:
+    async def update_frontmatter(self, path: FilePath, updates: Dict[str, Any]) -> str:
         """
         Update frontmatter fields in a file while preserving all content.
-        """
 
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        Args:
+            path: Path to the file (Path or string)
+            updates: Dictionary of frontmatter fields to update
+
+        Returns:
+            Checksum of updated file
+        """
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         return await file_utils.update_frontmatter(full_path, updates)
 
-    async def compute_checksum(self, path: Union[str, Path]) -> str:
-        """Compute checksum for a file."""
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+    async def compute_checksum(self, path: FilePath) -> str:
+        """Compute checksum for a file.
+
+        Args:
+            path: Path to the file (Path or string)
+
+        Returns:
+            Checksum of the file content
+
+        Raises:
+            FileError: If checksum computation fails
+        """
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
+
         try:
             if self.is_markdown(path):
                 # read str
@@ -202,28 +243,36 @@ class FileService:
             return await file_utils.compute_checksum(content)
 
         except Exception as e:  # pragma: no cover
-            logger.error(f"Failed to compute checksum for {path}: {e}")
+            logger.error("Failed to compute checksum", path=str(full_path), error=str(e))
             raise FileError(f"Failed to compute checksum for {path}: {e}")
 
-    def file_stats(self, path: Union[Path, str]) -> stat_result:
+    def file_stats(self, path: FilePath) -> stat_result:
+        """Return file stats for a given path.
+
+        Args:
+            path: Path to the file (Path or string)
+
+        Returns:
+            File statistics
         """
-        Return file stats for a given path.
-        :param path:
-        :return:
-        """
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         # get file timestamps
         return full_path.stat()
 
-    def content_type(self, path: Union[Path, str]) -> str:
+    def content_type(self, path: FilePath) -> str:
+        """Return content_type for a given path.
+
+        Args:
+            path: Path to the file (Path or string)
+
+        Returns:
+            MIME type of the file
         """
-        Return content_type for a given path.
-        :param path:
-        :return:
-        """
-        path = Path(path)
-        full_path = path if path.is_absolute() else self.base_path / path
+        # Convert string to Path if needed
+        path_obj = Path(path) if isinstance(path, str) else path
+        full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
         # get file timestamps
         mime_type, _ = mimetypes.guess_type(full_path.name)
 
@@ -234,10 +283,13 @@ class FileService:
         content_type = mime_type or "text/plain"
         return content_type
 
-    def is_markdown(self, path: Union[Path, str]) -> bool:
-        """
-        Return content_type for a given path.
-        :param path:
-        :return:
+    def is_markdown(self, path: FilePath) -> bool:
+        """Check if a file is a markdown file.
+
+        Args:
+            path: Path to the file (Path or string)
+
+        Returns:
+            True if the file is a markdown file, False otherwise
         """
         return self.content_type(path) == "text/markdown"

src/basic_memory/services/search_service.py

@@ -179,9 +179,16 @@ class SearchService:
         Each type gets its own row in the search index with appropriate metadata.
         """
 
-        assert entity.permalink is not None, (
-            "entity.permalink should not be None for markdown entities"
-        )
+        if entity.permalink is None:  # pragma: no cover
+            logger.error(
+                "Missing permalink for markdown entity",
+                entity_id=entity.id,
+                title=entity.title,
+                file_path=entity.file_path,
+            )
+            raise ValueError(
+                f"Entity permalink should not be None for markdown entity: {entity.id} ({entity.title})"
+            )
 
         content_stems = []
         content_snippet = ""
@@ -198,9 +205,16 @@ class SearchService:
 
         entity_content_stems = "\n".join(p for p in content_stems if p and p.strip())
 
-        assert entity.permalink is not None, (
-            "entity.permalink should not be None for markdown entities"
-        )
+        if entity.permalink is None:  # pragma: no cover
+            logger.error(
+                "Missing permalink for markdown entity",
+                entity_id=entity.id,
+                title=entity.title,
+                file_path=entity.file_path,
+            )
+            raise ValueError(
+                f"Entity permalink should not be None for markdown entity: {entity.id} ({entity.title})"
+            )
 
         # Index entity
         await self.repository.index_item(

src/basic_memory/sync/sync_service.py

@@ -1,12 +1,15 @@
 """Service for syncing files between filesystem and database."""
 
+# Suppress logfire warnings
 import os
+
+os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
+
 from dataclasses import dataclass
 from dataclasses import field
 from datetime import datetime
 from pathlib import Path
-from typing import Set, Dict
-from typing import Tuple
+from typing import Dict, Optional, Set, Tuple
 
 import logfire
 from loguru import logger
@@ -78,22 +81,126 @@ class SyncService:
         self.search_service = search_service
         self.file_service = file_service
 
-    async def sync(self, directory: Path) -> SyncReport:
+    @logfire.instrument(extract_args=False)
+    async def sync(self, directory: Path, show_progress: bool = True) -> SyncReport:
         """Sync all files with database."""
+        import time
+        from rich.progress import Progress, TextColumn, BarColumn, TaskProgressColumn
 
-        with logfire.span(f"sync {directory}", directory=directory):  # pyright: ignore [reportGeneralTypeIssues]
-            # initial paths from db to sync
-            # path -> checksum
-            report = await self.scan(directory)
+        start_time = time.time()
+        console = None
+        progress = None  # Will be initialized if show_progress is True
 
-            # order of sync matters to resolve relations effectively
+        logger.info("Sync operation started", directory=str(directory))
 
+        # initial paths from db to sync
+        # path -> checksum
+        if show_progress:
+            from rich.console import Console
+
+            console = Console()
+            console.print(f"Scanning directory: {directory}")
+
+        report = await self.scan(directory)
+
+        # Initialize progress tracking if requested
+        if show_progress and report.total > 0:
+            progress = Progress(
+                TextColumn("[bold blue]{task.description}"),
+                BarColumn(),
+                TaskProgressColumn(),
+                console=console,
+                expand=True,
+            )
+
+        # order of sync matters to resolve relations effectively
+        logger.info(
+            "Sync changes detected",
+            new_files=len(report.new),
+            modified_files=len(report.modified),
+            deleted_files=len(report.deleted),
+            moved_files=len(report.moves),
+        )
+
+        if show_progress and report.total > 0:
+            with progress:  # pyright: ignore
+                # Track each category separately
+                move_task = None
+                if report.moves:  # pragma: no cover
+                    move_task = progress.add_task("[blue]Moving files...", total=len(report.moves))  # pyright: ignore
+
+                delete_task = None
+                if report.deleted:  # pragma: no cover
+                    delete_task = progress.add_task(  # pyright: ignore
+                        "[red]Deleting files...", total=len(report.deleted)
+                    )
+
+                new_task = None
+                if report.new:
+                    new_task = progress.add_task(  # pyright: ignore
+                        "[green]Adding new files...", total=len(report.new)
+                    )
+
+                modify_task = None
+                if report.modified:  # pragma: no cover
+                    modify_task = progress.add_task(  # pyright: ignore
+                        "[yellow]Updating modified files...", total=len(report.modified)
+                    )
+
+                # sync moves first
+                for i, (old_path, new_path) in enumerate(report.moves.items()):
+                    # in the case where a file has been deleted and replaced by another file
+                    # it will show up in the move and modified lists, so handle it in modified
+                    if new_path in report.modified:  # pragma: no cover
+                        report.modified.remove(new_path)
+                        logger.debug(
+                            "File marked as moved and modified",
+                            old_path=old_path,
+                            new_path=new_path,
+                            action="processing as modified",
+                        )
+                    else:  # pragma: no cover
+                        await self.handle_move(old_path, new_path)
+
+                    if move_task is not None:  # pragma: no cover
+                        progress.update(move_task, advance=1)  # pyright: ignore
+
+                # deleted next
+                for i, path in enumerate(report.deleted):  # pragma: no cover
+                    await self.handle_delete(path)
+                    if delete_task is not None:  # pragma: no cover
+                        progress.update(delete_task, advance=1)  # pyright: ignore
+
+                # then new and modified
+                for i, path in enumerate(report.new):
+                    await self.sync_file(path, new=True)
+                    if new_task is not None:
+                        progress.update(new_task, advance=1)  # pyright: ignore
+
+                for i, path in enumerate(report.modified):  # pragma: no cover
+                    await self.sync_file(path, new=False)
+                    if modify_task is not None:  # pragma: no cover
+                        progress.update(modify_task, advance=1)  # pyright: ignore
+
+                # Final step - resolving relations
+                if report.total > 0:
+                    relation_task = progress.add_task("[cyan]Resolving relations...", total=1)  # pyright: ignore
+                    await self.resolve_relations()
+                    progress.update(relation_task, advance=1)  # pyright: ignore
+        else:
+            # No progress display - proceed with normal sync
             # sync moves first
             for old_path, new_path in report.moves.items():
                 # in the case where a file has been deleted and replaced by another file
                 # it will show up in the move and modified lists, so handle it in modified
                 if new_path in report.modified:
                     report.modified.remove(new_path)
+                    logger.debug(
+                        "File marked as moved and modified",
+                        old_path=old_path,
+                        new_path=new_path,
+                        action="processing as modified",
+                    )
                 else:
                     await self.handle_move(old_path, new_path)
 
@@ -109,7 +216,16 @@ class SyncService:
                 await self.sync_file(path, new=False)
 
             await self.resolve_relations()
-            return report
+
+        duration_ms = int((time.time() - start_time) * 1000)
+        logger.info(
+            "Sync operation completed",
+            directory=str(directory),
+            total_changes=report.total,
+            duration_ms=duration_ms,
+        )
+
+        return report
 
     async def scan(self, directory):
         """Scan directory for changes compared to database state."""
@@ -167,25 +283,55 @@ class SyncService:
         db_records = await self.entity_repository.find_all()
         return {r.file_path: r.checksum or "" for r in db_records}
 
-    async def sync_file(self, path: str, new: bool = True) -> Tuple[Entity, str]:
-        """Sync a single file."""
+    async def sync_file(
+        self, path: str, new: bool = True
+    ) -> Tuple[Optional[Entity], Optional[str]]:
+        """Sync a single file.
 
+        Args:
+            path: Path to file to sync
+            new: Whether this is a new file
+
+        Returns:
+            Tuple of (entity, checksum) or (None, None) if sync fails
+        """
         try:
+            logger.debug(
+                "Syncing file",
+                path=path,
+                is_new=new,
+                is_markdown=self.file_service.is_markdown(path),
+            )
+
             if self.file_service.is_markdown(path):
                 entity, checksum = await self.sync_markdown_file(path, new)
             else:
                 entity, checksum = await self.sync_regular_file(path, new)
-            await self.search_service.index_entity(entity)
+
+            if entity is not None:
+                await self.search_service.index_entity(entity)
+
+                logger.debug(
+                    "File sync completed", path=path, entity_id=entity.id, checksum=checksum
+                )
             return entity, checksum
 
         except Exception as e:  # pragma: no cover
-            logger.exception(f"Failed to sync {path}: {e}")
-            return None, None  # pyright: ignore
+            logger.exception("Failed to sync file", path=path, error=str(e))
+            return None, None
 
-    async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Entity, str]:
-        """Sync a markdown file with full proces    sing."""
+    async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Optional[Entity], str]:
+        """Sync a markdown file with full processing.
 
+        Args:
+            path: Path to markdown file
+            new: Whether this is a new file
+
+        Returns:
+            Tuple of (entity, checksum)
+        """
         # Parse markdown first to get any existing permalink
+        logger.debug("Parsing markdown file", path=path)
         entity_markdown = await self.entity_parser.parse_file(path)
 
         # Resolve permalink - this handles all the cases including conflicts
@@ -193,7 +339,13 @@ class SyncService:
 
         # If permalink changed, update the file
         if permalink != entity_markdown.frontmatter.permalink:
-            logger.info(f"Updating permalink in {path}: {permalink}")
+            logger.info(
+                "Updating permalink",
+                path=path,
+                old_permalink=entity_markdown.frontmatter.permalink,
+                new_permalink=permalink,
+            )
+
             entity_markdown.frontmatter.metadata["permalink"] = permalink
             checksum = await self.file_service.update_frontmatter(path, {"permalink": permalink})
         else:
@@ -202,12 +354,14 @@ class SyncService:
         # if the file is new, create an entity
         if new:
             # Create entity with final permalink
-            logger.debug(f"Creating new entity from markdown: {path}")
+            logger.debug("Creating new entity from markdown", path=path, permalink=permalink)
+
             await self.entity_service.create_entity_from_markdown(Path(path), entity_markdown)
 
         # otherwise we need to update the entity and observations
         else:
-            logger.debug(f"Updating entity from markdown: {path}")
+            logger.debug("Updating entity from markdown", path=path, permalink=permalink)
+
             await self.entity_service.update_entity_and_observations(Path(path), entity_markdown)
 
         # Update relations and search index
@@ -215,11 +369,27 @@ class SyncService:
 
         # set checksum
         await self.entity_repository.update(entity.id, {"checksum": checksum})
+
+        logger.debug(
+            "Markdown sync completed",
+            path=path,
+            entity_id=entity.id,
+            observation_count=len(entity.observations),
+            relation_count=len(entity.relations),
+        )
+
         return entity, checksum
 
-    async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Entity, str]:
-        """Sync a non-markdown file with basic tracking."""
+    async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Optional[Entity], str]:
+        """Sync a non-markdown file with basic tracking.
 
+        Args:
+            path: Path to file
+            new: Whether this is a new file
+
+        Returns:
+            Tuple of (entity, checksum)
+        """
         checksum = await self.file_service.compute_checksum(path)
         if new:
             # Generate permalink from path
@@ -248,11 +418,18 @@ class SyncService:
             return entity, checksum
         else:
             entity = await self.entity_repository.get_by_file_path(path)
-            assert entity is not None, "entity should not be None for existing file"
+            if entity is None:  # pragma: no cover
+                logger.error("Entity not found for existing file", path=path)
+                raise ValueError(f"Entity not found for existing file: {path}")
+
             updated = await self.entity_repository.update(
                 entity.id, {"file_path": path, "checksum": checksum}
             )
-            assert updated is not None, "entity should be updated"
+
+            if updated is None:  # pragma: no cover
+                logger.error("Failed to update entity", entity_id=entity.id, path=path)
+                raise ValueError(f"Failed to update entity with ID {entity.id}")
+
             return updated, checksum
 
     async def handle_delete(self, file_path: str):
@@ -261,7 +438,12 @@ class SyncService:
         # First get entity to get permalink before deletion
         entity = await self.entity_repository.get_by_file_path(file_path)
         if entity:
-            logger.debug(f"Deleting entity and cleaning up search index: {file_path}")
+            logger.info(
+                "Deleting entity",
+                file_path=file_path,
+                entity_id=entity.id,
+                permalink=entity.permalink,
+            )
 
             # Delete from db (this cascades to observations/relations)
             await self.entity_service.delete_entity_by_file_path(file_path)
@@ -272,7 +454,14 @@ class SyncService:
                 + [o.permalink for o in entity.observations]
                 + [r.permalink for r in entity.relations]
             )
-            logger.debug(f"Deleting from search index: {permalinks}")
+
+            logger.debug(
+                "Cleaning up search index",
+                entity_id=entity.id,
+                file_path=file_path,
+                index_entries=len(permalinks),
+            )
+
             for permalink in permalinks:
                 if permalink:
                     await self.search_service.delete_by_permalink(permalink)
@@ -280,12 +469,30 @@ class SyncService:
                     await self.search_service.delete_by_entity_id(entity.id)
 
     async def handle_move(self, old_path, new_path):
-        logger.debug(f"Moving entity: {old_path} -> {new_path}")
+        logger.info("Moving entity", old_path=old_path, new_path=new_path)
+
         entity = await self.entity_repository.get_by_file_path(old_path)
         if entity:
             # Update file_path but keep the same permalink for link stability
             updated = await self.entity_repository.update(entity.id, {"file_path": new_path})
-            assert updated is not None, "entity should be updated"
+
+            if updated is None:  # pragma: no cover
+                logger.error(
+                    "Failed to update entity path",
+                    entity_id=entity.id,
+                    old_path=old_path,
+                    new_path=new_path,
+                )
+                raise ValueError(f"Failed to update entity path for ID {entity.id}")
+
+            logger.debug(
+                "Entity path updated",
+                entity_id=entity.id,
+                permalink=entity.permalink,
+                old_path=old_path,
+                new_path=new_path,
+            )
+
             # update search index
             await self.search_service.index_entity(updated)
 
@@ -293,14 +500,28 @@ class SyncService:
         """Try to resolve any unresolved relations"""
 
         unresolved_relations = await self.relation_repository.find_unresolved_relations()
-        logger.debug(f"Attempting to resolve {len(unresolved_relations)} forward references")
+
+        logger.info("Resolving forward references", count=len(unresolved_relations))
+
         for relation in unresolved_relations:
+            logger.debug(
+                "Attempting to resolve relation",
+                relation_id=relation.id,
+                from_id=relation.from_id,
+                to_name=relation.to_name,
+            )
+
             resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name)
 
             # ignore reference to self
             if resolved_entity and resolved_entity.id != relation.from_id:
                 logger.debug(
-                    f"Resolved forward reference: {relation.to_name} -> {resolved_entity.title}"
+                    "Resolved forward reference",
+                    relation_id=relation.id,
+                    from_id=relation.from_id,
+                    to_name=relation.to_name,
+                    resolved_id=resolved_entity.id,
+                    resolved_title=resolved_entity.title,
                 )
                 try:
                     await self.relation_repository.update(
@@ -311,7 +532,12 @@ class SyncService:
                         },
                     )
                 except IntegrityError:  # pragma: no cover
-                    logger.debug(f"Ignoring duplicate relation {relation}")
+                    logger.debug(
+                        "Ignoring duplicate relation",
+                        relation_id=relation.id,
+                        from_id=relation.from_id,
+                        to_name=relation.to_name,
+                    )
 
                 # update search index
                 await self.search_service.index_entity(resolved_entity)
@@ -326,8 +552,11 @@ class SyncService:
         Returns:
             ScanResult containing found files and any errors
         """
+        import time
 
-        logger.debug(f"Scanning directory: {directory}")
+        start_time = time.time()
+
+        logger.debug("Scanning directory", directory=str(directory))
         result = ScanResult()
 
         for root, dirnames, filenames in os.walk(str(directory)):
@@ -344,6 +573,15 @@ class SyncService:
                 checksum = await self.file_service.compute_checksum(rel_path)
                 result.files[rel_path] = checksum
                 result.checksums[checksum] = rel_path
-                logger.debug(f"Found file: {rel_path} with checksum: {checksum}")
+
+                logger.debug("Found file", path=rel_path, checksum=checksum)
+
+        duration_ms = int((time.time() - start_time) * 1000)
+        logger.debug(
+            "Directory scan completed",
+            directory=str(directory),
+            files_found=len(result.files),
+            duration_ms=duration_ms,
+        )
 
         return result

src/basic_memory/sync/watch_service.py

@@ -1,6 +1,5 @@
 """Watch service for Basic Memory."""
 
-import dataclasses
 import os
 from datetime import datetime
 from pathlib import Path
@@ -29,8 +28,8 @@ class WatchEvent(BaseModel):
 class WatchServiceState(BaseModel):
     # Service status
     running: bool = False
-    start_time: datetime = dataclasses.field(default_factory=datetime.now)
-    pid: int = dataclasses.field(default_factory=os.getpid)
+    start_time: datetime = datetime.now()  # Use directly with Pydantic model
+    pid: int = os.getpid()  # Use directly with Pydantic model
 
     # Stats
     error_count: int = 0
@@ -41,7 +40,7 @@ class WatchServiceState(BaseModel):
     synced_files: int = 0
 
     # Recent activity
-    recent_events: List[WatchEvent] = dataclasses.field(default_factory=list)
+    recent_events: List[WatchEvent] = []  # Use directly with Pydantic model
 
     def add_event(
         self,
@@ -81,10 +80,17 @@ class WatchService:
 
     async def run(self):  # pragma: no cover
         """Watch for file changes and sync them"""
-        logger.info("Watching for sync changes")
+        logger.info(
+            "Watch service started",
+            directory=str(self.config.home),
+            debounce_ms=self.config.sync_delay,
+            pid=os.getpid(),
+        )
+
         self.state.running = True
         self.state.start_time = datetime.now()
         await self.write_status()
+
         try:
             async for changes in awatch(
                 self.config.home,
@@ -95,14 +101,23 @@ class WatchService:
                 await self.handle_changes(self.config.home, changes)
 
         except Exception as e:
+            logger.exception("Watch service error", error=str(e), directory=str(self.config.home))
+
             self.state.record_error(str(e))
             await self.write_status()
             raise
+
         finally:
+            logger.info(
+                "Watch service stopped",
+                directory=str(self.config.home),
+                runtime_seconds=int((datetime.now() - self.state.start_time).total_seconds()),
+            )
+
             self.state.running = False
             await self.write_status()
 
-    def filter_changes(self, change: Change, path: str) -> bool:
+    def filter_changes(self, change: Change, path: str) -> bool:  # pragma: no cover
         """Filter to only watch non-hidden files and directories.
 
         Returns:
@@ -112,6 +127,7 @@ class WatchService:
         try:
             relative_path = Path(path).relative_to(self.config.home)
         except ValueError:
+            # This is a defensive check for paths outside our home directory
             return False
 
         # Skip hidden directories and files
@@ -128,12 +144,17 @@ class WatchService:
 
     async def handle_changes(self, directory: Path, changes: Set[FileChange]):
         """Process a batch of file changes"""
-        logger.debug(f"handling {len(changes)} changes in directory: {directory} ...")
+        import time
+        from typing import List, Set
+
+        start_time = time.time()
+
+        logger.info("Processing file changes", change_count=len(changes), directory=str(directory))
 
         # Group changes by type
-        adds = []
-        deletes = []
-        modifies = []
+        adds: List[str] = []
+        deletes: List[str] = []
+        modifies: List[str] = []
 
         for change, path in changes:
             # convert to relative path
@@ -145,25 +166,44 @@ class WatchService:
             elif change == Change.modified:
                 modifies.append(relative_path)
 
+        logger.debug(
+            "Grouped file changes", added=len(adds), deleted=len(deletes), modified=len(modifies)
+        )
+
         # Track processed files to avoid duplicates
-        processed = set()
+        processed: Set[str] = set()
 
         # First handle potential moves
         for added_path in adds:
             if added_path in processed:
                 continue  # pragma: no cover
 
+            # Skip directories for added paths
+            # We don't need to process directories, only the files inside them
+            # This prevents errors when trying to compute checksums or read directories as files
+            added_full_path = directory / added_path
+            if added_full_path.is_dir():
+                logger.debug("Skipping directory for move detection", path=added_path)
+                processed.add(added_path)
+                continue
+
             for deleted_path in deletes:
                 if deleted_path in processed:
                     continue  # pragma: no cover
 
+                # Skip directories for deleted paths (based on entity type in db)
+                deleted_entity = await self.sync_service.entity_repository.get_by_file_path(
+                    deleted_path
+                )
+                if deleted_entity is None:
+                    # If this was a directory, it wouldn't have an entity
+                    logger.debug("Skipping unknown path for move detection", path=deleted_path)
+                    continue
+
                 if added_path != deleted_path:
                     # Compare checksums to detect moves
                     try:
                         added_checksum = await self.file_service.compute_checksum(added_path)
-                        deleted_entity = await self.sync_service.entity_repository.get_by_file_path(
-                            deleted_path
-                        )
 
                         if deleted_entity and deleted_entity.checksum == added_checksum:
                             await self.sync_service.handle_move(deleted_path, added_path)
@@ -172,48 +212,131 @@ class WatchService:
                                 action="moved",
                                 status="success",
                             )
-                            self.console.print(
-                                f"[blue]→[/blue] Moved: {deleted_path} → {added_path}"
-                            )
+                            self.console.print(f"[blue]→[/blue] {deleted_path} → {added_path}")
                             processed.add(added_path)
                             processed.add(deleted_path)
                             break
                     except Exception as e:  # pragma: no cover
-                        logger.warning(f"Error checking for move: {e}")
+                        logger.warning(
+                            "Error checking for move",
+                            old_path=deleted_path,
+                            new_path=added_path,
+                            error=str(e),
+                        )
 
-        # Handle remaining changes
+        # Handle remaining changes - group them by type for concise output
+        moved_count = len([p for p in processed if p in deletes or p in adds])
+        delete_count = 0
+        add_count = 0
+        modify_count = 0
+
+        # Process deletes
         for path in deletes:
             if path not in processed:
+                logger.debug("Processing deleted file", path=path)
                 await self.sync_service.handle_delete(path)
                 self.state.add_event(path=path, action="deleted", status="success")
-                self.console.print(f"[red]✕[/red] Deleted: {path}")
+                self.console.print(f"[red]✕[/red] {path}")
                 processed.add(path)
+                delete_count += 1
 
+        # Process adds
         for path in adds:
             if path not in processed:
-                _, checksum = await self.sync_service.sync_file(path, new=True)
+                # Skip directories - only process files
+                full_path = directory / path
+                if full_path.is_dir():  # pragma: no cover
+                    logger.debug("Skipping directory", path=path)
+                    processed.add(path)
+                    continue
+
+                logger.debug("Processing new file", path=path)
+                entity, checksum = await self.sync_service.sync_file(path, new=True)
                 if checksum:
                     self.state.add_event(
                         path=path, action="new", status="success", checksum=checksum
                     )
-                    self.console.print(f"[green]✓[/green] Added: {path}")
+                    self.console.print(f"[green]✓[/green] {path}")
+                    logger.debug(
+                        "Added file processed",
+                        path=path,
+                        entity_id=entity.id if entity else None,
+                        checksum=checksum,
+                    )
                     processed.add(path)
-                else:
-                    self.console.print(f"[orange]?[/orange] Error syncing: {path}")
+                    add_count += 1
+                else:  # pragma: no cover
+                    logger.warning("Error syncing new file", path=path)  # pragma: no cover
+                    self.console.print(
+                        f"[orange]?[/orange] Error syncing: {path}"
+                    )  # pragma: no cover
+
+        # Process modifies - detect repeats
+        last_modified_path = None
+        repeat_count = 0
 
         for path in modifies:
             if path not in processed:
-                _, checksum = await self.sync_service.sync_file(path, new=False)
+                # Skip directories - only process files
+                full_path = directory / path
+                if full_path.is_dir():
+                    logger.debug("Skipping directory", path=path)
+                    processed.add(path)
+                    continue
+
+                logger.debug("Processing modified file", path=path)
+                entity, checksum = await self.sync_service.sync_file(path, new=False)
                 self.state.add_event(
                     path=path, action="modified", status="success", checksum=checksum
                 )
-                self.console.print(f"[yellow]✎[/yellow] Modified: {path}")
+
+                # Check if this is a repeat of the last modified file
+                if path == last_modified_path:  # pragma: no cover
+                    repeat_count += 1  # pragma: no cover
+                    # Only show a message for the first repeat
+                    if repeat_count == 1:  # pragma: no cover
+                        self.console.print(
+                            f"[yellow]...[/yellow] Repeated changes to {path}"
+                        )  # pragma: no cover
+                else:
+                    # New file being modified
+                    self.console.print(f"[yellow]✎[/yellow] {path}")
+                    last_modified_path = path
+                    repeat_count = 0
+                    modify_count += 1
+
+                logger.debug(
+                    "Modified file processed",
+                    path=path,
+                    entity_id=entity.id if entity else None,
+                    checksum=checksum,
+                )
                 processed.add(path)
 
-        # Add a divider if we processed any files
+        # Add a concise summary instead of a divider
         if processed:
-            self.console.print("─" * 80, style="dim")
+            changes = []  # pyright: ignore
+            if add_count > 0:
+                changes.append(f"[green]{add_count} added[/green]")  # pyright: ignore
+            if modify_count > 0:
+                changes.append(f"[yellow]{modify_count} modified[/yellow]")  # pyright: ignore
+            if moved_count > 0:
+                changes.append(f"[blue]{moved_count} moved[/blue]")  # pyright: ignore
+            if delete_count > 0:
+                changes.append(f"[red]{delete_count} deleted[/red]")  # pyright: ignore
 
+            if changes:
+                self.console.print(f"{', '.join(changes)}", style="dim")  # pyright: ignore
+
+        duration_ms = int((time.time() - start_time) * 1000)
         self.state.last_scan = datetime.now()
         self.state.synced_files += len(processed)
+
+        logger.info(
+            "File change processing completed",
+            processed_files=len(processed),
+            total_synced_files=self.state.synced_files,
+            duration_ms=duration_ms,
+        )
+
         await self.write_status()

src/basic_memory/utils.py

@@ -1,30 +1,43 @@
 """Utility functions for basic-memory."""
 
-import logging
+# Set environment variable before importing logfire to suppress warnings
 import os
+
+os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
+
+import logging
 import re
 import sys
 from pathlib import Path
-from typing import Optional, Union
+from typing import Optional, Protocol, Union, runtime_checkable
 
 from loguru import logger
 from unidecode import unidecode
 
 import basic_memory
-
 import logfire
 
+
+@runtime_checkable
+class PathLike(Protocol):
+    """Protocol for objects that can be used as paths."""
+
+    def __str__(self) -> str: ...
+
+
+# In type annotations, use Union[Path, str] instead of FilePath for now
+# This preserves compatibility with existing code while we migrate
+FilePath = Union[Path, str]
+
 # Disable the "Queue is full" warning
 logging.getLogger("opentelemetry.sdk.metrics._internal.instrument").setLevel(logging.ERROR)
-# Disable logfire prompts in CI/automated environments
-os.environ["LOGFIRE_IGNORE_NO_CONFIG"] = "1"
 
 
-def generate_permalink(file_path: Union[Path, str]) -> str:
+def generate_permalink(file_path: Union[Path, str, PathLike]) -> str:
     """Generate a stable permalink from a file path.
 
     Args:
-        file_path: Original file path
+        file_path: Original file path (str, Path, or PathLike)
 
     Returns:
         Normalized permalink that matches validation rules. Converts spaces and underscores
@@ -78,44 +91,45 @@ def setup_logging(
 ) -> None:  # pragma: no cover
     """
     Configure logging for the application.
-    :param home_dir: the root directory for the application
-    :param log_file: the name of the log file to write to
-    :param app: the fastapi application instance
-    :param console: whether to log to the console
-    """
 
+    Args:
+        env: The environment name (dev, test, prod)
+        home_dir: The root directory for the application
+        log_file: The name of the log file to write to
+        log_level: The logging level to use
+        console: Whether to log to the console
+    """
     # Remove default handler and any existing handlers
     logger.remove()
 
-    # Add file handler if we are not running tests
+    # Add file handler if we are not running tests and a log file is specified
     if log_file and env != "test":
         try:
-            # Skip logfire configuration if LOGFIRE_API_KEY is not set
-            # This avoids interactive prompts when running automated tasks
-            if "LOGFIRE_API_KEY" in os.environ:
-                # enable pydantic logfire
+            # Only configure logfire if API key is set - avoids interactive prompts
+            if "LOGFIRE_TOKEN" in os.environ:
+                # Configure logfire with code source info
                 logfire.configure(
                     code_source=logfire.CodeSource(
                         repository="https://github.com/basicmachines-co/basic-memory",
-                        revision=basic_memory.__version__,
+                        revision=f"v{basic_memory.__version__}" if env != "dev" else "HEAD",
                     ),
                     environment=env,
                     console=False,
                 )
                 logger.configure(handlers=[logfire.loguru_handler()])
 
-                # instrument code spans
+                # Instrument code spans for better observability
                 logfire.instrument_sqlite3()
                 logfire.instrument_httpx()
         except Exception as e:
             logger.warning(f"Failed to configure logfire: {e}")
 
-        # setup logger
+        # Setup file logger
         log_path = home_dir / log_file
         logger.add(
             str(log_path),
             level=log_level,
-            rotation="100 MB",
+            rotation="10 MB",
             retention="10 days",
             backtrace=True,
             diagnose=True,
@@ -123,26 +137,26 @@ def setup_logging(
             colorize=False,
         )
 
+    # Add console logger if requested or in test mode
     if env == "test" or console:
-        # Add stderr handler
         logger.add(sys.stderr, level=log_level, backtrace=True, diagnose=True, colorize=True)
 
     logger.info(f"ENV: '{env}' Log level: '{log_level}' Logging to {log_file}")
 
-    # Get the logger for 'httpx'
-    httpx_logger = logging.getLogger("httpx")
-    # Set the logging level to WARNING to ignore INFO and DEBUG logs
-    httpx_logger.setLevel(logging.WARNING)
+    # Reduce noise from third-party libraries
+    noisy_loggers = {
+        # HTTP client logs
+        "httpx": logging.WARNING,
+        # File watching logs
+        "watchfiles.main": logging.WARNING,
+        # Instrumentation noise
+        "instrumentor": logging.ERROR,
+        "opentelemetry.instrumentation.instrumentor": logging.ERROR,
+        "opentelemetry.instrumentation": logging.ERROR,
+        "logfire.instrumentor": logging.ERROR,
+        "opentelemetry.sdk.metrics._internal.instrument": logging.ERROR,
+    }
 
-    # turn watchfiles to WARNING
-    logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
-
-    # Disable all instrumentor-related warnings
-    for logger_name in [
-        "instrumentor",
-        "opentelemetry.instrumentation.instrumentor",
-        "opentelemetry.instrumentation",
-        "logfire.instrumentor",
-        "opentelemetry.sdk.metrics._internal.instrument",
-    ]:
-        logging.getLogger(logger_name).setLevel(logging.ERROR)
+    # Set log levels for noisy loggers
+    for logger_name, level in noisy_loggers.items():
+        logging.getLogger(logger_name).setLevel(level)