Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 67 additions & 15 deletions src/basic_memory/mcp/project_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -798,6 +798,67 @@ async def ensure_workspace_project_index(
return await _ensure_workspace_project_index(context=context)


def _match_workspace_identifier(
workspaces: tuple[WorkspaceInfo, ...],
workspace_identifier: str,
) -> WorkspaceInfo:
"""Resolve the first segment of a qualified route to a single workspace.

The edit_note/write_note contract advertises that the workspace segment may be a
slug, tenant_id, or display name. We honor those forms in a fixed priority order so
that adding tenant_id/name support never changes the meaning of an identifier that
already resolves today:

1. slug (casefold) — existing behavior, checked first so working routes are stable.
2. tenant_id — exact match against the opaque id (no casefolding, mirroring the
precedent in ``workspace_matches_exact_identifier``).
3. display name (casefold) — names are not guaranteed unique, so a name that matches
multiple workspaces is rejected rather than silently picking one.
"""
# Trigger: identifier equals a workspace slug (casefold).
# Why: slug is the canonical routing key; resolving it first guarantees a workspace
# whose display name collides with another workspace's slug yields to the slug owner.
# Outcome: return the slug owner before tenant_id/name are considered.
slug_matches = [
workspace
for workspace in workspaces
if workspace.slug.casefold() == workspace_identifier.casefold()
]
if slug_matches:
return slug_matches[0]

# Trigger: identifier exactly equals a workspace tenant_id (an opaque id).
# Why: tenant_ids are unique, so an exact hit is unambiguous and needs no tie-break.
tenant_matches = [
workspace for workspace in workspaces if workspace.tenant_id == workspace_identifier
]
if tenant_matches:
return tenant_matches[0]

# Trigger: identifier matches one or more workspace display names (casefold).
# Why: names are not guaranteed unique; failing fast on collisions keeps routing
# deterministic and tells the caller exactly how to disambiguate.
name_matches = [
workspace
for workspace in workspaces
if workspace.name.casefold() == workspace_identifier.casefold()
]
if len(name_matches) > 1:
candidates = ", ".join(workspace.slug for workspace in name_matches)
raise ValueError(
f"Workspace name '{workspace_identifier}' matched multiple workspaces "
f"(slugs: {candidates}). Use the workspace slug or tenant_id to disambiguate."
)
if name_matches:
return name_matches[0]

available = ", ".join(workspace.slug for workspace in workspaces)
raise ValueError(
f"Workspace '{workspace_identifier}' was not found by slug, tenant_id, or name. "
f"Available workspace slugs: {available}"
)


async def resolve_workspace_project_identifier(
project: str,
context: Optional[Context] = None,
Expand All @@ -816,23 +877,14 @@ async def resolve_workspace_project_identifier(
except ValueError:
pass

workspace_slug, project_identifier = _split_qualified_project_identifier(project)
workspace_identifier, project_identifier = _split_qualified_project_identifier(project)
project_permalink = generate_permalink(project_identifier)

if workspace_slug:
workspace_matches = [
workspace
for workspace in index.workspaces
if workspace.slug.casefold() == workspace_slug.casefold()
]
if not workspace_matches:
available = ", ".join(workspace.slug for workspace in index.workspaces)
raise ValueError(
f"Workspace '{workspace_slug}' was not found. "
f"Available workspace slugs: {available}"
)

workspace = workspace_matches[0]
if workspace_identifier:
# Honor the documented "slug, name, or tenant_id" contract for the workspace
# segment; _match_workspace_identifier raises a clear error on ambiguous names
# and unknown identifiers, listing what forms were tried.
workspace = _match_workspace_identifier(index.workspaces, workspace_identifier)
matches = [
entry
for entry in index.entries_by_permalink.get(project_permalink, ())
Expand Down
164 changes: 164 additions & 0 deletions tests/mcp/test_project_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -694,6 +694,170 @@ async def fake_index(context=None):
assert resolved.project.external_id == "personal-project-id"


def _patch_index(monkeypatch, workspaces, entries):
"""Install a fake workspace/project index for resolver tests."""
import basic_memory.mcp.project_context as project_context
from basic_memory.mcp.project_context import _build_workspace_project_index

index = _build_workspace_project_index(workspaces, entries)

async def fake_index(context=None):
return index

monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)


@pytest.mark.asyncio
async def test_resolve_workspace_identifier_by_slug_tenant_id_and_name(monkeypatch):
"""Qualified routes resolve the workspace segment by slug, tenant_id, or display name."""
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
resolve_workspace_project_identifier,
)

acme = _workspace(
tenant_id="acme-tenant-uuid",
workspace_type="organization",
slug="acme-slug",
name="Acme Corp",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=acme,
project=_project("Meeting Notes", id=1, external_id="acme-project-id"),
),
)
_patch_index(monkeypatch, (acme,), entries)

# slug (existing behavior, stays green)
by_slug = await resolve_workspace_project_identifier("acme-slug/meeting-notes")
assert by_slug.project.external_id == "acme-project-id"

# tenant_id (exact, opaque id)
by_tenant = await resolve_workspace_project_identifier("acme-tenant-uuid/meeting-notes")
assert by_tenant.project.external_id == "acme-project-id"

# display name, case-insensitive
by_name = await resolve_workspace_project_identifier("ACME corp/meeting-notes")
assert by_name.project.external_id == "acme-project-id"


@pytest.mark.asyncio
async def test_resolve_workspace_identifier_ambiguous_name_lists_candidates(monkeypatch):
"""A display name shared by multiple workspaces fails fast naming the candidate slugs."""
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
resolve_workspace_project_identifier,
)

alpha = _workspace(
tenant_id="alpha-tenant",
workspace_type="organization",
slug="research-alpha",
name="Research",
role="editor",
)
beta = _workspace(
tenant_id="beta-tenant",
workspace_type="organization",
slug="research-beta",
name="Research",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=alpha,
project=_project("Notes", id=1, external_id="alpha-project-id"),
),
WorkspaceProjectEntry(
workspace=beta,
project=_project("Notes", id=2, external_id="beta-project-id"),
),
)
_patch_index(monkeypatch, (alpha, beta), entries)

with pytest.raises(ValueError) as exc_info:
await resolve_workspace_project_identifier("Research/notes")

message = str(exc_info.value)
assert "matched multiple workspaces" in message
assert "research-alpha" in message
assert "research-beta" in message
assert "slug or tenant_id" in message


@pytest.mark.asyncio
async def test_resolve_workspace_identifier_unknown_lists_tried_forms(monkeypatch):
"""An unknown workspace identifier reports the slug/tenant_id/name forms that were tried."""
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
resolve_workspace_project_identifier,
)

acme = _workspace(
tenant_id="acme-tenant",
workspace_type="organization",
slug="acme",
name="Acme",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=acme,
project=_project("Notes", id=1, external_id="acme-project-id"),
),
)
_patch_index(monkeypatch, (acme,), entries)

with pytest.raises(ValueError) as exc_info:
await resolve_workspace_project_identifier("nonexistent/notes")

message = str(exc_info.value)
assert "was not found by slug, tenant_id, or name" in message
assert "acme" in message


@pytest.mark.asyncio
async def test_resolve_workspace_identifier_slug_wins_over_name_collision(monkeypatch):
"""A name that equals another workspace's slug resolves to the slug owner."""
from basic_memory.mcp.project_context import (
WorkspaceProjectEntry,
resolve_workspace_project_identifier,
)

# slug_owner.slug == "shared"; name_owner.name == "shared" — the slug owner must win.
slug_owner = _workspace(
tenant_id="slug-owner-tenant",
workspace_type="organization",
slug="shared",
name="Slug Owner",
role="editor",
)
name_owner = _workspace(
tenant_id="name-owner-tenant",
workspace_type="organization",
slug="name-owner-slug",
name="shared",
role="editor",
)
entries = (
WorkspaceProjectEntry(
workspace=slug_owner,
project=_project("Notes", id=1, external_id="slug-owner-project-id"),
),
WorkspaceProjectEntry(
workspace=name_owner,
project=_project("Notes", id=2, external_id="name-owner-project-id"),
),
)
_patch_index(monkeypatch, (slug_owner, name_owner), entries)

resolved = await resolve_workspace_project_identifier("shared/notes")
assert resolved.workspace.slug == "shared"
assert resolved.project.external_id == "slug-owner-project-id"


@pytest.mark.asyncio
async def test_detect_project_from_memory_url_prefix_resolves_workspace_slug(monkeypatch):
import basic_memory.mcp.project_context as project_context
Expand Down
Loading