From dd35e43f3b100c5c307bb1742a7cf05430e6e3ee Mon Sep 17 00:00:00 2001 From: Drew Cain Date: Thu, 11 Jun 2026 15:05:52 -0500 Subject: [PATCH] fix(mcp): resolve workspace display names and tenant ids in qualified project routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The edit_note docstring (and write_note in PR #964) advertise that the workspace segment of a "workspace/project" route may be a slug, name, or tenant_id, but resolve_workspace_project_identifier() only matched against WorkspaceInfo.slug. Cloud routes using a display name or tenant UUID failed with "Workspace ... was not found" despite the documented contract. Extend the first-segment matching (only the matching logic; the overall resolution flow is unchanged) to honor, in priority order: 1. slug (casefold) — unchanged, checked first so working routes keep meaning 2. tenant_id — exact match on the opaque id 3. display name (casefold) — fails fast on collisions, listing candidate slugs A name that collides with another workspace's slug resolves to the slug owner. Unknown identifiers raise a not-found error naming the forms that were tried. Co-Authored-By: Claude Signed-off-by: Drew Cain --- src/basic_memory/mcp/project_context.py | 82 +++++++++--- tests/mcp/test_project_context.py | 164 ++++++++++++++++++++++++ 2 files changed, 231 insertions(+), 15 deletions(-) diff --git a/src/basic_memory/mcp/project_context.py b/src/basic_memory/mcp/project_context.py index 314d944ae..212b60487 100644 --- a/src/basic_memory/mcp/project_context.py +++ b/src/basic_memory/mcp/project_context.py @@ -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, @@ -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, ()) diff --git a/tests/mcp/test_project_context.py b/tests/mcp/test_project_context.py index 70142b434..b84931650 100644 --- a/tests/mcp/test_project_context.py +++ b/tests/mcp/test_project_context.py @@ -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