Compare commits

...

7 Commits

Author SHA1 Message Date
Drew Cain 87616924ff chore: update version to 0.20.0 for v0.20.0 release 2026-03-10 22:08:00 -05:00
Drew Cain 5cb0502ed2 docs: add v0.20.0 changelog entry
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-03-10 22:07:49 -05:00
Paul Hernandez a94a717b1b feat(cli): add default-on auto-update system and bm update command (#643)
Signed-off-by: phernandez <paul@basicmachines.co>
Signed-off-by: Drew Cain <groksrc@users.noreply.github.com>
Co-authored-by: Drew Cain <groksrc@users.noreply.github.com>
2026-03-10 22:06:33 -05:00
phernandez 6e4bb72f10 chore: update version to 0.19.2 for v0.19.2 release 2026-03-09 23:42:10 -05:00
phernandez 11b0e31e24 docs: add v0.19.2 changelog entry
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-03-09 23:41:40 -05:00
Paul Hernandez a5c9e77f16 fix: coerce string params to list/dict in MCP tools (#657)
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-09 22:19:07 -05:00
Paul Hernandez 30a89357cb fix(core): handle SQLite and Windows semantic regressions (#655)
Signed-off-by: phernandez <paul@basicmachines.co>
2026-03-09 22:18:17 -05:00
26 changed files with 1692 additions and 74 deletions
+25
View File
@@ -2,6 +2,31 @@
## Unreleased
## v0.20.0 (2026-03-10)
### Features
- **#643**: Default-on auto-update system and `bm update` command
- Automatic background update checks for CLI installs (uv tool, Homebrew)
- Install-source detection (homebrew, uv_tool, uvx, unknown) with uvx skip behavior
- Periodic check gating via `auto_update_last_checked_at` + `update_check_interval` config
- Manager-specific update flows: Homebrew (`brew upgrade`) and uv tool (`uv tool upgrade`)
- Silent, non-blocking MCP behavior via daemon thread before server run
- Manual commands: `bm update` (force check + apply) and `bm update --check` (check only)
- New config fields: `auto_update`, `update_check_interval`, `auto_update_last_checked_at`
## 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
+30
View File
@@ -75,6 +75,36 @@ uv tool install basic-memory
You can view shared context via files in `~/basic-memory` (default directory location).
## Automatic Updates
Basic Memory includes a default-on auto-update flow for CLI installs.
- **Auto-install supported:** `uv tool` and Homebrew installs
- **Default check interval:** every 24 hours (`86400` seconds)
- **MCP-safe behavior:** update checks run silently in `basic-memory mcp` mode
- **`uvx` behavior:** skipped (runtime is ephemeral and managed by `uvx`)
Manual update commands:
```bash
# Check now and install if supported
bm update
# Check only, do not install
bm update --check
```
Config options in `~/.basic-memory/config.json`:
```json
{
"auto_update": true,
"update_check_interval": 86400
}
```
To disable automatic updates, set `"auto_update": false`.
## Why Basic Memory?
Most LLM interactions are ephemeral - you ask a question, get an answer, and everything is forgotten. Each conversation
+17 -1
View File
@@ -54,6 +54,22 @@ Or for a one-time sync:
basic-memory sync
```
### 4. Updating Basic Memory
Basic Memory supports automatic updates by default for `uv tool` and Homebrew installs.
For manual checks and upgrades:
```bash
# Check now and install if supported
bm update
# Check only, do not install
bm update --check
```
To disable automatic updates, set `"auto_update": false` in `~/.basic-memory/config.json`.
## Configuration Options
### Custom Directory
@@ -125,4 +141,4 @@ If you encounter issues:
cat ~/.basic-memory/basic-memory.log
```
For more detailed information, refer to the [full documentation](https://memory.basicmachines.co/).
For more detailed information, refer to the [full documentation](https://docs.basicmemory.com/).
+2 -2
View File
@@ -6,12 +6,12 @@
"url": "https://github.com/basicmachines-co/basic-memory.git",
"source": "github"
},
"version": "0.19.1",
"version": "0.20.0",
"packages": [
{
"registryType": "pypi",
"identifier": "basic-memory",
"version": "0.19.1",
"version": "0.20.0",
"runtimeHint": "uvx",
"runtimeArguments": [
{"type": "positional", "value": "basic-memory"},
+1 -1
View File
@@ -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.20.0"
# API version for FastAPI - independent of package version
__api_version__ = "v0"
+10 -4
View File
@@ -8,6 +8,7 @@ from typing import Optional # noqa: E402
import typer # noqa: E402
from basic_memory.cli.auto_update import maybe_run_periodic_auto_update # noqa: E402
from basic_memory.cli.container import CliContainer, set_container # noqa: E402
from basic_memory.cli.promo import maybe_show_cloud_promo, maybe_show_init_line # noqa: E402
from basic_memory.config import init_cli_logging # noqa: E402
@@ -52,10 +53,14 @@ def app_callback(
# Outcome: one-time plain line printed before the subcommand runs.
maybe_show_init_line(ctx.invoked_subcommand)
# Trigger: register promo as a post-command callback.
# Why: promo output should appear after the command's own output, not before.
# Outcome: promo panel renders below the command results (status tree, table, etc.).
ctx.call_on_close(lambda: maybe_show_cloud_promo(ctx.invoked_subcommand))
# Trigger: register post-command messaging callbacks.
# Why: informational/promo/update output belongs below command results.
# Outcome: command output remains primary, with optional follow-up notices afterwards.
def _post_command_messages() -> None:
maybe_show_cloud_promo(ctx.invoked_subcommand)
maybe_run_periodic_auto_update(ctx.invoked_subcommand)
ctx.call_on_close(_post_command_messages)
# Run initialization for commands that don't use the API
# Skip for 'mcp' command - it has its own lifespan that handles initialization
@@ -70,6 +75,7 @@ def app_callback(
"tool",
"reset",
"reindex",
"update",
"watch",
}
if (
+391
View File
@@ -0,0 +1,391 @@
"""Automatic update checks and upgrades for the Basic Memory CLI."""
from __future__ import annotations
import json
import subprocess
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from loguru import logger
from packaging.version import InvalidVersion, Version
from rich.console import Console
import basic_memory
from basic_memory.config import ConfigManager
PACKAGE_NAME = "basic-memory"
PYPI_JSON_URL = "https://pypi.org/pypi/basic-memory/json"
PYPI_TIMEOUT_SECONDS = 5
BREW_OUTDATED_TIMEOUT_SECONDS = 15
UV_UPGRADE_TIMEOUT_SECONDS = 180
BREW_UPGRADE_TIMEOUT_SECONDS = 600
class InstallSource(str, Enum):
"""How the running CLI appears to have been installed."""
HOMEBREW = "homebrew"
UV_TOOL = "uv_tool"
UVX = "uvx"
UNKNOWN = "unknown"
class AutoUpdateStatus(str, Enum):
"""Result classification for update checks and installs."""
SKIPPED = "skipped"
UP_TO_DATE = "up_to_date"
UPDATE_AVAILABLE = "update_available"
UPDATED = "updated"
FAILED = "failed"
@dataclass(frozen=True)
class AutoUpdateResult:
"""Structured result for update checks/install attempts."""
status: AutoUpdateStatus
source: InstallSource
checked: bool
update_available: bool
updated: bool
latest_version: str | None = None
message: str | None = None
error: str | None = None
restart_recommended: bool = False
def detect_install_source(executable: str | None = None) -> InstallSource:
"""Infer installation source from the active interpreter path."""
active_executable = executable or sys.executable
normalized = active_executable.lower().replace("\\", "/")
if "cellar/basic-memory" in normalized:
return InstallSource.HOMEBREW
if "uv/tools/basic-memory" in normalized:
return InstallSource.UV_TOOL
if "/uv/archive-" in normalized:
return InstallSource.UVX
return InstallSource.UNKNOWN
def _is_interactive_session() -> bool:
"""Return whether stdin/stdout are interactive terminals."""
try:
return sys.stdin.isatty() and sys.stdout.isatty()
except ValueError:
# Trigger: stdin/stdout may be closed during transport teardown.
# Why: isatty() raises ValueError on closed descriptors.
# Outcome: treat as non-interactive and suppress periodic output.
return False
def _run_subprocess(
command: list[str],
*,
timeout_seconds: int,
silent: bool,
capture_output: bool,
) -> subprocess.CompletedProcess[str]:
"""Run a subprocess with explicit stdio behavior for protocol safety."""
# Trigger: silent operation (MCP/background) with no need for subprocess output.
# Why: prevent protocol/terminal pollution from child process output.
# Outcome: stdout/stderr are discarded unless explicit capture is requested.
use_devnull = silent and not capture_output
stdout_target = subprocess.DEVNULL if use_devnull else subprocess.PIPE
stderr_target = subprocess.DEVNULL if use_devnull else subprocess.PIPE
return subprocess.run(
command,
stdin=subprocess.DEVNULL,
stdout=stdout_target,
stderr=stderr_target,
text=True,
timeout=timeout_seconds,
check=False,
)
def _version_from_pypi() -> str:
"""Fetch the latest published package version from PyPI."""
request = urllib.request.Request(
PYPI_JSON_URL,
headers={"User-Agent": f"basic-memory-cli/{basic_memory.__version__}"},
)
with urllib.request.urlopen(request, timeout=PYPI_TIMEOUT_SECONDS) as response:
payload = json.loads(response.read().decode("utf-8"))
latest = payload.get("info", {}).get("version")
if not latest:
raise RuntimeError("PyPI JSON response did not include info.version")
return str(latest)
def _check_homebrew_update_available(silent: bool) -> tuple[bool, str | None]:
"""Check whether Homebrew reports an outdated basic-memory formula."""
result = _run_subprocess(
["brew", "outdated", "--quiet", PACKAGE_NAME],
timeout_seconds=BREW_OUTDATED_TIMEOUT_SECONDS,
silent=silent,
capture_output=True,
)
if result.returncode != 0:
stderr = (result.stderr or "").strip()
stdout = (result.stdout or "").strip()
detail = stderr or stdout or "brew outdated failed"
raise RuntimeError(detail)
is_outdated = bool((result.stdout or "").strip())
return is_outdated, None
def _check_pypi_update_available() -> tuple[bool, str]:
"""Compare installed package version with PyPI latest version."""
latest = _version_from_pypi()
try:
current_version = Version(basic_memory.__version__)
latest_version = Version(latest)
except InvalidVersion as exc:
raise RuntimeError(
f"Could not compare versions (current={basic_memory.__version__}, latest={latest})"
) from exc
return latest_version > current_version, latest
def _manual_update_hint(source: InstallSource) -> str:
"""Return manager-appropriate manual update instructions."""
if source == InstallSource.UV_TOOL:
return "Run `uv tool upgrade basic-memory`."
if source == InstallSource.HOMEBREW:
return "Run `brew upgrade basic-memory`."
return (
"Automatic install is not supported for this environment. "
"Update with your package manager (for pip: `python3 -m pip install -U basic-memory`)."
)
def _save_last_checked_timestamp(config_manager: ConfigManager, checked_at: datetime) -> None:
"""Persist the timestamp for the most recent attempted update check."""
config = config_manager.load_config()
config.auto_update_last_checked_at = checked_at
config_manager.save_config(config)
def run_auto_update(
*,
force: bool = False,
check_only: bool = False,
silent: bool = False,
config_manager: ConfigManager | None = None,
now: datetime | None = None,
executable: str | None = None,
) -> AutoUpdateResult:
"""Run update check/install flow and return a structured result."""
manager = config_manager or ConfigManager()
config = manager.load_config()
source = detect_install_source(executable)
checked_at = now or datetime.now()
if source == InstallSource.UVX:
return AutoUpdateResult(
status=AutoUpdateStatus.SKIPPED,
source=source,
checked=False,
update_available=False,
updated=False,
message="uvx runtime detected; updates are managed by uvx cache resolution.",
)
if not force and not config.auto_update:
return AutoUpdateResult(
status=AutoUpdateStatus.SKIPPED,
source=source,
checked=False,
update_available=False,
updated=False,
message="Auto-update is disabled in config.",
)
if not force and config.auto_update_last_checked_at is not None:
try:
elapsed = checked_at - config.auto_update_last_checked_at
except TypeError:
# Trigger: mixed naive/aware datetimes from manual config edits.
# Why: datetime subtraction fails for mixed tz-awareness.
# Outcome: ignore the gate once and continue with a forced check path.
logger.warning("Auto-update interval gate skipped due to incompatible timestamp format")
else:
if elapsed < timedelta(seconds=config.update_check_interval):
return AutoUpdateResult(
status=AutoUpdateStatus.SKIPPED,
source=source,
checked=False,
update_available=False,
updated=False,
message="Update check interval has not elapsed.",
)
try:
# --- Availability check ---
latest_version: str | None = None
if source == InstallSource.HOMEBREW:
update_available, latest_version = _check_homebrew_update_available(silent=silent)
else:
update_available, latest_version = _check_pypi_update_available()
if not update_available:
return AutoUpdateResult(
status=AutoUpdateStatus.UP_TO_DATE,
source=source,
checked=True,
update_available=False,
updated=False,
latest_version=latest_version,
message=f"Basic Memory is up to date ({basic_memory.__version__}).",
)
if check_only:
return AutoUpdateResult(
status=AutoUpdateStatus.UPDATE_AVAILABLE,
source=source,
checked=True,
update_available=True,
updated=False,
latest_version=latest_version,
message=(
f"Update available (latest: {latest_version or 'unknown'}). "
f"{_manual_update_hint(source)}"
),
)
if source == InstallSource.UNKNOWN:
return AutoUpdateResult(
status=AutoUpdateStatus.UPDATE_AVAILABLE,
source=source,
checked=True,
update_available=True,
updated=False,
latest_version=latest_version,
message=(
f"Update available (latest: {latest_version or 'unknown'}). "
f"{_manual_update_hint(source)}"
),
)
# --- Automatic install ---
command = (
["uv", "tool", "upgrade", PACKAGE_NAME]
if source == InstallSource.UV_TOOL
else ["brew", "upgrade", PACKAGE_NAME]
)
timeout = (
UV_UPGRADE_TIMEOUT_SECONDS
if source == InstallSource.UV_TOOL
else BREW_UPGRADE_TIMEOUT_SECONDS
)
install_result = _run_subprocess(
command,
timeout_seconds=timeout,
silent=silent,
capture_output=not silent,
)
if install_result.returncode != 0:
stderr = (install_result.stderr or "").strip() if install_result.stderr else ""
stdout = (install_result.stdout or "").strip() if install_result.stdout else ""
detail = stderr or stdout or "update command failed"
return AutoUpdateResult(
status=AutoUpdateStatus.FAILED,
source=source,
checked=True,
update_available=True,
updated=False,
latest_version=latest_version,
message="Automatic update failed.",
error=detail,
)
return AutoUpdateResult(
status=AutoUpdateStatus.UPDATED,
source=source,
checked=True,
update_available=True,
updated=True,
latest_version=latest_version,
message=(
"Basic Memory was updated successfully. "
"Restart running sessions to use the new version."
),
restart_recommended=True,
)
except (
RuntimeError,
urllib.error.URLError,
ValueError,
TimeoutError,
subprocess.SubprocessError,
OSError,
) as exc:
logger.warning(f"Auto-update check failed: {exc}")
return AutoUpdateResult(
status=AutoUpdateStatus.FAILED,
source=source,
checked=True,
update_available=False,
updated=False,
message="Automatic update check failed.",
error=str(exc),
)
finally:
# Trigger: we attempted a check path (including failures).
# Why: repeated failing checks on every command create noise and unnecessary network load.
# Outcome: next periodic check is gated by update_check_interval.
try:
_save_last_checked_timestamp(manager, checked_at)
except Exception as exc: # pragma: no cover
logger.warning(f"Failed to persist auto-update timestamp: {exc}")
def maybe_run_periodic_auto_update(
invoked_subcommand: str | None,
*,
config_manager: ConfigManager | None = None,
is_interactive: bool | None = None,
console: Console | None = None,
) -> AutoUpdateResult | None:
"""Run a periodic auto-update check for interactive CLI sessions."""
interactive = _is_interactive_session() if is_interactive is None else is_interactive
if not interactive:
return None
if invoked_subcommand in {None, "mcp", "update"}:
return None
result = run_auto_update(
force=False,
check_only=False,
silent=False,
config_manager=config_manager,
)
if result.status in {
AutoUpdateStatus.UPDATE_AVAILABLE,
AutoUpdateStatus.UPDATED,
AutoUpdateStatus.FAILED,
}:
out = console or Console()
if result.status == AutoUpdateStatus.UPDATED:
out.print(f"[green]{result.message}[/green]")
elif result.status == AutoUpdateStatus.FAILED:
error_detail = f" {result.error}" if result.error else ""
out.print(f"[yellow]{result.message}{error_detail}[/yellow]")
elif result.message:
out.print(f"[cyan]{result.message}[/cyan]")
return result
@@ -8,6 +8,7 @@ from . import (
project,
format,
schema,
update,
)
__all__ = [
@@ -23,4 +24,5 @@ __all__ = [
"project",
"format",
"schema",
"update",
]
+18
View File
@@ -1,12 +1,14 @@
"""MCP server command with streamable HTTP transport."""
import os
import threading
from typing import Any, Optional
import typer
from loguru import logger
from basic_memory.cli.app import app
from basic_memory.cli.auto_update import AutoUpdateStatus, run_auto_update
from basic_memory.config import ConfigManager, init_mcp_logging
@@ -80,6 +82,22 @@ def mcp(
os.environ["BASIC_MEMORY_MCP_PROJECT"] = project_name
logger.info(f"MCP server constrained to project: {project_name}")
def _run_background_auto_update() -> None:
result = run_auto_update(force=False, check_only=False, silent=True)
if result.restart_recommended:
logger.info(
"A newer Basic Memory version was installed and will apply on next restart."
)
elif result.status == AutoUpdateStatus.FAILED and result.error:
logger.warning(f"MCP background auto-update failed: {result.error}")
# Trigger: stdio transport corresponds to local user installs.
# Why: server transports (HTTP/SSE) run in managed environments where
# package-manager self-upgrades are inappropriate.
# Outcome: background auto-update runs only for local stdio MCP sessions.
if transport == "stdio":
threading.Thread(target=_run_background_auto_update, daemon=True).start()
# Run the MCP server (blocks)
# Lifespan handles: initialization, migrations, file sync, cleanup
logger.info(f"Starting MCP server with {transport.upper()} transport")
+40
View File
@@ -0,0 +1,40 @@
"""Manual update command for Basic Memory CLI."""
import typer
from rich.console import Console
from basic_memory.cli.app import app
from basic_memory.cli.auto_update import AutoUpdateStatus, run_auto_update
console = Console()
@app.command("update")
def update(
check: bool = typer.Option(
False,
"--check",
help="Check for updates only (do not install).",
),
) -> None:
"""Check for updates and install when supported."""
result = run_auto_update(force=True, check_only=check, silent=False)
if result.status == AutoUpdateStatus.FAILED:
detail = f" {result.error}" if result.error else ""
console.print(f"[red]{result.message or 'Update failed.'}{detail}[/red]")
raise typer.Exit(1)
if result.status == AutoUpdateStatus.UPDATED:
console.print(f"[green]{result.message or 'Basic Memory updated successfully.'}[/green]")
return
if result.status == AutoUpdateStatus.UP_TO_DATE:
console.print(f"[green]{result.message or 'Basic Memory is up to date.'}[/green]")
return
if result.status == AutoUpdateStatus.UPDATE_AVAILABLE:
console.print(f"[cyan]{result.message or 'Update available.'}[/cyan]")
return
console.print(f"[dim]{result.message or 'No update action was performed.'}[/dim]")
+1
View File
@@ -28,6 +28,7 @@ if not _version_only_invocation(sys.argv[1:]):
schema,
status,
tool,
update,
)
warnings.filterwarnings("ignore") # pragma: no cover
+16
View File
@@ -351,6 +351,22 @@ class BasicMemoryConfig(BaseSettings):
description="Most recent cloud promo version shown in CLI.",
)
auto_update: bool = Field(
default=True,
description="Enable automatic CLI update checks and installs when supported.",
)
update_check_interval: int = Field(
default=86400,
description="Seconds between automatic update checks.",
gt=0,
)
auto_update_last_checked_at: Optional[datetime] = Field(
default=None,
description="Timestamp of the last attempted automatic update check.",
)
cloud_api_key: Optional[str] = Field(
default=None,
description="API key for cloud access (bmc_ prefixed). Account-level, not per-project.",
+5 -3
View File
@@ -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,
+12 -2
View File
@@ -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,
+4 -3
View File
@@ -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,
@@ -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 ---
+72 -46
View File
@@ -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
+63 -3
View File
@@ -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.
@@ -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
+374
View File
@@ -0,0 +1,374 @@
"""Tests for CLI auto-update behavior."""
from __future__ import annotations
import subprocess
from datetime import datetime, timedelta, timezone
from io import StringIO
from rich.console import Console
from basic_memory.cli.auto_update import (
AutoUpdateResult,
AutoUpdateStatus,
InstallSource,
_is_interactive_session,
detect_install_source,
maybe_run_periodic_auto_update,
run_auto_update,
)
from basic_memory.config import BasicMemoryConfig
class StubConfigManager:
"""Simple in-memory ConfigManager stub for updater tests."""
def __init__(self, config: BasicMemoryConfig):
self._config = config
self.save_calls = 0
def load_config(self) -> BasicMemoryConfig:
return self._config
def save_config(self, config: BasicMemoryConfig) -> None:
self._config = config
self.save_calls += 1
def _capture_console() -> tuple[Console, StringIO]:
"""Create a Console that writes to an in-memory buffer."""
buf = StringIO()
return Console(file=buf, force_terminal=True), buf
def _base_config(tmp_path) -> BasicMemoryConfig:
return BasicMemoryConfig(projects={"main": {"path": str(tmp_path / "main")}})
def _result(
status: AutoUpdateStatus,
*,
message: str | None,
error: str | None = None,
) -> AutoUpdateResult:
return AutoUpdateResult(
status=status,
source=InstallSource.UV_TOOL,
checked=True,
update_available=status in {AutoUpdateStatus.UPDATE_AVAILABLE, AutoUpdateStatus.UPDATED},
updated=status == AutoUpdateStatus.UPDATED,
latest_version="9.9.9",
message=message,
error=error,
restart_recommended=status == AutoUpdateStatus.UPDATED,
)
def test_detect_install_source_variants():
assert (
detect_install_source("/opt/homebrew/Cellar/basic-memory/0.18.0/bin/python")
== InstallSource.HOMEBREW
)
assert (
detect_install_source("/Users/me/.local/share/uv/tools/basic-memory/bin/python")
== InstallSource.UV_TOOL
)
assert (
detect_install_source("/Users/me/.cache/uv/archive-v0/abc123/bin/python")
== InstallSource.UVX
)
assert (
detect_install_source("/Users/me/Library/Caches/uv/archive-v0/abc123/bin/python")
== InstallSource.UVX
)
assert detect_install_source("/usr/local/bin/python3") == InstallSource.UNKNOWN
def test_interval_gate_skips_check_when_recent(tmp_path):
config = _base_config(tmp_path)
config.auto_update_last_checked_at = datetime.now() - timedelta(seconds=30)
config.update_check_interval = 3600
manager = StubConfigManager(config)
result = run_auto_update(config_manager=manager)
assert result.status == AutoUpdateStatus.SKIPPED
assert result.checked is False
assert manager.save_calls == 0
def test_auto_update_disabled_skips_periodic(tmp_path):
config = _base_config(tmp_path)
config.auto_update = False
manager = StubConfigManager(config)
result = run_auto_update(config_manager=manager)
assert result.status == AutoUpdateStatus.SKIPPED
assert result.checked is False
def test_force_bypasses_auto_update_disabled(monkeypatch, tmp_path):
config = _base_config(tmp_path)
config.auto_update = False
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (False, "0.0.0"),
)
result = run_auto_update(
force=True,
config_manager=manager,
executable="/Users/me/.local/share/uv/tools/basic-memory/bin/python",
)
assert result.status == AutoUpdateStatus.UP_TO_DATE
assert result.checked is True
assert manager.save_calls == 1
def test_homebrew_outdated_triggers_upgrade(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_homebrew_update_available",
lambda silent: (True, None),
)
calls: list[list[str]] = []
def _fake_run_subprocess(command, **kwargs):
calls.append(command)
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _fake_run_subprocess)
result = run_auto_update(
config_manager=manager,
executable="/opt/homebrew/Cellar/basic-memory/0.18.0/bin/python",
)
assert result.status == AutoUpdateStatus.UPDATED
assert calls == [["brew", "upgrade", "basic-memory"]]
def test_uv_tool_pypi_check_triggers_upgrade(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (True, "9.9.9"),
)
calls: list[list[str]] = []
def _fake_run_subprocess(command, **kwargs):
calls.append(command)
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _fake_run_subprocess)
result = run_auto_update(
config_manager=manager,
executable="/Users/me/.local/share/uv/tools/basic-memory/bin/python",
)
assert result.status == AutoUpdateStatus.UPDATED
assert result.latest_version == "9.9.9"
assert calls == [["uv", "tool", "upgrade", "basic-memory"]]
def test_unknown_manager_returns_manual_update_guidance(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (True, "9.9.9"),
)
result = run_auto_update(
force=True,
config_manager=manager,
executable="/usr/local/bin/python3",
)
assert result.status == AutoUpdateStatus.UPDATE_AVAILABLE
assert result.updated is False
assert "Automatic install is not supported" in (result.message or "")
def test_uvx_runtime_is_skipped(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
result = run_auto_update(
config_manager=manager,
executable="/Users/me/.cache/uv/archive-v0/abc123/bin/python",
)
assert result.status == AutoUpdateStatus.SKIPPED
assert result.source == InstallSource.UVX
assert result.checked is False
assert manager.save_calls == 0
def test_mcp_silent_mode_suppresses_subprocess_output(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (True, "9.9.9"),
)
captured_kwargs: list[dict] = []
def _fake_run_subprocess(command, **kwargs):
captured_kwargs.append(kwargs)
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _fake_run_subprocess)
result = run_auto_update(
config_manager=manager,
executable="/Users/me/.local/share/uv/tools/basic-memory/bin/python",
silent=True,
)
assert result.status == AutoUpdateStatus.UPDATED
assert captured_kwargs
assert captured_kwargs[0]["silent"] is True
assert captured_kwargs[0]["capture_output"] is False
def test_subprocess_oserror_is_non_fatal(monkeypatch, tmp_path):
config = _base_config(tmp_path)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (True, "9.9.9"),
)
def _raise_oserror(command, **kwargs):
raise FileNotFoundError(command[0])
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _raise_oserror)
result = run_auto_update(
config_manager=manager,
executable="/Users/me/.local/share/uv/tools/basic-memory/bin/python",
)
assert result.status == AutoUpdateStatus.FAILED
assert result.checked is True
def test_mixed_timezone_timestamp_does_not_crash_interval_gate(monkeypatch, tmp_path):
config = _base_config(tmp_path)
config.auto_update_last_checked_at = datetime.now(timezone.utc)
manager = StubConfigManager(config)
monkeypatch.setattr(
"basic_memory.cli.auto_update._check_pypi_update_available",
lambda: (False, "0.0.0"),
)
result = run_auto_update(
config_manager=manager,
executable="/Users/me/.local/share/uv/tools/basic-memory/bin/python",
)
assert result.status == AutoUpdateStatus.UP_TO_DATE
assert result.checked is True
def test_maybe_run_periodic_auto_update_non_interactive_has_no_console_output():
console, buf = _capture_console()
result = maybe_run_periodic_auto_update(
"status",
is_interactive=False,
console=console,
)
assert result is None
assert buf.getvalue() == ""
def test_maybe_run_periodic_auto_update_prints_updated(monkeypatch):
console, buf = _capture_console()
monkeypatch.setattr(
"basic_memory.cli.auto_update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UPDATED,
message="Basic Memory was updated successfully.",
),
)
result = maybe_run_periodic_auto_update("status", is_interactive=True, console=console)
assert result is not None
assert result.status == AutoUpdateStatus.UPDATED
assert "updated successfully" in buf.getvalue().lower()
def test_maybe_run_periodic_auto_update_prints_available(monkeypatch):
console, buf = _capture_console()
monkeypatch.setattr(
"basic_memory.cli.auto_update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UPDATE_AVAILABLE,
message="Update available (latest: 9.9.9).",
),
)
result = maybe_run_periodic_auto_update("status", is_interactive=True, console=console)
assert result is not None
assert result.status == AutoUpdateStatus.UPDATE_AVAILABLE
assert "update available" in buf.getvalue().lower()
def test_maybe_run_periodic_auto_update_prints_failed_with_error(monkeypatch):
console, buf = _capture_console()
monkeypatch.setattr(
"basic_memory.cli.auto_update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.FAILED,
message="Automatic update check failed.",
error="network timeout",
),
)
result = maybe_run_periodic_auto_update("status", is_interactive=True, console=console)
assert result is not None
assert result.status == AutoUpdateStatus.FAILED
output = buf.getvalue().lower()
assert "automatic update check failed" in output
assert "network timeout" in output
def test_maybe_run_periodic_auto_update_uses_interactive_probe_when_not_overridden(monkeypatch):
console, buf = _capture_console()
monkeypatch.setattr("basic_memory.cli.auto_update._is_interactive_session", lambda: True)
monkeypatch.setattr(
"basic_memory.cli.auto_update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UP_TO_DATE,
message="Basic Memory is up to date.",
),
)
result = maybe_run_periodic_auto_update("status", console=console)
assert result is not None
assert result.status == AutoUpdateStatus.UP_TO_DATE
# UP_TO_DATE is intentionally silent for periodic checks.
assert buf.getvalue() == ""
def test_is_interactive_session_handles_closed_stdio(monkeypatch):
class _BrokenStream:
def isatty(self) -> bool:
raise ValueError("I/O operation on closed file")
monkeypatch.setattr("basic_memory.cli.auto_update.sys.stdin", _BrokenStream())
monkeypatch.setattr("basic_memory.cli.auto_update.sys.stdout", _BrokenStream())
assert _is_interactive_session() is False
+90
View File
@@ -0,0 +1,90 @@
"""Tests for `bm update` command."""
from typer.testing import CliRunner
from basic_memory.cli.app import app
from basic_memory.cli.auto_update import AutoUpdateResult, AutoUpdateStatus, InstallSource
def _result(
status: AutoUpdateStatus,
*,
message: str | None,
error: str | None = None,
) -> AutoUpdateResult:
return AutoUpdateResult(
status=status,
source=InstallSource.UV_TOOL,
checked=True,
update_available=status in {AutoUpdateStatus.UPDATE_AVAILABLE, AutoUpdateStatus.UPDATED},
updated=status == AutoUpdateStatus.UPDATED,
latest_version="9.9.9",
message=message,
error=error,
restart_recommended=status == AutoUpdateStatus.UPDATED,
)
def test_update_command_applies_upgrade(monkeypatch):
runner = CliRunner()
monkeypatch.setattr(
"basic_memory.cli.commands.update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UPDATED,
message="Basic Memory was updated successfully.",
),
)
result = runner.invoke(app, ["update"])
assert result.exit_code == 0
assert "updated successfully" in result.stdout.lower()
def test_update_command_check_only_shows_available(monkeypatch):
runner = CliRunner()
monkeypatch.setattr(
"basic_memory.cli.commands.update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UPDATE_AVAILABLE,
message="Update available (latest: 9.9.9). Run `uv tool upgrade basic-memory`.",
),
)
result = runner.invoke(app, ["update", "--check"])
assert result.exit_code == 0
assert "update available" in result.stdout.lower()
def test_update_command_reports_up_to_date(monkeypatch):
runner = CliRunner()
monkeypatch.setattr(
"basic_memory.cli.commands.update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.UP_TO_DATE,
message="Basic Memory is up to date.",
),
)
result = runner.invoke(app, ["update"])
assert result.exit_code == 0
assert "up to date" in result.stdout.lower()
def test_update_command_failure_exits_nonzero(monkeypatch):
runner = CliRunner()
monkeypatch.setattr(
"basic_memory.cli.commands.update.run_auto_update",
lambda **kwargs: _result(
AutoUpdateStatus.FAILED,
message="Automatic update failed.",
error="network timeout",
),
)
result = runner.invoke(app, ["update"])
assert result.exit_code == 1
assert "automatic update failed" in result.stdout.lower()
@@ -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 ---
@@ -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."""
+58
View File
@@ -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
+44
View File
@@ -1237,3 +1237,47 @@ class TestLocalSyncPathMigration:
assert result["projects"]["local-proj"]["path"] == local_path
assert result["projects"]["cloud-only"]["path"] == "cloud-only"
assert result["projects"]["cloud-bisync"]["path"] == bisync_path
class TestAutoUpdateConfig:
"""Test auto-update configuration fields."""
def test_auto_update_defaults(self):
"""Auto-update should default on with a daily check interval."""
config = BasicMemoryConfig()
assert config.auto_update is True
assert config.update_check_interval == 86400
assert config.auto_update_last_checked_at is None
def test_auto_update_env_overrides(self, monkeypatch):
"""Environment variables should override auto-update defaults."""
monkeypatch.setenv("BASIC_MEMORY_AUTO_UPDATE", "false")
monkeypatch.setenv("BASIC_MEMORY_UPDATE_CHECK_INTERVAL", "3600")
config = BasicMemoryConfig()
assert config.auto_update is False
assert config.update_check_interval == 3600
def test_auto_update_round_trip_persistence(self):
"""Auto-update values should survive save/load cycle."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
config_manager = ConfigManager()
config_manager.config_dir = temp_path / "basic-memory"
config_manager.config_file = config_manager.config_dir / "config.json"
config_manager.config_dir.mkdir(parents=True, exist_ok=True)
checked_at = datetime.now()
test_config = BasicMemoryConfig(
projects={"main": {"path": str(temp_path / "main")}},
auto_update=False,
update_check_interval=7200,
auto_update_last_checked_at=checked_at,
)
config_manager.save_config(test_config)
loaded = config_manager.load_config()
assert loaded.auto_update is False
assert loaded.update_check_interval == 7200
assert loaded.auto_update_last_checked_at == checked_at
+162
View File
@@ -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)