Compare commits

..

11 Commits

Author SHA1 Message Date
Drew Cain 148e07c580 chore: update version to 0.20.1 for v0.20.1 release 2026-03-10 23:06:21 -05:00
Drew Cain 21334cc29b docs: add v0.20.1 changelog entry
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-03-10 23:06:15 -05:00
Drew Cain db60942267 fix(core): invalidate config cache when file is modified by another process (#662)
Signed-off-by: Drew Cain <groksrc@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-10 23:05:55 -05:00
Drew Cain 7bfac158df fix(cli): project list MCP column shows transport type instead of DB presence (#661)
Signed-off-by: Drew Cain <groksrc@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-10 23:05:48 -05:00
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
37 changed files with 1953 additions and 84 deletions
+38
View File
@@ -2,6 +2,44 @@
## Unreleased
## v0.20.1 (2026-03-10)
### Bug Fixes
- **#661**: Fix `bm project list` MCP column to show transport type (stdio/https) instead of DB presence
- Renamed "MCP (stdio)" column to "MCP"
- Shows actual routing mode: `stdio` for local, `https` for cloud projects
- Clears local path display for cloud-mode projects
- **#662**: Invalidate config cache when file is modified by another process
- Adds mtime-based cache validation to `ConfigManager.load_config()`
- Long-lived processes (MCP stdio server) now detect external config changes
- Fixes `bm project set-cloud` having no effect on running MCP server
## 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.1",
"packages": [
{
"registryType": "pypi",
"identifier": "basic-memory",
"version": "0.19.1",
"version": "0.20.1",
"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.1"
# 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")
+14 -3
View File
@@ -128,7 +128,7 @@ def list_projects(
table.add_column("Cloud Path", style="green")
table.add_column("Workspace", style="green")
table.add_column("CLI Route", style="blue")
table.add_column("MCP (stdio)", style="blue")
table.add_column("MCP", style="blue")
table.add_column("Sync", style="green")
table.add_column("Default", style="magenta")
@@ -164,6 +164,11 @@ def list_projects(
elif entry and entry.mode == ProjectMode.LOCAL and entry.path:
local_path = format_path(normalize_project_path(entry.path))
# Clear local path for cloud-mode projects — only local projects
# should display a local path
if entry and entry.mode == ProjectMode.CLOUD:
local_path = ""
cloud_path = ""
if cloud_project is not None:
cloud_path = normalize_project_path(cloud_project.path)
@@ -182,7 +187,13 @@ def list_projects(
is_default = config.default_project == project_name
has_sync = bool(entry and entry.local_sync_path)
mcp_stdio_target = "local" if local_project is not None else "n/a"
# Determine MCP transport based on project routing mode
if entry and entry.mode == ProjectMode.CLOUD:
mcp_transport = "https"
elif entry is None and cloud_project is not None:
mcp_transport = "https"
else:
mcp_transport = "stdio"
# Show workspace name (type) for cloud-sourced projects
ws_label = ""
@@ -195,7 +206,7 @@ def list_projects(
"local_path": local_path,
"cloud_path": cloud_path,
"cli_route": cli_route,
"mcp_stdio": mcp_stdio_target,
"mcp_stdio": mcp_transport,
"sync": has_sync,
"is_default": is_default,
}
+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
+63 -5
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.",
@@ -629,6 +645,12 @@ class BasicMemoryConfig(BaseSettings):
# Module-level cache for configuration
_CONFIG_CACHE: Optional[BasicMemoryConfig] = None
# Track config file mtime+size so cross-process changes (e.g. `bm project set-cloud`
# in a separate terminal) invalidate the cache in long-lived processes like the
# MCP stdio server. Using both mtime and size guards against coarse-granularity
# filesystems where two writes within the same second share the same mtime.
_CONFIG_MTIME: Optional[float] = None
_CONFIG_SIZE: Optional[int] = None
class ConfigManager:
@@ -662,13 +684,38 @@ class ConfigManager:
Environment variables take precedence over file config values,
following Pydantic Settings best practices.
Uses module-level cache for performance across ConfigManager instances.
Uses module-level cache with file mtime validation so that
cross-process config changes (e.g. `bm project set-cloud` in a
separate terminal) are picked up by long-lived processes like
the MCP stdio server.
"""
global _CONFIG_CACHE
global _CONFIG_CACHE, _CONFIG_MTIME, _CONFIG_SIZE
# Return cached config if available
# Trigger: cached config exists but the on-disk file may have been
# modified by another process (CLI command in a different terminal).
# Why: the MCP server is long-lived; without this check it would
# serve stale project routing forever.
# Outcome: cheap os.stat() per access; re-read only when mtime or size differs.
if _CONFIG_CACHE is not None:
return _CONFIG_CACHE
try:
st = self.config_file.stat()
current_mtime = st.st_mtime
current_size = st.st_size
except OSError:
current_mtime = None
current_size = None
if (
current_mtime is not None
and current_mtime == _CONFIG_MTIME
and current_size == _CONFIG_SIZE
):
return _CONFIG_CACHE
# mtime/size changed or file gone — invalidate and fall through to re-read
_CONFIG_CACHE = None
_CONFIG_MTIME = None
_CONFIG_SIZE = None
if self.config_file.exists():
try:
@@ -723,6 +770,15 @@ class ConfigManager:
_CONFIG_CACHE = BasicMemoryConfig(**merged_data)
# Record mtime+size so subsequent calls detect cross-process changes
try:
st = self.config_file.stat()
_CONFIG_MTIME = st.st_mtime
_CONFIG_SIZE = st.st_size
except OSError:
_CONFIG_MTIME = None
_CONFIG_SIZE = None
# Re-save to normalize legacy config into current format
if needs_resave:
# Create backup before overwriting so users can revert if needed
@@ -753,10 +809,12 @@ class ConfigManager:
def save_config(self, config: BasicMemoryConfig) -> None:
"""Save configuration to file and invalidate cache."""
global _CONFIG_CACHE
global _CONFIG_CACHE, _CONFIG_MTIME, _CONFIG_SIZE
save_basic_memory_config(self.config_file, config)
# Invalidate cache so next load_config() reads fresh data
_CONFIG_CACHE = None
_CONFIG_MTIME = None
_CONFIG_SIZE = None
@property
def projects(self) -> Dict[str, str]:
+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.
+2
View File
@@ -258,6 +258,8 @@ def config_manager(app_config: BasicMemoryConfig, config_home) -> ConfigManager:
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_manager = ConfigManager()
# Update its paths to use the test directory
@@ -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
+2
View File
@@ -25,6 +25,8 @@ def isolated_home(tmp_path, monkeypatch) -> Path:
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
monkeypatch.setenv("HOME", str(tmp_path))
if os.name == "nt":
+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
+2
View File
@@ -350,6 +350,8 @@ def write_config(tmp_path, monkeypatch):
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
@@ -27,6 +27,8 @@ def mock_config(tmp_path, monkeypatch):
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
+5 -2
View File
@@ -29,6 +29,8 @@ def write_config(tmp_path, monkeypatch):
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
@@ -122,15 +124,16 @@ def test_project_list_shows_local_cloud_presence_and_routes(
assert "Local Path" in result.stdout
assert "Cloud Path" in result.stdout
assert "CLI Route" in result.stdout
assert "MCP (stdio)" in result.stdout
assert "MCP" in result.stdout
lines = result.stdout.splitlines()
alpha_line = next(line for line in lines if "│ alpha" in line)
beta_line = next(line for line in lines if "│ beta" in line)
assert "local" in alpha_line # CLI route for alpha
assert "stdio" in alpha_line # Local projects use stdio transport
assert "cloud" in beta_line # CLI route for beta
assert "n/a" in beta_line # MCP stdio route is unavailable for cloud-only projects
assert "https" in beta_line # Cloud projects use HTTPS transport
assert "alpha-local" in result.stdout
assert "/alpha" in result.stdout
assert "/beta" in result.stdout
+24
View File
@@ -22,6 +22,8 @@ def mock_config(tmp_path, monkeypatch):
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
@@ -68,6 +70,8 @@ class TestSetCloud:
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
@@ -91,6 +95,8 @@ class TestSetCloud:
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_dir = tmp_path / ".basic-memory"
config_dir.mkdir(parents=True, exist_ok=True)
@@ -161,11 +167,15 @@ class TestSetLocal:
# Manually set workspace_id on the project
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
config_data = json.loads(mock_config.read_text())
config_data["projects"]["research"]["mode"] = "cloud"
config_data["projects"]["research"]["workspace_id"] = "11111111-1111-1111-1111-111111111111"
mock_config.write_text(json.dumps(config_data, indent=2))
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
# Set back to local
result = runner.invoke(app, ["project", "set-local", "research"])
@@ -173,6 +183,8 @@ class TestSetLocal:
# Verify workspace_id was cleared
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
updated_data = json.loads(mock_config.read_text())
assert updated_data["projects"]["research"]["workspace_id"] is None
assert updated_data["projects"]["research"]["mode"] == "local"
@@ -187,6 +199,8 @@ class TestSetCloudWithWorkspace:
from basic_memory.schemas.cloud import WorkspaceInfo
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
async def fake_get_available_workspaces():
return [
@@ -210,6 +224,8 @@ class TestSetCloudWithWorkspace:
# Verify workspace_id was persisted
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
updated_data = json.loads(mock_config.read_text())
assert (
updated_data["projects"]["research"]["workspace_id"]
@@ -222,6 +238,8 @@ class TestSetCloudWithWorkspace:
from basic_memory.schemas.cloud import WorkspaceInfo
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
async def fake_get_available_workspaces():
return [
@@ -249,17 +267,23 @@ class TestSetCloudWithWorkspace:
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
# Set default_workspace in config
config_data = json.loads(mock_config.read_text())
config_data["default_workspace"] = "global-default-tenant-id"
mock_config.write_text(json.dumps(config_data, indent=2))
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
result = runner.invoke(app, ["project", "set-cloud", "research"])
assert result.exit_code == 0
# Verify workspace_id was set from default
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
updated_data = json.loads(mock_config.read_text())
assert updated_data["projects"]["research"]["workspace_id"] == "global-default-tenant-id"
+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()
+4
View File
@@ -76,6 +76,8 @@ class TestWorkspaceSetDefault:
monkeypatch.setenv("HOME", str(temp_path))
monkeypatch.setenv("BASIC_MEMORY_CONFIG_DIR", str(config_dir))
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
config_manager = ConfigManager()
test_config = BasicMemoryConfig(
@@ -106,6 +108,8 @@ class TestWorkspaceSetDefault:
# Verify config was updated
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
config = ConfigManager().config
assert config.default_workspace == "11111111-1111-1111-1111-111111111111"
+2
View File
@@ -138,6 +138,8 @@ def config_manager(app_config: BasicMemoryConfig, config_home: Path, monkeypatch
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
# Create a new ConfigManager that uses the test home directory
config_manager = ConfigManager()
+8
View File
@@ -1257,6 +1257,11 @@ class TestWriteNoteOverwriteGuard:
# Set config to allow overwrites by default
app_config.write_note_overwrite_default = True
config_module._CONFIG_CACHE = app_config
# Pin mtime+size to the on-disk file so the cache guard sees a match
# and keeps our injected config instead of re-reading from disk.
_st = config_manager.config_file.stat()
config_module._CONFIG_MTIME = _st.st_mtime
config_module._CONFIG_SIZE = _st.st_size
try:
await write_note(
@@ -1281,6 +1286,9 @@ class TestWriteNoteOverwriteGuard:
# Restore config
app_config.write_note_overwrite_default = False
config_module._CONFIG_CACHE = app_config
_st = config_manager.config_file.stat()
config_module._CONFIG_MTIME = _st.st_mtime
config_module._CONFIG_SIZE = _st.st_size
@pytest.mark.asyncio
async def test_write_note_new_note_unaffected(self, app, test_project):
@@ -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 ---
+10
View File
@@ -778,6 +778,8 @@ async def test_add_project_with_project_root_sanitizes_paths(
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
test_cases = [
# (project_name, user_path, expected_sanitized_name)
@@ -845,6 +847,8 @@ async def test_add_project_with_project_root_rejects_escape_attempts(
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
# All of these should succeed by being sanitized to paths under project_root
# The sanitization removes dangerous patterns, so they don't escape
@@ -931,6 +935,8 @@ async def test_add_project_with_project_root_normalizes_case(
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
test_cases = [
# (input_path, expected_normalized_path)
@@ -985,6 +991,8 @@ async def test_add_project_with_project_root_detects_case_collisions(
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
# First, create a project with lowercase path
first_project = "documents-project"
@@ -1159,6 +1167,8 @@ async def test_add_project_nested_validation_with_project_root(
from basic_memory import config as config_module
config_module._CONFIG_CACHE = None
config_module._CONFIG_MTIME = None
config_module._CONFIG_SIZE = None
parent_project_name = f"cloud-parent-{os.urandom(4).hex()}"
child_project_name = f"cloud-child-{os.urandom(4).hex()}"
@@ -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
+170
View File
@@ -213,6 +213,8 @@ class TestBasicMemoryConfig:
}
config_manager.config_file.write_text(json.dumps(config_data, indent=2))
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
loaded = config_manager.load_config()
assert loaded.default_project == "research"
@@ -238,6 +240,8 @@ class TestBasicMemoryConfig:
}
config_manager.config_file.write_text(json.dumps(config_data, indent=2))
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
loaded = config_manager.load_config()
assert loaded.default_project == "work"
@@ -545,6 +549,8 @@ class TestConfigManager:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
# Should load successfully with migration to ProjectEntry
config = config_manager.load_config()
@@ -585,6 +591,8 @@ class TestConfigManager:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
config = config_manager.load_config()
@@ -617,6 +625,8 @@ class TestConfigManager:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
loaded = config_manager.load_config()
assert isinstance(loaded, BasicMemoryConfig)
@@ -647,6 +657,8 @@ class TestConfigManager:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
config_manager.load_config()
@@ -678,6 +690,8 @@ class TestConfigManager:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
config_manager.load_config()
@@ -1099,6 +1113,8 @@ class TestProjectMode:
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
# Should load successfully with migration
config = config_manager.load_config()
@@ -1174,6 +1190,116 @@ class TestProjectMode:
assert loaded.projects["main"].workspace_id is None
class TestConfigCacheMtimeInvalidation:
"""Test that config cache is invalidated when file is modified externally."""
def test_cache_returns_same_config_when_file_unchanged(self, config_home):
"""Verify cache hit when config file mtime has not changed."""
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
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)
test_config = BasicMemoryConfig(
projects={"main": {"path": str(temp_path / "main")}},
default_project="main",
)
config_manager.save_config(test_config)
# First load populates cache
config1 = config_manager.load_config()
assert config1.default_project == "main"
# Second load should return cached config (same object)
config2 = config_manager.load_config()
assert config1 is config2
def test_cache_invalidated_when_file_modified(self, config_home):
"""Verify cache miss when config file is modified by another process."""
import json
import os
import time
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
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)
test_config = BasicMemoryConfig(
projects={"main": {"path": str(temp_path / "main")}},
default_project="main",
)
config_manager.save_config(test_config)
# First load populates cache
config1 = config_manager.load_config()
assert config1.get_project_mode("main") == ProjectMode.LOCAL
# Simulate external process modifying the config file
config_data = json.loads(config_manager.config_file.read_text())
config_data["projects"]["main"]["mode"] = "cloud"
# Ensure mtime actually changes (some filesystems have 1s granularity)
time.sleep(0.05)
config_manager.config_file.write_text(json.dumps(config_data, indent=2))
# Force mtime change on filesystems with coarse granularity
new_mtime = os.path.getmtime(config_manager.config_file) + 1
os.utime(config_manager.config_file, (new_mtime, new_mtime))
# Next load should detect mtime change and re-read
config2 = config_manager.load_config()
assert config2.get_project_mode("main") == ProjectMode.CLOUD
assert config1 is not config2
def test_save_config_resets_mtime(self, config_home):
"""Verify save_config clears both cache and mtime."""
import basic_memory.config
basic_memory.config._CONFIG_CACHE = None
basic_memory.config._CONFIG_MTIME = None
basic_memory.config._CONFIG_SIZE = None
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)
test_config = BasicMemoryConfig(
projects={"main": {"path": str(temp_path / "main")}},
)
config_manager.save_config(test_config)
# Load to populate cache
config_manager.load_config()
assert basic_memory.config._CONFIG_CACHE is not None
assert basic_memory.config._CONFIG_MTIME is not None
assert basic_memory.config._CONFIG_SIZE is not None
# Save should clear all cache state
config_manager.save_config(test_config)
assert basic_memory.config._CONFIG_CACHE is None
assert basic_memory.config._CONFIG_MTIME is None
assert basic_memory.config._CONFIG_SIZE is None
class TestLocalSyncPathMigration:
"""Test migration that promotes local_sync_path into path for cloud projects."""
@@ -1237,3 +1363,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)