chore: update version to 0.19.2 for v0.19.2 release

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 ## Unreleased
 
+## v0.19.2 (2026-03-09)
+
+### Bug Fixes
+
+- **#657**: Coerce string params to list/dict in MCP tools
+  - MCP clients that serialize `list`/`dict` arguments as JSON strings no longer fail Pydantic validation
+  - Adds `BeforeValidator` coercion to `search_notes` (`entity_types`, `note_types`, `tags`, `metadata_filters`), `write_note` (`metadata`), and `canvas` (`nodes`, `edges`)
+- **#655**: Handle SQLite and Windows semantic search regressions
+  - Fix embedding status query for non-semantic SQLite databases
+  - Windows-safe log file rotation with per-process log filenames
+  - Robust `setup_logging` that handles all environments cleanly
+
 ## v0.19.1 (2026-03-08)
 
 ### Bug Fixes

server.json

@@ -6,12 +6,12 @@
     "url": "https://github.com/basicmachines-co/basic-memory.git",
     "source": "github"
   },
-  "version": "0.19.1",
+  "version": "0.19.2",
   "packages": [
     {
       "registryType": "pypi",
       "identifier": "basic-memory",
-      "version": "0.19.1",
+      "version": "0.19.2",
       "runtimeHint": "uvx",
       "runtimeArguments": [
         {"type": "positional", "value": "basic-memory"},

src/basic_memory/__init__.py

@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
 # Package version - updated by release automation
-__version__ = "0.19.1"
+__version__ = "0.19.2"
 
 # API version for FastAPI - independent of package version
 __api_version__ = "v0"

src/basic_memory/mcp/tools/canvas.py

@@ -4,12 +4,14 @@ This tool creates Obsidian canvas files (.canvas) using the JSON Canvas 1.0 spec
 """
 
 import json
-from typing import Dict, List, Any, Optional
+from typing import Annotated, Dict, List, Any, Optional
 
 from loguru import logger
 from fastmcp import Context
+from pydantic import BeforeValidator
 
 from basic_memory.mcp.project_context import get_project_client
+from basic_memory.utils import coerce_list
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.utils import call_put, call_post, resolve_entity_id
 
@@ -19,8 +21,8 @@ from basic_memory.mcp.tools.utils import call_put, call_post, resolve_entity_id
     annotations={"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
 )
 async def canvas(
-    nodes: List[Dict[str, Any]],
-    edges: List[Dict[str, Any]],
+    nodes: Annotated[List[Dict[str, Any]], BeforeValidator(coerce_list)],
+    edges: Annotated[List[Dict[str, Any]], BeforeValidator(coerce_list)],
     title: str,
     directory: str,
     project: Optional[str] = None,

src/basic_memory/mcp/tools/search.py

@@ -6,8 +6,10 @@ from typing import Annotated, List, Optional, Dict, Any, Literal
 
 from loguru import logger
 from fastmcp import Context
+from pydantic import BeforeValidator
 
 from basic_memory.config import ConfigManager
+from basic_memory.utils import coerce_dict, coerce_list
 from basic_memory.mcp.container import get_container
 from basic_memory.mcp.project_context import (
     detect_project_from_url_prefix,
@@ -307,18 +309,26 @@ async def search_notes(
     output_format: Literal["text", "json"] = "text",
     note_types: Annotated[
         List[str] | None,
+        BeforeValidator(coerce_list),
         "Filter by the 'type' field in note frontmatter (e.g. 'note', 'chapter', 'person'). "
         "Case-insensitive.",
     ] = None,
     entity_types: Annotated[
         List[str] | None,
+        BeforeValidator(coerce_list),
         "Filter by knowledge graph item type: 'entity' (whole notes), 'observation', or "
         "'relation'. Defaults to 'entity'. Do NOT pass schema/frontmatter types like "
         "'Chapter' here — use note_types instead.",
     ] = None,
     after_date: Optional[str] = None,
-    metadata_filters: Optional[Dict[str, Any]] = None,
-    tags: Optional[List[str]] = None,
+    metadata_filters: Annotated[
+        Dict[str, Any] | None,
+        BeforeValidator(coerce_dict),
+    ] = None,
+    tags: Annotated[
+        List[str] | None,
+        BeforeValidator(coerce_list),
+    ] = None,
     status: Optional[str] = None,
     min_similarity: Optional[float] = None,
     context: Context | None = None,

src/basic_memory/mcp/tools/write_note.py

@@ -1,16 +1,17 @@
 """Write note tool for Basic Memory MCP server."""
 
 import textwrap
-from typing import List, Union, Optional, Literal
+from typing import Annotated, List, Union, Optional, Literal
 
 from loguru import logger
+from pydantic import BeforeValidator
 
 from basic_memory.config import ConfigManager
 from basic_memory.mcp.project_context import get_project_client, add_project_metadata
 from basic_memory.mcp.server import mcp
 from fastmcp import Context
 from basic_memory.schemas.base import Entity
-from basic_memory.utils import parse_tags, validate_project_path
+from basic_memory.utils import coerce_dict, parse_tags, validate_project_path
 
 # Define TagType as a Union that can accept either a string or a list of strings or None
 TagType = Union[List[str], str, None]
@@ -28,7 +29,7 @@ async def write_note(
     workspace: Optional[str] = None,
     tags: list[str] | str | None = None,
     note_type: str = "note",
-    metadata: dict | None = None,
+    metadata: Annotated[dict | None, BeforeValidator(coerce_dict)] = None,
     overwrite: bool | None = None,
     output_format: Literal["text", "json"] = "text",
     context: Context | None = None,

src/basic_memory/repository/search_repository_base.py

@@ -451,21 +451,36 @@ class SearchRepositoryBase(ABC):
         return "\n\n".join(part for part in row_parts if part)
 
     def _build_chunk_records(self, rows) -> list[dict[str, str]]:
-        records: list[dict[str, str]] = []
+        records_by_key: dict[str, dict[str, str]] = {}
+        duplicate_chunk_keys = 0
         for row in rows:
             source_text = self._compose_row_source_text(row)
             chunks = self._split_text_into_chunks(source_text)
             for chunk_index, chunk_text in enumerate(chunks):
                 chunk_key = f"{row.type}:{row.id}:{chunk_index}"
                 source_hash = hashlib.sha256(chunk_text.encode("utf-8")).hexdigest()
-                records.append(
-                    {
-                        "chunk_key": chunk_key,
-                        "chunk_text": chunk_text,
-                        "source_hash": source_hash,
-                    }
-                )
-        return records
+                # Trigger: SQLite FTS5 can accumulate duplicate logical rows for the
+                # same search_index id because it does not enforce relational uniqueness.
+                # Why: duplicate chunk keys would schedule duplicate writes for the same
+                # chunk row and eventually trip UNIQUE(rowid) in search_vector_embeddings.
+                # Outcome: collapse chunk work to one deterministic record per chunk key.
+                if chunk_key in records_by_key:
+                    duplicate_chunk_keys += 1
+                records_by_key[chunk_key] = {
+                    "chunk_key": chunk_key,
+                    "chunk_text": chunk_text,
+                    "source_hash": source_hash,
+                }
+
+        if duplicate_chunk_keys:
+            logger.warning(
+                "Collapsed duplicate vector chunk keys before embedding sync: "
+                "project_id={project_id} duplicate_chunk_keys={duplicate_chunk_keys}",
+                project_id=self.project_id,
+                duplicate_chunk_keys=duplicate_chunk_keys,
+            )
+
+        return list(records_by_key.values())
 
     # --- Text splitting ---
 

src/basic_memory/services/project_service.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Dict, Optional, Sequence
 
 from loguru import logger
 from sqlalchemy import text
+from sqlalchemy.exc import OperationalError as SAOperationalError
 
 from basic_memory.models import Project
 from basic_memory.repository.project_repository import ProjectRepository
@@ -1004,56 +1005,81 @@ class ProjectService:
         )
         total_indexed_entities = si_result.scalar() or 0
 
-        chunks_result = await self.repository.execute_query(
-            text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :project_id"),
-            {"project_id": project_id},
-        )
-        total_chunks = chunks_result.scalar() or 0
-
-        entities_with_chunks_result = await self.repository.execute_query(
-            text(
-                "SELECT COUNT(DISTINCT entity_id) FROM search_vector_chunks "
-                "WHERE project_id = :project_id"
-            ),
-            {"project_id": project_id},
-        )
-        total_entities_with_chunks = entities_with_chunks_result.scalar() or 0
-
-        # Embeddings count — join pattern differs between SQLite and Postgres
-        if is_postgres:
-            embeddings_sql = text(
-                "SELECT COUNT(*) FROM search_vector_chunks c "
-                "JOIN search_vector_embeddings e ON e.chunk_id = c.id "
-                "WHERE c.project_id = :project_id"
-            )
-        else:
-            embeddings_sql = text(
-                "SELECT COUNT(*) FROM search_vector_chunks c "
-                "JOIN search_vector_embeddings e ON e.rowid = c.id "
-                "WHERE c.project_id = :project_id"
+        try:
+            chunks_result = await self.repository.execute_query(
+                text("SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :project_id"),
+                {"project_id": project_id},
             )
+            total_chunks = chunks_result.scalar() or 0
 
-        embeddings_result = await self.repository.execute_query(
-            embeddings_sql, {"project_id": project_id}
-        )
-        total_embeddings = embeddings_result.scalar() or 0
-
-        # Orphaned chunks (chunks without embeddings — indicates interrupted indexing)
-        if is_postgres:
-            orphan_sql = text(
-                "SELECT COUNT(*) FROM search_vector_chunks c "
-                "LEFT JOIN search_vector_embeddings e ON e.chunk_id = c.id "
-                "WHERE c.project_id = :project_id AND e.chunk_id IS NULL"
-            )
-        else:
-            orphan_sql = text(
-                "SELECT COUNT(*) FROM search_vector_chunks c "
-                "LEFT JOIN search_vector_embeddings e ON e.rowid = c.id "
-                "WHERE c.project_id = :project_id AND e.rowid IS NULL"
+            entities_with_chunks_result = await self.repository.execute_query(
+                text(
+                    "SELECT COUNT(DISTINCT entity_id) FROM search_vector_chunks "
+                    "WHERE project_id = :project_id"
+                ),
+                {"project_id": project_id},
             )
+            total_entities_with_chunks = entities_with_chunks_result.scalar() or 0
 
-        orphan_result = await self.repository.execute_query(orphan_sql, {"project_id": project_id})
-        orphaned_chunks = orphan_result.scalar() or 0
+            # Embeddings count — join pattern differs between SQLite and Postgres
+            if is_postgres:
+                embeddings_sql = text(
+                    "SELECT COUNT(*) FROM search_vector_chunks c "
+                    "JOIN search_vector_embeddings e ON e.chunk_id = c.id "
+                    "WHERE c.project_id = :project_id"
+                )
+            else:
+                embeddings_sql = text(
+                    "SELECT COUNT(*) FROM search_vector_chunks c "
+                    "JOIN search_vector_embeddings e ON e.rowid = c.id "
+                    "WHERE c.project_id = :project_id"
+                )
+
+            embeddings_result = await self.repository.execute_query(
+                embeddings_sql, {"project_id": project_id}
+            )
+            total_embeddings = embeddings_result.scalar() or 0
+
+            # Orphaned chunks (chunks without embeddings — indicates interrupted indexing)
+            if is_postgres:
+                orphan_sql = text(
+                    "SELECT COUNT(*) FROM search_vector_chunks c "
+                    "LEFT JOIN search_vector_embeddings e ON e.chunk_id = c.id "
+                    "WHERE c.project_id = :project_id AND e.chunk_id IS NULL"
+                )
+            else:
+                orphan_sql = text(
+                    "SELECT COUNT(*) FROM search_vector_chunks c "
+                    "LEFT JOIN search_vector_embeddings e ON e.rowid = c.id "
+                    "WHERE c.project_id = :project_id AND e.rowid IS NULL"
+                )
+
+            orphan_result = await self.repository.execute_query(
+                orphan_sql, {"project_id": project_id}
+            )
+            orphaned_chunks = orphan_result.scalar() or 0
+        except SAOperationalError as exc:
+            # Trigger: sqlite_master can list vec0 virtual tables even when sqlite-vec
+            # is not loaded in the current Python runtime.
+            # Why: project info should degrade gracefully instead of crashing on stats queries.
+            # Outcome: report vector tables as unavailable and point the user to install the
+            # missing dependency before rebuilding embeddings.
+            if is_postgres or "no such module: vec0" not in str(exc).lower():
+                raise
+
+            return EmbeddingStatus(
+                semantic_search_enabled=True,
+                embedding_provider=provider,
+                embedding_model=model,
+                embedding_dimensions=dimensions,
+                total_indexed_entities=total_indexed_entities,
+                vector_tables_exist=False,
+                reindex_recommended=True,
+                reindex_reason=(
+                    "SQLite vector tables exist but sqlite-vec is unavailable in this Python "
+                    "environment — install/update basic-memory, then run: bm reindex --embeddings"
+                ),
+            )
 
         # --- Reindex recommendation logic (priority order) ---
         reindex_recommended = False

src/basic_memory/utils.py

@@ -1,5 +1,6 @@
 """Utility functions for basic-memory."""
 
+import json
 import os
 
 import logging
@@ -7,7 +8,7 @@ import re
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Protocol, Union, runtime_checkable, List, Optional
+from typing import Any, Protocol, Union, runtime_checkable, List, Optional
 
 from loguru import logger
 from unidecode import unidecode
@@ -66,6 +67,7 @@ class PathLike(Protocol):
 # 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]
+WINDOWS_LOG_FILE_RETENTION = 5
 
 
 def generate_permalink(file_path: Union[Path, str, PathLike], split_extension: bool = True) -> str:
@@ -250,7 +252,7 @@ def setup_logging(
     log_to_file: bool = False,
     log_to_stdout: bool = False,
     structured_context: bool = False,
-) -> None:  # pragma: no cover
+) -> None:
     """Configure logging with explicit settings.
 
     This function provides a simple, explicit interface for configuring logging.
@@ -273,8 +275,14 @@ def setup_logging(
 
     # Add file handler with rotation
     if log_to_file:
-        log_path = Path.home() / ".basic-memory" / "basic-memory.log"
+        # Trigger: Windows does not allow renaming an open file held by another process.
+        # Why: multiple basic-memory processes can share the same log directory at once.
+        # Outcome: use per-process log files on Windows so log rotation stays local.
+        log_filename = f"basic-memory-{os.getpid()}.log" if os.name == "nt" else "basic-memory.log"
+        log_path = Path.home() / ".basic-memory" / log_filename
         log_path.parent.mkdir(parents=True, exist_ok=True)
+        if os.name == "nt":
+            _cleanup_windows_log_files(log_path.parent, log_path.name)
         # Keep logging synchronous (enqueue=False) to avoid background logging threads.
         # Background threads are a common source of "hang on exit" issues in CLI/test runs.
         logger.add(
@@ -308,6 +316,28 @@ def setup_logging(
     logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
 
 
+def _cleanup_windows_log_files(log_dir: Path, current_log_name: str) -> None:
+    """Trim stale per-process Windows log files so the directory stays bounded."""
+    stale_logs = [
+        path
+        for path in log_dir.glob("basic-memory-*.log*")
+        if path.is_file() and path.name != current_log_name
+    ]
+
+    if len(stale_logs) <= WINDOWS_LOG_FILE_RETENTION - 1:
+        return
+
+    # Trigger: per-process log filenames avoid Windows rename contention but fragment retention.
+    # Why: loguru retention applies per sink, not across the whole basic-memory log directory.
+    # Outcome: keep only the newest stale PID logs so repeated CLI/server launches stay bounded.
+    stale_logs.sort(key=lambda path: path.stat().st_mtime, reverse=True)
+    for stale_log in stale_logs[WINDOWS_LOG_FILE_RETENTION - 1 :]:
+        try:
+            stale_log.unlink()
+        except OSError:
+            logger.debug("Failed to delete stale Windows log file: {path}", path=stale_log)
+
+
 def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
     """Parse tags from various input formats into a consistent list.
 
@@ -356,6 +386,36 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
         return []
 
 
+def coerce_list(v: Any) -> Any:
+    """Coerce string input to list for MCP clients that serialize lists as strings."""
+    if v is None:
+        return v
+    if isinstance(v, str):
+        try:
+            parsed = json.loads(v)
+            if isinstance(parsed, list):
+                return parsed
+        except (json.JSONDecodeError, TypeError):
+            pass
+        # Single string value — wrap in a list
+        return [v]
+    return v
+
+
+def coerce_dict(v: Any) -> Any:
+    """Coerce string input to dict for MCP clients that serialize dicts as strings."""
+    if v is None:
+        return v
+    if isinstance(v, str):
+        try:
+            parsed = json.loads(v)
+            if isinstance(parsed, dict):
+                return parsed
+        except (json.JSONDecodeError, TypeError):
+            pass
+    return v
+
+
 def normalize_newlines(multiline: str) -> str:
     """Replace any \r\n, \r, or \n with the native newline.
 

test-int/mcp/test_string_params_integration.py

@@ -0,0 +1,167 @@
+"""Integration tests for MCP tools accepting string-serialized list/dict params.
+
+Goes through the full FastMCP Client → validate_call → tool function path,
+which is where Pydantic rejects strings for list/dict params.
+"""
+
+import pytest
+from fastmcp import Client
+
+
+@pytest.mark.asyncio
+async def test_search_notes_entity_types_as_string(mcp_server, app, test_project):
+    """search_notes should accept entity_types as a JSON string via MCP protocol."""
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Entity Type Coerce Test",
+                "directory": "test",
+                "content": "# Test\nContent for entity type coercion",
+            },
+        )
+
+        # MCP client sends entity_types as a string
+        result = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "coercion",
+                "entity_types": '["entity"]',
+            },
+        )
+        text = result.content[0].text
+        assert "Search Failed" not in text
+
+
+@pytest.mark.asyncio
+async def test_search_notes_note_types_as_string(mcp_server, app, test_project):
+    """search_notes should accept note_types as a JSON string via MCP protocol."""
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Note Type Coerce Test",
+                "directory": "test",
+                "content": "# Test\nContent for note type coercion",
+            },
+        )
+
+        result = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "coercion",
+                "note_types": '["note"]',
+            },
+        )
+        text = result.content[0].text
+        assert "Search Failed" not in text
+
+
+@pytest.mark.asyncio
+async def test_search_notes_tags_as_string(mcp_server, app, test_project):
+    """search_notes should accept tags as a JSON string via MCP protocol."""
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Tags Coerce Test",
+                "directory": "test",
+                "content": "# Test\nTagged content for coercion",
+                "tags": "alpha",
+            },
+        )
+
+        result = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "tagged",
+                "tags": '["alpha"]',
+            },
+        )
+        text = result.content[0].text
+        assert "Search Failed" not in text
+
+
+@pytest.mark.asyncio
+async def test_search_notes_metadata_filters_as_string(mcp_server, app, test_project):
+    """search_notes should accept metadata_filters as a JSON string via MCP protocol."""
+    async with Client(mcp_server) as client:
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Metadata Coerce Test",
+                "directory": "test",
+                "content": "# Test\nMetadata content for coercion",
+            },
+        )
+
+        result = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "metadata",
+                "metadata_filters": '{"type": "note"}',
+            },
+        )
+        text = result.content[0].text
+        assert "Search Failed" not in text
+
+
+@pytest.mark.asyncio
+async def test_write_note_metadata_as_string(mcp_server, app, test_project):
+    """write_note should accept metadata as a JSON string via MCP protocol."""
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "String Metadata Note",
+                "directory": "test",
+                "content": "# Test\nWith string metadata",
+                "metadata": '{"priority": "high"}',
+            },
+        )
+        text = result.content[0].text
+        assert "Created note" in text or "Updated note" in text
+
+
+@pytest.mark.asyncio
+async def test_canvas_nodes_edges_as_string(mcp_server, app, test_project):
+    """canvas should accept nodes and edges as JSON strings via MCP protocol."""
+    import json
+
+    nodes = [
+        {
+            "id": "n1",
+            "type": "text",
+            "text": "Hello",
+            "x": 0,
+            "y": 0,
+            "width": 200,
+            "height": 100,
+        }
+    ]
+    edges = [
+        {"id": "e1", "fromNode": "n1", "toNode": "n1", "label": "self"}
+    ]
+
+    async with Client(mcp_server) as client:
+        result = await client.call_tool(
+            "canvas",
+            {
+                "project": test_project.name,
+                "title": "Coerce Canvas Test",
+                "directory": "test",
+                "nodes": json.dumps(nodes),
+                "edges": json.dumps(edges),
+            },
+        )
+        text = result.content[0].text
+        assert "Created" in text or "Updated" in text

tests/repository/test_semantic_search_base.py

@@ -237,6 +237,29 @@ class TestBuildChunkRecords:
         records = self.repo._build_chunk_records(rows)
         assert any("99" in r["chunk_key"] for r in records)
 
+    def test_duplicate_rows_collapse_to_unique_chunk_keys(self):
+        rows = [
+            _make_row(
+                row_type=SearchItemType.ENTITY.value,
+                title="Spec",
+                permalink="spec",
+                content_snippet="shared content",
+                row_id=77,
+            ),
+            _make_row(
+                row_type=SearchItemType.ENTITY.value,
+                title="Spec",
+                permalink="spec",
+                content_snippet="shared content",
+                row_id=77,
+            ),
+        ]
+
+        records = self.repo._build_chunk_records(rows)
+
+        assert len(records) == 1
+        assert records[0]["chunk_key"] == "entity:77:0"
+
 
 # --- SQLite SemanticSearchDisabledError ---
 

tests/services/test_project_service_embedding_status.py

@@ -5,6 +5,7 @@ from unittest.mock import patch
 
 import pytest
 from sqlalchemy import text
+from sqlalchemy.exc import OperationalError as SAOperationalError
 
 from basic_memory.schemas.project_info import EmbeddingStatus
 from basic_memory.services.project_service import ProjectService
@@ -142,6 +143,46 @@ async def test_embedding_status_orphaned_chunks(
     assert "orphaned chunks" in (status.reindex_reason or "")
 
 
+@pytest.mark.asyncio
+async def test_embedding_status_handles_sqlite_vec_unavailable(
+    project_service: ProjectService, test_graph, test_project
+):
+    """Unreadable vec0 tables should degrade to unavailable status instead of crashing."""
+    # Trigger: Postgres test matrix executes the same unit suite.
+    # Why: sqlite-vec loading failures are specific to SQLite virtual tables, not Postgres joins.
+    # Outcome: keep the regression focused on the backend that can actually hit this path.
+    if _is_postgres():
+        pytest.skip("sqlite-vec unavailable handling is SQLite-specific.")
+
+    original_execute_query = project_service.repository.execute_query
+
+    async def _execute_query_with_vec0_failure(query, params):
+        query_text = str(query)
+        if "JOIN search_vector_embeddings" in query_text:
+            raise SAOperationalError(query_text, params, Exception("no such module: vec0"))
+        return await original_execute_query(query, params)
+
+    with patch.object(
+        type(project_service),
+        "config_manager",
+        new_callable=lambda: property(
+            lambda self: _config_manager_with(semantic_search_enabled=True)
+        ),
+    ):
+        with patch.object(
+            project_service.repository,
+            "execute_query",
+            side_effect=_execute_query_with_vec0_failure,
+        ):
+            status = await project_service.get_embedding_status(test_project.id)
+
+    assert status.semantic_search_enabled is True
+    assert status.total_indexed_entities > 0
+    assert status.vector_tables_exist is False
+    assert status.reindex_recommended is True
+    assert "sqlite-vec is unavailable" in (status.reindex_reason or "")
+
+
 @pytest.mark.asyncio
 async def test_embedding_status_healthy(project_service: ProjectService, test_graph, test_project):
     """When all entities have embeddings, no reindex recommended."""

tests/test_coerce.py

@@ -0,0 +1,58 @@
+"""Tests for coerce_list and coerce_dict utility functions.
+
+These must fail until the helpers are implemented in utils.py.
+"""
+
+
+from basic_memory.utils import coerce_list, coerce_dict
+
+
+class TestCoerceList:
+    """Tests for coerce_list."""
+
+    def test_none_passthrough(self):
+        assert coerce_list(None) is None
+
+    def test_native_list_passthrough(self):
+        assert coerce_list(["a", "b"]) == ["a", "b"]
+
+    def test_json_array_string(self):
+        assert coerce_list('["entity", "observation"]') == ["entity", "observation"]
+
+    def test_single_string_wrapped(self):
+        assert coerce_list("entity") == ["entity"]
+
+    def test_non_json_string_wrapped(self):
+        assert coerce_list("not-json") == ["not-json"]
+
+    def test_json_object_string_wrapped(self):
+        """A JSON object string is not a list, so wrap it."""
+        assert coerce_list('{"key": "val"}') == ['{"key": "val"}']
+
+    def test_int_passthrough(self):
+        """Non-string, non-None values pass through unchanged."""
+        assert coerce_list(42) == 42
+
+
+class TestCoerceDict:
+    """Tests for coerce_dict."""
+
+    def test_none_passthrough(self):
+        assert coerce_dict(None) is None
+
+    def test_native_dict_passthrough(self):
+        assert coerce_dict({"k": "v"}) == {"k": "v"}
+
+    def test_json_object_string(self):
+        assert coerce_dict('{"status": "draft"}') == {"status": "draft"}
+
+    def test_non_json_string_passthrough(self):
+        """Non-parseable strings pass through (Pydantic will reject them)."""
+        assert coerce_dict("not-json") == "not-json"
+
+    def test_json_array_string_passthrough(self):
+        """A JSON array string is not a dict, so pass through."""
+        assert coerce_dict('["a", "b"]') == '["a", "b"]'
+
+    def test_int_passthrough(self):
+        assert coerce_dict(42) == 42

tests/utils/test_setup_logging.py

@@ -0,0 +1,162 @@
+"""Tests for logging setup helpers."""
+
+import os
+import sys
+
+from basic_memory import utils
+
+
+def test_setup_logging_uses_shared_log_file_off_windows(monkeypatch, tmp_path) -> None:
+    """Non-Windows platforms should keep the shared log filename."""
+    added_sinks: list[str] = []
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setattr(utils.os, "name", "posix")
+    monkeypatch.setattr(utils.Path, "home", lambda: tmp_path)
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(
+        utils.logger,
+        "add",
+        lambda sink, **kwargs: added_sinks.append(str(sink)),
+    )
+
+    utils.setup_logging(log_to_file=True)
+
+    assert added_sinks == [str(tmp_path / ".basic-memory" / "basic-memory.log")]
+
+
+def test_setup_logging_uses_per_process_log_file_on_windows(monkeypatch, tmp_path) -> None:
+    """Windows uses per-process logs so rotation never contends across processes."""
+    added_sinks: list[str] = []
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setattr(utils.os, "name", "nt")
+    monkeypatch.setattr(utils.os, "getpid", lambda: 4242)
+    monkeypatch.setattr(utils.Path, "home", lambda: tmp_path)
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(
+        utils.logger,
+        "add",
+        lambda sink, **kwargs: added_sinks.append(str(sink)),
+    )
+
+    utils.setup_logging(log_to_file=True)
+
+    assert added_sinks == [str(tmp_path / ".basic-memory" / "basic-memory-4242.log")]
+
+
+def test_setup_logging_trims_stale_windows_pid_logs(monkeypatch, tmp_path) -> None:
+    """Windows cleanup should bound stale PID-specific log files across runs."""
+    log_dir = tmp_path / ".basic-memory"
+    log_dir.mkdir()
+
+    stale_logs = []
+    for index in range(6):
+        log_path = log_dir / f"basic-memory-{1000 + index}.log"
+        log_path.write_text("old log", encoding="utf-8")
+        mtime = 1_000 + index
+        os.utime(log_path, (mtime, mtime))
+        stale_logs.append(log_path)
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setattr(utils.os, "name", "nt")
+    monkeypatch.setattr(utils.os, "getpid", lambda: 4242)
+    monkeypatch.setattr(utils.Path, "home", lambda: tmp_path)
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(utils.logger, "add", lambda *args, **kwargs: None)
+
+    utils.setup_logging(log_to_file=True)
+
+    remaining = sorted(path.name for path in log_dir.glob("basic-memory-*.log*"))
+    assert remaining == [
+        "basic-memory-1002.log",
+        "basic-memory-1003.log",
+        "basic-memory-1004.log",
+        "basic-memory-1005.log",
+    ]
+
+
+def test_setup_logging_test_env_uses_stderr_only(monkeypatch) -> None:
+    """Test mode should add one stderr sink and return before other branches run."""
+    added_sinks: list[object] = []
+    configured_calls: list[dict] = []
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "test")
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(utils.logger, "add", lambda sink, **kwargs: added_sinks.append(sink))
+    monkeypatch.setattr(
+        utils.logger,
+        "configure",
+        lambda **kwargs: configured_calls.append(kwargs),
+    )
+
+    utils.setup_logging(log_to_file=True, log_to_stdout=True, structured_context=True)
+
+    assert added_sinks == [sys.stderr]
+    assert configured_calls == []
+
+
+def test_setup_logging_log_to_stdout(monkeypatch) -> None:
+    """stdout logging should attach a stderr sink outside test mode."""
+    added_sinks: list[object] = []
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(utils.logger, "add", lambda sink, **kwargs: added_sinks.append(sink))
+
+    utils.setup_logging(log_to_stdout=True)
+
+    assert added_sinks == [sys.stderr]
+
+
+def test_setup_logging_structured_context(monkeypatch) -> None:
+    """Structured context should bind cloud metadata into loguru extras."""
+    configured_extras: list[dict[str, str]] = []
+
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setenv("BASIC_MEMORY_TENANT_ID", "tenant-123")
+    monkeypatch.setenv("FLY_APP_NAME", "bm-app")
+    monkeypatch.setenv("FLY_MACHINE_ID", "machine-123")
+    monkeypatch.setenv("FLY_REGION", "ord")
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(utils.logger, "add", lambda *args, **kwargs: None)
+    monkeypatch.setattr(
+        utils.logger,
+        "configure",
+        lambda **kwargs: configured_extras.append(kwargs["extra"]),
+    )
+
+    utils.setup_logging(structured_context=True)
+
+    assert configured_extras == [
+        {
+            "tenant_id": "tenant-123",
+            "fly_app_name": "bm-app",
+            "fly_machine_id": "machine-123",
+            "fly_region": "ord",
+        }
+    ]
+
+
+def test_setup_logging_suppresses_noisy_loggers(monkeypatch) -> None:
+    """Third-party HTTP/file-watch loggers should be raised to WARNING."""
+    monkeypatch.setenv("BASIC_MEMORY_ENV", "dev")
+    monkeypatch.setattr(utils.logger, "remove", lambda *args, **kwargs: None)
+    monkeypatch.setattr(utils.logger, "add", lambda *args, **kwargs: None)
+
+    httpx_logger = utils.logging.getLogger("httpx")
+    watchfiles_logger = utils.logging.getLogger("watchfiles.main")
+    original_httpx_level = httpx_logger.level
+    original_watchfiles_level = watchfiles_logger.level
+
+    try:
+        httpx_logger.setLevel(utils.logging.DEBUG)
+        watchfiles_logger.setLevel(utils.logging.INFO)
+
+        utils.setup_logging()
+
+        assert httpx_logger.level == utils.logging.WARNING
+        assert watchfiles_logger.level == utils.logging.WARNING
+    finally:
+        httpx_logger.setLevel(original_httpx_level)
+        watchfiles_logger.setLevel(original_watchfiles_level)