admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mcp): attach local state to one workspace project row (#854)

Browse Source 
Signed-off-by: Drew Cain <groksrc@gmail.com>
This commit is contained in:
Drew Cain
2026-05-26 10:54:44 -05:00
committed by GitHub
parent 36b51b676e
commit b109b7337f
2 changed files with 307 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/mcp/tools/project_management.py
+91 -9
View File
@@ -10,7 +10,12 @@ from typing import Literal
from fastmcp import Context
from loguru import logger
from basic_memory.config import ConfigManager, has_cloud_credentials
from basic_memory.config import (
BasicMemoryConfig,
ConfigManager,
ProjectEntry,
has_cloud_credentials,
)
from basic_memory.mcp.async_client import (
_explicit_routing,
_force_local_mode,
@@ -131,9 +136,66 @@ def _merge_projects(
return merged
def _workspace_entry_priority(entry: WorkspaceProjectEntry) -> tuple[bool, int, str, str]:
"""Prefer default/personal workspaces when duplicate project permalinks exist."""
workspace_type_rank = 0 if entry.workspace.workspace_type == "personal" else 1
return (
# False sorts before True, so the cloud/default workspace comes first.
not entry.workspace.is_default,
workspace_type_rank,
entry.workspace.name.casefold(),
entry.workspace.tenant_id,
)
def _select_attached_cloud_entry(
cloud_entries: tuple[WorkspaceProjectEntry, ...],
*,
config_entry: ProjectEntry | None,
config: BasicMemoryConfig | None,
) -> WorkspaceProjectEntry | None:
"""Choose the single cloud row that should inherit local project state."""
if not cloud_entries:
return None
preferred_workspace_ids: list[str] = []
if config_entry and config_entry.workspace_id:
preferred_workspace_ids.append(config_entry.workspace_id)
if (
config
and config.default_workspace
and config.default_workspace not in preferred_workspace_ids
):
preferred_workspace_ids.append(config.default_workspace)
# The configured default workspace can differ from the cloud-side default.
# Use the cloud default only after explicit local config preferences.
default_workspace_entry = next(
(entry for entry in cloud_entries if entry.workspace.is_default),
None,
)
if (
default_workspace_entry is not None
and default_workspace_entry.workspace.tenant_id not in preferred_workspace_ids
):
preferred_workspace_ids.append(default_workspace_entry.workspace.tenant_id)
for workspace_id in preferred_workspace_ids:
for entry in cloud_entries:
if entry.workspace.tenant_id == workspace_id:
return entry
if len(cloud_entries) == 1:
return cloud_entries[0]
return sorted(cloud_entries, key=_workspace_entry_priority)[0]
def _merge_workspace_projects(
local_list: ProjectList | None,
cloud_entries: tuple[WorkspaceProjectEntry, ...],
*,
config: BasicMemoryConfig | None = None,
) -> list[dict]:
"""Merge local projects with cloud projects from every accessible workspace."""
local_by_permalink: dict[str, ProjectItem] = {}
@@ -141,20 +203,40 @@ def _merge_workspace_projects(
for project in local_list.projects:
local_by_permalink[project.permalink] = project
config_by_permalink: dict[str, ProjectEntry] = {}
if config:
config_by_permalink = {
generate_permalink(project_name): entry
for project_name, entry in config.projects.items()
}
cloud_entries_by_permalink: dict[str, list[WorkspaceProjectEntry]] = {}
for entry in cloud_entries:
cloud_entries_by_permalink.setdefault(entry.project.permalink, []).append(entry)
attached_entry_by_permalink: dict[str, WorkspaceProjectEntry | None] = {}
for permalink in local_by_permalink:
attached_entry_by_permalink[permalink] = _select_attached_cloud_entry(
tuple(cloud_entries_by_permalink.get(permalink, ())),
config_entry=config_by_permalink.get(permalink),
config=config,
)
cloud_permalinks = {entry.project.permalink for entry in cloud_entries}
merged: list[dict] = []
for entry in sorted(
cloud_entries,
key=lambda item: (
not item.workspace.is_default,
item.workspace.workspace_type != "personal",
item.workspace.name.casefold(),
item.project.permalink,
),
key=lambda item: (*_workspace_entry_priority(item), item.project.permalink),
):
permalink = entry.project.permalink
local_proj = local_by_permalink.get(permalink)
local_proj = (
local_by_permalink.get(permalink)
# WorkspaceProjectEntry is a frozen dataclass containing Pydantic
# models, so value equality is the intended comparison here.
if attached_entry_by_permalink.get(permalink) == entry
else None
)
cloud_proj = entry.project
source = "local+cloud" if local_proj else "cloud"
local_path = local_proj.path if local_proj else None
@@ -339,7 +421,7 @@ async def list_memory_projects(
)
if cloud_entries:
merged = _merge_workspace_projects(local_list, cloud_entries)
merged = _merge_workspace_projects(local_list, cloud_entries, config=config)
else:
merged = _merge_projects(
local_list,
tests/mcp/test_tool_project_management.py
+216 -1
View File
@@ -9,7 +9,8 @@ from sqlalchemy import select
from basic_memory import db
from basic_memory.mcp.tools import list_memory_projects, create_memory_project, delete_project
from basic_memory.mcp.tools.project_management import _merge_projects
from basic_memory.config import BasicMemoryConfig, ProjectEntry
from basic_memory.mcp.tools.project_management import _merge_projects, _merge_workspace_projects
from basic_memory.models.project import Project
from basic_memory.schemas.project_info import ProjectItem, ProjectList
@@ -909,6 +910,220 @@ def test_merge_projects_overlap():
assert merged[0]["workspace_tenant_id"] == "org-456"
def test_merge_workspace_projects_attaches_local_state_to_one_duplicate_workspace(tmp_path):
"""A same-name team workspace project should stay cloud-only (#848)."""
local_path = str(tmp_path / "main")
local_main = _make_project("main", local_path, is_default=True)
local_list = _make_list([local_main], default="main")
personal_main = _make_project(
"main",
"/cloud/personal-main",
id=10,
external_id="personal-main-uuid",
)
team_main = _make_project(
"main",
"/cloud/team-main",
id=11,
external_id="team-main-uuid",
)
personal_ws = _make_workspace(
"personal-tenant",
"Personal",
slug="personal",
is_default=True,
)
team_ws = _make_workspace(
"team-tenant",
"Team",
workspace_type="organization",
slug="team",
)
workspace_index = _make_workspace_index(
[
(personal_ws, [personal_main]),
(team_ws, [team_main]),
]
)
config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
by_qualified_name = {project["qualified_name"]: project for project in merged}
personal_project = by_qualified_name["personal/main"]
team_project = by_qualified_name["team/main"]
assert personal_project["source"] == "local+cloud"
assert personal_project["local_path"] == local_path
assert personal_project["path"] == local_path
assert team_project["source"] == "cloud"
assert team_project["local_path"] is None
assert team_project["path"] == "/cloud/team-main"
def test_merge_workspace_projects_uses_configured_workspace_for_local_state(tmp_path):
"""Per-project workspace_id should select the attached duplicate row."""
local_path = str(tmp_path / "main")
local_main = _make_project("main", local_path, is_default=True)
local_list = _make_list([local_main], default="main")
personal_main = _make_project(
"main",
"/cloud/personal-main",
id=10,
external_id="personal-main-uuid",
)
team_main = _make_project(
"main",
"/cloud/team-main",
id=11,
external_id="team-main-uuid",
)
personal_ws = _make_workspace(
"personal-tenant",
"Personal",
slug="personal",
is_default=True,
)
team_ws = _make_workspace(
"team-tenant",
"Team",
workspace_type="organization",
slug="team",
)
workspace_index = _make_workspace_index(
[
(personal_ws, [personal_main]),
(team_ws, [team_main]),
]
)
config = BasicMemoryConfig(
projects={
"main": ProjectEntry(
path=local_path,
workspace_id="team-tenant",
)
}
)
merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
by_qualified_name = {project["qualified_name"]: project for project in merged}
personal_project = by_qualified_name["personal/main"]
team_project = by_qualified_name["team/main"]
assert personal_project["source"] == "cloud"
assert personal_project["local_path"] is None
assert personal_project["path"] == "/cloud/personal-main"
assert team_project["source"] == "local+cloud"
assert team_project["local_path"] == local_path
assert team_project["path"] == local_path
def test_merge_workspace_projects_uses_default_workspace_for_local_state(tmp_path):
"""Global default_workspace should attach local state before cloud default fallback."""
local_path = str(tmp_path / "main")
local_main = _make_project("main", local_path, is_default=True)
local_list = _make_list([local_main], default="main")
personal_main = _make_project(
"main",
"/cloud/personal-main",
id=10,
external_id="personal-main-uuid",
)
team_main = _make_project(
"main",
"/cloud/team-main",
id=11,
external_id="team-main-uuid",
)
personal_ws = _make_workspace(
"personal-tenant",
"Personal",
slug="personal",
is_default=True,
)
team_ws = _make_workspace(
"team-tenant",
"Team",
workspace_type="organization",
slug="team",
)
workspace_index = _make_workspace_index(
[
(personal_ws, [personal_main]),
(team_ws, [team_main]),
]
)
config = BasicMemoryConfig(
projects={"main": ProjectEntry(path=local_path)},
default_workspace="team-tenant",
)
merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
by_qualified_name = {project["qualified_name"]: project for project in merged}
personal_project = by_qualified_name["personal/main"]
team_project = by_qualified_name["team/main"]
assert personal_project["source"] == "cloud"
assert personal_project["local_path"] is None
assert personal_project["path"] == "/cloud/personal-main"
assert team_project["source"] == "local+cloud"
assert team_project["local_path"] == local_path
assert team_project["path"] == local_path
def test_merge_workspace_projects_sorted_fallback_attaches_personal_workspace(tmp_path):
"""When config has no preference and no cloud default exists, use stable priority."""
local_path = str(tmp_path / "main")
local_main = _make_project("main", local_path, is_default=True)
local_list = _make_list([local_main], default="main")
personal_main = _make_project(
"main",
"/cloud/personal-main",
id=10,
external_id="personal-main-uuid",
)
team_main = _make_project(
"main",
"/cloud/team-main",
id=11,
external_id="team-main-uuid",
)
personal_ws = _make_workspace(
"personal-tenant",
"Personal",
slug="personal",
is_default=False,
)
team_ws = _make_workspace(
"team-tenant",
"Team",
workspace_type="organization",
slug="team",
)
workspace_index = _make_workspace_index(
[
(team_ws, [team_main]),
(personal_ws, [personal_main]),
]
)
config = BasicMemoryConfig(projects={"main": ProjectEntry(path=local_path)})
merged = _merge_workspace_projects(local_list, workspace_index.entries, config=config)
by_qualified_name = {project["qualified_name"]: project for project in merged}
personal_project = by_qualified_name["personal/main"]
team_project = by_qualified_name["team/main"]
assert personal_project["source"] == "local+cloud"
assert personal_project["local_path"] == local_path
assert personal_project["path"] == local_path
assert team_project["source"] == "cloud"
assert team_project["local_path"] is None
assert team_project["path"] == "/cloud/team-main"
# --- Workspace passthrough tests ---