feat: enhanced search

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

README.md

@@ -396,6 +396,8 @@ list_directory(dir_name, depth) - Browse directory contents with filtering
 **Search & Discovery:**
 ```
 search(query, page, page_size) - Search across your knowledge base
+search_notes(query, page, page_size, search_type, types, entity_types, after_date, metadata_filters, tags, status, project) - Search with filters
+search_by_metadata(filters, limit, offset, project) - Structured frontmatter search
 ```
 
 **Project Management:**

docs/ARCHITECTURE.md

@@ -214,15 +214,28 @@ Example tool using typed client:
 
 ```python
 @mcp.tool()
-async def search_notes(query: str, project: str | None = None) -> SearchResponse:
+async def search_notes(
+    query: str,
+    project: str | None = None,
+    metadata_filters: dict | None = None,
+    tags: list[str] | None = None,
+    status: str | None = None,
+) -> SearchResponse:
     async with get_client() as client:
         active_project = await get_active_project(client, project)
 
         # Import client inside function to avoid circular imports
         from basic_memory.mcp.clients import SearchClient
+        from basic_memory.schemas.search import SearchQuery
 
+        search_query = SearchQuery(
+            text=query,
+            metadata_filters=metadata_filters,
+            tags=tags,
+            status=status,
+        )
         search_client = SearchClient(client, active_project.external_id)
-        return await search_client.search(query)
+        return await search_client.search(search_query.model_dump())
 ```
 
 ## Sync Coordination

docs/ai-assistant-guide-extended.md

@@ -1038,6 +1038,35 @@ recent_decisions = await search_notes(
 )
 ```
 
+**Structured frontmatter filters**:
+
+```python
+# Filter by tags and status
+results = await search_notes(
+    query="authentication",
+    tags=["security"],
+    status="in-progress",
+    project="main"
+)
+
+# Complex metadata filters (supports $in, $gt, $gte, $lt, $lte, $between)
+results = await search_notes(
+    query="api design",
+    metadata_filters={
+        "type": "spec",
+        "priority": {"$in": ["high", "critical"]},
+        "tags": ["architecture"]
+    },
+    project="main"
+)
+
+# Metadata-only search
+results = await search_by_metadata(
+    filters={"type": "spec", "status": "in-progress"},
+    project="main"
+)
+```
+
 ### Search Types
 
 **Text search (default)**:
@@ -2861,7 +2890,7 @@ contents = await list_directory(
 
 ### Search & Discovery
 
-**search_notes(query, page, page_size, search_type, types, entity_types, after_date, project)**
+**search_notes(query, page, page_size, search_type, types, entity_types, after_date, metadata_filters, tags, status, project)**
 - Search across knowledge base
 - Parameters:
   - `query` (required): Search query
@@ -2871,6 +2900,9 @@ contents = await list_directory(
   - `types` (optional): Entity type filter
   - `entity_types` (optional): Observation category filter
   - `after_date` (optional): Date filter (ISO format)
+  - `metadata_filters` (optional): Structured frontmatter filters (dict)
+  - `tags` (optional): Frontmatter tags filter (list)
+  - `status` (optional): Frontmatter status filter (string)
   - `project` (required unless default_project_mode): Target project
 - Returns: Matching entities with scores
 - Example:
@@ -2883,6 +2915,22 @@ results = await search_notes(
 )
 ```
 
+**search_by_metadata(filters, limit, offset, project)**
+- Metadata-only search using structured frontmatter
+- Parameters:
+  - `filters` (required): Dict of field -> value (supports $in, $gt/$gte/$lt/$lte, $between)
+  - `limit` (optional): Max results (default: 20)
+  - `offset` (optional): Pagination offset (default: 0)
+  - `project` (required unless default_project_mode): Target project
+- Returns: Matching entities
+- Example:
+```python
+results = await search_by_metadata(
+    filters={"type": "spec", "status": "in-progress"},
+    project="main"
+)
+```
+
 ### Project Management
 
 **list_memory_projects()**

src/basic_memory/alembic/versions/d7e8f9a0b1c2_add_structured_metadata_indexes.py

@@ -0,0 +1,153 @@
+"""Add structured metadata indexes for entity frontmatter
+
+Revision ID: d7e8f9a0b1c2
+Revises: g9a0b3c4d5e6
+Create Date: 2026-01-31 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import text
+
+
+def column_exists(connection, table: str, column: str) -> bool:
+    """Check if a column exists in a table (idempotent migration support)."""
+    if connection.dialect.name == "postgresql":
+        result = connection.execute(
+            text(
+                "SELECT 1 FROM information_schema.columns "
+                "WHERE table_name = :table AND column_name = :column"
+            ),
+            {"table": table, "column": column},
+        )
+        return result.fetchone() is not None
+    # SQLite
+    result = connection.execute(text(f"PRAGMA table_info({table})"))
+    columns = [row[1] for row in result]
+    return column in columns
+
+
+def index_exists(connection, index_name: str) -> bool:
+    """Check if an index exists (idempotent migration support)."""
+    if connection.dialect.name == "postgresql":
+        result = connection.execute(
+            text("SELECT 1 FROM pg_indexes WHERE indexname = :index_name"),
+            {"index_name": index_name},
+        )
+        return result.fetchone() is not None
+    # SQLite
+    result = connection.execute(
+        text("SELECT 1 FROM sqlite_master WHERE type='index' AND name = :index_name"),
+        {"index_name": index_name},
+    )
+    return result.fetchone() is not None
+
+
+# revision identifiers, used by Alembic.
+revision: str = "d7e8f9a0b1c2"
+down_revision: Union[str, None] = "g9a0b3c4d5e6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add JSONB/GiN indexes for Postgres and generated columns for SQLite."""
+    connection = op.get_bind()
+    dialect = connection.dialect.name
+
+    if dialect == "postgresql":
+        # Ensure JSONB for efficient indexing
+        result = connection.execute(
+            text(
+                "SELECT data_type FROM information_schema.columns "
+                "WHERE table_name = 'entity' AND column_name = 'entity_metadata'"
+            )
+        ).fetchone()
+        if result and result[0] != "jsonb":
+            op.execute(
+                "ALTER TABLE entity ALTER COLUMN entity_metadata "
+                "TYPE jsonb USING entity_metadata::jsonb"
+            )
+
+        # General JSONB GIN index
+        op.execute(
+            "CREATE INDEX IF NOT EXISTS idx_entity_metadata_gin "
+            "ON entity USING GIN (entity_metadata jsonb_path_ops)"
+        )
+
+        # Common field indexes
+        op.execute(
+            "CREATE INDEX IF NOT EXISTS idx_entity_tags_json "
+            "ON entity USING GIN ((entity_metadata -> 'tags'))"
+        )
+        op.execute(
+            "CREATE INDEX IF NOT EXISTS idx_entity_frontmatter_type "
+            "ON entity ((entity_metadata ->> 'type'))"
+        )
+        op.execute(
+            "CREATE INDEX IF NOT EXISTS idx_entity_frontmatter_status "
+            "ON entity ((entity_metadata ->> 'status'))"
+        )
+        return
+
+    # SQLite: add generated columns for common frontmatter fields
+    if not column_exists(connection, "entity", "tags_json"):
+        op.add_column(
+            "entity",
+            sa.Column(
+                "tags_json",
+                sa.Text(),
+                sa.Computed("json_extract(entity_metadata, '$.tags')", persisted=True),
+            ),
+        )
+    if not column_exists(connection, "entity", "frontmatter_status"):
+        op.add_column(
+            "entity",
+            sa.Column(
+                "frontmatter_status",
+                sa.Text(),
+                sa.Computed("json_extract(entity_metadata, '$.status')", persisted=True),
+            ),
+        )
+    if not column_exists(connection, "entity", "frontmatter_type"):
+        op.add_column(
+            "entity",
+            sa.Column(
+                "frontmatter_type",
+                sa.Text(),
+                sa.Computed("json_extract(entity_metadata, '$.type')", persisted=True),
+            ),
+        )
+
+    # Index generated columns
+    if not index_exists(connection, "idx_entity_tags_json"):
+        op.create_index("idx_entity_tags_json", "entity", ["tags_json"])
+    if not index_exists(connection, "idx_entity_frontmatter_status"):
+        op.create_index("idx_entity_frontmatter_status", "entity", ["frontmatter_status"])
+    if not index_exists(connection, "idx_entity_frontmatter_type"):
+        op.create_index("idx_entity_frontmatter_type", "entity", ["frontmatter_type"])
+
+
+def downgrade() -> None:
+    """Best-effort downgrade (drop indexes, revert JSONB on Postgres)."""
+    connection = op.get_bind()
+    dialect = connection.dialect.name
+
+    if dialect == "postgresql":
+        op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_status")
+        op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_type")
+        op.execute("DROP INDEX IF EXISTS idx_entity_tags_json")
+        op.execute("DROP INDEX IF EXISTS idx_entity_metadata_gin")
+        op.execute(
+            "ALTER TABLE entity ALTER COLUMN entity_metadata "
+            "TYPE json USING entity_metadata::json"
+        )
+        return
+
+    # SQLite: drop indexes (dropping generated columns requires table rebuild)
+    op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_status")
+    op.execute("DROP INDEX IF EXISTS idx_entity_frontmatter_type")
+    op.execute("DROP INDEX IF EXISTS idx_entity_tags_json")

src/basic_memory/cli/commands/tool.py

@@ -1,5 +1,6 @@
 """CLI tool commands for Basic Memory."""
 
+import json
 import sys
 from typing import Annotated, List, Optional
 
@@ -288,7 +289,10 @@ def recent_activity(
 
 @tool_app.command("search-notes")
 def search_notes(
-    query: str,
+    query: Annotated[
+        Optional[str],
+        typer.Argument(help="Search query string (optional when using metadata filters)"),
+    ] = "",
     permalink: Annotated[bool, typer.Option("--permalink", help="Search permalink values")] = False,
     title: Annotated[bool, typer.Option("--title", help="Search title values")] = False,
     project: Annotated[
@@ -301,6 +305,26 @@ def search_notes(
         Optional[str],
         typer.Option("--after_date", help="Search results after date, eg. '2d', '1 week'"),
     ] = None,
+    tags: Annotated[
+        Optional[List[str]],
+        typer.Option("--tag", help="Filter by frontmatter tag (repeatable)"),
+    ] = None,
+    status: Annotated[
+        Optional[str],
+        typer.Option("--status", help="Filter by frontmatter status"),
+    ] = None,
+    note_types: Annotated[
+        Optional[List[str]],
+        typer.Option("--type", help="Filter by frontmatter type (repeatable)"),
+    ] = None,
+    meta: Annotated[
+        Optional[List[str]],
+        typer.Option("--meta", help="Filter by frontmatter key=value (repeatable)"),
+    ] = None,
+    filter_json: Annotated[
+        Optional[str],
+        typer.Option("--filter", help="JSON metadata filter (advanced)"),
+    ] = None,
     page: int = 1,
     page_size: int = 10,
     local: bool = typer.Option(
@@ -335,21 +359,57 @@ def search_notes(
             )
             raise typer.Exit(1)
 
+        # Build metadata filters from --filter and --meta
+        metadata_filters = {}
+        if filter_json:
+            try:
+                metadata_filters = json.loads(filter_json)
+                if not isinstance(metadata_filters, dict):
+                    raise ValueError("Metadata filter JSON must be an object")
+            except json.JSONDecodeError as e:
+                typer.echo(f"Invalid JSON for --filter: {e}", err=True)
+                raise typer.Exit(1)
+
+        if meta:
+            for item in meta:
+                if "=" not in item:
+                    typer.echo(
+                        f"Invalid --meta entry '{item}'. Use key=value format.",
+                        err=True,
+                    )
+                    raise typer.Exit(1)
+                key, value = item.split("=", 1)
+                key = key.strip()
+                if not key:
+                    typer.echo(f"Invalid --meta entry '{item}'.", err=True)
+                    raise typer.Exit(1)
+                metadata_filters[key] = value
+
+        if not metadata_filters:
+            metadata_filters = None
+
         # set search type
-        search_type = ("permalink" if permalink else None,)
-        search_type = ("permalink_match" if permalink and "*" in query else None,)
-        search_type = ("title" if title else None,)
-        search_type = "text" if search_type is None else search_type
+        search_type = "text"
+        if permalink:
+            search_type = "permalink"
+            if query and "*" in query:
+                search_type = "permalink"
+        if title:
+            search_type = "title"
 
         with force_routing(local=local, cloud=cloud):
             results = run_with_cleanup(
                 mcp_search.fn(
-                    query,
+                    query or "",
                     project_name,
                     search_type=search_type,
                     page=page,
                     after_date=after_date,
                     page_size=page_size,
+                    types=note_types,
+                    metadata_filters=metadata_filters,
+                    tags=tags,
+                    status=status,
                 )
             )
         # Use json module for more controlled serialization