fix(mcp): preserve legacy permalink routing

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

Author: phernandez

src/basic_memory/mcp/async_client.py

@@ -43,6 +43,8 @@ def _asgi_client(timeout: Timeout) -> AsyncClient:
         transport=ASGITransport(app=fastapi_app),
         base_url="http://test",
         timeout=timeout,
+        # Local ASGI calls still cross the HTTP boundary, so request handlers need
+        # the same workspace permalink metadata that cloud proxy calls receive.
         headers=workspace_permalink_headers(),
     )
 

src/basic_memory/mcp/project_context.py

@@ -432,6 +432,9 @@ def _split_workspace_identifier_segments(identifier: str) -> tuple[str, str, str
     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
@@ -1331,6 +1334,14 @@ async def detect_project_from_identifier_prefix(
     if local_project is not None:
         return local_project
 
+    normalized_identifier = normalize_project_reference(_identifier_path(identifier)).strip("/")
+    if "/" not in normalized_identifier:
+        # Trigger: plain text search query or single-segment title/permalink.
+        # Why: cloud project discovery can build a workspace index; only path-shaped
+        #   identifiers carry enough structure to justify that cost.
+        # Outcome: keep unqualified search/title input on the active/default project route.
+        return None
+
     if _cloud_workspace_discovery_available(config):
         workspace_resolution = await resolve_workspace_qualified_identifier(
             identifier,
@@ -1339,8 +1350,7 @@ async def detect_project_from_identifier_prefix(
         if workspace_resolution is not None:
             return workspace_resolution.project_identifier
 
-        normalized = normalize_project_reference(_identifier_path(identifier))
-        project_prefix, _ = _split_project_prefix(normalized)
+        project_prefix, _ = _split_project_prefix(normalized_identifier)
         if project_prefix is None:
             return None
 

src/basic_memory/mcp/tools/utils.py

@@ -8,7 +8,7 @@
 from typing import Any, Optional
 
 import logfire
-from httpx import Response, URL, AsyncClient, HTTPStatusError
+from httpx import Response, URL, AsyncClient, HTTPStatusError, Headers
 from httpx._client import UseClientDefault, USE_CLIENT_DEFAULT
 from httpx._types import (
     RequestContent,
@@ -58,6 +58,22 @@ def _transport_error_span_attrs(exc: Exception) -> dict[str, Any]:
     }
 
 
+def _request_headers(headers: HeaderTypes | None) -> HeaderTypes | None:
+    """Merge request-local workspace permalink headers into outbound API calls."""
+    from basic_memory.workspace_context import workspace_permalink_headers
+
+    workspace_headers = workspace_permalink_headers()
+    if not workspace_headers:
+        return headers
+
+    if headers is None:
+        return workspace_headers
+
+    merged_headers = Headers(headers)
+    merged_headers.update(workspace_headers)
+    return merged_headers
+
+
 def get_error_message(
     status_code: int, url: URL | str, method: str, msg: Optional[str] = None
 ) -> str:
@@ -224,7 +240,7 @@ async def call_get(
             response = await client.get(
                 url,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -328,7 +344,7 @@ async def call_put(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -432,7 +448,7 @@ async def call_patch(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -542,7 +558,7 @@ async def call_post(
                 files=files,
                 json=json,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,
@@ -667,7 +683,7 @@ async def call_delete(
             response = await client.delete(
                 url=url,
                 params=params,
-                headers=headers,
+                headers=_request_headers(headers),
                 cookies=cookies,
                 auth=auth,
                 follow_redirects=follow_redirects,

src/basic_memory/utils.py

@@ -386,6 +386,12 @@ def add_candidate(value: str | None) -> None:
             remainder = normalized_path.removeprefix(f"{normalized_project}/")
             add_candidate(remainder)
 
+    if not include_project and normalized_path.startswith(f"{normalized_project}/"):
+        # Trigger: caller supplied `project/path` while legacy short permalinks are stored.
+        # Why: routing uses the project prefix, but strict lookup still needs the short row.
+        # Outcome: try `path` after the exact project-qualified candidate.
+        add_candidate(normalized_path.removeprefix(f"{normalized_project}/"))
+
     return candidates
 
 

tests/mcp/test_tool_utils.py

@@ -13,6 +13,11 @@
     call_put,
     get_error_message,
 )
+from basic_memory.workspace_context import (
+    WORKSPACE_SLUG_HEADER,
+    WORKSPACE_TYPE_HEADER,
+    workspace_permalink_context,
+)
 
 
 @pytest.fixture
@@ -170,6 +175,26 @@ async def test_call_get_with_params(mock_response):
     assert kwargs["params"] == params
 
 
+@pytest.mark.asyncio
+async def test_call_post_adds_workspace_permalink_headers_at_request_time(mock_response):
+    client = _Client()
+    client.set_response("post", mock_response())
+
+    with workspace_permalink_context("team-paul", "organization"):
+        await call_post(
+            _client(client),
+            "http://test.com",
+            headers={"X-Existing": "value"},
+        )
+
+    assert len(client.calls) == 1
+    _, _args, kwargs = client.calls[0]
+    request_headers = kwargs["headers"]
+    assert request_headers["X-Existing"] == "value"
+    assert request_headers[WORKSPACE_SLUG_HEADER] == "team-paul"
+    assert request_headers[WORKSPACE_TYPE_HEADER] == "organization"
+
+
 @pytest.mark.asyncio
 async def test_get_error_message():
     """Test the get_error_message function."""

tests/test_permalink_utils.py

@@ -20,6 +20,19 @@ def test_workspace_qualified_permalink_candidates_include_legacy_without_rewrapp
     ]
 
 
+def test_project_prefixed_candidate_includes_short_legacy_when_project_prefix_disabled():
+    candidates = build_permalink_resolution_candidates(
+        "main/notes/example",
+        "main",
+        include_project=False,
+    )
+
+    assert candidates == [
+        "main/notes/example",
+        "notes/example",
+    ]
+
+
 def test_short_permalink_candidates_include_workspace_and_project_forms():
     candidates = build_permalink_resolution_candidates(
         "notes/example",