fix(mcp): qualify ChatGPT search fetch permalinks

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/mcp/project_context.py

@@ -432,15 +432,13 @@ def _canonical_memory_path_for_workspace(
 ) -> str:
     """Return the stored canonical path for a workspace-qualified memory URL."""
     normalized_remainder = remainder.strip("/")
-    if workspace_type == "organization":
-        prefix = f"{generate_permalink(workspace_slug)}/{project_permalink}"
-    elif workspace_type == "personal":
-        prefix = project_permalink if include_project else ""
-    else:
+    if workspace_type not in {"organization", "personal"}:
         raise ValueError(f"Unsupported workspace_type for memory URL routing: {workspace_type}")
 
-    if not prefix:
-        return normalized_remainder
+    # 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.
+    prefix = f"{generate_permalink(workspace_slug)}/{project_permalink}"
     if not normalized_remainder:
         return prefix
     return f"{prefix}/{normalized_remainder}"

src/basic_memory/mcp/tools/chatgpt_tools.py

@@ -6,29 +6,168 @@
 """
 
 import json
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, cast
+from uuid import UUID
 
 from fastmcp import Context
 from loguru import logger
 
 from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.project_management import list_memory_projects
 from basic_memory.mcp.tools.read_note import read_note
 from basic_memory.mcp.tools.search import search_notes
 from basic_memory.schemas.search import SearchResponse, SearchResult
 
 
+def _valid_project_id(value: object) -> str | None:
+    """Return a UUID project id string when one is present."""
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        return str(UUID(value))
+    except ValueError:
+        return None
+
+
+def _matches_constrained_project(project: dict[str, Any], constrained_project: object) -> bool:
+    """Return True when a project list row satisfies BASIC_MEMORY_MCP_PROJECT."""
+    if not isinstance(constrained_project, str) or not constrained_project.strip():
+        return True
+
+    candidates = {
+        value
+        for value in (
+            project.get("name"),
+            project.get("qualified_name"),
+            project.get("external_id"),
+        )
+        if isinstance(value, str)
+    }
+    return constrained_project in candidates
+
+
+def _search_project_refs(projects_payload: object) -> list[dict[str, str | None]]:
+    """Extract project routing refs for account-scoped ChatGPT search."""
+    if not isinstance(projects_payload, dict):
+        return []
+
+    payload = cast(dict[str, Any], projects_payload)
+    projects = payload.get("projects")
+    if not isinstance(projects, list):
+        return []
+
+    refs: list[dict[str, str | None]] = []
+    seen: set[tuple[str | None, str | None]] = set()
+    constrained_project = payload.get("constrained_project")
+    for item in projects:
+        if not isinstance(item, dict) or not _matches_constrained_project(
+            item, constrained_project
+        ):
+            continue
+
+        project = item.get("qualified_name") or item.get("name")
+        project_name = project if isinstance(project, str) and project.strip() else None
+        project_id = _valid_project_id(item.get("external_id"))
+        if project_name is None and project_id is None:
+            continue
+
+        key = (project_name, project_id)
+        if key in seen:
+            continue
+        seen.add(key)
+        refs.append({"project": project_name, "project_id": project_id})
+    return refs
+
+
+def _raw_results_from_search_payload(
+    results: SearchResponse | list[SearchResult | dict[str, Any]] | dict[str, Any],
+) -> list[SearchResult | dict[str, Any]]:
+    """Return the result list from any search_notes JSON-compatible payload."""
+    if isinstance(results, SearchResponse):
+        return list(results.results)
+    if isinstance(results, dict):
+        nested_results = results.get("results")
+        return (
+            cast(list[SearchResult | dict[str, Any]], nested_results)
+            if isinstance(nested_results, list)
+            else []
+        )
+    return list(results)
+
+
+def _qualify_permalink_for_project(permalink: object, project: str | None) -> object:
+    """Return a workspace-qualified permalink when the project ref supplies one."""
+    if not isinstance(permalink, str) or not permalink.strip():
+        return permalink
+    if not isinstance(project, str) or "/" not in project.strip("/"):
+        return permalink
+
+    normalized_permalink = permalink.strip("/")
+    qualified_project = project.strip("/")
+    if normalized_permalink == qualified_project or normalized_permalink.startswith(
+        f"{qualified_project}/"
+    ):
+        return normalized_permalink
+
+    workspace_slug, project_permalink = qualified_project.split("/", 1)
+    if normalized_permalink == project_permalink or normalized_permalink.startswith(
+        f"{project_permalink}/"
+    ):
+        return f"{workspace_slug}/{normalized_permalink}"
+    return f"{qualified_project}/{normalized_permalink}"
+
+
+def _qualify_results_for_project(
+    results: list[SearchResult | dict[str, Any]],
+    project_ref: dict[str, str | None],
+) -> list[dict[str, Any]]:
+    """Attach the searched workspace/project prefix to each ChatGPT result id."""
+    qualified: list[dict[str, Any]] = []
+    for result in results:
+        if isinstance(result, SearchResult):
+            result_data = result.model_dump()
+        else:
+            result_data = dict(result)
+        result_data["permalink"] = _qualify_permalink_for_project(
+            result_data.get("permalink"),
+            project_ref.get("project"),
+        )
+        qualified.append(result_data)
+    return qualified
+
+
+def _result_score(result: SearchResult | dict[str, Any]) -> float:
+    """Return a comparable search score for merged project results."""
+    if isinstance(result, SearchResult):
+        return result.score
+    score = result.get("score")
+    return float(score) if isinstance(score, int | float) else 0.0
+
+
+def _identifier_for_read_note(identifier: str) -> str:
+    """Convert ChatGPT result ids into routable Basic Memory identifiers."""
+    stripped = identifier.strip()
+    if stripped.startswith("memory://") or "/" not in stripped:
+        return identifier
+    return f"memory://{stripped}"
+
+
 def _format_search_results_for_chatgpt(
-    results: SearchResponse | list[SearchResult] | list[dict[str, Any]] | dict[str, Any],
+    results: SearchResponse | list[SearchResult | dict[str, Any]] | dict[str, Any],
 ) -> List[Dict[str, Any]]:
     """Format search results according to ChatGPT's expected schema.
 
     Returns a list of result objects with id, title, and url fields.
     """
     if isinstance(results, SearchResponse):
-        raw_results: list[SearchResult] | list[dict[str, Any]] = results.results
+        raw_results: list[SearchResult | dict[str, Any]] = list(results.results)
     elif isinstance(results, dict):
         nested_results = results.get("results")
-        raw_results = nested_results if isinstance(nested_results, list) else []
+        raw_results = (
+            cast(list[SearchResult | dict[str, Any]], nested_results)
+            if isinstance(nested_results, list)
+            else []
+        )
     else:
         raw_results = results
 
@@ -113,34 +252,77 @@ async def search(
     logger.info(f"ChatGPT search request: query='{query}'")
 
     try:
-        # Let search_notes resolve the default project via get_project_client(),
-        # which works in both local mode (ConfigManager) and cloud mode (database).
-        results = await search_notes(
-            query=query,
-            page=1,
-            page_size=10,
-            output_format="json",
-            context=context,
+        project_refs = _search_project_refs(
+            await list_memory_projects(output_format="json", context=context)
         )
 
-        # Handle string error responses from search_notes
-        if isinstance(results, str):
-            logger.warning(f"Search failed with error: {results[:100]}...")
-            search_results = {
-                "results": [],
-                "error": "Search failed",
-                "error_details": results[:500],  # Truncate long error messages
-            }
+        raw_results: list[SearchResult | dict[str, Any]] = []
+        if project_refs:
+            # Trigger: ChatGPT only has a strict search(query) -> fetch(id) contract.
+            # Why: default-project search strands notes in other workspaces/projects.
+            # Outcome: query every discovered project, then merge the top matches.
+            for project_ref in project_refs:
+                results = await search_notes(
+                    query=query,
+                    project=project_ref["project"],
+                    project_id=project_ref["project_id"],
+                    page=1,
+                    page_size=10,
+                    output_format="json",
+                    context=context,
+                )
+
+                if isinstance(results, str):
+                    logger.warning(f"Search failed with error: {results[:100]}...")
+                    search_results = {
+                        "results": [],
+                        "error": "Search failed",
+                        "error_details": results[:500],
+                    }
+                    return [
+                        {
+                            "type": "text",
+                            "text": json.dumps(search_results, ensure_ascii=False),
+                        }
+                    ]
+
+                raw_results.extend(
+                    _qualify_results_for_project(
+                        _raw_results_from_search_payload(results),
+                        project_ref,
+                    )
+                )
+            raw_results = sorted(raw_results, key=_result_score, reverse=True)[:10]
         else:
-            # Format successful results for ChatGPT
-            raw_results = results.get("results", []) if isinstance(results, dict) else []
-            formatted_results = _format_search_results_for_chatgpt(raw_results)
-            search_results = {
-                "results": formatted_results,
-                "total_count": len(raw_results),  # Use actual count from results
-                "query": query,
-            }
-            logger.info(f"Search completed: {len(formatted_results)} results returned")
+            # Trigger: project discovery returned no structured rows.
+            # Why: preserve the legacy single-project behavior when discovery is unavailable.
+            # Outcome: let search_notes resolve the default project as before.
+            results = await search_notes(
+                query=query,
+                page=1,
+                page_size=10,
+                output_format="json",
+                context=context,
+            )
+
+            if isinstance(results, str):
+                logger.warning(f"Search failed with error: {results[:100]}...")
+                search_results = {
+                    "results": [],
+                    "error": "Search failed",
+                    "error_details": results[:500],  # Truncate long error messages
+                }
+                return [{"type": "text", "text": json.dumps(search_results, ensure_ascii=False)}]
+
+            raw_results = _raw_results_from_search_payload(results)
+
+        formatted_results = _format_search_results_for_chatgpt(raw_results)
+        search_results = {
+            "results": formatted_results,
+            "total_count": len(raw_results),  # Use actual count from results
+            "query": query,
+        }
+        logger.info(f"Search completed: {len(formatted_results)} results returned")
 
         # Return in MCP content array format as required by OpenAI
         return [{"type": "text", "text": json.dumps(search_results, ensure_ascii=False)}]
@@ -180,7 +362,7 @@ async def fetch(
         # which works in both local mode (ConfigManager) and cloud mode (database).
         content = str(
             await read_note(
-                identifier=id,
+                identifier=_identifier_for_read_note(id),
                 context=context,
             )
         )

src/basic_memory/workspace_context.py

@@ -14,14 +14,14 @@
 
 @dataclass(frozen=True)
 class WorkspacePermalinkContext:
-    """Workspace metadata needed to build canonical organization permalinks."""
+    """Workspace metadata needed to build canonical workspace permalinks."""
 
     workspace_slug: str
     workspace_type: str
 
     @property
     def should_prefix_permalinks(self) -> bool:
-        return self.workspace_type == "organization" and bool(self.workspace_slug)
+        return bool(self.workspace_slug)
 
 
 _workspace_permalink_context: ContextVar[WorkspacePermalinkContext | None] = ContextVar(

tests/mcp/test_project_context.py

@@ -826,7 +826,7 @@ async def fake_index(context=None):
     resolved = await resolve_workspace_qualified_memory_url("memory://personal/main/notes/foo")
 
     assert resolved is not None
-    assert resolved.canonical_path == "main/notes/foo"
+    assert resolved.canonical_path == "personal/main/notes/foo"
 
 
 @pytest.mark.asyncio
@@ -1373,7 +1373,7 @@ async def fake_call_post(*args, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_resolve_project_and_path_strips_personal_workspace_prefix(
+async def test_resolve_project_and_path_preserves_personal_workspace_prefix(
     config_manager,
     monkeypatch,
 ):
@@ -1417,7 +1417,7 @@ async def fake_call_post(*args, **kwargs):
     )
 
     assert active_project == cached_project
-    assert resolved_path == "main/notes/foo"
+    assert resolved_path == "personal/main/notes/foo"
     assert is_memory_url is True
 
 

tests/mcp/test_tool_read_note.py

@@ -380,14 +380,14 @@ async def test_team_workspace_write_stores_complete_permalink_when_project_prefi
 
 
 @pytest.mark.asyncio
-async def test_personal_workspace_write_keeps_project_scoped_permalink(
+async def test_personal_workspace_write_stores_complete_canonical_permalink(
     app,
     test_project,
     entity_repository,
 ):
     from basic_memory.workspace_context import workspace_permalink_context
 
-    expected_permalink = f"{test_project.name}/personal/personal-workspace-note"
+    expected_permalink = f"personal/{test_project.name}/personal/personal-workspace-note"
 
     with workspace_permalink_context(workspace_slug="personal", workspace_type="personal"):
         write_result = await write_note(