Compare commits

...

20 Commits

Author SHA1 Message Date
Drew Cain dd91b49054 chore: update version to 0.20.2 for v0.20.2 release 2026-03-10 23:13:59 -05:00
Drew Cain 7c96a0777d fix(cli): handle brew outdated exit code 1 as outdated, not error
Signed-off-by: Drew Cain <groksrc@gmail.com>
2026-03-10 23:13:54 -05:00
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
phernandez 222ec5d3b6 chore: update version to 0.19.1 for v0.19.1 release 2026-03-08 18:09:04 -05:00
phernandez d42aec7ea9 docs: add v0.19.1 changelog entry
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-03-08 18:04:30 -05:00
Paul Hernandez 9809b469c6 fix: enforce strict entity resolution in destructive MCP tools (#650)
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 15:56:07 -05:00
phernandez 76ac880f2d feat(api): add GET /knowledge/graph endpoint for full graph visualization
Returns all entities and resolved relations in a flat node/edge format
optimized for graph rendering. Replaces the frontend's use of the
recent memory endpoint which only returned a subset of relations.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-03-08 11:52:17 -05:00
Paul Hernandez ad3f2650d9 feat: add insert_before_section and insert_after_section edit operations (#648)
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 10:54:10 -05:00
dependabot[bot] d6508d985c chore(deps): bump authlib from 1.6.6 to 1.6.7 in the uv group across 1 directory (#645)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-03-08 10:53:52 -05:00
phernandez 7b95b9f37b docs: add What's New in v0.19.0 section to README
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>
2026-03-08 10:29:06 -05:00
56 changed files with 2925 additions and 114 deletions
+62
View File
@@ -2,6 +2,68 @@
## Unreleased
## v0.20.2 (2026-03-10)
### Bug Fixes
- Fix auto-update Homebrew detection: `brew outdated` exits 1 when a formula is outdated, not on error
- Previously treated exit code 1 as a failure, causing "Automatic update check failed" instead of detecting the available update
## 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
- **#649**: Enforce strict entity resolution in destructive MCP tools (`edit_note`, `move_note`, `delete_note`)
- Prevents fuzzy-match fallback from silently editing/moving/deleting the wrong note
- DST-related timeframe validation fix (round instead of truncate days)
### Features
- **#648**: Add `insert_before_section` and `insert_after_section` edit operations
- Add `GET /knowledge/graph` endpoint for full graph visualization
### Dependencies
- Bump authlib from 1.6.6 to 1.6.7
## v0.19.0 (2026-03-07)
### Highlights
+42
View File
@@ -23,6 +23,18 @@ Basic Memory lets you build persistent knowledge through natural conversations w
Claude, while keeping everything in simple Markdown files on your computer. It uses the Model Context Protocol (MCP) to
enable any compatible LLM to read and write to your local knowledge base.
## What's New in v0.19.0
- **Semantic Vector Search** — find notes by meaning, not just keywords. Combines full-text and vector similarity for hybrid search with FastEmbed embeddings.
- **Schema System** — infer, validate, and diff the structure of your knowledge base with `schema_infer`, `schema_validate`, and `schema_diff` tools.
- **Per-Project Cloud Routing** — route individual projects through the cloud while others stay local, using API key authentication (`basic-memory project set-cloud`).
- **FastMCP 3.0** — upgraded to FastMCP 3.0 with tool annotations for better client integration.
- **CLI Overhaul** — JSON output mode (`--json`) for scripting, workspace-aware commands, and an htop-inspired project dashboard.
- **Smarter Editing** — `edit_note` append/prepend auto-creates notes if they don't exist; `write_note` has an overwrite guard to prevent accidental data loss.
- **Richer Search Results** — matched chunk text returned in search results for better context.
See the full [CHANGELOG](CHANGELOG.md) for details.
- Website: [basicmemory.com](https://basicmemory.com?utm_source=github&utm_medium=referral&utm_campaign=readme)
- Documentation: [docs.basicmemory.com](https://docs.basicmemory.com?utm_source=github&utm_medium=referral&utm_campaign=readme)
- Community: [Discord](https://discord.gg/tyvKNccgqN?utm_source=github&utm_medium=referral&utm_campaign=readme)
@@ -63,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.0",
"version": "0.20.2",
"packages": [
{
"registryType": "pypi",
"identifier": "basic-memory",
"version": "0.19.0",
"version": "0.20.2",
"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.0"
__version__ = "0.20.2"
# API version for FastAPI - independent of package version
__api_version__ = "v0"
@@ -20,6 +20,7 @@ from basic_memory.deps import (
ProjectConfigV2ExternalDep,
AppConfigDep,
EntityRepositoryV2ExternalDep,
RelationRepositoryV2ExternalDep,
ProjectExternalIdPathDep,
TaskSchedulerDep,
FileServiceV2ExternalDep,
@@ -31,6 +32,9 @@ from basic_memory.schemas.v2 import (
EntityResolveRequest,
EntityResolveResponse,
EntityResponseV2,
GraphEdge,
GraphNode,
GraphResponse,
MoveEntityRequestV2,
MoveDirectoryRequestV2,
DeleteDirectoryRequestV2,
@@ -56,6 +60,50 @@ def _schedule_vector_sync_if_enabled(
)
## Graph endpoint
@router.get("/graph", response_model=GraphResponse)
async def get_graph(
project_id: ProjectExternalIdPathDep,
entity_repository: EntityRepositoryV2ExternalDep,
relation_repository: RelationRepositoryV2ExternalDep,
) -> GraphResponse:
"""Return all entities and resolved relations for knowledge graph visualization.
Returns a flat node/edge structure optimized for rendering with graph libraries.
Only includes resolved relations (where to_id is not null).
"""
logger.info("API v2 request: get_graph")
# Fetch all entities for this project
entities = await entity_repository.find_all(use_load_options=False)
nodes = [
GraphNode(
external_id=entity.external_id,
title=entity.title,
note_type=entity.note_type,
file_path=entity.file_path,
)
for entity in entities
]
# Fetch all resolved relations (to_id is not null) with eager-loaded entities
relations = await relation_repository.find_all()
edges = [
GraphEdge(
from_id=relation.from_entity.external_id,
to_id=relation.to_entity.external_id,
relation_type=relation.relation_type,
)
for relation in relations
if relation.to_entity is not None
]
logger.info(f"API v2 response: graph with {len(nodes)} nodes and {len(edges)} edges")
return GraphResponse(nodes=nodes, edges=edges)
## Resolution endpoint
+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 (
+389
View File
@@ -0,0 +1,389 @@
"""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,
)
# Trigger: brew outdated exits 1 when the formula IS outdated (with name on stdout).
# Why: non-zero exit here means "outdated", not "error".
# Outcome: check stdout for the package name to determine outdated status.
stdout = (result.stdout or "").strip()
is_outdated = PACKAGE_NAME in stdout
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,
+1 -1
View File
@@ -318,7 +318,7 @@ delete_note("path/to/file.md")
note_file_path = None
try:
# Resolve identifier to entity ID
entity_id = await knowledge_client.resolve_entity(identifier)
entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
if output_format == "json":
entity = await knowledge_client.get_entity(entity_id)
note_title = entity.title
+19 -5
View File
@@ -158,7 +158,7 @@ Error editing note '{identifier}': {error_message}
@mcp.tool(
description="Edit an existing markdown note using various operations like append, prepend, find_replace, or replace_section.",
description="Edit an existing markdown note using various operations like append, prepend, find_replace, replace_section, insert_before_section, or insert_after_section.",
annotations={"destructiveHint": False, "openWorldHint": False},
)
async def edit_note(
@@ -190,6 +190,8 @@ async def edit_note(
- "prepend": Add content to the beginning of the note (creates the note if it doesn't exist)
- "find_replace": Replace occurrences of find_text with content (note must exist)
- "replace_section": Replace content under a specific markdown header (note must exist)
- "insert_before_section": Insert content before a section heading without consuming it (note must exist)
- "insert_after_section": Insert content after a section heading without consuming it (note must exist)
content: The content to add or use for replacement
project: Project name to edit in. Optional - server will resolve using hierarchy.
If unknown, use list_memory_projects() to discover available projects.
@@ -257,7 +259,14 @@ async def edit_note(
logger.info("MCP tool call", tool="edit_note", identifier=identifier, operation=operation)
# Validate operation
valid_operations = ["append", "prepend", "find_replace", "replace_section"]
valid_operations = [
"append",
"prepend",
"find_replace",
"replace_section",
"insert_before_section",
"insert_after_section",
]
if operation not in valid_operations:
raise ValueError(
f"Invalid operation '{operation}'. Must be one of: {', '.join(valid_operations)}"
@@ -266,8 +275,9 @@ async def edit_note(
# Validate required parameters for specific operations
if operation == "find_replace" and not find_text:
raise ValueError("find_text parameter is required for find_replace operation")
if operation == "replace_section" and not section:
raise ValueError("section parameter is required for replace_section operation")
section_ops = ("replace_section", "insert_before_section", "insert_after_section")
if operation in section_ops and not section:
raise ValueError("section parameter is required for section-based operations")
# Use the PATCH endpoint to edit the entity
try:
@@ -283,7 +293,7 @@ async def edit_note(
# Try to resolve the entity; for append/prepend, create it if not found
try:
entity_id = await knowledge_client.resolve_entity(identifier)
entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
except Exception as resolve_error:
# Trigger: entity does not exist yet
# Why: append/prepend can meaningfully create a new note from the content,
@@ -389,6 +399,10 @@ async def edit_note(
summary.append("operation: Find and replace operation completed")
elif operation == "replace_section":
summary.append(f"operation: Replaced content under section '{section}'")
elif operation == "insert_before_section":
summary.append(f"operation: Inserted content before section '{section}'")
elif operation == "insert_after_section":
summary.append(f"operation: Inserted content after section '{section}'")
# Count observations by category (reuse logic from write_note)
categories = {}
+21 -2
View File
@@ -6,6 +6,7 @@ from typing import Optional, Literal
from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.server import mcp
from basic_memory.mcp.project_context import get_project_client
@@ -637,7 +638,7 @@ move_note("path/to/file.md", "{destination_path}/file.md")
"""Resolve and cache the source entity ID for the duration of this move."""
nonlocal resolved_entity_id
if resolved_entity_id is None:
resolved_entity_id = await knowledge_client.resolve_entity(identifier)
resolved_entity_id = await knowledge_client.resolve_entity(identifier, strict=True)
return resolved_entity_id
try:
@@ -645,8 +646,26 @@ move_note("path/to/file.md", "{destination_path}/file.md")
source_entity = await knowledge_client.get_entity(resolved_entity_id)
if "." in source_entity.file_path:
source_ext = source_entity.file_path.split(".")[-1]
except ToolError as e:
# Trigger: strict=True resolve_entity raised because the entity was not found.
# Why: fail fast with a formatted error instead of silently falling through
# to extension defaults and failing later with a confusing message.
# Outcome: move_note returns a user-facing not-found error immediately.
logger.error(f"Move failed for '{identifier}' to '{destination_path}': {e}")
if output_format == "json":
return {
"moved": False,
"title": None,
"permalink": None,
"file_path": None,
"source": identifier,
"destination": destination_path,
"error": str(e),
}
return _format_move_error_response(str(e), identifier, destination_path)
except Exception as e:
# If we can't fetch source metadata, continue with extension defaults.
# If we can't fetch source metadata (e.g. get_entity or file_path parsing fails),
# continue with extension defaults — the entity was at least resolved.
logger.debug(f"Could not fetch source entity for extension check: {e}")
# --- Resolve destination_folder into destination_path ---
+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 ---
+5 -3
View File
@@ -140,10 +140,12 @@ def validate_timeframe(timeframe: str) -> str:
if parsed > now:
raise ValueError("Timeframe cannot be in the future") # pragma: no cover
# Could format the duration back to our standard format
days = (now - parsed).days
# Round to nearest day to handle DST transitions where an hour shift
# can cause e.g. "7d" to compute as 6 days + 23 hours
total_seconds = (now - parsed).total_seconds()
days = round(total_seconds / 86400)
# Could enforce reasonable limits
# Enforce reasonable limits
if days > 365:
raise ValueError("Timeframe should be <= 1 year")
+18 -3
View File
@@ -65,7 +65,14 @@ class EditEntityRequest(BaseModel):
Supports various operation types for different editing scenarios.
"""
operation: Literal["append", "prepend", "find_replace", "replace_section"]
operation: Literal[
"append",
"prepend",
"find_replace",
"replace_section",
"insert_before_section",
"insert_after_section",
]
content: str
section: Optional[str] = None
find_text: Optional[str] = None
@@ -75,8 +82,16 @@ class EditEntityRequest(BaseModel):
@classmethod
def validate_section_for_replace_section(cls, v, info):
"""Ensure section is provided for replace_section operation."""
if info.data.get("operation") == "replace_section" and not v:
raise ValueError("section parameter is required for replace_section operation")
if (
info.data.get("operation")
in (
"replace_section",
"insert_before_section",
"insert_after_section",
)
and not v
):
raise ValueError("section parameter is required for section-based operations")
return v
@field_validator("find_text")
+8
View File
@@ -10,6 +10,11 @@ from basic_memory.schemas.v2.entity import (
ProjectResolveRequest,
ProjectResolveResponse,
)
from basic_memory.schemas.v2.graph import (
GraphEdge,
GraphNode,
GraphResponse,
)
from basic_memory.schemas.v2.resource import (
CreateResourceRequest,
UpdateResourceRequest,
@@ -25,6 +30,9 @@ __all__ = [
"DeleteDirectoryRequestV2",
"ProjectResolveRequest",
"ProjectResolveResponse",
"GraphEdge",
"GraphNode",
"GraphResponse",
"CreateResourceRequest",
"UpdateResourceRequest",
"ResourceResponse",
+31
View File
@@ -0,0 +1,31 @@
"""Graph visualization schemas for the knowledge graph endpoint."""
from typing import Optional
from pydantic import BaseModel, Field
class GraphNode(BaseModel):
"""A node in the knowledge graph visualization."""
external_id: str = Field(..., description="Entity external ID (UUID)")
title: str = Field(..., description="Entity title")
note_type: Optional[str] = Field(None, description="Note type (e.g., note, spec, task)")
file_path: str = Field(..., description="Relative file path")
class GraphEdge(BaseModel):
"""An edge in the knowledge graph visualization."""
from_id: str = Field(..., description="External ID of source entity")
to_id: str = Field(..., description="External ID of target entity")
relation_type: str = Field(..., description="Type of relation")
class GraphResponse(BaseModel):
"""Complete knowledge graph for visualization."""
nodes: list[GraphNode] = Field(default_factory=list, description="All entities as nodes")
edges: list[GraphEdge] = Field(
default_factory=list, description="All resolved relations as edges"
)
@@ -888,6 +888,14 @@ class EntityService(BaseService[EntityModel]):
raise ValueError("section cannot be empty or whitespace only")
return self.replace_section_content(current_content, section, content)
elif operation in ("insert_before_section", "insert_after_section"):
if not section:
raise ValueError("section is required for insert section operations")
if not section.strip():
raise ValueError("section cannot be empty or whitespace only")
position = "before" if operation == "insert_before_section" else "after"
return self.insert_relative_to_section(current_content, section, content, position)
else:
raise ValueError(f"Unsupported operation: {operation}")
@@ -979,6 +987,73 @@ class EntityService(BaseService[EntityModel]):
return "\n".join(result_lines)
def insert_relative_to_section(
self,
current_content: str,
section_header: str,
new_content: str,
position: str,
) -> str:
"""Insert content before or after a section heading without consuming it.
Unlike replace_section_content, this preserves the section heading and its
existing content. The new content is inserted immediately before or after
the heading line.
Args:
current_content: The current markdown content
section_header: The section header to anchor on (e.g., "## Section Name")
new_content: The content to insert
position: "before" to insert above the heading, "after" to insert below it
Returns:
The updated content with new_content inserted relative to the heading
Raises:
ValueError: If the section header is not found or appears more than once
"""
# Normalize the section header (ensure it starts with #)
if not section_header.startswith("#"):
section_header = "## " + section_header
lines = current_content.split("\n")
matching_indices = [
i for i, line in enumerate(lines) if line.strip() == section_header.strip()
]
if len(matching_indices) == 0:
raise ValueError(
f"Section '{section_header}' not found in document. "
f"Use replace_section to create a new section."
)
if len(matching_indices) > 1:
raise ValueError(
f"Multiple sections found with header '{section_header}'. "
f"Section insertion requires unique headers."
)
idx = matching_indices[0]
if position == "before":
# Insert new content before the section heading
before = lines[:idx]
after = lines[idx:]
# Ensure blank line separation
insert_lines = new_content.rstrip("\n").split("\n")
if before and before[-1].strip() != "":
insert_lines = [""] + insert_lines
return "\n".join(before + insert_lines + [""] + after)
else:
# Insert new content after the section heading line
before = lines[: idx + 1]
after = lines[idx + 1 :]
insert_lines = new_content.rstrip("\n").split("\n")
# Ensure blank line separation so inserted text doesn't merge
# with existing section content into a single paragraph
if after and after[0].strip() != "":
insert_lines = insert_lines + [""]
return "\n".join(before + insert_lines + after)
def _prepend_after_frontmatter(self, current_content: str, content: str) -> str:
"""Prepend content after frontmatter, preserving frontmatter structure."""
+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.
@@ -208,7 +208,7 @@ def test_edit_note_replace_section_fails_without_section(
)
assert result.exit_code != 0
assert "section parameter is required for replace_section operation" in result.output
assert "section parameter is required for section-based operations" in result.output
def test_edit_note_append_creates_nonexistent_note_cli(
+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
+30 -5
View File
@@ -307,8 +307,13 @@ async def test_delete_note_by_file_path(mcp_server, app, test_project):
@pytest.mark.asyncio
async def test_delete_note_case_insensitive(mcp_server, app, test_project):
"""Test that note deletion is case insensitive for titles."""
async def test_delete_note_rejects_case_mismatch(mcp_server, app, test_project):
"""Test that delete_note with wrong case does not fuzzy-match to an existing note.
Strict resolution (#649) prevents destructive operations from silently
resolving to a different note via fuzzy search. Case-mismatched titles
should be rejected, not resolved to the nearest match.
"""
async with Client(mcp_server) as client:
# Create a note with mixed case
@@ -323,7 +328,7 @@ async def test_delete_note_case_insensitive(mcp_server, app, test_project):
},
)
# Try to delete with different case
# Try to delete with different case — should NOT find the note
delete_result = await client.call_tool(
"delete_note",
{
@@ -332,8 +337,28 @@ async def test_delete_note_case_insensitive(mcp_server, app, test_project):
},
)
# Should return True for successful deletion
assert "true" in delete_result.content[0].text.lower()
# Should return False (not found) — strict mode rejects fuzzy matches
assert "false" in delete_result.content[0].text.lower()
# Verify the note still exists using the exact title
read_result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "CamelCase Note Title",
},
)
assert "Testing case sensitivity" in read_result.content[0].text
# Delete with exact title should succeed
delete_result2 = await client.call_tool(
"delete_note",
{
"project": test_project.name,
"identifier": "CamelCase Note Title",
},
)
assert "true" in delete_result2.content[0].text.lower()
@pytest.mark.asyncio
@@ -710,3 +710,81 @@ async def test_edit_note_using_different_identifiers(mcp_server, app, test_proje
assert "Edited by title." in content
assert "Edited by permalink." in content
assert "Edited by folder/title." in content
@pytest.mark.asyncio
async def test_edit_note_append_autocreate_does_not_fuzzy_match(mcp_server, app, test_project):
"""Reproduces #649: edit_note append must auto-create, not fuzzy-match to an existing note.
Creates two notes, then attempts to append to a nonexistent identifier.
The tool should create a new note, and neither existing note should be modified.
"""
async with Client(mcp_server) as client:
# Create two notes that could be fuzzy-matched
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Routing Test A",
"directory": "test",
"content": "# Routing Test A\n\nContent A.",
},
)
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Routing Test B",
"directory": "test",
"content": "# Routing Test B\n\nContent B.",
},
)
# Attempt to edit a nonexistent note — should error, not silently edit A or B
edit_result = await client.call_tool(
"edit_note",
{
"project": test_project.name,
"identifier": "Routing Test NONEXISTENT",
"operation": "append",
"content": "\n\nThis should NOT appear in any note.",
},
)
edit_text = edit_result.content[0].text
# append to nonexistent creates a new note — verify it did NOT edit A or B
assert "Created note (append)" in edit_text
assert "fileCreated: true" in edit_text
# Verify neither A nor B was modified
read_a = await client.call_tool(
"read_note",
{"project": test_project.name, "identifier": "Routing Test A"},
)
content_a = read_a.content[0].text
assert "Content A" in content_a
assert "This should NOT appear" not in content_a
read_b = await client.call_tool(
"read_note",
{"project": test_project.name, "identifier": "Routing Test B"},
)
content_b = read_b.content[0].text
assert "Content B" in content_b
assert "This should NOT appear" not in content_b
# Now test find_replace on nonexistent — should error
edit_result2 = await client.call_tool(
"edit_note",
{
"project": test_project.name,
"identifier": "Routing Test NONEXISTENT AGAIN",
"operation": "find_replace",
"content": "replaced",
"find_text": "Content",
},
)
error_text = edit_result2.content[0].text
assert "Edit Failed" in error_text
@@ -716,3 +716,56 @@ async def test_move_note_destination_folder_mutually_exclusive(mcp_server, app,
error_text = move_result.content[0].text
assert "# Move Failed - Invalid Parameters" in error_text
assert "Cannot specify both" in error_text
@pytest.mark.asyncio
async def test_move_note_strict_resolution_rejects_fuzzy_match(mcp_server, app, test_project):
"""move_note must not fuzzy-match a nonexistent identifier to an existing note (#649)."""
async with Client(mcp_server) as client:
# Create two notes that could be fuzzy-matched
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Move Strict Test A",
"directory": "test",
"content": "# Move Strict Test A\n\nContent A.",
},
)
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Move Strict Test B",
"directory": "test",
"content": "# Move Strict Test B\n\nContent B.",
},
)
# Attempt to move a nonexistent note — should error, not move A or B
move_result = await client.call_tool(
"move_note",
{
"project": test_project.name,
"identifier": "Move Strict Test NONEXISTENT",
"destination_path": "archive/Moved.md",
},
)
assert len(move_result.content) == 1
error_text = move_result.content[0].text
assert "# Move Failed" in error_text
# Verify neither A nor B was moved
read_a = await client.call_tool(
"read_note",
{"project": test_project.name, "identifier": "Move Strict Test A"},
)
assert "Content A" in read_a.content[0].text
read_b = await client.call_tool(
"read_note",
{"project": test_project.name, "identifier": "Move Strict Test B"},
)
assert "Content B" in read_b.content[0].text
@@ -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":
+399
View File
@@ -0,0 +1,399 @@
"""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,
_check_homebrew_update_available,
_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_check_homebrew_update_available_exit_code_1_means_outdated(monkeypatch):
"""brew outdated exits 1 when the formula is outdated, not on error."""
def _fake_run(command, **kwargs):
return subprocess.CompletedProcess(
command, 1, stdout="basicmachines-co/basic-memory/basic-memory\n", stderr=""
)
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _fake_run)
is_outdated, _ = _check_homebrew_update_available(silent=False)
assert is_outdated is True
def test_check_homebrew_update_available_exit_code_0_means_up_to_date(monkeypatch):
"""brew outdated exits 0 when the formula is up to date."""
def _fake_run(command, **kwargs):
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr("basic_memory.cli.auto_update._run_subprocess", _fake_run)
is_outdated, _ = _check_homebrew_update_available(silent=False)
assert is_outdated is False
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()
+27 -3
View File
@@ -1,6 +1,10 @@
"""Tests for delete_note MCP tool."""
from basic_memory.mcp.tools.delete_note import _format_delete_error_response
import pytest
from basic_memory.mcp.tools.delete_note import delete_note, _format_delete_error_response
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.write_note import write_note
class TestDeleteNoteErrorFormatting:
@@ -94,5 +98,25 @@ class TestDeleteNoteErrorFormatting:
assert "folder/note-title" in result # Permalink format
# Integration tests removed to focus on error formatting coverage
# The error formatting tests above provide the necessary coverage for MCP tool error messaging
@pytest.mark.asyncio
async def test_delete_note_rejects_fuzzy_match(client, test_project):
"""delete_note must reject nonexistent identifiers, not fuzzy-match to a similar note."""
await write_note(
project=test_project.name,
title="Delete Target Note",
directory="test",
content="# Delete Target Note\nShould not be deleted.",
)
# Attempt to delete a nonexistent note — should return False, not silently delete the existing note
result = await delete_note(
project=test_project.name,
identifier="Delete Target NONEXISTENT",
)
# Should indicate not found (False or error string)
assert result is False or (isinstance(result, str) and "not found" in result.lower())
# Verify the existing note was NOT deleted
content = await read_note("Delete Target Note", project=test_project.name)
assert "Should not be deleted" in content
+160 -1
View File
@@ -1,8 +1,10 @@
"""Tests for the edit_note MCP tool."""
import pytest
from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.write_note import write_note
@@ -320,7 +322,7 @@ async def test_edit_note_replace_section_missing_section(client, test_project):
content="new content",
)
assert "section parameter is required for replace_section operation" in str(exc_info.value)
assert "section parameter is required for section-based operations" in str(exc_info.value)
@pytest.mark.asyncio
@@ -611,3 +613,160 @@ async def test_edit_note_preserves_permalink_when_frontmatter_missing(client, te
assert f"permalink: {test_project.name}/test/test-note" in second_result
assert f"[Session: Using project '{test_project.name}']" in second_result
# The edit should succeed without validation errors
@pytest.mark.asyncio
async def test_edit_note_find_replace_rejects_fuzzy_match(client, test_project):
"""find_replace must reject nonexistent identifiers, not fuzzy-match to a similar note."""
# Create two notes that could be fuzzy-matched
await write_note(
project=test_project.name,
title="Routing Test A",
directory="test",
content="# Routing Test A\nContent A.",
)
await write_note(
project=test_project.name,
title="Routing Test B",
directory="test",
content="# Routing Test B\nContent B.",
)
# Attempt to edit a nonexistent note — should error, not silently edit A or B
result = await edit_note(
project=test_project.name,
identifier="Routing Test NONEXISTENT",
operation="find_replace",
content="replaced",
find_text="Content",
)
assert isinstance(result, str)
assert "# Edit Failed" in result
# Verify neither A nor B was modified
content_a = await read_note("Routing Test A", project=test_project.name)
assert "Content A" in content_a
content_b = await read_note("Routing Test B", project=test_project.name)
assert "Content B" in content_b
@pytest.mark.asyncio
async def test_edit_note_append_autocreate_not_fuzzy_match(client, test_project):
"""append to a nonexistent note should auto-create it, not fuzzy-match an existing note."""
await write_note(
project=test_project.name,
title="Existing Note Alpha",
directory="test",
content="# Existing Note Alpha\nOriginal content.",
)
# Append to a nonexistent note — should create a new note, not edit "Existing Note Alpha"
result = await edit_note(
project=test_project.name,
identifier="Existing Note ZZZZZ",
operation="append",
content="# New Note\nBrand new content.",
)
assert isinstance(result, str)
assert "Created note (append)" in result
assert "fileCreated: true" in result
# Verify original note was NOT modified
content = await read_note("Existing Note Alpha", project=test_project.name)
assert "Original content" in content
assert "Brand new content" not in content
@pytest.mark.asyncio
async def test_edit_note_insert_before_section_operation(client, test_project):
"""Test inserting content before a section heading."""
# Create initial note with sections
await write_note(
project=test_project.name,
title="Insert Before Doc",
directory="docs",
content="# Doc\n\n## Overview\nOverview content.\n\n## Details\nDetail content.",
)
result = await edit_note(
project=test_project.name,
identifier="docs/insert-before-doc",
operation="insert_before_section",
content="--- inserted divider ---",
section="## Details",
)
assert isinstance(result, str)
assert "Edited note (insert_before_section)" in result
assert f"project: {test_project.name}" in result
assert "Inserted content before section '## Details'" in result
assert f"[Session: Using project '{test_project.name}']" in result
@pytest.mark.asyncio
async def test_edit_note_insert_after_section_operation(client, test_project):
"""Test inserting content after a section heading."""
# Create initial note with sections
await write_note(
project=test_project.name,
title="Insert After Doc",
directory="docs",
content="# Doc\n\n## Overview\nOverview content.\n\n## Details\nDetail content.",
)
result = await edit_note(
project=test_project.name,
identifier="docs/insert-after-doc",
operation="insert_after_section",
content="Inserted after overview heading",
section="## Overview",
)
assert isinstance(result, str)
assert "Edited note (insert_after_section)" in result
assert f"project: {test_project.name}" in result
assert "Inserted content after section '## Overview'" in result
assert f"[Session: Using project '{test_project.name}']" in result
@pytest.mark.asyncio
async def test_edit_note_insert_before_section_missing_section(client, test_project):
"""Test insert_before_section without section parameter raises ValueError."""
await write_note(
project=test_project.name,
title="Test Note",
directory="test",
content="# Test\nContent here.",
)
with pytest.raises(ValueError, match="section parameter is required"):
await edit_note(
project=test_project.name,
identifier="test/test-note",
operation="insert_before_section",
content="new content",
)
@pytest.mark.asyncio
async def test_edit_note_insert_before_section_not_found(client, test_project):
"""Test insert_before_section when section doesn't exist returns error."""
await write_note(
project=test_project.name,
title="Test Note",
directory="test",
content="# Test\n\n## Existing\nContent here.",
)
result = await edit_note(
project=test_project.name,
identifier="test/test-note",
operation="insert_before_section",
content="new content",
section="## Nonexistent",
)
assert isinstance(result, str)
assert "# Edit Failed" in result
+25
View File
@@ -590,6 +590,31 @@ async def test_move_note_preserves_frontmatter(app, client, test_project):
assert "Content with custom metadata" in content
@pytest.mark.asyncio
async def test_move_note_rejects_fuzzy_match(client, test_project):
"""move_note must reject nonexistent identifiers, not fuzzy-match to a similar note."""
await write_note(
project=test_project.name,
title="Move Target Note",
directory="source",
content="# Move Target Note\nShould not be moved.",
)
# Attempt to move a nonexistent note — should error, not silently move the existing note
result = await move_note(
project=test_project.name,
identifier="Move Target NONEXISTENT",
destination_path="target/Moved.md",
)
assert isinstance(result, str)
assert "# Move Failed" in result
# Verify the existing note was NOT moved
content = await read_note("Move Target Note", project=test_project.name)
assert "Should not be moved" in content
class TestMoveNoteErrorFormatting:
"""Test move note error formatting for better user experience."""
+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 ---
+43 -3
View File
@@ -345,7 +345,7 @@ def test_edit_entity_request_find_replace_empty_find_text():
def test_edit_entity_request_replace_section_empty_section():
"""Test that replace_section operation requires non-empty section parameter."""
with pytest.raises(
ValueError, match="section parameter is required for replace_section operation"
ValueError, match="section parameter is required for section-based operations"
):
EditEntityRequest.model_validate(
{
@@ -356,6 +356,46 @@ def test_edit_entity_request_replace_section_empty_section():
)
def test_edit_entity_request_insert_before_section():
"""Test insert_before_section is a valid operation."""
edit_request = EditEntityRequest.model_validate(
{
"operation": "insert_before_section",
"content": "content to insert",
"section": "## Target Section",
}
)
assert edit_request.operation == "insert_before_section"
assert edit_request.section == "## Target Section"
def test_edit_entity_request_insert_after_section():
"""Test insert_after_section is a valid operation."""
edit_request = EditEntityRequest.model_validate(
{
"operation": "insert_after_section",
"content": "content to insert",
"section": "## Target Section",
}
)
assert edit_request.operation == "insert_after_section"
assert edit_request.section == "## Target Section"
def test_edit_entity_request_insert_before_section_empty_section():
"""Test that insert_before_section requires non-empty section parameter."""
with pytest.raises(
ValueError, match="section parameter is required for section-based operations"
):
EditEntityRequest.model_validate(
{
"operation": "insert_before_section",
"content": "content",
"section": "",
}
)
# New tests for timeframe parsing functions
class TestTimeframeParsing:
"""Test cases for parse_timeframe() and validate_timeframe() functions."""
@@ -391,7 +431,7 @@ class TestTimeframeParsing:
result_1d = parse_timeframe("1d")
expected_1d = now - timedelta(days=1)
diff = abs((result_1d - expected_1d).total_seconds())
assert diff < 3600 # Within 1 hour tolerance (accounts for DST transitions)
assert diff <= 3610 # Within 1 hour tolerance + execution margin (DST transitions)
assert result_1d.tzinfo is not None
# Test yesterday - should be yesterday at same time
@@ -404,7 +444,7 @@ class TestTimeframeParsing:
result_week = parse_timeframe("1 week ago")
expected_week = now - timedelta(weeks=1)
diff = abs((result_week - expected_week).total_seconds())
assert diff < 3600 # Within 1 hour tolerance
assert diff <= 3610 # Within 1 hour tolerance + execution margin (DST transitions)
assert result_week.tzinfo is not None
def test_parse_timeframe_invalid(self):
+261
View File
@@ -1402,6 +1402,267 @@ async def test_edit_entity_replace_section_strips_duplicate_header(
assert "## Another Section" in file_content # Other sections preserved
# Insert before/after section tests
@pytest.mark.asyncio
async def test_edit_entity_insert_before_section(
entity_service: EntityService, file_service: FileService
):
"""Test inserting content before a section heading."""
content = dedent("""
# Main Title
## Section 1
Section 1 content
## Section 2
Section 2 content
""").strip()
entity = await entity_service.create_entity(
EntitySchema(
title="Insert Before Test",
directory="docs",
note_type="note",
content=content,
)
)
updated = await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_before_section",
content="Inserted before section 2",
section="## Section 2",
)
file_path = file_service.get_entity_path(updated)
file_content, _ = await file_service.read_file(file_path)
assert "Inserted before section 2" in file_content
assert "## Section 2" in file_content
assert "Section 2 content" in file_content
# Inserted content should appear before the section heading
assert file_content.index("Inserted before section 2") < file_content.index("## Section 2")
@pytest.mark.asyncio
async def test_edit_entity_insert_after_section(
entity_service: EntityService, file_service: FileService
):
"""Test inserting content after a section heading."""
content = dedent("""
# Main Title
## Section 1
Section 1 content
## Section 2
Section 2 content
""").strip()
entity = await entity_service.create_entity(
EntitySchema(
title="Insert After Test",
directory="docs",
note_type="note",
content=content,
)
)
updated = await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_after_section",
content="Inserted after section 1 heading",
section="## Section 1",
)
file_path = file_service.get_entity_path(updated)
file_content, _ = await file_service.read_file(file_path)
assert "Inserted after section 1 heading" in file_content
assert "## Section 1" in file_content
assert "Section 1 content" in file_content
# Inserted content should appear after the heading but content is also preserved
assert file_content.index("## Section 1") < file_content.index(
"Inserted after section 1 heading"
)
@pytest.mark.asyncio
async def test_edit_entity_insert_before_section_not_found(entity_service: EntityService):
"""Test insert_before_section raises ValueError when section not found."""
entity = await entity_service.create_entity(
EntitySchema(
title="Test Note",
directory="test",
note_type="note",
content="# Main Title\n\nSome content",
)
)
with pytest.raises(ValueError, match="Section '## Missing' not found"):
await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_before_section",
content="new content",
section="## Missing",
)
@pytest.mark.asyncio
async def test_edit_entity_insert_after_section_not_found(entity_service: EntityService):
"""Test insert_after_section raises ValueError when section not found."""
entity = await entity_service.create_entity(
EntitySchema(
title="Test Note",
directory="test",
note_type="note",
content="# Main Title\n\nSome content",
)
)
with pytest.raises(ValueError, match="Section '## Missing' not found"):
await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_after_section",
content="new content",
section="## Missing",
)
@pytest.mark.asyncio
async def test_edit_entity_insert_before_section_multiple_sections_error(
entity_service: EntityService,
):
"""Test insert_before_section raises ValueError with duplicate sections."""
entity = await entity_service.create_entity(
EntitySchema(
title="Test Note",
directory="test",
note_type="note",
content="# Title\n\n## Dup\nFirst\n\n## Dup\nSecond",
)
)
with pytest.raises(ValueError, match="Multiple sections found"):
await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_before_section",
content="new content",
section="## Dup",
)
@pytest.mark.asyncio
async def test_edit_entity_insert_before_section_missing_section_param(
entity_service: EntityService,
):
"""Test insert_before_section raises ValueError when section param is missing."""
entity = await entity_service.create_entity(
EntitySchema(
title="Test Note",
directory="test",
note_type="note",
content="# Title\n\nContent",
)
)
with pytest.raises(ValueError, match="section is required"):
await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_before_section",
content="new content",
)
@pytest.mark.asyncio
async def test_edit_entity_insert_before_section_empty_section(entity_service: EntityService):
"""Test insert_before_section raises ValueError when section is empty/whitespace."""
entity = await entity_service.create_entity(
EntitySchema(
title="Test Note",
directory="test",
note_type="note",
content="# Title\n\nContent",
)
)
with pytest.raises(ValueError, match="section cannot be empty"):
await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_before_section",
content="new content",
section=" ",
)
@pytest.mark.asyncio
async def test_edit_entity_insert_after_section_at_end_of_document(
entity_service: EntityService, file_service: FileService
):
"""Test inserting after the last section in a document."""
content = dedent("""
# Main Title
## Only Section
Some content here
""").strip()
entity = await entity_service.create_entity(
EntitySchema(
title="Insert End Test",
directory="docs",
note_type="note",
content=content,
)
)
updated = await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_after_section",
content="Inserted after the last section heading",
section="## Only Section",
)
file_path = file_service.get_entity_path(updated)
file_content, _ = await file_service.read_file(file_path)
assert "Inserted after the last section heading" in file_content
assert "## Only Section" in file_content
assert "Some content here" in file_content
@pytest.mark.asyncio
async def test_edit_entity_insert_after_section_preserves_paragraph_separation(
entity_service: EntityService, file_service: FileService
):
"""Test that insert_after_section adds blank line so inserted text doesn't merge
with existing section content into a single markdown paragraph."""
content = dedent("""
# Main Title
## Section
Existing paragraph text
""").strip()
entity = await entity_service.create_entity(
EntitySchema(
title="Paragraph Sep Test",
directory="docs",
note_type="note",
content=content,
)
)
updated = await entity_service.edit_entity(
identifier=entity.permalink,
operation="insert_after_section",
content="Inserted line",
section="## Section",
)
file_path = file_service.get_entity_path(updated)
file_content, _ = await file_service.read_file(file_path)
# The inserted line and existing content should be separated by a blank line
assert "Inserted line\n\nExisting paragraph text" in file_content
# Move entity tests
@pytest.mark.asyncio
async def test_move_entity_success(
+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)
Generated
+9 -3
View File
@@ -142,14 +142,14 @@ wheels = [
[[package]]
name = "authlib"
version = "1.6.6"
version = "1.6.7"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "cryptography" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bb/9b/b1661026ff24bc641b76b78c5222d614776b0c085bcfdac9bd15a1cb4b35/authlib-1.6.6.tar.gz", hash = "sha256:45770e8e056d0f283451d9996fbb59b70d45722b45d854d58f32878d0a40c38e", size = 164894, upload-time = "2025-12-12T08:01:41.464Z" }
sdist = { url = "https://files.pythonhosted.org/packages/49/dc/ed1681bf1339dd6ea1ce56136bad4baabc6f7ad466e375810702b0237047/authlib-1.6.7.tar.gz", hash = "sha256:dbf10100011d1e1b34048c9d120e83f13b35d69a826ae762b93d2fb5aafc337b", size = 164950, upload-time = "2026-02-06T14:04:14.171Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/51/321e821856452f7386c4e9df866f196720b1ad0c5ea1623ea7399969ae3b/authlib-1.6.6-py2.py3-none-any.whl", hash = "sha256:7d9e9bc535c13974313a87f53e8430eb6ea3d1cf6ae4f6efcd793f2e949143fd", size = 244005, upload-time = "2025-12-12T08:01:40.209Z" },
{ url = "https://files.pythonhosted.org/packages/f8/00/3ed12264094ec91f534fae429945efbaa9f8c666f3aa7061cc3b2a26a0cd/authlib-1.6.7-py2.py3-none-any.whl", hash = "sha256:c637340d9a02789d2efa1d003a7437d10d3e565237bcb5fcbc6c134c7b95bab0", size = 244115, upload-time = "2026-02-06T14:04:12.141Z" },
]
[[package]]
@@ -304,10 +304,16 @@ sdist = { url = "https://files.pythonhosted.org/packages/92/88/b8527e1b00c1811db
wheels = [
{ url = "https://files.pythonhosted.org/packages/d3/25/79c98ebe12df31548ba4eaf44db11b7cad6b3e7b4203718335620939083c/caio-0.9.25-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:fb7ff95af4c31ad3f03179149aab61097a71fd85e05f89b4786de0359dffd044", size = 36983, upload-time = "2025-12-26T15:21:36.075Z" },
{ url = "https://files.pythonhosted.org/packages/a3/2b/21288691f16d479945968a0a4f2856818c1c5be56881d51d4dac9b255d26/caio-0.9.25-cp312-cp312-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:97084e4e30dfa598449d874c4d8e0c8d5ea17d2f752ef5e48e150ff9d240cd64", size = 82012, upload-time = "2025-12-26T15:22:20.983Z" },
{ url = "https://files.pythonhosted.org/packages/03/c4/8a1b580875303500a9c12b9e0af58cb82e47f5bcf888c2457742a138273c/caio-0.9.25-cp312-cp312-manylinux_2_34_aarch64.whl", hash = "sha256:4fa69eba47e0f041b9d4f336e2ad40740681c43e686b18b191b6c5f4c5544bfb", size = 81502, upload-time = "2026-03-04T22:08:22.381Z" },
{ url = "https://files.pythonhosted.org/packages/d1/1c/0fe770b8ffc8362c48134d1592d653a81a3d8748d764bec33864db36319d/caio-0.9.25-cp312-cp312-manylinux_2_34_x86_64.whl", hash = "sha256:6bebf6f079f1341d19f7386db9b8b1f07e8cc15ae13bfdaff573371ba0575d69", size = 80200, upload-time = "2026-03-04T22:08:23.382Z" },
{ url = "https://files.pythonhosted.org/packages/31/57/5e6ff127e6f62c9f15d989560435c642144aa4210882f9494204bc892305/caio-0.9.25-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:d6c2a3411af97762a2b03840c3cec2f7f728921ff8adda53d7ea2315a8563451", size = 36979, upload-time = "2025-12-26T15:21:35.484Z" },
{ url = "https://files.pythonhosted.org/packages/a3/9f/f21af50e72117eb528c422d4276cbac11fb941b1b812b182e0a9c70d19c5/caio-0.9.25-cp313-cp313-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:0998210a4d5cd5cb565b32ccfe4e53d67303f868a76f212e002a8554692870e6", size = 81900, upload-time = "2025-12-26T15:22:21.919Z" },
{ url = "https://files.pythonhosted.org/packages/9c/12/c39ae2a4037cb10ad5eb3578eb4d5f8c1a2575c62bba675f3406b7ef0824/caio-0.9.25-cp313-cp313-manylinux_2_34_aarch64.whl", hash = "sha256:1a177d4777141b96f175fe2c37a3d96dec7911ed9ad5f02bac38aaa1c936611f", size = 81523, upload-time = "2026-03-04T22:08:25.187Z" },
{ url = "https://files.pythonhosted.org/packages/22/59/f8f2e950eb4f1a5a3883e198dca514b9d475415cb6cd7b78b9213a0dd45a/caio-0.9.25-cp313-cp313-manylinux_2_34_x86_64.whl", hash = "sha256:9ed3cfb28c0e99fec5e208c934e5c157d0866aa9c32aa4dc5e9b6034af6286b7", size = 80243, upload-time = "2026-03-04T22:08:26.449Z" },
{ url = "https://files.pythonhosted.org/packages/69/ca/a08fdc7efdcc24e6a6131a93c85be1f204d41c58f474c42b0670af8c016b/caio-0.9.25-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:fab6078b9348e883c80a5e14b382e6ad6aabbc4429ca034e76e730cf464269db", size = 36978, upload-time = "2025-12-26T15:21:41.055Z" },
{ url = "https://files.pythonhosted.org/packages/5e/6c/d4d24f65e690213c097174d26eda6831f45f4734d9d036d81790a27e7b78/caio-0.9.25-cp314-cp314-manylinux2010_x86_64.manylinux2014_x86_64.manylinux_2_12_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:44a6b58e52d488c75cfaa5ecaa404b2b41cc965e6c417e03251e868ecd5b6d77", size = 81832, upload-time = "2025-12-26T15:22:22.757Z" },
{ url = "https://files.pythonhosted.org/packages/87/a4/e534cf7d2d0e8d880e25dd61e8d921ffcfe15bd696734589826f5a2df727/caio-0.9.25-cp314-cp314-manylinux_2_34_aarch64.whl", hash = "sha256:628a630eb7fb22381dd8e3c8ab7f59e854b9c806639811fc3f4310c6bd711d79", size = 81565, upload-time = "2026-03-04T22:08:27.483Z" },
{ url = "https://files.pythonhosted.org/packages/3f/ed/bf81aeac1d290017e5e5ac3e880fd56ee15e50a6d0353986799d1bc5cfd5/caio-0.9.25-cp314-cp314-manylinux_2_34_x86_64.whl", hash = "sha256:0ba16aa605ccb174665357fc729cf500679c2d94d5f1458a6f0d5ca48f2060a7", size = 80071, upload-time = "2026-03-04T22:08:28.751Z" },
{ url = "https://files.pythonhosted.org/packages/86/93/1f76c8d1bafe3b0614e06b2195784a3765bbf7b0a067661af9e2dd47fc33/caio-0.9.25-py3-none-any.whl", hash = "sha256:06c0bb02d6b929119b1cfbe1ca403c768b2013a369e2db46bfa2a5761cf82e40", size = 19087, upload-time = "2025-12-26T15:22:00.221Z" },
]