fix(mcp): keep ChatGPT search as wrapper

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

Author: phernandez

src/basic_memory/mcp/tools/chatgpt_tools.py

@@ -7,143 +7,16 @@
 
 import json
 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()
@@ -252,69 +125,25 @@ async def search(
     logger.info(f"ChatGPT search request: query='{query}'")
 
     try:
-        project_refs = _search_project_refs(
-            await list_memory_projects(output_format="json", context=context)
+        # Keep this adapter tiny: the real search behavior lives in search_notes.
+        results = await search_notes(
+            query=query,
+            page=1,
+            page_size=10,
+            output_format="json",
+            context=context,
         )
 
-        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:
-            # 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)}]
+        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)
+        raw_results = results.get("results", []) if isinstance(results, dict) else []
 
         formatted_results = _format_search_results_for_chatgpt(raw_results)
         search_results = {