diff --git a/src/basic_memory/mcp/project_context.py b/src/basic_memory/mcp/project_context.py index 2c5c7801a..ed2628149 100644 --- a/src/basic_memory/mcp/project_context.py +++ b/src/basic_memory/mcp/project_context.py @@ -21,11 +21,8 @@ Callable, List, Optional, - Sequence, Tuple, - cast, ) -from dataclasses import dataclass, field from uuid import UUID from httpx import AsyncClient @@ -48,15 +45,53 @@ from basic_memory.schemas.project_info import ProjectItem, ProjectList from basic_memory.schemas.v2 import ProjectResolveResponse from basic_memory.schemas.memory import memory_url_path -from basic_memory.utils import ( - build_qualified_permalink_reference, - generate_permalink, - normalize_project_reference, -) +from basic_memory.utils import generate_permalink, normalize_project_reference from basic_memory.workspace_context import ( current_workspace_permalink_context, workspace_permalink_context, ) +from basic_memory.mcp.project_context_identifiers import ( + UnresolvedProjectRouteError, + WorkspaceMemoryUrlResolution, + add_project_metadata as _add_project_metadata, + canonical_memory_path_for_active_route as _canonical_memory_path_for_active_route, + canonical_memory_path_for_workspace as _canonical_memory_path_for_workspace, + canonicalize_project_name as _canonicalize_project_name, + detect_project_from_url_prefix, + identifier_path as _identifier_path, + project_matches_identifier as _project_matches_identifier, + split_project_prefix as _split_project_prefix, + split_qualified_project_identifier as _split_qualified_project_identifier_impl, + split_workspace_identifier_segments as _split_workspace_identifier_segments, + split_workspace_memory_url_segments as _split_workspace_memory_url_segments, + unqualified_project_identifier as _unqualified_project_identifier, +) +from basic_memory.mcp.workspace_project_index import ( + WORKSPACE_PROJECT_INDEX_STATE_KEY as _WORKSPACE_PROJECT_INDEX_STATE_KEY, + WorkspaceProjectEntry, + WorkspaceProjectIndex, + WorkspaceProjectLookupMiss, + build_workspace_project_index as _build_workspace_project_index, + clear_cached_active_project as _clear_cached_active_project, + clear_cached_active_workspace_for_local_route as _clear_cached_active_workspace_for_local_route, + format_qualified_choices as _format_qualified_choices_impl, + get_cached_active_project as _get_cached_active_project, + get_cached_active_workspace as _get_cached_active_workspace, + get_cached_default_project as _get_cached_default_project, + match_workspace_identifier as _match_workspace_identifier_impl, + resolve_workspace_project_from_index as _resolve_workspace_project_from_index, + set_cached_active_project as _set_cached_active_project, + set_cached_active_workspace as _set_cached_active_workspace, + workspace_project_index_from_state as _workspace_project_index_from_state, + workspace_project_index_to_state as _workspace_project_index_to_state, +) + +# Keep the original module's helper surface intact for callers and tests while +# the implementations live in focused, dependency-light modules. +add_project_metadata = _add_project_metadata +_format_qualified_choices = _format_qualified_choices_impl +_match_workspace_identifier = _match_workspace_identifier_impl +_split_qualified_project_identifier = _split_qualified_project_identifier_impl if TYPE_CHECKING: from fastmcp import Context @@ -66,62 +101,6 @@ # The cloud MCP server sets a provider that queries its own database directly, # avoiding the control-plane HTTP round-trip that requires local credentials. _workspace_provider: Optional[Callable[[], Awaitable[list[WorkspaceInfo]]]] = None -_WORKSPACE_PROJECT_INDEX_STATE_KEY = "workspace_project_index" - - -class WorkspaceProjectLookupMiss(ValueError): - """A project was absent from the workspace index (as opposed to ambiguous). - - Misses are retried once against a freshly rebuilt index, because the - session cache may simply predate an out-of-band project creation (#956). - """ - - -class UnresolvedProjectRouteError(ValueError): - """A mutating memory URL named a project prefix that could not be resolved.""" - - def __init__(self, identifier: str, project_prefix: str): - self.identifier = identifier - self.project_prefix = project_prefix - super().__init__( - f"Memory URL project route '{project_prefix}' could not be resolved; " - "refusing to treat the URL as a path in the active project." - ) - - -@dataclass(frozen=True) -class WorkspaceProjectEntry: - """A cloud project resolved together with the workspace that owns it.""" - - workspace: WorkspaceInfo - project: ProjectItem - - @property - def qualified_name(self) -> str: - return f"{self.workspace.slug}/{self.project.permalink}" - - -@dataclass(frozen=True) -class WorkspaceProjectIndex: - """Session-local cloud project lookup index keyed by project permalink and external_id.""" - - workspaces: tuple[WorkspaceInfo, ...] - entries: tuple[WorkspaceProjectEntry, ...] - entries_by_permalink: dict[str, tuple[WorkspaceProjectEntry, ...]] - entries_by_external_id: dict[str, WorkspaceProjectEntry] = field(default_factory=dict) - failed_workspaces: tuple[WorkspaceInfo, ...] = () - - -@dataclass(frozen=True) -class WorkspaceMemoryUrlResolution: - """Resolved workspace/project route for a workspace-qualified memory URL.""" - - entry: WorkspaceProjectEntry - canonical_path: str - - @property - def project_identifier(self) -> str: - return self.entry.qualified_name def set_workspace_provider(provider: Callable[[], Awaitable[list[WorkspaceInfo]]]) -> None: @@ -153,124 +132,6 @@ async def _resolve_default_project_from_api() -> Optional[str]: return None -async def _get_cached_active_project(context: Optional[Context]) -> Optional[ProjectItem]: - """Return the cached active project from context when available.""" - if not context: - return None - - cached_raw = await context.get_state("active_project") - if isinstance(cached_raw, dict): - return ProjectItem.model_validate(cached_raw) - return None - - -async def _set_cached_active_project( - context: Optional[Context], - active_project: ProjectItem, -) -> None: - """Persist the active project and known default-project metadata in context.""" - if not context: - return - - await context.set_state("active_project", active_project.model_dump()) - if active_project.is_default: - await context.set_state("default_project_name", active_project.name) - - -async def _clear_cached_active_project(context: Optional[Context]) -> None: - """Clear cached project metadata that may no longer match the active route.""" - if not context: - return - - await context.set_state("active_project", None) - await context.set_state("default_project_name", None) - - -async def _get_cached_active_workspace(context: Optional[Context]) -> Optional[WorkspaceInfo]: - """Return the cached active workspace from context when available.""" - if not context: - return None - - cached_raw = await context.get_state("active_workspace") - if isinstance(cached_raw, dict): - return WorkspaceInfo.model_validate(cached_raw) - return None - - -async def _set_cached_active_workspace( - context: Optional[Context], - active_workspace: WorkspaceInfo, -) -> None: - """Persist workspace context and clear project cache when the tenant changes.""" - if not context: - return - - cached_workspace = await _get_cached_active_workspace(context) - if cached_workspace and cached_workspace.tenant_id != active_workspace.tenant_id: - # Trigger: project routing moved to another workspace - # Why: project names are only unique inside one workspace, so a cached - # ProjectItem from the previous tenant can point at the wrong project - # Outcome: force the next validation call to resolve within the new tenant - await _clear_cached_active_project(context) - - await context.set_state("active_workspace", active_workspace.model_dump()) - - -async def _clear_cached_active_workspace_for_local_route(context: Optional[Context]) -> None: - """Drop tenant workspace metadata before routing through a local project.""" - if not context: - return - - # Trigger: local routing follows a cloud route in the same MCP session - # Why: active_workspace is tenant metadata, not part of local project identity - # Outcome: memory:// resolution uses project-only local permalinks - await context.set_state("active_workspace", None) - - -async def _get_cached_default_project(context: Optional[Context]) -> Optional[str]: - """Return the cached default project name from context when available.""" - if not context: - return None - - cached_default = await context.get_state("default_project_name") - if isinstance(cached_default, str): - return cached_default - return None - - -def _canonicalize_project_name( - project_name: Optional[str], - config: BasicMemoryConfig, -) -> Optional[str]: - """Return the configured project name when the identifier matches by permalink. - - Project routing happens before API validation, so we normalize explicit inputs - here to keep local/cloud routing aligned with the database's case-insensitive - project resolver. - """ - if project_name is None: - return None - - requested_permalink = generate_permalink(project_name) - for configured_name in config.projects: - if generate_permalink(configured_name) == requested_permalink: - return configured_name - - return project_name - - -def _project_matches_identifier(project_item: ProjectItem, identifier: Optional[str]) -> bool: - """Return True when the identifier refers to the cached project.""" - if identifier is None: - return True - - normalized_identifier = generate_permalink(identifier) - return normalized_identifier in { - generate_permalink(project_item.name), - project_item.permalink, - } - - async def resolve_project_parameter( project: Optional[str] = None, allow_discovery: bool = False, @@ -349,233 +210,6 @@ async def get_project_names(client: AsyncClient, headers: HeaderTypes | None = N return [project.name for project in project_list.projects] -def _workspace_project_index_from_state(raw: object) -> WorkspaceProjectIndex | None: - """Deserialize a cached workspace project index from MCP context state.""" - if not isinstance(raw, dict): - return None - - raw_mapping = cast(dict[str, object], raw) - workspaces_raw = raw_mapping.get("workspaces") - entries_raw = raw_mapping.get("entries") - if not isinstance(workspaces_raw, list) or not isinstance(entries_raw, list): - return None - - workspaces = tuple(WorkspaceInfo.model_validate(item) for item in workspaces_raw) - failed_workspaces_raw = raw_mapping.get("failed_workspaces") - failed_workspaces = ( - tuple(WorkspaceInfo.model_validate(item) for item in failed_workspaces_raw) - if isinstance(failed_workspaces_raw, list) - else () - ) - entries_list: list[WorkspaceProjectEntry] = [] - for item in entries_raw: - if not isinstance(item, dict): - continue - item_mapping = cast(dict[str, object], item) - workspace_raw = item_mapping.get("workspace") - project_raw = item_mapping.get("project") - if workspace_raw is None or project_raw is None: - continue - entries_list.append( - WorkspaceProjectEntry( - workspace=WorkspaceInfo.model_validate(workspace_raw), - project=ProjectItem.model_validate(project_raw), - ) - ) - entries = tuple(entries_list) - return _build_workspace_project_index( - workspaces, - entries, - failed_workspaces=failed_workspaces, - ) - - -def _workspace_project_index_to_state(index: WorkspaceProjectIndex) -> dict: - """Serialize a workspace project index for MCP context state.""" - return { - "workspaces": [workspace.model_dump() for workspace in index.workspaces], - "failed_workspaces": [workspace.model_dump() for workspace in index.failed_workspaces], - "entries": [ - { - "workspace": entry.workspace.model_dump(), - "project": entry.project.model_dump(), - } - for entry in index.entries - ], - } - - -def _build_workspace_project_index( - workspaces: tuple[WorkspaceInfo, ...], - entries: tuple[WorkspaceProjectEntry, ...], - *, - failed_workspaces: tuple[WorkspaceInfo, ...] = (), -) -> WorkspaceProjectIndex: - """Build the permalink and external_id lookup tables for workspace-project entries.""" - grouped: dict[str, list[WorkspaceProjectEntry]] = {} - by_external_id: dict[str, WorkspaceProjectEntry] = {} - for entry in entries: - grouped.setdefault(entry.project.permalink, []).append(entry) - by_external_id[entry.project.external_id] = entry - - return WorkspaceProjectIndex( - workspaces=workspaces, - entries=entries, - entries_by_permalink={ - permalink: tuple(items) - for permalink, items in sorted(grouped.items(), key=lambda item: item[0]) - }, - entries_by_external_id=by_external_id, - failed_workspaces=failed_workspaces, - ) - - -def _split_qualified_project_identifier(identifier: str) -> tuple[str | None, str]: - """Split ``/`` identifiers for cloud routing.""" - cleaned = identifier.strip() - if "/" not in cleaned: - return None, cleaned - - workspace_slug, project_identifier = cleaned.split("/", 1) - if not workspace_slug or not project_identifier: - return None, cleaned - return workspace_slug, project_identifier - - -def _unqualified_project_identifier(identifier: str) -> str: - """Return the project segment from an optional qualified project identifier.""" - _, project_identifier = _split_qualified_project_identifier(identifier) - return project_identifier - - -def _identifier_path(identifier: str) -> str: - """Return the routable path portion of a raw identifier or memory URL.""" - stripped = identifier.strip() - return memory_url_path(stripped) if stripped.startswith("memory://") else stripped - - -def _split_workspace_identifier_segments(identifier: str) -> tuple[str, str, str] | None: - """Split ``//`` identifiers into route segments.""" - normalized = normalize_project_reference(_identifier_path(identifier)).strip("/") - parts = normalized.split("/", 2) - if len(parts) != 3: - # Trigger: two-segment identifiers such as `workspace/project` or `project/path`. - # Why: without a remainder, the shape is ambiguous with existing project-prefix routing. - # Outcome: fall through so the normal project-prefix/default-project resolver decides. - return None - - workspace_slug, project_identifier, remainder = parts - if not workspace_slug or not project_identifier or not remainder: - return None - return workspace_slug, project_identifier, remainder - - -def _split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str] | None: - """Split ``memory:////`` into route segments.""" - if not identifier.strip().startswith("memory://"): - return None - - return _split_workspace_identifier_segments(identifier) - - -def _canonical_memory_path_for_workspace( - *, - workspace_slug: str, - workspace_type: str, - project_permalink: str, - remainder: str, -) -> str: - """Return the stored canonical path for a workspace-qualified memory URL.""" - normalized_remainder = remainder.strip("/") - if workspace_type not in {"organization", "personal"}: - raise ValueError(f"Unsupported workspace_type for memory URL routing: {workspace_type}") - - # Trigger: a caller supplied a workspace-qualified memory URL. - # Why: the first two path segments are the global route, even for Personal. - # Outcome: lookups preserve the complete workspace/project canonical permalink. - if not normalized_remainder: - normalized_remainder = project_permalink - - # Same index-form rule as _canonical_memory_path_for_active_route (#957): - # without an active workspace permalink context, stored permalinks are - # project-qualified and a workspace-prefixed pattern cannot match. - if "*" in normalized_remainder and current_workspace_permalink_context() is None: - return build_qualified_permalink_reference( - project_permalink, - normalized_remainder, - include_project=True, - ) - - return build_qualified_permalink_reference( - project_permalink, - normalized_remainder, - include_project=True, - workspace_permalink=workspace_slug, - ) - - -def _canonical_memory_path_for_active_route( - active_project: ProjectItem, - path: str, - *, - include_project: bool, - cached_workspace: WorkspaceInfo | None = None, -) -> str: - """Return the canonical permalink path for the currently routed project/workspace.""" - project_prefix = active_project.permalink - - # Trigger: the path contains a glob wildcard (folder/*) and no server-side - # workspace permalink context is active. - # Why: patterns match raw against the search index, so they must mirror the - # stored permalink form. The contextvar is what qualified permalinks at - # write time — when it is absent, stored rows are project-qualified and a - # workspace prefix (from the client's cached_workspace display state) - # guarantees zero matches (#957). Direct lookups keep full qualification - # because the link resolver understands it; patterns have no fallback. - # Outcome: without the contextvar, qualify patterns with the project prefix - # only; with it, fall through to normal workspace canonicalization. - if "*" in path and current_workspace_permalink_context() is None: - if not include_project: - return path - if path == project_prefix or path.startswith(f"{project_prefix}/"): - return path - return f"{project_prefix}/{path}" - - workspace_remainder = path - if include_project and (path == project_prefix or path.startswith(f"{project_prefix}/")): - # Trigger: the memory URL already names the active project root/prefix - # Why: workspace canonicalization adds the project prefix itself, so - # keeping it in the remainder would produce // - # Outcome: keep project-root and project-prefixed URLs canonical once - workspace_remainder = ( - "" if path == project_prefix else path.removeprefix(f"{project_prefix}/") - ) - - workspace_context = current_workspace_permalink_context() - if workspace_context is not None: - return _canonical_memory_path_for_workspace( - workspace_slug=workspace_context.workspace_slug, - workspace_type=workspace_context.workspace_type, - project_permalink=active_project.permalink, - remainder=workspace_remainder, - ) - - if cached_workspace is not None: - return _canonical_memory_path_for_workspace( - workspace_slug=cached_workspace.slug, - workspace_type=cached_workspace.workspace_type, - project_permalink=active_project.permalink, - remainder=workspace_remainder, - ) - - if not include_project: - return path - - if path == project_prefix or path.startswith(f"{project_prefix}/"): - return path - return f"{project_prefix}/{path}" - - def _cloud_workspace_discovery_available(config: BasicMemoryConfig) -> bool: """Return True when workspace discovery can be used without forcing local routing.""" from basic_memory.mcp.async_client import ( @@ -702,11 +336,6 @@ async def _resolve_workspace_segments( return WorkspaceMemoryUrlResolution(entry=entry, canonical_path=canonical_path) -def _format_qualified_choices(entries: Sequence[WorkspaceProjectEntry]) -> str: - """Format qualified project choices for collision errors.""" - return " or ".join(entry.qualified_name for entry in entries) - - async def get_available_workspaces(context: Optional[Context] = None) -> list[WorkspaceInfo]: """Load available cloud workspaces for the current authenticated user.""" if context: @@ -874,67 +503,6 @@ 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, @@ -957,119 +525,6 @@ async def resolve_workspace_project_identifier( return await _resolve_workspace_project_from_index(refreshed, project, context) -async def _resolve_workspace_project_from_index( - index: WorkspaceProjectIndex, - project: str, - context: Optional[Context] = None, -) -> WorkspaceProjectEntry: - """Resolve a project against one concrete index snapshot. - - Raises WorkspaceProjectLookupMiss for absent projects (retryable via index - refresh) and plain ValueError for ambiguity, which a refresh cannot fix. - """ - # Fast path: direct lookup by external_id when the identifier is a UUID - # Canonicalize via str(UUID(...)) so uppercase, brace-wrapped, or urn:uuid forms - # all hash to the same lowercase-hyphenated key as the stored external_ids. - try: - canonical_external_id = str(UUID(project)) - entry = index.entries_by_external_id.get(canonical_external_id) - if entry: - return entry - except ValueError: - pass - - workspace_identifier, project_identifier = _split_qualified_project_identifier(project) - project_permalink = generate_permalink(project_identifier) - - 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, ()) - if entry.workspace.tenant_id == workspace.tenant_id - ] - if not matches: - if any( - failed_workspace.tenant_id == workspace.tenant_id - for failed_workspace in index.failed_workspaces - ): - raise WorkspaceProjectLookupMiss( - f"Projects for workspace '{workspace.name}' ({workspace.slug}) " - "could not be loaded. Retry after workspace discovery recovers." - ) - available = ", ".join( - entry.qualified_name - for entry in index.entries - if entry.workspace.tenant_id == workspace.tenant_id - ) - raise WorkspaceProjectLookupMiss( - f"Project '{project_identifier}' was not found in workspace " - f"'{workspace.name}' ({workspace.slug}). Available projects: {available}" - ) - if len(matches) > 1: - details = ", ".join( - f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches - ) - raise ValueError( - f"Project '{project_identifier}' matched multiple projects in workspace " - f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. " - f"Matches: {details}" - ) - return matches[0] - - matches = list(index.entries_by_permalink.get(project_permalink, ())) - if not matches: - failed_note = "" - if index.failed_workspaces: - failed = ", ".join(workspace.slug for workspace in index.failed_workspaces) - failed_note = ( - f" Project discovery failed for workspace(s): {failed}; " - "retry or use a qualified project from an indexed workspace." - ) - available = ", ".join(entry.qualified_name for entry in index.entries) - raise WorkspaceProjectLookupMiss( - f"Project '{project}' was not found in indexed cloud workspaces. " - f"Available projects: {available}.{failed_note}" - ) - - cached_workspace = await _get_cached_active_workspace(context) - if cached_workspace: - cached_matches = [ - entry for entry in matches if entry.workspace.tenant_id == cached_workspace.tenant_id - ] - if cached_matches: - return cached_matches[0] - - if len(matches) > 1: - # Prefer the project in the default workspace when name is ambiguous - default_match = next((entry for entry in matches if entry.workspace.is_default), None) - if default_match: - return default_match - - choices = _format_qualified_choices(matches) - details = "\n".join( - f"- {entry.workspace.name} ({entry.workspace.slug}): {entry.qualified_name}" - for entry in matches - ) - raise ValueError( - f"Project '{project}' exists in multiple workspaces. Use: {choices}\n{details}" - ) - - if index.failed_workspaces: - qualified_name = matches[0].qualified_name - failed = ", ".join(workspace.slug for workspace in index.failed_workspaces) - raise ValueError( - f"Project '{project}' was found as {qualified_name}, but project discovery " - f"failed for workspace(s): {failed}. Use '{qualified_name}' to route " - "explicitly, or retry after discovery recovers." - ) - - return matches[0] - - async def _default_workspace_project_entry( context: Optional[Context] = None, ) -> WorkspaceProjectEntry | None: @@ -1275,21 +730,6 @@ async def get_active_project( return active_project -def _split_project_prefix(path: str) -> tuple[Optional[str], str]: - """Split a possible project prefix from a memory URL path.""" - if "/" not in path: - return None, path - - project_prefix, remainder = path.split("/", 1) - if not project_prefix or not remainder: - return None, path - - if "*" in project_prefix: - return None, path - - return project_prefix, remainder - - async def resolve_project_and_path( client: AsyncClient, identifier: str, @@ -1474,51 +914,6 @@ async def resolve_project_and_path( return active_project, resolved_path, True -def add_project_metadata(result: str, project_name: str) -> str: - """Add project context as metadata footer for assistant session tracking. - - Provides clear project context to help the assistant remember which - project is being used throughout the conversation session. - - Args: - result: The tool result string - project_name: The project name that was used - - Returns: - Result with project session tracking metadata - """ - return f"{result}\n\n[Session: Using project '{project_name}']" - - -def detect_project_from_url_prefix(identifier: str, config: BasicMemoryConfig) -> Optional[str]: - """Check if a memory URL's first path segment matches a known project in config. - - This enables automatic project routing from memory URLs like - ``memory://specs/in-progress`` without requiring the caller to pass - an explicit ``project`` parameter. - - Uses local config only — no network calls. - - Args: - identifier: Raw identifier string (may or may not start with ``memory://``). - config: Current BasicMemoryConfig with project entries. - - Returns: - Matching project name from config, or None if no match. - """ - path = memory_url_path(identifier) if identifier.strip().startswith("memory://") else identifier - normalized = normalize_project_reference(path) - prefix, _ = _split_project_prefix(normalized) - if prefix is None: - return None - - prefix_permalink = generate_permalink(prefix) - for project_name in config.projects: - if generate_permalink(project_name) == prefix_permalink: - return project_name - return None - - async def detect_project_from_memory_url_prefix( identifier: str, config: BasicMemoryConfig, diff --git a/src/basic_memory/mcp/project_context_identifiers.py b/src/basic_memory/mcp/project_context_identifiers.py new file mode 100644 index 000000000..af1e33154 --- /dev/null +++ b/src/basic_memory/mcp/project_context_identifiers.py @@ -0,0 +1,213 @@ +"""Pure project/workspace identifier parsing and canonical path construction.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +from basic_memory.config import BasicMemoryConfig +from basic_memory.mcp.workspace_project_index import WorkspaceProjectEntry +from basic_memory.schemas.cloud import WorkspaceInfo +from basic_memory.schemas.memory import memory_url_path +from basic_memory.schemas.project_info import ProjectItem +from basic_memory.utils import ( + build_qualified_permalink_reference, + generate_permalink, + normalize_project_reference, +) +from basic_memory.workspace_context import current_workspace_permalink_context + + +class UnresolvedProjectRouteError(ValueError): + """A mutating memory URL named a project prefix that could not be resolved.""" + + def __init__(self, identifier: str, project_prefix: str): + self.identifier = identifier + self.project_prefix = project_prefix + super().__init__( + f"Memory URL project route '{project_prefix}' could not be resolved; " + "refusing to treat the URL as a path in the active project." + ) + + +@dataclass(frozen=True) +class WorkspaceMemoryUrlResolution: + """Resolved workspace/project route for a workspace-qualified memory URL.""" + + entry: WorkspaceProjectEntry + canonical_path: str + + @property + def project_identifier(self) -> str: + return self.entry.qualified_name + + +def canonicalize_project_name( + project_name: Optional[str], + config: BasicMemoryConfig, +) -> Optional[str]: + """Return the configured name when an identifier matches by permalink.""" + if project_name is None: + return None + + requested_permalink = generate_permalink(project_name) + for configured_name in config.projects: + if generate_permalink(configured_name) == requested_permalink: + return configured_name + return project_name + + +def project_matches_identifier(project_item: ProjectItem, identifier: Optional[str]) -> bool: + """Return True when the identifier refers to the cached project.""" + if identifier is None: + return True + normalized_identifier = generate_permalink(identifier) + return normalized_identifier in { + generate_permalink(project_item.name), + project_item.permalink, + } + + +def split_qualified_project_identifier(identifier: str) -> tuple[str | None, str]: + """Split ``/`` identifiers for cloud routing.""" + cleaned = identifier.strip() + if "/" not in cleaned: + return None, cleaned + workspace_slug, project_identifier = cleaned.split("/", 1) + if not workspace_slug or not project_identifier: + return None, cleaned + return workspace_slug, project_identifier + + +def unqualified_project_identifier(identifier: str) -> str: + """Return the project segment from an optional qualified identifier.""" + _, project_identifier = split_qualified_project_identifier(identifier) + return project_identifier + + +def identifier_path(identifier: str) -> str: + """Return the routable path portion of a raw identifier or memory URL.""" + stripped = identifier.strip() + return memory_url_path(stripped) if stripped.startswith("memory://") else stripped + + +def split_workspace_identifier_segments(identifier: str) -> tuple[str, str, str] | None: + """Split ``//`` identifiers into route segments.""" + normalized = normalize_project_reference(identifier_path(identifier)).strip("/") + parts = normalized.split("/", 2) + if len(parts) != 3: + return None + workspace_slug, project_identifier, remainder = parts + if not workspace_slug or not project_identifier or not remainder: + return None + return workspace_slug, project_identifier, remainder + + +def split_workspace_memory_url_segments(identifier: str) -> tuple[str, str, str] | None: + """Split ``memory:////`` into route segments.""" + if not identifier.strip().startswith("memory://"): + return None + return split_workspace_identifier_segments(identifier) + + +def canonical_memory_path_for_workspace( + *, + workspace_slug: str, + workspace_type: str, + project_permalink: str, + remainder: str, +) -> str: + """Return the stored canonical path for a workspace-qualified memory URL.""" + normalized_remainder = remainder.strip("/") + if workspace_type not in {"organization", "personal"}: + raise ValueError(f"Unsupported workspace_type for memory URL routing: {workspace_type}") + if not normalized_remainder: + normalized_remainder = project_permalink + if "*" in normalized_remainder and current_workspace_permalink_context() is None: + return build_qualified_permalink_reference( + project_permalink, + normalized_remainder, + include_project=True, + ) + return build_qualified_permalink_reference( + project_permalink, + normalized_remainder, + include_project=True, + workspace_permalink=workspace_slug, + ) + + +def canonical_memory_path_for_active_route( + active_project: ProjectItem, + path: str, + *, + include_project: bool, + cached_workspace: WorkspaceInfo | None = None, +) -> str: + """Return the canonical permalink path for the active project/workspace.""" + project_prefix = active_project.permalink + if "*" in path and current_workspace_permalink_context() is None: + if not include_project: + return path + if path == project_prefix or path.startswith(f"{project_prefix}/"): + return path + return f"{project_prefix}/{path}" + + workspace_remainder = path + if include_project and (path == project_prefix or path.startswith(f"{project_prefix}/")): + workspace_remainder = ( + "" if path == project_prefix else path.removeprefix(f"{project_prefix}/") + ) + + workspace_context = current_workspace_permalink_context() + if workspace_context is not None: + return canonical_memory_path_for_workspace( + workspace_slug=workspace_context.workspace_slug, + workspace_type=workspace_context.workspace_type, + project_permalink=active_project.permalink, + remainder=workspace_remainder, + ) + if cached_workspace is not None: + return canonical_memory_path_for_workspace( + workspace_slug=cached_workspace.slug, + workspace_type=cached_workspace.workspace_type, + project_permalink=active_project.permalink, + remainder=workspace_remainder, + ) + if not include_project: + return path + if path == project_prefix or path.startswith(f"{project_prefix}/"): + return path + return f"{project_prefix}/{path}" + + +def split_project_prefix(path: str) -> tuple[Optional[str], str]: + """Split a possible project prefix from a memory URL path.""" + if "/" not in path: + return None, path + project_prefix, remainder = path.split("/", 1) + if not project_prefix or not remainder or "*" in project_prefix: + return None, path + return project_prefix, remainder + + +def add_project_metadata(result: str, project_name: str) -> str: + """Add project context as a metadata footer for session tracking.""" + return f"{result}\n\n[Session: Using project '{project_name}']" + + +def detect_project_from_url_prefix( + identifier: str, + config: BasicMemoryConfig, +) -> Optional[str]: + """Return the local config project matching a memory URL path prefix.""" + path = memory_url_path(identifier) if identifier.strip().startswith("memory://") else identifier + normalized = normalize_project_reference(path) + prefix, _ = split_project_prefix(normalized) + if prefix is None: + return None + prefix_permalink = generate_permalink(prefix) + for project_name in config.projects: + if generate_permalink(project_name) == prefix_permalink: + return project_name + return None diff --git a/src/basic_memory/mcp/workspace_project_index.py b/src/basic_memory/mcp/workspace_project_index.py new file mode 100644 index 000000000..16a47158c --- /dev/null +++ b/src/basic_memory/mcp/workspace_project_index.py @@ -0,0 +1,358 @@ +"""Workspace project index values, context cache helpers, and pure lookup logic.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Optional, Sequence, cast +from uuid import UUID + +from basic_memory.schemas.cloud import WorkspaceInfo +from basic_memory.schemas.project_info import ProjectItem +from basic_memory.utils import generate_permalink + +if TYPE_CHECKING: + from fastmcp import Context + +WORKSPACE_PROJECT_INDEX_STATE_KEY = "workspace_project_index" + + +class WorkspaceProjectLookupMiss(ValueError): + """A project was absent from the workspace index rather than ambiguous. + + Misses are retried once against a freshly rebuilt index because the session + cache may predate an out-of-band project creation. + """ + + +@dataclass(frozen=True) +class WorkspaceProjectEntry: + """A cloud project resolved together with the workspace that owns it.""" + + workspace: WorkspaceInfo + project: ProjectItem + + @property + def qualified_name(self) -> str: + return f"{self.workspace.slug}/{self.project.permalink}" + + +@dataclass(frozen=True) +class WorkspaceProjectIndex: + """Session-local project lookup keyed by permalink and external_id.""" + + workspaces: tuple[WorkspaceInfo, ...] + entries: tuple[WorkspaceProjectEntry, ...] + entries_by_permalink: dict[str, tuple[WorkspaceProjectEntry, ...]] + entries_by_external_id: dict[str, WorkspaceProjectEntry] = field(default_factory=dict) + failed_workspaces: tuple[WorkspaceInfo, ...] = () + + +async def get_cached_active_project(context: Optional[Context]) -> Optional[ProjectItem]: + """Return the cached active project from context when available.""" + if not context: + return None + + cached_raw = await context.get_state("active_project") + if isinstance(cached_raw, dict): + return ProjectItem.model_validate(cached_raw) + return None + + +async def set_cached_active_project( + context: Optional[Context], + active_project: ProjectItem, +) -> None: + """Persist the active project and known default-project metadata in context.""" + if not context: + return + + await context.set_state("active_project", active_project.model_dump()) + if active_project.is_default: + await context.set_state("default_project_name", active_project.name) + + +async def clear_cached_active_project(context: Optional[Context]) -> None: + """Clear cached project metadata that may no longer match the active route.""" + if not context: + return + + await context.set_state("active_project", None) + await context.set_state("default_project_name", None) + + +async def get_cached_active_workspace(context: Optional[Context]) -> Optional[WorkspaceInfo]: + """Return the cached active workspace from context when available.""" + if not context: + return None + + cached_raw = await context.get_state("active_workspace") + if isinstance(cached_raw, dict): + return WorkspaceInfo.model_validate(cached_raw) + return None + + +async def set_cached_active_workspace( + context: Optional[Context], + active_workspace: WorkspaceInfo, +) -> None: + """Persist workspace context and clear project cache when the tenant changes.""" + if not context: + return + + cached_workspace = await get_cached_active_workspace(context) + if cached_workspace and cached_workspace.tenant_id != active_workspace.tenant_id: + # Project names are only unique inside one workspace. A cached project + # from the previous tenant must not survive a workspace route change. + await clear_cached_active_project(context) + + await context.set_state("active_workspace", active_workspace.model_dump()) + + +async def clear_cached_active_workspace_for_local_route( + context: Optional[Context], +) -> None: + """Drop tenant workspace metadata before routing through a local project.""" + if not context: + return + + await context.set_state("active_workspace", None) + + +async def get_cached_default_project(context: Optional[Context]) -> Optional[str]: + """Return the cached default project name from context when available.""" + if not context: + return None + + cached_default = await context.get_state("default_project_name") + if isinstance(cached_default, str): + return cached_default + return None + + +def workspace_project_index_from_state(raw: object) -> WorkspaceProjectIndex | None: + """Deserialize a cached workspace project index from MCP context state.""" + if not isinstance(raw, dict): + return None + + raw_mapping = cast(dict[str, object], raw) + workspaces_raw = raw_mapping.get("workspaces") + entries_raw = raw_mapping.get("entries") + if not isinstance(workspaces_raw, list) or not isinstance(entries_raw, list): + return None + + workspaces = tuple(WorkspaceInfo.model_validate(item) for item in workspaces_raw) + failed_workspaces_raw = raw_mapping.get("failed_workspaces") + failed_workspaces = ( + tuple(WorkspaceInfo.model_validate(item) for item in failed_workspaces_raw) + if isinstance(failed_workspaces_raw, list) + else () + ) + entries_list: list[WorkspaceProjectEntry] = [] + for item in entries_raw: + if not isinstance(item, dict): + continue + item_mapping = cast(dict[str, object], item) + workspace_raw = item_mapping.get("workspace") + project_raw = item_mapping.get("project") + if workspace_raw is None or project_raw is None: + continue + entries_list.append( + WorkspaceProjectEntry( + workspace=WorkspaceInfo.model_validate(workspace_raw), + project=ProjectItem.model_validate(project_raw), + ) + ) + return build_workspace_project_index( + workspaces, + tuple(entries_list), + failed_workspaces=failed_workspaces, + ) + + +def workspace_project_index_to_state(index: WorkspaceProjectIndex) -> dict: + """Serialize a workspace project index for MCP context state.""" + return { + "workspaces": [workspace.model_dump() for workspace in index.workspaces], + "failed_workspaces": [workspace.model_dump() for workspace in index.failed_workspaces], + "entries": [ + { + "workspace": entry.workspace.model_dump(), + "project": entry.project.model_dump(), + } + for entry in index.entries + ], + } + + +def build_workspace_project_index( + workspaces: tuple[WorkspaceInfo, ...], + entries: tuple[WorkspaceProjectEntry, ...], + *, + failed_workspaces: tuple[WorkspaceInfo, ...] = (), +) -> WorkspaceProjectIndex: + """Build permalink and external_id lookup tables for workspace-project entries.""" + grouped: dict[str, list[WorkspaceProjectEntry]] = {} + by_external_id: dict[str, WorkspaceProjectEntry] = {} + for entry in entries: + grouped.setdefault(entry.project.permalink, []).append(entry) + by_external_id[entry.project.external_id] = entry + + return WorkspaceProjectIndex( + workspaces=workspaces, + entries=entries, + entries_by_permalink={ + permalink: tuple(items) + for permalink, items in sorted(grouped.items(), key=lambda item: item[0]) + }, + entries_by_external_id=by_external_id, + failed_workspaces=failed_workspaces, + ) + + +def format_qualified_choices(entries: Sequence[WorkspaceProjectEntry]) -> str: + """Format qualified project choices for collision errors.""" + return " or ".join(entry.qualified_name for entry in entries) + + +def match_workspace_identifier( + workspaces: tuple[WorkspaceInfo, ...], + workspace_identifier: str, +) -> WorkspaceInfo: + """Resolve a qualified route segment by slug, tenant_id, then display name.""" + slug_matches = [ + workspace + for workspace in workspaces + if workspace.slug.casefold() == workspace_identifier.casefold() + ] + if slug_matches: + return slug_matches[0] + + tenant_matches = [ + workspace for workspace in workspaces if workspace.tenant_id == workspace_identifier + ] + if tenant_matches: + return tenant_matches[0] + + 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_from_index( + index: WorkspaceProjectIndex, + project: str, + context: Optional[Context] = None, +) -> WorkspaceProjectEntry: + """Resolve a project against one concrete workspace-index snapshot.""" + try: + canonical_external_id = str(UUID(project)) + entry = index.entries_by_external_id.get(canonical_external_id) + if entry: + return entry + except ValueError: + pass + + from basic_memory.mcp.project_context_identifiers import split_qualified_project_identifier + + workspace_identifier, project_identifier = split_qualified_project_identifier(project) + project_permalink = generate_permalink(project_identifier) + + if workspace_identifier: + workspace = match_workspace_identifier(index.workspaces, workspace_identifier) + matches = [ + entry + for entry in index.entries_by_permalink.get(project_permalink, ()) + if entry.workspace.tenant_id == workspace.tenant_id + ] + if not matches: + if any( + failed_workspace.tenant_id == workspace.tenant_id + for failed_workspace in index.failed_workspaces + ): + raise WorkspaceProjectLookupMiss( + f"Projects for workspace '{workspace.name}' ({workspace.slug}) " + "could not be loaded. Retry after workspace discovery recovers." + ) + available = ", ".join( + entry.qualified_name + for entry in index.entries + if entry.workspace.tenant_id == workspace.tenant_id + ) + raise WorkspaceProjectLookupMiss( + f"Project '{project_identifier}' was not found in workspace " + f"'{workspace.name}' ({workspace.slug}). Available projects: {available}" + ) + if len(matches) > 1: + details = ", ".join( + f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches + ) + raise ValueError( + f"Project '{project_identifier}' matched multiple projects in workspace " + f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. " + f"Matches: {details}" + ) + return matches[0] + + matches = list(index.entries_by_permalink.get(project_permalink, ())) + if not matches: + failed_note = "" + if index.failed_workspaces: + failed = ", ".join(workspace.slug for workspace in index.failed_workspaces) + failed_note = ( + f" Project discovery failed for workspace(s): {failed}; " + "retry or use a qualified project from an indexed workspace." + ) + available = ", ".join(entry.qualified_name for entry in index.entries) + raise WorkspaceProjectLookupMiss( + f"Project '{project}' was not found in indexed cloud workspaces. " + f"Available projects: {available}.{failed_note}" + ) + + cached_workspace = await get_cached_active_workspace(context) + if cached_workspace: + cached_matches = [ + entry for entry in matches if entry.workspace.tenant_id == cached_workspace.tenant_id + ] + if cached_matches: + return cached_matches[0] + + if len(matches) > 1: + default_match = next((entry for entry in matches if entry.workspace.is_default), None) + if default_match: + return default_match + + choices = format_qualified_choices(matches) + details = "\n".join( + f"- {entry.workspace.name} ({entry.workspace.slug}): {entry.qualified_name}" + for entry in matches + ) + raise ValueError( + f"Project '{project}' exists in multiple workspaces. Use: {choices}\n{details}" + ) + + if index.failed_workspaces: + qualified_name = matches[0].qualified_name + failed = ", ".join(workspace.slug for workspace in index.failed_workspaces) + raise ValueError( + f"Project '{project}' was found as {qualified_name}, but project discovery " + f"failed for workspace(s): {failed}. Use '{qualified_name}' to route " + "explicitly, or retry after discovery recovers." + ) + + return matches[0]