Flexible task search with prefix matching, slug search, and ranked results

nerve/agent/tools.py

@@ -81,9 +81,9 @@ def _done_dir() -> Path:
 
 @tool(
     "task_search",
-    "Search tasks by keyword in title. Returns matching tasks. Use this before creating tasks to check for duplicates.",
+    "Search tasks by keyword in title, content, tags, or slug. Supports partial words and task ID lookup. Returns matching tasks ranked by relevance. Use this before creating tasks to check for duplicates.",
     {
-        "query": {"type": "string", "description": "Search keyword(s) to match in task titles"},
+        "query": {"type": "string", "description": "Search keyword(s), partial words, or task ID/slug to match against title, content, tags, and task ID"},
         "status": {"type": "string", "description": "Filter: 'all' (include done), specific status, or empty (open tasks only)", "default": ""},
         "tag": {"type": "string", "description": "Filter by tag name (exact match)", "default": ""},
     },
@@ -308,7 +308,7 @@ async def task_update(args: dict) -> dict:
                 if new_title:
                     # Replace the H1 heading (first line starting with #)
                     content = _re.sub(r"^# .+", f"# {new_title}", content, count=1)
-                    # Sync title to SQLite
+                    # Sync title to SQLite — pass content to preserve FTS index
                     await _db.upsert_task(
                         task_id=task_id,
                         file_path=task["file_path"],
@@ -318,6 +318,7 @@ async def task_update(args: dict) -> dict:
                         source_url=task.get("source_url"),
                         deadline=deadline or task.get("deadline"),
                         tags=new_tags_str if raw_tags else (task.get("tags") or ""),
+                        content=content,
                     )
                 if note:
                     today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

nerve/db/migrations/v023_tasks_fts_slug.py

@@ -0,0 +1,29 @@
+"""V23: Rebuild tasks_fts with task_id as a searchable column.
+
+Previously task_id was marked UNINDEXED, making slug-based search impossible.
+The new schema makes task_id searchable so queries like "distribution" match
+the slug "2026-03-10-distribution-documentation".
+"""
+
+from __future__ import annotations
+
+import logging
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+
+async def up(db: aiosqlite.Connection) -> None:
+    # Drop and recreate FTS table with task_id as searchable column
+    await db.execute("DROP TABLE IF EXISTS tasks_fts")
+    await db.execute(
+        "CREATE VIRTUAL TABLE tasks_fts USING fts5(task_id, title, content)"
+    )
+    # Seed from existing tasks — content will be filled on next reindex
+    await db.execute(
+        "INSERT INTO tasks_fts (task_id, title, content) "
+        "SELECT id, title, '' FROM tasks"
+    )
+    await db.commit()
+    logger.info("V23 migration: rebuilt tasks_fts with searchable task_id column")

nerve/db/tasks.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from datetime import datetime, timezone
 
 
@@ -31,12 +32,15 @@ async def upsert_task(
                        deadline=excluded.deadline, tags=excluded.tags, updated_at=?""",
                 (task_id, file_path, title, status, source, source_url, deadline, tags, now, now, now),
             )
-            # Sync FTS index — include tags so tag names are searchable
+            # Sync FTS index — include tags and slug so they're all searchable
             fts_content = f"{content} {tags.replace(',', ' ')}" if tags else content
-            await self.db.execute("DELETE FROM tasks_fts WHERE task_id = ?", (task_id,))
+            # Normalize slug: replace hyphens with spaces so "2026-03-10-distribution"
+            # becomes searchable as individual words
+            fts_slug = task_id.replace("-", " ")
+            await self.db.execute("DELETE FROM tasks_fts WHERE task_id = ?", (fts_slug,))
             await self.db.execute(
                 "INSERT INTO tasks_fts (task_id, title, content) VALUES (?, ?, ?)",
-                (task_id, title, fts_content),
+                (fts_slug, title, fts_content),
             )
 
     async def get_task(self, task_id: str) -> dict | None:
@@ -85,67 +89,177 @@ async def update_task_tags(self, task_id: str, tags: str) -> None:
         )
         await self.db.commit()
 
-    # Short words and common stop words that add noise to FTS searches.
+    # ── FTS query building ───────────────────────────────────────────────
+
+    # Characters that are FTS5 syntax or punctuation — replaced with spaces.
+    _FTS_CLEAN_RE = re.compile(r'["\*\(\)\-:/\\#\.\,\;\'\[\]\{\}@!?\^~`]')
+
+    # Common stop words filtered from search queries.
     _FTS_STOP_WORDS = frozenset({
         "a", "an", "the", "is", "at", "by", "on", "in", "to", "of", "for",
         "and", "or", "not", "it", "be", "as", "do", "if", "so", "no", "up",
         "my", "we", "he", "me",
     })
 
+    @classmethod
+    def _tokenize_query(cls, query: str) -> list[str]:
+        """Clean and tokenize a raw search string into FTS-safe words."""
+        clean = cls._FTS_CLEAN_RE.sub(" ", query)
+        return [
+            w for w in clean.split()
+            if w.strip() and len(w) > 1 and w.lower() not in cls._FTS_STOP_WORDS
+        ]
+
     @classmethod
     def _build_fts_query(cls, query: str, mode: str = "and") -> str:
-        """Build an FTS5 query from a user search string.
+        """Build an FTS5 MATCH expression with prefix matching.
+
+        Each word is searched as both exact and prefix: (word OR word*)
+        so "distrib" matches "distribution", "distributed", etc.
 
         Args:
             query: Raw search text.
-            mode: 'and' — all terms must match (strict, good for user search).
-                  'or'  — any term can match (permissive, good for dedup).
+            mode: 'and' — all terms must match (strict search).
+                  'or'  — any term can match (permissive dedup).
         """
-        import re
-        clean = re.sub(r'["\*\(\)\-:/\\#]', " ", query)
-        words = [
-            w for w in clean.split()
-            if w.strip() and len(w) > 1 and w.lower() not in cls._FTS_STOP_WORDS
-        ]
+        words = cls._tokenize_query(query)
         if not words:
             return ""
-        joiner = " OR " if mode == "or" else " "
-        return joiner.join(f'"{w}"' for w in words)
 
-    async def search_tasks(
-        self, query: str, status: str | None = None, tag: str | None = None, limit: int = 20,
-    ) -> list[dict]:
-        """Search tasks using FTS5 full-text search on title and content.
+        # Each term: exact OR prefix match.  FTS5 prefix syntax is word*
+        terms = [f'("{w}" OR {w}*)' for w in words]
 
-        Args:
-            query: Search words — tokenized and matched via FTS5.
-            status: Filter by status. None = non-done, 'all' = everything.
-            tag: Filter by exact tag name.
-            limit: Max results.
-        """
-        fts_query = self._build_fts_query(query)
-        if not fts_query:
-            return []
+        joiner = " OR " if mode == "or" else " AND "
+        return joiner.join(terms)
+
+    # ── Status/tag filter helpers ────────────────────────────────────────
 
-        conditions = ["t.id IN (SELECT task_id FROM tasks_fts WHERE tasks_fts MATCH ?)"]
-        params: list = [fts_query]
+    @staticmethod
+    def _apply_status_filter(
+        conditions: list[str], params: list, status: str | None,
+    ) -> None:
+        """Append status filter clause to conditions/params in place."""
         if status == "all":
-            pass  # no status filter
+            pass
         elif status:
             conditions.append("t.status = ?")
             params.append(status)
         else:
             conditions.append("t.status != 'done'")
+
+    @staticmethod
+    def _apply_tag_filter(
+        conditions: list[str], params: list, tag: str | None,
+    ) -> None:
+        """Append tag filter clause to conditions/params in place."""
         if tag:
             conditions.append("',' || t.tags || ',' LIKE ?")
             params.append(f"%,{tag.strip().lower()},%")
-        where = " AND ".join(conditions)
-        params.append(limit)
-        async with self.db.execute(
-            f"SELECT t.* FROM tasks t WHERE {where} ORDER BY t.updated_at DESC LIMIT ?",
-            tuple(params),
-        ) as cursor:
-            return [dict(row) async for row in cursor]
+
+    # ── Search ───────────────────────────────────────────────────────────
+
+    async def search_tasks(
+        self, query: str, status: str | None = None, tag: str | None = None, limit: int = 20,
+    ) -> list[dict]:
+        """Flexible task search with multiple strategies and relevance ranking.
+
+        Strategies (in priority order):
+          1. Exact task ID match
+          2. FTS5 prefix search ranked by BM25
+          3. Task ID substring match (LIKE)
+          4. Title substring fallback (LIKE)
+
+        Results are merged and deduplicated — earlier strategies rank higher.
+        """
+        query = query.strip()
+        if not query:
+            return []
+
+        seen: set[str] = set()
+        results: list[dict] = []
+
+        def _add(rows: list[dict]) -> None:
+            for row in rows:
+                tid = row["id"]
+                if tid not in seen:
+                    seen.add(tid)
+                    results.append(row)
+
+        # ── Strategy 1: exact task ID ────────────────────────────────────
+        exact = await self.get_task(query)
+        if exact and self._row_matches_filters(exact, status, tag):
+            _add([exact])
+
+        # ── Strategy 2: FTS5 prefix match + BM25 ranking ────────────────
+        fts_query = self._build_fts_query(query)
+        if fts_query:
+            conditions: list[str] = []
+            params: list = []
+            self._apply_status_filter(conditions, params, status)
+            self._apply_tag_filter(conditions, params, tag)
+
+            where_extra = (" AND " + " AND ".join(conditions)) if conditions else ""
+            fts_params = [fts_query] + params + [limit]
+
+            async with self.db.execute(
+                f"SELECT t.* FROM tasks t "
+                f"JOIN tasks_fts f ON f.task_id = REPLACE(t.id, '-', ' ') "
+                f"WHERE tasks_fts MATCH ?{where_extra} "
+                f"ORDER BY f.rank LIMIT ?",
+                tuple(fts_params),
+            ) as cursor:
+                _add([dict(row) async for row in cursor])
+
+        # ── Strategy 3: task ID substring (slug search) ──────────────────
+        if len(results) < limit:
+            conditions = ["t.id LIKE ?"]
+            params = [f"%{query.lower()}%"]
+            self._apply_status_filter(conditions, params, status)
+            self._apply_tag_filter(conditions, params, tag)
+            remaining = limit - len(results)
+            params.append(remaining)
+
+            async with self.db.execute(
+                f"SELECT t.* FROM tasks t WHERE {' AND '.join(conditions)} "
+                f"ORDER BY t.updated_at DESC LIMIT ?",
+                tuple(params),
+            ) as cursor:
+                _add([dict(row) async for row in cursor])
+
+        # ── Strategy 4: title LIKE fallback ──────────────────────────────
+        if len(results) < limit:
+            conditions = ["lower(t.title) LIKE ?"]
+            params = [f"%{query.lower()}%"]
+            self._apply_status_filter(conditions, params, status)
+            self._apply_tag_filter(conditions, params, tag)
+            remaining = limit - len(results)
+            params.append(remaining)
+
+            async with self.db.execute(
+                f"SELECT t.* FROM tasks t WHERE {' AND '.join(conditions)} "
+                f"ORDER BY t.updated_at DESC LIMIT ?",
+                tuple(params),
+            ) as cursor:
+                _add([dict(row) async for row in cursor])
+
+        return results[:limit]
+
+    @staticmethod
+    def _row_matches_filters(row: dict, status: str | None, tag: str | None) -> bool:
+        """Check if a single task row passes the status/tag filters."""
+        if status == "all":
+            pass
+        elif status:
+            if row.get("status") != status:
+                return False
+        else:
+            if row.get("status") == "done":
+                return False
+        if tag:
+            tags_csv = row.get("tags", "") or ""
+            if tag.strip().lower() not in [t.strip().lower() for t in tags_csv.split(",")]:
+                return False
+        return True
 
     async def search_tasks_similar(
         self, query: str, limit: int = 10,
@@ -162,7 +276,7 @@ async def search_tasks_similar(
 
         async with self.db.execute(
             "SELECT t.* FROM tasks t "
-            "JOIN tasks_fts f ON f.task_id = t.id "
+            "JOIN tasks_fts f ON f.task_id = REPLACE(t.id, '-', ' ') "
             "WHERE tasks_fts MATCH ? "
             "ORDER BY f.rank LIMIT ?",
             (fts_query, limit),

nerve/gateway/routes/tasks.py

@@ -96,6 +96,8 @@ async def update_task(task_id: str, req: TaskUpdateRequest, user: dict = Depends
                 source=task.get("source"),
                 source_url=task.get("source_url"),
                 deadline=fields.get("deadline") or task.get("deadline"),
+                tags=fields.get("tags") or task.get("tags", ""),
+                content=req.content,
             )
 
     # Update status/note/deadline/title via agent tool (may move file for "done")

tests/test_db.py

@@ -454,13 +454,18 @@ async def test_fts_updated_on_upsert(self, db: Database):
         assert len(results) == 1
 
     async def test_rebuild_fts(self, db: Database):
-        await self._create_task(db, "t1", "Some task")
-        results = await db.search_tasks("task")
+        await self._create_task(db, "t1", "Some task", content="secret keyword xyzzy")
+        # Content-only word is findable via FTS
+        results = await db.search_tasks("xyzzy")
         assert len(results) == 1
-        # Rebuild clears the index
+        # Rebuild clears the FTS index
         await db.rebuild_fts()
-        results = await db.search_tasks("task")
+        # Content-only word no longer found (not in title or slug)
+        results = await db.search_tasks("xyzzy")
         assert len(results) == 0
+        # But title LIKE fallback still works
+        results = await db.search_tasks("task")
+        assert len(results) == 1
 
     async def test_build_fts_query_sanitizes_special_chars(self, db: Database):
         """FTS5 special characters should be stripped, not cause errors."""
@@ -480,6 +485,51 @@ async def test_empty_query_returns_empty(self, db: Database):
         results = await db.search_tasks("   ")
         assert len(results) == 0
 
+    async def test_search_prefix_matching(self, db: Database):
+        """Partial word should match via FTS5 prefix syntax."""
+        await self._create_task(db, "t1", "Distribution documentation update")
+        results = await db.search_tasks("distrib")
+        assert len(results) == 1
+        assert results[0]["id"] == "t1"
+
+    async def test_search_by_exact_task_id(self, db: Database):
+        """Exact task ID should be found."""
+        await self._create_task(db, "2026-03-10-distribution-docs", "Distribution documentation")
+        results = await db.search_tasks("2026-03-10-distribution-docs")
+        assert len(results) == 1
+        assert results[0]["id"] == "2026-03-10-distribution-docs"
+
+    async def test_search_by_slug_substring(self, db: Database):
+        """Partial slug should find the task."""
+        await self._create_task(db, "2026-03-10-azure-migration", "Azure eastus to centralus")
+        results = await db.search_tasks("azure-migration")
+        assert len(results) == 1
+        assert results[0]["id"] == "2026-03-10-azure-migration"
+
+    async def test_search_slug_words_via_fts(self, db: Database):
+        """Words from the slug should be searchable via FTS."""
+        await self._create_task(db, "2026-04-01-payment-failures", "Google billing issue")
+        # "payment" is only in the slug, not the title
+        results = await db.search_tasks("payment")
+        assert len(results) == 1
+
+    async def test_search_deduplicates_across_strategies(self, db: Database):
+        """A task matching multiple strategies should appear only once."""
+        await self._create_task(db, "billing-fix", "Fix billing issue", content="billing problem")
+        results = await db.search_tasks("billing")
+        assert len(results) == 1
+
+    async def test_search_relevance_ordering(self, db: Database):
+        """FTS matches should rank higher than LIKE-only matches."""
+        # Task with exact FTS match on title
+        await self._create_task(db, "t1", "Optimize database queries")
+        # Task where "optim" only matches via slug LIKE
+        await self._create_task(db, "2026-01-01-optimize-cache", "Cache layer improvements")
+        results = await db.search_tasks("optimize", status="all")
+        assert len(results) == 2
+        # FTS match (t1 has "Optimize" in title) should come first
+        assert results[0]["id"] == "t1"
+
 
 # --- Consumer Cursors ---